Avoid repeating Manager Inbox requests in the Progressive first read

docs/reference/required-work-context.md

@@ -213,12 +213,15 @@ corrections, retirement/expiry and empty replacements retain their provenance,
 revision and advisory authority. This captured pre-work view does not replace the
 preference owner's fresh exact-scope read before a later preference-dependent
 external action, and neither delivery nor a preference grants authority.
-The existing bounded `unfinished_delegations` projection and `read_complete`
-flag also remain visible. Adoption or a queued return does not establish
-completion. Pending peer results, counts, omitted items and unavailable
-observations retain their source semantics. Full request bodies stay on demand and must be
-read before assessing, executing or returning a request; the initial view does
-not consume a result or grant authority.
+The initial read includes the bounded `unfinished_delegations` projection and
+`read_complete` flag. For an admitted normal work Turn with a large, complete
+agenda and a matching required Manager Inbox read, the initial view keeps the
+incoming and outgoing request counts, scan status and required command. The
+required read delivers the rows and their guidance. The original rows remain
+available from the same capture. Adoption and
+queued returns are intermediate states. Read the full current request before
+assessing, executing or returning it. Use the existing scoped request and
+consumption routes for those actions.
 It also exposes existing outcome material addresses and the current settlement
 plan's obligations, including unknown step fields. Only attachments explicitly
 marked `required=false` defer their guidance/templates; their routes stay visible
@@ -366,9 +369,11 @@ full/delta 字节比较计入这些 reader 命令元数据；既有进程内引
 首读也给出现有验收材料地址和当前结算义务。仅明确 `required=false` 的附件延后完整
 说明与模板，使用前须读原附件；未知计划保留全文。首读分页不重复正文中的指针，
 选读摘要仍阻止将它记作完整输入。
-已有的有界 `unfinished_delegations` 投影和 `read_complete` 标记也直接呈现；采纳和
-回传入队不等于完成。待对端结果、计数、遗漏和不可用观测保留原来源语义。
-完整请求按需读取，评估、执行或回传前仍须读其全文；首读不消费结果，也不授予权限。
+首读包含有界 `unfinished_delegations` 投影和 `read_complete` 标记。已准入的普通
+工作 Turn 若有完整的大型委派摘要，且必读列表中有同一条管家收件箱命令，首读保留
+收发请求计数、扫描状态和必读命令，逐条记录及其说明由必做的收件箱读取交付。原记录仍可从
+同一捕获回读。采纳和回传入队是中间状态，评估、执行或回传前须读当前请求全文，
+并沿已有范围约束和消费入口办理。
 
 既有 acceptance owner 给出的完整 `outcome_context` 导航，与当前 capture 的 Goal、
 Agent、所选 Todo、成果版本及 provider revision 一致时，首读完整展开其建议成果在

loopx/control_plane/quota/progressive_context.ts

@@ -498,6 +498,29 @@ export function initialContextReadPointers(context: JsonObject, decisionFields:
   return [...opening, ...remaining];
 }
 
