Read write scopes on demand before a progressive Turn is ready

docs/reference/contracts/interface-budget-contract.md

@@ -470,6 +470,18 @@ effect 日志不变与原捕获被改写后的拒绝验证。分别计量导航
 冒充完整决策或模型成本收益；生产默认不变。两端各自导入已固定的产品版本，并共用
 同一份冻结的当前调用夹具，避免将夹具修订误计为产品输出变化。
 
+When a normal Turn still needs explicit selection or an ordinary claim, the
+progressive view can point to the captured write scope through
+`goal_boundary.write_scope_pointer`. Read that pointer using the same
+capture's supporting-source reference if the choice depends on write scope.
+Stop and approval rules remain inline. The claim checks scope atomically, and
+the ready decision supplies the complete current scope before delivery.
+
+正常 Turn 仍需选择任务或普通认领时，渐进视图可用
+`goal_boundary.write_scope_pointer` 指向原捕获中的写入范围。选择依赖写入范围时，
+按同一捕获给出的源引用读取该指针。停止和审批规则继续内联。认领会原子校验范围，
+进入可执行阶段后从当前决策读取完整写入范围。
+
 Ordinary CLI reads require canonical authority. Both differential arms explicitly
 back up and import the same complete synthetic source before measuring; fixture
 setup is outside the measured commands. On base `887d66551` and the read-retirement

loopx/control_plane/quota/progressive_context.ts

@@ -6,7 +6,8 @@ import {visionAuthoringContract} from "../goals/vision_checkpoint.ts";
 import type {JsonObject} from "../effect_program.ts";
 import {jsonObject, requireBoolean, requireJsonObject} from "../runtime_decode.ts";
 import {GOAL_OUTCOMES_SCHEMA, OUTCOME_CONTEXT_NAVIGATION_SCHEMA} from "../goals/outcome_requirements.ts";
-import {FULL_OUTCOME_READ_GUIDANCE, PROGRESSIVE_OUTCOME_READ_GUIDANCE} from "../work_items/interaction_contract.ts";
+import {FULL_OUTCOME_READ_GUIDANCE, PROGRESSIVE_OUTCOME_READ_GUIDANCE}
+  from "../work_items/interaction_contract.ts";
 import {quotaSummaryReferences, contextSourceIdentity, contextProjectionChanges,
   type ProjectionSourceValues} from "./diagnostic_projection.ts";
 import {encodeContextDelivery, fileNavigationPaths} from "./context_delivery.ts";
@@ -165,7 +166,8 @@ function inlineReadSections(request: JsonObject, required: JsonObject): JsonObje
 
 /** Return paths instead of round-tripping the caller's JSON numeric values
  * through JS. The IO adapter performs only these typed-owner-selected edits. */
-export function progressiveContextPlan(request: JsonObject): JsonObject {
+export function progressiveContextPlan(request: JsonObject,
+    scope?: {path: Path; marker_path: Path; instruction: string}): JsonObject {
   const payload = requireJsonObject(request.payload, "captured context");
   const deferred: JsonObject[] = [], instructions: JsonObject[] = [];
   const unchanged = {deferred, instructions};
@@ -222,7 +224,6 @@ export function progressiveContextPlan(request: JsonObject): JsonObject {
         deferred.push({path: [...parent, "requirements", j, "agent_review", field], parent, index_key});
     });
   }
-  if (!deferred.length && !references.length) return unchanged;
   // Replace only the exact source-owned obligation. Other merged obligations,
   // objective adoption, unknown provider prose and pending reads remain intact.
   for (const listPath of deferred.length ? [sourcePath, [...agent, "required_reads"]] : []) {
@@ -236,6 +237,11 @@ export function progressiveContextPlan(request: JsonObject): JsonObject {
           row.reason.replace(FULL_OUTCOME_READ_GUIDANCE, PROGRESSIVE_OUTCOME_READ_GUIDANCE)});
     });
   }
