refactor(collaboration): compose delegation inventory pages in TypeScript

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.
+
 
 <a id="current-delivery-frontier-2026-10-02"></a>
 

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`；不可读操作触发告警，展示/可用数量说明
+剩余页，重放保留原始观察时间。
+每页复用既有代码版本请求作用域；操作记录仍实时读取，下一页重新解析当前代码。
+
 
 <a id="当前交付边界2026-10-02"></a>
 

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;

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)
 

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");

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,14 +38,18 @@ 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.
             if directory.exists():
                 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)

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",

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)],

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",

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