diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
index 112ac1d6f3..58e3806358 100644
--- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
+++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md
@@ -28,6 +28,23 @@ identity. No second inbox store, per-field bridge expansion or full rewrite is
required for the first App outcome. These are planned product consumers of
T0–T4, not additional provider promotion or completed migration claims.
+Requester-scoped delegation inventory now composes page readback in
+`collaboration/delegation.ts`. Python retains bounded address scanning, requester
+registration and fresh `Delegations.read` IO, including canonical acceptance and
+artifact checks. TypeScript validates page order and item observations, selects
+the next cursor and reports completeness. The per-item projection RPC and Python
+page decisions are retired: query admission and final page projection use two
+requests per page; provider reads and validation calls remain. The existing
+private snapshot transport carries complete pages above the inline wire budget.
+CLI, MCP and Goal Chat keep their existing live-page contract. Async lifecycle
+cutover and receiver-adoption qualification remain separate.
+The same page reduction now seals actual directory/operation read times through
+the existing `projection_envelope.ts` owner. Completeness is scoped to
+`requester_page`; unreadable operations alert, while shown/available counts
+disclose remaining pages. Replay preserves the original observation time.
+Each page reuses the existing executable-revision request scope; operation
+records remain live reads, and the next page resolves the current code again.
+
diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md
index 4d7cb3a032..410bd1af34 100644
--- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md
+++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md
@@ -26,6 +26,18 @@ R1–R3 的 TS 消费者包括 App 产品路径，不只 CLI 结算。
首个 App 结果不需要第二份 inbox 存储、逐字段 RPC 或全量重写。
这里是 T0–T4 的产品消费计划，不新增 provider promotion，也不声称迁移完成。
+请求者范围内的委派清单现在由 `collaboration/delegation.ts` 一次组装整页读回。
+Python 保留有界地址扫描、请求者注册检查和实时 `Delegations.read` IO，
+包括 canonical 验收与产物检查。TS 校验页内顺序和逐项观察，选择下一游标并报告
+完整性。逐项投影 RPC 与 Python 页级决定已退役：查询准入和整页投影每页共两次
+请求；provider 读取与验证调用仍保留。超过内联传输预算的完整页复用既有私有
+快照传输。CLI、MCP 与 Goal Chat 保持原有实时分页契约。异步生命周期切换和
+接收方采用资格仍需单独验收。
+同一次整页投影现经既有 `projection_envelope.ts` owner 封装真实目录与操作读取
+时间。完整性仅限 `requester_page`；不可读操作触发告警，展示/可用数量说明
+剩余页，重放保留原始观察时间。
+每页复用既有代码版本请求作用域；操作记录仍实时读取，下一页重新解析当前代码。
+
diff --git a/docs/reference/contracts/projection-envelope-contract.md b/docs/reference/contracts/projection-envelope-contract.md
index 8e11333134..5a3ea5e61a 100644
--- a/docs/reference/contracts/projection-envelope-contract.md
+++ b/docs/reference/contracts/projection-envelope-contract.md
@@ -23,9 +23,19 @@ Python-owned projections do so through `loopx/control_plane/projection_envelope_
| `loopx global-gates` | `global_gates` | `global` |
| `loopx global-todos` | `global_todos` | `global` |
| `loopx global-risks` | `global_risks` | `global` |
+| Delegation operations (CLI, MCP, Goal Chat) | `delegation_inventory` | `requester_page` |
| Goal task map in delivery review | `goal_task_map` | `goal_task_map` |
| `GET /api/chat/delivery-review` | `delivery_review` | `selected_goal_delivery` |
+The delegation inventory seals directory and operation read facts in its existing
+native page reduction. A complete `requester_page` certifies only the requested
+page: `has_more`, `next_cursor` and the shown/available counts disclose the
+remaining requester inventory, not whole-Goal completion. Unavailable operations
+remain visible and count as omissions with unreadable-source alerts. A missing
+journal is a freshly observed empty inventory; a denied directory read still
+fails rather than returning an apparently complete empty page. No cache or
+cross-page snapshot is introduced; restart paging to discover earlier additions.
+
The Goal map seals actual Todo-read time and source/display omissions in its
existing native composition. Its envelope does not certify task acceptance or
execution readiness. The outer review inherits both the status and map sources;
diff --git a/loopx/collaboration_mcp.py b/loopx/collaboration_mcp.py
index 6994599d72..a850904a07 100644
--- a/loopx/collaboration_mcp.py
+++ b/loopx/collaboration_mcp.py
@@ -538,7 +538,10 @@ def _read_stop(path: Path) -> dict | None:
def operations(self, *, limit: int = 20, cursor: str | None = None) -> dict:
from .control_plane.collaboration.delegation_inventory import read_delegation_inventory
- return read_delegation_inventory(self, limit=limit, cursor=cursor)
+ # One page uses one executable revision; records are still read live,
+ # and the next page resolves its own current revision.
+ with effect_runtime_request_scope():
+ return read_delegation_inventory(self, limit=limit, cursor=cursor)
def inspect(self, binding_id: str) -> dict:
"""Observe the real Turn preflight; never create a request or run a host."""
@@ -1861,6 +1864,8 @@ def list_delegations(limit: int = 20, cursor: str | None = None) -> dict:
Accepted items are rechecked; unavailable requires reconciliation, not duplicate
dispatch. Read the original operation for full artifacts. Listing starts no work.
+ Read projection_envelope before claiming current or complete state: its scope
+ is this requester page, and alerts and omitted operations must be disclosed.
"""
return delegations.operations(limit=limit, cursor=cursor)
diff --git a/loopx/control_plane/collaboration/delegation.ts b/loopx/control_plane/collaboration/delegation.ts
index 0c0c5d5a34..718fb929c3 100644
--- a/loopx/control_plane/collaboration/delegation.ts
+++ b/loopx/control_plane/collaboration/delegation.ts
@@ -1,13 +1,14 @@
/** Explicit local execution bindings. Registration/messages alone grant no launch.
* These are host observations; canonical task/Turn/acceptance remain authoritative. */
import type {JsonObject} from "../effect_program.ts";
-import {requireJsonObject} from "../runtime_decode.ts";
+import {requireInteger, requireJsonObject, requireStringArray} from "../runtime_decode.ts";
import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts";
import {canonicalAuthoritySha256} from "../coordination/authority_store_codec.ts";
import {acceptanceValidationEffects, goalAcceptanceTodoDigest, type AcceptanceCompletionRequirements} from "../goals/acceptance_contract.ts";
import {normalizeTodoCompletionValidationDeclaration} from "../todos/completion_validation_declaration.ts";
import {readTurnSelectionRejection, turnSelectionRejectionState} from "../turn_driver/selection_rejection.ts";
import { BARE_SHA256_PATTERN, ENVELOPED_SHA256_PATTERN } from "../content_digest.ts";
+import {PROJECTION_ENVELOPE_SEAL_REQUEST, sealProjectionEnvelope} from "../projection_envelope.ts";
function requireThat(ok: unknown, message: string): asserts ok {
if (!ok) throw new EffectRuntimeRequestError(message);
@@ -414,6 +415,63 @@ export function delegationInventoryItem(params: JsonObject): JsonObject {
return result;
}
+function inventoryPageItem(raw: unknown, recordId: string): JsonObject {
+ const entry = requireJsonObject(raw, "delegation page observation");
+ const record = requireJsonObject(entry.record, "delegation inventory record");
+ requireThat(record.record_id === recordId, "delegation inventory page identity mismatch");
+ try {
+ return delegationInventoryItem({record, observation: entry.observation});
+ } catch (error) {
+ // A malformed current observation cannot hide healthy siblings or imply acceptance.
+ if (error instanceof EffectRuntimeRequestError || error instanceof TypeError)
+ return delegationInventoryItem({record, observation: null});
+ throw error;
+ }
+}
+
+/** The lookahead address proves another page exists; its operation is not read. */
+export function delegationInventoryPage(params: JsonObject): JsonObject {
+ const query = delegationInventoryQuery(requireJsonObject(params.query, "delegation inventory query"));
+ const limit = Number(query.limit);
+ const recordIds = requireStringArray(params.record_ids, "delegation inventory addresses");
+ requireThat(recordIds.length <= limit + 1 && recordIds.every((id, index) =>
+ BARE_SHA256_PATTERN.test(id) && (query.cursor === null || id > String(query.cursor))
+ && (index === 0 || id > recordIds[index - 1]!)), "invalid delegation inventory page addresses");
+ const observations = params.observations;
+ requireThat(Array.isArray(observations) && observations.length === Math.min(limit, recordIds.length),
+ "delegation inventory page observations must match its addresses");
+ const items = observations.map((entry, index) => inventoryPageItem(entry, recordIds[index]!));
+ const facts = requireJsonObject(params.read_facts, "delegation inventory read facts");
+ const available = requireInteger(facts.available_count, "delegation inventory available_count");
+ requireThat(recordIds.length === Math.min(available, limit + 1),
+ "delegation inventory count does not match its page addresses");
+ const unavailable = items.filter(item => item.status === "unavailable").length;
+ const projectionEnvelope = sealProjectionEnvelope({
+ schema_version: PROJECTION_ENVELOPE_SEAL_REQUEST, projection: "delegation_inventory",
+ observed_at: facts.observed_at,
+ sources: [
+ {source_id: "delegation_directory", read_status: "read", last_read_at: facts.directory_read_at,
+ item_count: available},
+ {source_id: "delegation_operations", read_status: "read", last_read_at: facts.operations_read_at,
+ item_count: items.length, unreadable_count: unavailable},
+ ],
+ coverage: {scope: "requester_page", expected_count: items.length,
+ included_count: items.length - unavailable,
+ omitted: unavailable ? [{reason: "unavailable_operations", count: unavailable}] : [],
+ shown_count: items.length, available_count: available},
+ });
+ const more = recordIds.length > limit;
+ return {
+ schema_version: "loopx_delegation_inventory_v0", items, has_more: more,
+ next_cursor: more ? recordIds[limit - 1]! : null,
+ page_readback_complete: unavailable === 0,
+ projection_envelope: projectionEnvelope,
+ note: "Live requester-scoped page, not a snapshot or proof that the whole Goal is complete. "
+ + "Read original operations for full results; resume only when recovery is required. "
+ + "Restart paging to discover work added before the cursor. Listing starts no work.",
+ };
+}
+
export function transitionDelegationObservation(params: JsonObject): JsonObject {
const from = params.from as Observation, to = params.to as Observation;
requireThat(Object.hasOwn(transitions, from) && Object.hasOwn(transitions, to), "invalid delegation observation");
diff --git a/loopx/control_plane/collaboration/delegation_inventory.py b/loopx/control_plane/collaboration/delegation_inventory.py
index 4fb2fc3a84..d76a40a792 100644
--- a/loopx/control_plane/collaboration/delegation_inventory.py
+++ b/loopx/control_plane/collaboration/delegation_inventory.py
@@ -8,6 +8,7 @@
from .inbox import _read
from .peers import _goal, require_operation_id
from ..content_digest import BARE_SHA256_PATTERN
+from ..runtime.time import now_utc_iso
if TYPE_CHECKING:
from ...collaboration_mcp import Delegations
@@ -25,8 +26,10 @@ def read_delegation_inventory(service: Delegations, *, limit: int = 20,
{"limit": limit, "cursor": cursor})
_goal(service.registry, service.goal_id, service.agent_id)
directory = service.path("inventory").parent
+ available_count = 0
def addresses():
+ nonlocal available_count
try:
entries = directory.iterdir()
for path in entries:
@@ -35,6 +38,7 @@ def addresses():
if not BARE_SHA256_PATTERN.fullmatch(path.stem):
raise ValueError("unexpected delegation record address; reconcile inventory storage")
if query["cursor"] is None or path.stem > query["cursor"]:
+ available_count += 1
yield path.stem
except FileNotFoundError:
# No journal yet is a valid empty inventory; other IO errors propagate.
@@ -42,7 +46,10 @@ def addresses():
raise
keys = heapq.nsmallest(query["limit"] + 1, addresses())
- items = []
+ directory_read_at = now_utc_iso()
+ # Bound the aggregate by its earliest read, never by final page assembly.
+ operations_read_at = now_utc_iso() if keys else directory_read_at
+ observations = []
for key in keys[:query["limit"]]:
record = {"record_id": key, "operation_id": None}
try:
@@ -54,20 +61,12 @@ def addresses():
raise ValueError("delegation record identity mismatch")
record["operation_id"] = operation_id
observed = service.read(operation_id)
- item = effect_runtime_result("collaboration.delegation.inventory_item",
- {"record": record, "observation": observed})
except (OSError, ValueError, KeyError, TypeError, EffectRuntimeRemoteError):
- # A corrupt/stale branch must not hide healthy siblings or be called accepted.
- item = effect_runtime_result("collaboration.delegation.inventory_item",
- {"record": record, "observation": None})
- items.append(item)
- more = len(keys) > query["limit"]
- return {
- "schema_version": "loopx_delegation_inventory_v0",
- "items": items, "has_more": more,
- "next_cursor": keys[query["limit"] - 1] if more else None,
- "page_readback_complete": all(item["status"] != "unavailable" for item in items),
- "note": "Live requester-scoped page, not a snapshot or proof that the whole Goal is complete. "
- "Read original operations for full results; resume only when recovery is required. "
- "Restart paging to discover work added before the cursor. Listing starts no work.",
- }
+ observed = None
+ observations.append({"record": record, "observation": observed})
+ return effect_runtime_result("collaboration.delegation.inventory_page", {
+ "query": query, "record_ids": keys, "observations": observations,
+ "read_facts": {"observed_at": now_utc_iso(), "directory_read_at": directory_read_at,
+ "operations_read_at": operations_read_at,
+ "available_count": available_count},
+ }, large_local_snapshot=True)
diff --git a/loopx/control_plane/effect_runtime.py b/loopx/control_plane/effect_runtime.py
index 248c37a586..bda89d378c 100644
--- a/loopx/control_plane/effect_runtime.py
+++ b/loopx/control_plane/effect_runtime.py
@@ -42,6 +42,7 @@
MAX_REQUEST_BYTES = 2 * 1024 * 1024
MAX_LOCAL_SNAPSHOT_BYTES = 64 * 1024 * 1024
LOCAL_SNAPSHOT_METHODS = frozenset({
+ "collaboration.delegation.inventory_page",
"benchmark.trajectory.codex-usage",
"coordination.local_authority.todo_source",
"todo.context.page",
diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts
index a0653c9dc1..0a0bee20bb 100644
--- a/loopx/control_plane/effect_runtime_handlers.ts
+++ b/loopx/control_plane/effect_runtime_handlers.ts
@@ -605,7 +605,7 @@ export function createEffectRuntimeHandlers(
["collaboration.delegation.checked_artifacts", lazyHandler(() => import("./collaboration/delegation.ts"), ({delegationCheckedArtifacts}) => delegationCheckedArtifacts)],
["collaboration.delegation.turn_plan", lazyHandler(() => import("./collaboration/delegation.ts"), ({delegationTurnPlanDecision}) => delegationTurnPlanDecision)],
["collaboration.delegation.inventory_query", lazyHandler(() => import("./collaboration/delegation.ts"), ({delegationInventoryQuery}) => delegationInventoryQuery)],
- ["collaboration.delegation.inventory_item", lazyHandler(() => import("./collaboration/delegation.ts"), ({delegationInventoryItem}) => delegationInventoryItem)],
+ ["collaboration.delegation.inventory_page", lazyHandler(() => import("./collaboration/delegation.ts"), ({delegationInventoryPage}) => delegationInventoryPage)],
["collaboration.chat_mode", lazyHandler(() => import("./collaboration/chat_mode.ts"), ({planChatMode}) => planChatMode)],
["collaboration.goal_draft", lazyHandler(() => import("./collaboration/goal_draft.ts"), ({admitGoalDraft}) => (params) => ({draft: admitGoalDraft(params)}))],
["collaboration.conversation.trigger", lazyHandler(() => import("./collaboration/conversation_trigger.ts"), ({resolveConversationTrigger}) => resolveConversationTrigger)],
diff --git a/loopx/control_plane/effect_runtime_server.ts b/loopx/control_plane/effect_runtime_server.ts
index 0760268b9e..7ac30c191b 100644
--- a/loopx/control_plane/effect_runtime_server.ts
+++ b/loopx/control_plane/effect_runtime_server.ts
@@ -28,6 +28,7 @@ const MAX_REQUEST_BYTES = 2 * 1024 * 1024;
const MAX_INLINE_RESPONSE_BYTES = 2 * 1024 * 1024;
// Explicit opt-in: ordinary effects retain the 2 MiB request/response wire.
const LOCAL_SNAPSHOT_METHODS = new Set([
+ "collaboration.delegation.inventory_page",
"benchmark.trajectory.codex-usage",
"coordination.local_authority.todo_source",
"todo.context.page",
diff --git a/tests/control_plane/test_delegation_preflight_request_scope.py b/tests/control_plane/test_delegation_preflight_request_scope.py
index 23631aecf7..a4c16a57f2 100644
--- a/tests/control_plane/test_delegation_preflight_request_scope.py
+++ b/tests/control_plane/test_delegation_preflight_request_scope.py
@@ -6,11 +6,42 @@
import pytest
from loopx.control_plane import effect_runtime
-from test_local_delegation import service as delegation_service
+from test_local_delegation import brief, service as delegation_service
service = delegation_service
+def test_each_inventory_page_pins_once_and_reads_current_records(service, monkeypatch):
+ root, runner = service
+ monkeypatch.setattr(runner, "_spawn", lambda _: None)
+ for name in ["first", "second", "third"]:
+ runner.start("analysis", name, brief())
+ original = effect_runtime._runtime_fingerprint
+ revisions = []
+
+ def measured_fingerprint():
+ revision = original()
+ revisions.append(revision)
+ return revision
+
+ monkeypatch.setattr(effect_runtime, "_runtime_fingerprint", measured_fingerprint)
+ first = runner.operations(limit=2)
+ assert len(revisions) == 1
+ assert first["page_readback_complete"] and first["has_more"]
+ second = runner.operations(limit=2, cursor=first["next_cursor"])
+ assert len(revisions) == 2
+ assert second["page_readback_complete"] and len(second["items"]) == 1
+
+ # A page pins executable source, never a previous operation observation.
+ runner.path(first["items"][0]["operation_id"]).write_text("{")
+ damaged = runner.operations(limit=2)
+ assert len(revisions) == 3
+ assert not damaged["page_readback_complete"]
+ assert not damaged["projection_envelope"]["complete"]
+ assert len(damaged["items"]) == 2
+ assert not (root / "host-started").exists()
+
+
def test_each_inspection_pins_once_and_reads_current_binding(service, monkeypatch):
root, runner = service
original = effect_runtime._runtime_fingerprint
diff --git a/tests/control_plane_ts/delegation_inventory.test.ts b/tests/control_plane_ts/delegation_inventory.test.ts
new file mode 100644
index 0000000000..2ed8183138
--- /dev/null
+++ b/tests/control_plane_ts/delegation_inventory.test.ts
@@ -0,0 +1,97 @@
+import test from "node:test";
+import assert from "node:assert/strict";
+import {delegationInventoryPage} from "../../loopx/control_plane/collaboration/delegation.ts";
+import type {JsonObject} from "../../loopx/control_plane/effect_program.ts";
+import {PROJECTION_ENVELOPE_SERVE_REQUEST, sealProjectionEnvelope} from "../../loopx/control_plane/projection_envelope.ts";
+
+function inventoryPageFixture() {
+ const record_ids = ["a", "b", "c"].map(key => key.repeat(64));
+ const observations = record_ids.slice(0, 2).map((record_id, index) => ({
+ record: {record_id, operation_id: `review-${index}`},
+ observation: {operation_id: `review-${index}`, request_id: `request-${index}`,
+ agent_id: "reviewer", todo_id: "todo_review", status: "prepared",
+ worker_active: false, recovery_required: true},
+ }));
+ const read_facts = {observed_at: "2026-10-01T10:00:02Z", directory_read_at: "2026-10-01T10:00:00Z",
+ operations_read_at: "2026-10-01T10:00:01Z", available_count: 3};
+ return {query: {limit: 2, cursor: null}, record_ids, observations, read_facts};
+}
+
+test("inventory page uses an unread lookahead without skipping or inventing results", () => {
+ const request = inventoryPageFixture();
+ for (const [count, more] of [[0, false], [1, false], [2, false], [3, true]] as const) {
+ const page = delegationInventoryPage({...request, record_ids: request.record_ids.slice(0, count),
+ observations: request.observations.slice(0, Math.min(count, 2)),
+ read_facts: {...request.read_facts, available_count: count}});
+ assert.deepEqual((page.items as JsonObject[]).map(row => row.record_id), request.record_ids.slice(0, Math.min(count, 2)));
+ assert.equal(page.has_more, more);
+ assert.equal(page.next_cursor, more ? request.record_ids[1] : null);
+ assert.equal(page.page_readback_complete, true);
+ }
+});
+
+test("inventory envelopes distinguish complete page, oldest read, unreadable and later replay", () => {
+ const request = inventoryPageFixture();
+ const page = delegationInventoryPage(request);
+ const envelope = page.projection_envelope as JsonObject;
+ assert.equal(envelope.complete, true);
+ assert.equal(envelope.fresh, true);
+ assert.deepEqual(envelope.coverage, {scope: "requester_page", expected_count: 2,
+ included_count: 2, omitted: [], shown_count: 2, available_count: 3, truncated: true, complete: true});
+ const replay = sealProjectionEnvelope({schema_version: PROJECTION_ENVELOPE_SERVE_REQUEST,
+ envelope, served_at: "2026-10-01T10:06:00Z"});
+ assert.equal(replay.observed_at, envelope.observed_at);
+ assert.equal(replay.served_at, "2026-10-01T10:06:00Z");
+ assert.equal(replay.fresh, false);
+ assert.equal(replay.served_from_cache, true);
+ assert.deepEqual(replay.alert_reasons, ["stale_sources"]);
+ const slow = delegationInventoryPage({...request,
+ read_facts: {...request.read_facts, observed_at: "2026-10-01T10:06:00Z"}}).projection_envelope as JsonObject;
+ assert.equal(slow.fresh, false); // Assembly time cannot refresh earlier source reads.
+ const damaged = delegationInventoryPage({...request, observations: [
+ {...request.observations[0]!, observation: null}, request.observations[1]!],
+ }).projection_envelope as JsonObject;
+ assert.equal(damaged.complete, false);
+ assert.deepEqual(damaged.alert_reasons, ["incomplete_coverage", "unreadable_sources"]);
+});
+
+test("malformed observations stay unavailable without hiding a healthy sibling or the next page", () => {
+ const request = inventoryPageFixture();
+ const invalid: (JsonObject | null)[] = [null,
+ {...request.observations[0]!.observation, operation_id: "other"},
+ {...request.observations[0]!.observation, status: "not-an-observation"},
+ {...request.observations[0]!.observation, worker_active: null},
+ {...request.observations[0]!.observation, status: "accepted", artifacts: []},
+ ];
+ for (const observation of invalid) {
+ const page = delegationInventoryPage({...request, observations: [
+ {...request.observations[0]!, observation}, request.observations[1]!,
+ ]});
+ const rows = page.items as JsonObject[];
+ assert.equal(rows[0]!.status, "unavailable");
+ assert.equal(rows[0]!.operation_id, "review-0");
+ assert.equal(rows[0]!.recovery_required, null);
+ assert.equal(rows[0]!.artifacts, undefined);
+ assert.equal(rows[1]!.status, "prepared");
+ assert.equal(page.page_readback_complete, false);
+ assert.equal(page.has_more, true);
+ assert.equal(page.next_cursor, request.record_ids[1]);
+ }
+});
+
+test("inventory page refuses incomplete, reordered or out-of-cursor framing", () => {
+ const request = inventoryPageFixture();
+ for (const patch of [
+ {record_ids: ["invalid"]},
+ {record_ids: [request.record_ids[0]!, request.record_ids[0]!] },
+ {record_ids: [...request.record_ids].reverse()},
+ {record_ids: [...request.record_ids, "d".repeat(64)]},
+ {query: {limit: 2, cursor: request.record_ids[0]!}},
+ {observations: request.observations.slice(0, 1)},
+ {observations: [...request.observations].reverse()},
+ {read_facts: null},
+ {read_facts: {...request.read_facts, available_count: 1}},
+ {observations: [{...request.observations[0]!,
+ record: {...request.observations[0]!.record, operation_id: "../other"}}, request.observations[1]!] },
+ ]) assert.throws(() => delegationInventoryPage({...request, ...patch}));
+});
diff --git a/tests/test_chat_loopx_mode.py b/tests/test_chat_loopx_mode.py
index 0b0083e18e..270c387811 100644
--- a/tests/test_chat_loopx_mode.py
+++ b/tests/test_chat_loopx_mode.py
@@ -353,6 +353,7 @@ def test_pause_fences_dispatch_and_does_not_cancel_members(mode, monkeypatch):
inventory = adapter.session.read_tool_handler(TOOL["name"], {"action": "operations"})
assert inventory["ok"] and inventory["items"] == []
assert inventory["page_readback_complete"] and not inventory["has_more"]
+ assert inventory["projection_envelope"]["fresh"] and inventory["projection_envelope"]["complete"]
invalid = adapter.session.read_tool_handler(TOOL["name"], {"action": "operations", "operation_id": "one"})
assert invalid["error"] == "collaboration_request_rejected"
@@ -555,6 +556,7 @@ def test_owner_team_readback_is_configured_scoped_and_does_not_start_a_turn(mode
before = service.store.load_session(sid)
result = service.read_team(sid, {"operation": "operations"})
assert result["items"] == [] and result["page_readback_complete"]
+ assert result["projection_envelope"]["coverage"]["scope"] == "requester_page"
assert service.store.load_session(sid) == before
assert calls == []
with pytest.raises(ValueError, match="invalid team readback"):
diff --git a/tests/test_delegation_inventory.py b/tests/test_delegation_inventory.py
index 9ac402e4c4..2f30a5090d 100644
--- a/tests/test_delegation_inventory.py
+++ b/tests/test_delegation_inventory.py
@@ -1,14 +1,67 @@
"""Journal recovery must preserve scope and expose incomplete readback honestly."""
import json
+from hashlib import sha256
import pytest
+from loopx.control_plane.collaboration import delegation_inventory
from loopx.control_plane.collaboration.inbox import _read, _write
+from loopx.control_plane.effect_runtime import effect_runtime_result
from test_local_delegation import brief, service as delegation_service
service = delegation_service
+def test_cli_inventory_discloses_page_scope_and_unreadable_operations(service, monkeypatch):
+ from test_delegation_cli import cli
+
+ root, runner = service
+ monkeypatch.setattr(runner, "_spawn", lambda _: None)
+ for name in ["first", "second", "third", "fourth", "fifth"]:
+ runner.start("analysis", name, brief())
+ before = {path.name: path.read_bytes() for path in runner.path("inventory").parent.glob("*.json")}
+
+ status, page = cli(runner, "operations", "--limit", "2")
+ assert status == 0 and page["has_more"]
+ envelope = page["projection_envelope"]
+ assert envelope["fresh"] and envelope["complete"] and not envelope["alert"]
+ assert envelope["coverage"]["scope"] == "requester_page"
+ assert envelope["coverage"]["shown_count"] == 2
+ assert envelope["coverage"]["available_count"] == 5
+ assert envelope["coverage"]["truncated"] # Complete page is not the whole inventory.
+ # Lock-holder diagnostics change on reads; the operation records must not.
+ assert {path.name: path.read_bytes() for path in runner.path("inventory").parent.glob("*.json")} == before
+
+ runner.path(page["items"][0]["operation_id"]).write_text("{")
+ status, damaged = cli(runner, "operations", "--limit", "2")
+ assert status == 0 and len(damaged["items"]) == 2
+ assert not damaged["page_readback_complete"]
+ envelope = damaged["projection_envelope"]
+ assert not envelope["fresh"] and not envelope["complete"] and envelope["alert"]
+ assert envelope["alert_reasons"] == ["incomplete_coverage", "unreadable_sources"]
+ assert envelope["coverage"]["omitted"] == [{"reason": "unavailable_operations", "count": 1, "refs": []}]
+ assert not (root / "host-started").exists()
+
+
+def test_inventory_source_read_times_survive_assembly_and_replay(service, monkeypatch):
+ _, runner = service
+ monkeypatch.setattr(runner, "_spawn", lambda _: None)
+ runner.start("analysis", "slow-read", brief())
+ times = iter(["2026-10-01T10:00:00Z", "2026-10-01T10:00:01Z", "2026-10-01T10:06:00Z"])
+ monkeypatch.setattr(delegation_inventory, "now_utc_iso", lambda: next(times))
+ envelope = runner.operations()["projection_envelope"]
+ assert not envelope["fresh"] and envelope["complete"]
+ assert envelope["alert_reasons"] == ["stale_sources"]
+ assert [row["last_read_at"] for row in envelope["sources"]] == [
+ "2026-10-01T10:00:00Z", "2026-10-01T10:00:01Z"]
+ replay = effect_runtime_result("projection.envelope.seal", {
+ "schema_version": "loopx_projection_envelope_serve_request_v0", "envelope": envelope,
+ "served_at": "2026-10-01T10:07:00Z"})
+ assert replay["observed_at"] == envelope["observed_at"]
+ assert replay["served_at"] == "2026-10-01T10:07:00Z"
+ assert not replay["fresh"] and replay["age_seconds"] == 60
+
+
def test_pages_find_all_original_operations_without_dispatch(service, monkeypatch):
root, runner = service
starts = []
@@ -16,10 +69,20 @@ def test_pages_find_all_original_operations_without_dispatch(service, monkeypatc
for i in range(7):
runner.start("analysis", f"work-{i}", brief())
starts.clear()
+ projection_requests = []
+ real_projection = delegation_inventory.effect_runtime_result
+
+ def traced_projection(method, *args, **kwargs):
+ projection_requests.append(method)
+ return real_projection(method, *args, **kwargs)
+
+ monkeypatch.setattr(delegation_inventory, "effect_runtime_result", traced_projection)
seen = []
cursor = None
while True:
+ projection_requests.clear()
page = runner.operations(limit=2, cursor=cursor)
+ assert len(projection_requests) <= 2 # One preflight and one page reduction across the RPC boundary.
assert page["page_readback_complete"] and len(page["items"]) <= 2
seen.extend(row["operation_id"] for row in page["items"])
if not page["has_more"]:
@@ -33,7 +96,9 @@ def test_pages_find_all_original_operations_without_dispatch(service, monkeypatc
config = json.loads(runner.config.read_text())
config["bindings"] = []
runner.config.write_text(json.dumps(config))
+ projection_requests.clear()
page = runner.operations()
+ assert len(projection_requests) <= 2
assert not page["page_readback_complete"] and len(page["items"]) == 7
assert all(row["status"] == "unavailable" for row in page["items"])
@@ -41,7 +106,7 @@ def test_pages_find_all_original_operations_without_dispatch(service, monkeypatc
def test_corruption_and_stopped_worker_do_not_hide_healthy_sibling(service, monkeypatch):
_, runner = service
monkeypatch.setattr(runner, "_spawn", lambda _: None)
- for name in ["healthy", "stopped", "corrupt", "mismatched"]:
+ for name in ["healthy", "stopped", "corrupt", "mismatched", "malformed"]:
runner.start("analysis", name, brief())
row = _read(runner.path("stopped"))
row["created_at"] = 0
@@ -50,12 +115,16 @@ def test_corruption_and_stopped_worker_do_not_hide_healthy_sibling(service, monk
row = _read(runner.path("mismatched"))
row["identity"]["operation_id"] = "healthy"
_write(runner.path("mismatched"), row)
+ row = _read(runner.path("malformed"))
+ row["status"] = "not-an-observation"
+ _write(runner.path("malformed"), row)
page = runner.operations()
by_id = {row["operation_id"]: row for row in page["items"] if row["operation_id"]}
assert by_id["healthy"]["status"] == "prepared"
assert by_id["stopped"]["recovery_required"]
- assert len(page["items"]) == 4 and not page["page_readback_complete"]
- assert sum(row["status"] == "unavailable" for row in page["items"]) == 2
+ assert len(page["items"]) == 5 and not page["page_readback_complete"]
+ assert by_id["malformed"]["status"] == "unavailable"
+ assert sum(row["status"] == "unavailable" for row in page["items"]) == 3
# A stop receipt beside its record is not another record and reads back as stopped.
assert runner.stop("healthy", execute=True)["phase"] == "settled"
assert runner._stop_path(runner.path("healthy")).exists()
@@ -63,7 +132,7 @@ def test_corruption_and_stopped_worker_do_not_hide_healthy_sibling(service, monk
runner._host_process_record(runner.path("healthy")).write_text("{}")
page = runner.operations()
by_id = {row["operation_id"]: row for row in page["items"] if row["operation_id"]}
- assert len(page["items"]) == 4 and by_id["healthy"]["status"] == "stopped"
+ assert len(page["items"]) == 5 and by_id["healthy"]["status"] == "stopped"
assert not by_id["healthy"]["recovery_required"]
@@ -84,3 +153,34 @@ def denied(path):
monkeypatch.setattr(type(directory), "iterdir", denied)
with pytest.raises(PermissionError):
runner.operations()
+
+
+def test_full_page_preserves_accepted_references_above_inline_transport_budget(service):
+ record_ids = sorted(sha256(str(index).encode()).hexdigest() for index in range(50))
+ observations = [{
+ "record": {"record_id": record_id, "operation_id": f"result-{index}"},
+ "observation": {
+ "operation_id": f"result-{index}", "request_id": f"request-{index}",
+ "agent_id": "worker", "todo_id": f"todo_{index}", "status": "accepted",
+ "worker_active": False, "recovery_required": False,
+ "artifacts": [{"ref": f"output-{index}.json", "sha256": "a" * 64,
+ "text": "private body " * 4000}],
+ },
+ } for index, record_id in enumerate(record_ids)]
+ request = {"query": {"limit": 50, "cursor": None}, "record_ids": record_ids,
+ "observations": observations,
+ "read_facts": {"observed_at": "2026-10-01T10:00:02Z",
+ "directory_read_at": "2026-10-01T10:00:00Z",
+ "operations_read_at": "2026-10-01T10:00:01Z", "available_count": 50}}
+ assert len(json.dumps(request).encode()) > 2 * 1024 * 1024
+
+ page = effect_runtime_result("collaboration.delegation.inventory_page", request,
+ large_local_snapshot=True)
+
+ assert page["page_readback_complete"] and not page["has_more"]
+ assert page["next_cursor"] is None
+ assert [row["record_id"] for row in page["items"]] == record_ids
+ assert all(row["status"] == "accepted" for row in page["items"])
+ assert [row["artifacts"] for row in page["items"]] == [
+ [{"ref": f"output-{index}.json", "sha256": "a" * 64}] for index in range(50)
+ ]
diff --git a/tests/test_local_delegation.py b/tests/test_local_delegation.py
index 05887582f2..fa602e7ba5 100644
--- a/tests/test_local_delegation.py
+++ b/tests/test_local_delegation.py
@@ -393,6 +393,7 @@ async def disconnect_requester():
assert not (root / "host-started").exists()
inventory = await session.call_tool("list_delegations", {})
assert not inventory.isError and json.loads(inventory.content[0].text)["items"] == []
+ assert json.loads(inventory.content[0].text)["projection_envelope"]["complete"]
assert "stop_delegation" in {tool.name for tool in (await session.list_tools()).tools}
result = await session.call_tool("start_delegation", {
"binding_id": "analysis", "operation_id": "analysis-1", "brief": brief()})
@@ -400,6 +401,7 @@ async def disconnect_requester():
inventory = await session.call_tool("list_delegations", {})
assert not inventory.isError
assert json.loads(inventory.content[0].text)["items"][0]["operation_id"] == "analysis-1"
+ assert json.loads(inventory.content[0].text)["projection_envelope"]["fresh"]
return json.loads(result.content[0].text)
# Exiting the real stdio session closes the requesting MCP process.
first = asyncio.run(disconnect_requester())
diff --git a/tsconfig.control-plane.json b/tsconfig.control-plane.json
index 43a84b6831..12fc5478a2 100644
--- a/tsconfig.control-plane.json
+++ b/tsconfig.control-plane.json
@@ -39,6 +39,7 @@
"tests/control_plane_ts/goal_workspace_digest.test.ts",
"tests/control_plane_ts/interaction_contract.test.ts",
"tests/control_plane_ts/runtime_decode.test.ts",
+ "tests/control_plane_ts/delegation_inventory.test.ts",
"tests/control_plane_ts/quota_run_decision.test.ts",
"tests/control_plane_ts/authority_store.test.ts",
"tests/control_plane_ts/authority_state_log.test.ts",