+  if (scope) {
+    deferred.push({path: scope.path, read_instruction: scope.instruction});
+    instructions.push({path: scope.marker_path, value: pointer(scope.path)});
+  }
+  if (!deferred.length && !references.length) return unchanged;
   if (references.length) instructions.push({path: ["context_reference_instruction"], value:
     "A Todo row with context_ref is exactly the complete row at that JSON pointer in THIS required view. "
     + "Use that row for every referenced lane; all constraints, evidence, ownership and order still apply. "
@@ -289,9 +295,12 @@ function requiredNativeContext(request: JsonObject, sourceValues: ProjectionSour
       removed.push(["context", ...path]);
     }
     for (const edit of deferred) {
-      const path = edit.path as Path, owner = edit.parent as Path;
+      const path = edit.path as Path;
       delete parent(path)[path.at(-1)!];
-      parent([...owner, String(edit.index_key)])[String(edit.index_key)] = pointer(owner);
+      if (typeof edit.index_key === "string") {
+        const owner = edit.parent as Path;
+        parent([...owner, edit.index_key])[edit.index_key] = pointer(owner);
+      }
       removed.push(["context", ...path]);
     }
     for (const edit of instructions) {
@@ -325,7 +334,8 @@ function requiredNativeContext(request: JsonObject, sourceValues: ProjectionSour
           return jsonObject(source)?.content !== undefined && !hasRead(path) ? [pointer(path)] : [];
         }) : [],
       },
-      instruction: "This view retains complete requirements and current review/check states. "
+      instruction: deferred.map(edit => typeof edit.read_instruction === "string" ? edit.read_instruction : "").join("")
+        + "This view retains complete requirements and current review/check states. "
         + "Source history, review reasoning and evidence references are supporting material. "
         + "Read them from this same immutable capture when a decision, verification or review depends on them. "
         + "For source history use supporting_material_pointer plus /sources. For review detail append "
@@ -415,9 +425,9 @@ export function prepareProgressiveContextDelivery(request: JsonObject, legacyFie
 
 /** Stateless presentation uses the same plan without inventing a native session,
  * retained baseline or admission identity. Hosts may add delta independently. */
-export function prepareProgressiveContext(request: JsonObject): JsonObject {
+export function prepareProgressiveContext(request: JsonObject, plan?: JsonObject): JsonObject {
   const values: ProjectionSourceValues = new WeakMap();
-  const required = requiredNativeContext(request, values);
+  const required = requiredNativeContext(request, values, plan);
   const {operation: _operation, source_edits: _edits, source_number_tokens: _numbers, ...source} = request;
   return contextProjectionChanges(source, {context: required.context, projected: required.projected}, values);
 }

loopx/control_plane/quota/turn_envelope.ts

@@ -1,5 +1,6 @@
+import {EXECUTION_PREPARATION_SCHEMA_VERSION} from "../work_items/execution_preparation.ts";
 import {capturedContextSelection} from "./context_selection.ts";
-import {contextReadSectionDigests, initialContextReadPointers, prepareProgressiveContext, prepareProgressiveContextDelivery} from "./progressive_context.ts";
+import {contextReadSectionDigests, initialContextReadPointers, prepareProgressiveContext, prepareProgressiveContextDelivery, progressiveContextPlan} from "./progressive_context.ts";
 import {contextProjectionChanges, contextSourceIdentity, projectQuotaDiagnostics} from "./diagnostic_projection.ts";
 import type { LoopXTurnRouteValue } from "../turn_driver/turn_contract_generated.ts";
 import { EffectiveAction, type QuotaEffectiveActionValue } from "./effective_action.generated.ts";
@@ -21,7 +22,7 @@ import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts";
 import { projectTurnStartUnavailableContext } from "../capability_hooks.ts";
 import { jsonObject, requireBoolean, requireJsonObject, requireNonEmptyString, requireStringLiteral } from "../runtime_decode.ts";
 import { projectPendingCapabilityIntent } from "../work_items/pending_capability_intent.ts";
-import { projectInteractionRequiredReads, selectedWorkTodo, selectedWorkTodoId, USER_CHANNEL_NOTIFICATION_POLICIES } from "../work_items/interaction_contract.ts";
+import { INTERACTION_CONTRACT_SCHEMA_VERSION, projectInteractionRequiredReads, selectedWorkTodo, selectedWorkTodoId, USER_CHANNEL_NOTIFICATION_POLICIES } from "../work_items/interaction_contract.ts";
 import { measureTurnEnvelope, TURN_ENVELOPE_SECTION_TARGETS } from "./turn_envelope_budget.ts";
 export { TURN_ENVELOPE_BUDGET_BYTES } from "./turn_envelope_budget.ts";
 
@@ -1678,6 +1679,36 @@ function replanCliView(payload: JsonObject, identity: ReplanValueIdentity = cano
     replan_action_packet: replanActionPacket(payload, reads)};
 }
 
+/** Present the long scope list on demand only while an ordinary Turn prepares work. */
+function progressivePreparationScope(payload: JsonObject):
+  {path: string[]; marker_path: string[]; instruction: string} | undefined {
+  const interaction = jsonObject(payload.interaction_contract);
+  const cli = jsonObject(interaction?.cli_channel);
+  const preparation = jsonObject(cli?.execution_preparation);
+  if (interaction?.schema_version !== INTERACTION_CONTRACT_SCHEMA_VERSION ||
+      payload.should_run !== true || payload.effective_action !== EffectiveAction.NORMAL_RUN ||
+      jsonObject(interaction.agent_channel)?.delivery_allowed !== false ||
+      cli?.spend_allowed_now !== false || cli.spend_after_validation !== false) return undefined;
+  const selecting = cli.selection_required === true && !Object.hasOwn(cli, "execution_preparation") &&
+    jsonObject(cli.selection_command)?.schema_version === "action_selection_cli_command_v1";
+  const claiming = cli.selection_required !== true && payload.normal_delivery_allowed === false &&
+    preparation?.schema_version === EXECUTION_PREPARATION_SCHEMA_VERSION && preparation.state === "claim_required" &&
+    preparation.delivery_allowed === false && jsonObject(preparation.claim)?.host_execution_allowed === true;
+  if (!selecting && !claiming) return undefined;
+  const boundary = jsonObject(payload.goal_boundary);
+  const scopes = boundary?.write_scope;
+  // The recovery pointer and its guidance also cost bytes. Leave short lists inline.
+  if (!boundary || Object.hasOwn(boundary, "write_scope_pointer") ||
+      !Array.isArray(scopes) || scopes.some(scope => typeof scope !== "string") ||
+      Buffer.byteLength(JSON.stringify(scopes)) < 1024) return undefined;
+  return {
+    path: ["goal_boundary", "write_scope"],
+    marker_path: ["goal_boundary", "write_scope_pointer"],
+    instruction: "If choosing work depends on write scope, select the pointer in goal_boundary.write_scope_pointer from the source below. "
+      + "Use the complete current scope from the ready decision before delivery. ",
+  };
+}
+
 function nativeReplanPresentation(request: JsonObject): JsonObject | undefined {
   const path = object(request.native_projection).captured_decision_path;
   if (typeof path !== "string") return undefined;
@@ -1696,6 +1727,11 @@ function nativeReplanPresentation(request: JsonObject): JsonObject | undefined {
   return view === source ? undefined : {staged_context: view, deferred: [], instructions: []};
 }
 
+function contextPresentationPlan(request: JsonObject): JsonObject {
+  const context = requireJsonObject(request.context, "current projection");
+  return progressiveContextPlan({payload: context, surface: request.surface}, progressivePreparationScope(context));
+}
+
 export function evaluateTurnEnvelope(value: unknown): JsonObject {
   const request = requireJsonObject(value, "turn envelope request");
   const operation = request.operation;
@@ -1755,11 +1791,11 @@ export function evaluateTurnEnvelope(value: unknown): JsonObject {
     return {sections: contextReadSectionDigests(request,
       (selected.selections as {path: (string | number)[]}[]).map(row => row.path))};
   }
-  if (operation === "progressive_context_delivery") return prepareProgressiveContext(request);
+  if (operation === "progressive_context_delivery") return prepareProgressiveContext(request, contextPresentationPlan(request));
   if (operation === "context_delivery") return request.native_projection === undefined
     ? prepareContextDelivery(request, LEGACY_QUOTA_CONTEXT_FIELDS)
     : prepareProgressiveContextDelivery(request, LEGACY_QUOTA_CONTEXT_FIELDS, INITIAL_CONTEXT_DECISION_FIELDS,
-      nativeReplanPresentation(request));
+      nativeReplanPresentation(request) ?? contextPresentationPlan(request));
   if (operation === "native_context_lifecycle") return updateNativeContextLifecycle(request);
   if (operation === "context_recovery") return restoreContextDelivery(request, LEGACY_QUOTA_CONTEXT_FIELDS);
   if (operation === "bind_capture") return bindTurnDecisionCapture(request);

tests/control_plane/test_quota_execution_preparation.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from datetime import timedelta
+import hashlib
 import json
 import os
 from pathlib import Path
@@ -69,6 +70,7 @@ def _run(
     leases: list[dict[str, object]],
     *guard_args: str,
     todos: list[dict[str, object]] | None = None,
+    write_scopes: list[str] | None = None,
 ):
     isolate_sqlite_runtime(tmp_path, monkeypatch)
     runtime = tmp_path / "runtime"
@@ -93,6 +95,7 @@ def _run(
         extra_goal_fields={"coordination": {
             "registered_agents": [AGENT, "agent-b"],
             "handoff_mode": "hard_lease",
+            **({"write_scope": write_scopes, "requires_parent_approval": ["publish"]} if write_scopes else {}),
         }},
     )
     projection = build_todo_runtime_shadow_projection(
@@ -389,6 +392,7 @@ def hook():
 def test_projected_claim_prepares_the_same_receipt_bound_turn(
     tmp_path, monkeypatch, provider, multiple_candidates,
 ):
+    scopes = [f"src/synthetic-area-{index:03d}/**" for index in range(220)]
     unbound = _run(
         tmp_path,
         monkeypatch,
@@ -399,6 +403,7 @@ def test_projected_claim_prepares_the_same_receipt_bound_turn(
         "--begin-turn",
         todos=[_todo(), {**_todo(), "todo_id": "todo_other_candidate", "index": 2}]
         if multiple_candidates else None,
+        write_scopes=scopes,
     )
     assert "heartbeat_receipt" in unbound, json.dumps(
         unbound, ensure_ascii=False, indent=2,
@@ -433,6 +438,52 @@ def test_projected_claim_prepares_the_same_receipt_bound_turn(
         runtime_root=tmp_path / "runtime", goal_id=GOAL, include_leases=True,
     )["leases"]
 
+    def read_model_view(packet, stage, *, deferred):
+        source = tmp_path / f"{stage}.json"
+        raw = json.dumps(packet, ensure_ascii=False, separators=(",", ":")).encode()
+        source.write_bytes(raw)
+        registry, runtime = tmp_path / "absent-registry.json", tmp_path / "reader-runtime"
+        args = ["quota", "context-project", "--capture-file", str(source),
+            "--capture-sha256", hashlib.sha256(raw).hexdigest(), "--context-delivery", "on",
+            "--delta-delivery", "off", "--goal-id", GOAL, "--agent-id", AGENT, "--turn-instance-id", turn]
+
+        def read(*argv):
+            code, result = run_json_cli_result(*argv, registry_path=registry, runtime_root=runtime)
+            assert code == 0 and result["ok"], result
+            return result
+
+        navigation = read(*args)
+        command = shlex.split(navigation["first_read_command"])
+        page = read(*command[command.index("quota"):])
+        assert page["eof"]
+        values = {row["pointer"]: row["value"] for row in json.loads(page["content"])["selections"]}
+        boundary = packet["goal_boundary"]
+        assert boundary["write_scope"] == scopes and boundary["requires_parent_approval"] == ["publish"]
+        assert values["/heartbeat_receipt"] == packet["heartbeat_receipt"]
+        if deferred:
+            marker = "/goal_boundary/write_scope"
+            assert values["/goal_boundary"] == {
+                **{key: value for key, value in boundary.items() if key != "write_scope"},
+                "write_scope_pointer": marker,
+            }
+            detail = values["/context_details/unread_context_index/source"]
+            restored = read("quota", "context-read", "--capture-file", detail["path"],
+                "--capture-sha256", detail["sha256"], "--select", marker)
+            assert restored["eof"]
+            assert json.loads(restored["content"])["selections"] == [{"pointer": marker, "value": scopes}]
+            if stage == "claim":
+                args[args.index("--context-delivery") + 1] = "off"
+                plain = read(*args)
+                assert plain["mode"] == "full" and plain["recovery"]["sha256"] == hashlib.sha256(raw).hexdigest()
+                assert plain["recovery"]["path"] == str(source)
+        else:
+            assert values["/goal_boundary"] == boundary
+        assert source.read_bytes() == raw and not registry.exists()
+
+    if multiple_candidates:
+        read_model_view(unbound, "selection", deferred=True)
+        read_model_view(before, "claim", deferred=True)
+
     claim_code, claim = run_json_cli_result(
         *before["execution_preparation"]["claim"]["cli_args"],
         registry_path=tmp_path / "registry.json",
@@ -456,6 +507,8 @@ def test_projected_claim_prepares_the_same_receipt_bound_turn(
     assert after["selected_todo"]["todo_id"] == TODO
     assert after["execution_preparation"]["state"] == "ready"
     assert after["normal_delivery_allowed"] is True
+    if multiple_candidates:
+        read_model_view(after, "ready", deferred=False)
 
 
 @pytest.mark.parametrize("provider", ["file", "sqlite"])