+/** A pending Manager Inbox read already delivers these scoped request rows. */
+function canDeferManagerAgenda(value: unknown, requiredReads: unknown[]): boolean {
+  const {content, ...read} = jsonObject(value) ?? {};
+  const source = jsonObject(content), agenda = jsonObject(source?.unfinished_delegations);
+  if (read.source !== "turn_start_capability_hook" || read.kind !== "operator_inbox"
+      || read.capability_id !== "manager-context" || read.hook_id !== "manager.context_inbox"
+      || read.ordering !== "before_work" || typeof read.command !== "string" || !read.command
+      || source?.ok !== true || source.read_complete !== false
+      || !agenda || agenda.read_command !== read.command || agenda.count_unit !== "request"
+      || agenda.scan_complete !== true || agenda.omitted_count !== 0 || agenda.unavailable_count !== 0
+      || !Array.isArray(agenda.items)
+      || !known(agenda, ["count_unit", "items", "incoming_count", "outgoing_count", "omitted_count",
+        "scan_complete", "unavailable_count", "observed_at", "read_command", "instruction"])
+      || !requiredReads.some(required => isDeepStrictEqual(required, read))) return false;
+  const {incoming_count: incoming, outgoing_count: outgoing, items} = agenda;
+  if (typeof incoming !== "number" || !Number.isSafeInteger(incoming) || incoming < 0
+      || typeof outgoing !== "number" || !Number.isSafeInteger(outgoing) || outgoing < 0
+      || incoming + outgoing !== items.length) return false;
+  // Keep small agendas inline. Three pointers retain the starting counts and
+  // scan status without repeating the required read's details or guidance.
+  return Buffer.byteLength(JSON.stringify(items), "utf8") >= 2048;
+}
+
 function initialContextReadGroups(context: JsonObject, decisionFields: readonly string[]): {
   opening: string[]; remaining: string[];
 } {
@@ -568,6 +591,9 @@ function initialContextReadGroups(context: JsonObject, decisionFields: readonly
   const work = ["interaction_contract", "agent_channel", "work_context"];
   include(...work, "instruction");
   const channel = object(interaction.agent_channel);
+  const requiredInboxReads = context.effective_action === EffectiveAction.NORMAL_RUN && context.normal_delivery_allowed === true
+    && channel?.delivery_allowed === true && channel.must_attempt === true && Array.isArray(channel.required_reads)
+    ? channel.required_reads : [];
   const sources = object(channel?.work_context)?.sources;
   const sourceRows = Array.isArray(sources) ? sources.map((source, index) => ({source, index})) : [];
   const selectedTodo = object(context.selected_todo);
@@ -629,10 +655,11 @@ function initialContextReadGroups(context: JsonObject, decisionFields: readonly
     for (const field of ["ok", "status", "reason", "error", "recovery", "revision", "provider_revision", "source_authority",
       "decision_read_from_provider", "legacy_fallback_used", "graph_enabled", "harness_enabled", "boundary", "authority_read"])
       include(...source, "content", field);
-    // The capability already bounds this agenda. Keep pending returns and
-    // incomplete observations visible without expanding full request bodies.
     include(...source, "content", "read_complete");
-    include(...source, "content", "unfinished_delegations");
+    const agendaPath = [...source, "content", "unfinished_delegations"];
+    if (canDeferManagerAgenda(_source, requiredInboxReads)) {
+      for (const field of ["incoming_count", "outgoing_count", "scan_complete"]) include(...agendaPath, field);
+    } else include(...agendaPath);
     // The preference owner already supplies the complete current scoped view,
     // including corrections, retirement, provenance and advisory authority.
     // Its revision/instructions alone cannot fulfill that pre-work delivery.

tests/control_plane/test_context_first_read.py

@@ -648,6 +648,52 @@ def test_initial_read_preserves_unfinished_requests_and_incomplete_observation(t
     assert detail["ok"] and "FULL-REQUEST-DETAIL-ON-DEMAND" in detail["content"]
 
 
+@pytest.mark.parametrize("route", [
+    "matching", "different", "missing", "incomplete", "omitted", "unknown", "bad_counts", "small", "held",
+])
+def test_initial_read_leaves_large_agenda_rows_to_the_matching_required_read(tmp_path, decision, route):
+    command = "loopx manager-inbox read --goal-id fixture --agent-id author"
+    read = {"source": "turn_start_capability_hook", "kind": "operator_inbox",
+        "capability_id": "manager-context", "hook_id": "manager.context_inbox",
+        "ordering": "before_work", "command": command}
+    purpose = "保留原反馈与验收材料。" * (4 if route == "small" else 48)
+    agenda = {"count_unit": "request", "incoming_count": 3 if route == "bad_counts" else 2, "outgoing_count": 1,
+        "items": [{"request_id": f"request-{index}", "next_action": purpose} for index in range(3)],
+        "omitted_count": int(route == "omitted"), "scan_complete": route != "incomplete",
+        "unavailable_count": int(route == "incomplete"), "read_command": command,
+        "instruction": "Read the full current request before acting."}
+    if route == "unknown":
+        agenda["future_gate"] = "Check the retained restriction."
+    decision.update(should_run=True, normal_delivery_allowed=True, effective_action="normal_run")
+    channel = decision["interaction_contract"]["agent_channel"]
+    channel.update(delivery_allowed=route != "held", must_attempt=True)
+    channel["work_context"]["sources"].append({**read, "content": {
+        "ok": True, "read_complete": False, "unfinished_delegations": agenda,
+    }})
+    if route != "missing":
+        required = {**read, "command": command + " --agent-id other"} if route == "different" else read
+        channel["required_reads"].append(required)
+
+    view = native_context_capture_view(decision, tmp_path)
+    _, values = read_selection(view["first_read_command"])
+    prefix = "/interaction_contract/agent_channel/work_context/sources/3/content"
+    assert values[prefix + "/read_complete"] is False
+    agenda_path = prefix + "/unfinished_delegations"
+    if route == "matching":
+        assert agenda_path not in values and agenda_path + "/items" not in values
+        for field in ("incoming_count", "outgoing_count", "scan_complete"):
+            assert values[f"{agenda_path}/{field}"] == agenda[field]
+        assert agenda_path + "/instruction" not in values
+        assert read in values["/interaction_contract/agent_channel/required_reads"]
+    else:
+        assert values[agenda_path] == agenda
+    detail = read_context_capture(Namespace(capture_file=view["recovery"]["path"],
+        capture_sha256=view["recovery"]["sha256"], format="json", select=[agenda_path],
+        offset=0, page_bytes=64000, max_page_output_bytes=64000))
+    assert detail["ok"] and detail["eof"]
+    assert json.loads(detail["content"])["selections"][0]["value"] == agenda
+
+
 def test_unknown_settlement_plan_is_not_compacted(tmp_path, decision):
     plan = {"schema_version": "future", "new_stop": "Stop before every new effect."}
     decision["interaction_contract"]["cli_channel"]["settlement_plan"] = plan