Reduce repeated context in progressive Manager Inbox reads

docs/reference/required-work-context.md

@@ -214,12 +214,14 @@ revision and advisory authority. This captured pre-work view does not replace th
 preference owner's fresh exact-scope read before a later preference-dependent
 external action, and neither delivery nor a preference grants 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
+`read_complete` flag. For an admitted normal work Turn with a large
+agenda, a complete scan and a matching required Manager Inbox read, the initial
+view keeps the incoming and outgoing request counts, scan status and required
+command. A complete scan can still hit the summary's display limit. The initial
+view then also keeps the exact omitted count and reconciliation guidance. The
+required read and its existing cursor deliver the scoped request rows. The
+captured summary remains available at its original address. An incomplete scan, unavailable scan records or an unmatched read keeps the whole summary inline.
+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
@@ -417,10 +419,12 @@ full/delta 字节比较计入这些 reader 命令元数据；既有进程内引
 说明与模板，使用前须读原附件；未知计划保留全文。首读分页不重复正文中的指针，
 选读摘要仍阻止将它记作完整输入。
 首读包含有界 `unfinished_delegations` 投影和 `read_complete` 标记。已准入的普通
-工作 Turn 若有完整的大型委派摘要，且必读列表中有同一条管家收件箱命令，首读保留
-收发请求计数、扫描状态和必读命令，逐条记录及其说明由必做的收件箱读取交付。原记录仍可从
-同一捕获回读。采纳和回传入队是中间状态，评估、执行或回传前须读当前请求全文，
-并沿已有范围约束和消费入口办理。
+工作 Turn 若有大型委派摘要、扫描完整，且必读列表中有同一条管家收件箱命令，首读保留
+收发请求计数、扫描状态和必读命令。扫描完整但摘要达到展示上限时，首读同时保留精确的
+省略数量与核对指引。逐条请求通过必读命令及其现有分页游标交付。原摘要仍可从同一
+捕获取回。扫描不完整、扫描记录不可读或没有匹配必读命令时，首读保留完整摘要。
+采纳和回传入队是中间状态，评估、执行或回传前须读当前请求全文，并沿已有范围约束和
+消费入口办理。
 
 既有 acceptance owner 给出的完整 `outcome_context` 导航，与当前 capture 的 Goal、
 Agent、所选 Todo、成果版本及 provider revision 一致时，首读完整展开其建议成果在

examples/control_plane/control-plane-risk-characterization-smoke.py

@@ -107,7 +107,7 @@ def assert_agent_lane_delivery() -> None:
     ], packet
 
 
-def assert_scoped_operator_gate_safe_bypass() -> None:
+def assert_foreign_operator_gate_keeps_normal_admission() -> None:
     payload = status_payload(
         [
             todo_item(
@@ -136,7 +136,11 @@ def assert_scoped_operator_gate_safe_bypass() -> None:
     )
     assert quota["should_run"] is True, quota
     assert quota["effective_action"] == "normal_run", quota
-    assert quota["safe_bypass_allowed"] is True, quota
+    assert quota["safe_bypass_allowed"] is False, quota
+    scoped = quota["agent_scoped_user_gate_override"]
+    assert scoped["agent_id"] == AGENT_ID, scoped
+    assert scoped["from_state"] == "operator_gate", scoped
+    assert scoped["to_state"] == "eligible", scoped
     assert quota["interaction_contract"]["user_channel"]["action_required"] is False, quota
     assert quota["agent_lane_next_action"]["todo_id"] == "todo_refactor_slice", quota
 
@@ -467,7 +471,7 @@ def assert_reassignment_scheduler_contract() -> None:
 
 def main() -> None:
     assert_agent_lane_delivery()
-    assert_scoped_operator_gate_safe_bypass()
+    assert_foreign_operator_gate_keeps_normal_admission()
     assert_due_monitor_context_does_not_steal_advancement()
     assert_current_agent_claimed_advancement_beats_other_agent_frontier()
     assert_higher_priority_due_monitor_preempts_advancement()

loopx/capabilities/manager_context/README.md

@@ -349,6 +349,20 @@ LoopX quota/heartbeat Turn-start path; arbitrary host Turns that do not invoke
 that path are outside this integration. Frontend/Lark conversations retain their
 existing request and return views; no new screen or setting is introduced.
 
+The progressive quota hook's required command requests the Inbox Markdown view.
+When every item on a page has the same `instruction` or `next_action`, apply
+that exact value from `shared_item_guidance` to each item on that page. Differing
+values and all request bodies, current state, read receipts, warnings, and
+cursors are delivered in full. Use the same command with the returned
+`--cursor` or `--operation-cursor` as appropriate. Plain or JSON CLI reads retain the per-item fields.
+The command on an explicit context-delivery-off guard retains the JSON view.
+
+渐进 quota 的必读命令使用 Inbox Markdown 视图。当前页每条的 `instruction` 或
+`next_action` 完全相同时，从 `shared_item_guidance` 读取原值并应用到本页每条。
+不同指引、请求全文、当前状态、读回执、警告和游标均完整交付。用相同命令和
+对应的 `--cursor` 或 `--operation-cursor` 读完后续页。普通或 JSON CLI 读保留逐条字段。显式关闭上下文交付时，
+必读命令继续使用 JSON 视图。
+
 The local Inbox read transaction now lives in
 `loopx/control_plane/collaboration/inbox_read.ts`. It reuses the typed receipt,
 Goal-instance, request-revision and agenda owners inside one runtime request.

loopx/capabilities/manager_context/__init__.py

@@ -228,7 +228,7 @@ def publication_plan():
 
 
 def turn_start_hook(
-    runtime_root: Path, registry_path: Path, goal_id: str, agent_id: str
+    runtime_root: Path, registry_path: Path, goal_id: str, agent_id: str, *, model_read: bool = False
 ):
     from ...control_plane.capability_hooks import (
         TurnStartHookRegistration,
@@ -289,6 +289,7 @@ def produce():
             str(runtime_root),
             "manager-inbox",
             "read",
+            *(["--format", "markdown"] if model_read else []),
             "--goal-id",
             goal_id,
             "--agent-id",

loopx/cli_commands/manager_inbox.py

@@ -14,6 +14,27 @@
 from ..presentation.answer_instruction import collaboration_answer_instruction
 
 
+def _read_markdown(result: dict) -> str:
+    items = result.get("items")
+    shared = {}
+    if isinstance(items, list) and len(items) > 1 and all(isinstance(item, dict) for item in items):
+        for key in ("instruction", "next_action"):
+            text = items[0].get(key)
+            if isinstance(text, str) and text and all(item.get(key) == text for item in items[1:]):
+                shared[key] = text
+
+    lines = ["# Manager Inbox"]
+    if shared and "shared_item_guidance" not in result:
+        lines.append("Apply the exact fields in `shared_item_guidance` to every item on this page.")
+        result = {
+            "shared_item_guidance": shared,
+            **result,
+            "items": [{key: value for key, value in item.items() if key not in shared} for item in items],
+        }
+    body = json.dumps(result, ensure_ascii=False, separators=(",", ":"))
+    return "\n\n".join(lines) + "\n\n```json\n" + body + "\n```"
+
+
 def register_manager_inbox(subparsers, add_format):
     parser = subparsers.add_parser(
         "manager-inbox",
@@ -351,5 +372,9 @@ def handle_manager_inbox(args, registry_path, runtime_root):
         result = exc.result
     except (OSError, ValueError) as exc:
         result = {"ok": False, "error": str(exc)}
-    print(json.dumps(result, ensure_ascii=False, indent=2))
+    fmt = getattr(args, "subcommand_format", None) or getattr(args, "format", None)
+    if args.manager_inbox_action == "read" and fmt == "markdown" and result["ok"]:
+        print(_read_markdown(result))
+    else:
+        print(json.dumps(result, ensure_ascii=False, indent=2))
     return 0 if result["ok"] else 1

loopx/cli_commands/quota.py

@@ -247,7 +247,11 @@ def _dispatch_quota_turn_start_hooks(
             agent_id=args.agent_id,
             available=args.available_capabilities,
         )
-        context_dispatch = dispatch_turn_start_hooks((turn_start_hook(root, registry_path, args.goal_id, args.agent_id),))
+        model_read = (not getattr(args, "scheduler_observation", False)
+                      and native_context_activation(args)["progressive_enabled"])
+        context_dispatch = dispatch_turn_start_hooks((
+            turn_start_hook(root, registry_path, args.goal_id, args.agent_id, model_read=model_read),
+        ))
         from ..capabilities.semantic_preference.agent_preferences import extend_turn_start_dispatch as extend_preferences
         context_dispatch = extend_preferences(context_dispatch, runtime_root=root, registry_path=registry_path,
             goal_id=args.goal_id, agent_id=args.agent_id)

loopx/cli_runtime.py

@@ -143,6 +143,8 @@ def output_format(args: argparse.Namespace, *local_dests: str) -> str:
 def resolve_global_output_format(args: argparse.Namespace) -> str:
 	if getattr(args, "format", None):
 		return str(args.format)
+	if args.command == "manager-inbox":
+		return "json"
 	if args.command == "quota" and getattr(args, "quota_command", None) == "should-run":
 		return "json"
 	return "markdown"

loopx/control_plane/quota/progressive_context.ts

@@ -797,26 +797,33 @@ export function initialContextReadPointers(context: JsonObject, decisionFields:
 }
 
 /** A pending Manager Inbox read already delivers these scoped request rows. */
-function canDeferManagerAgenda(value: unknown, requiredReads: unknown[]): boolean {
+function managerAgendaOpeningFields(value: unknown, requiredReads: unknown[]): string[] | undefined {
   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
+      || agenda.scan_complete !== true || 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;
+      || !requiredReads.some(required => isDeepStrictEqual(required, read))) return undefined;
+  const {incoming_count: incoming, outgoing_count: outgoing, omitted_count: omitted, 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;
+      || typeof omitted !== "number" || !Number.isSafeInteger(omitted) || omitted < 0
+      || !Number.isSafeInteger(incoming + outgoing) || incoming + outgoing !== items.length + omitted
+      || Buffer.byteLength(JSON.stringify(items), "utf8") < 2048) return undefined;
+  const fields = ["incoming_count", "outgoing_count", "scan_complete"];
+  if (omitted === 0) return fields;
+  // A complete scan can have a bounded display. Preserve the omission count
+  // and reconciliation guidance while leaving the rows to the required read.
+  if (typeof agenda.instruction === "string" && agenda.instruction.trim()) {
+    return [...fields, "omitted_count", "instruction"];
+  }
+  return undefined;
 }
 
 function initialContextReadGroups(context: JsonObject, decisionFields: readonly string[]): {
@@ -955,8 +962,9 @@ function initialContextReadGroups(context: JsonObject, decisionFields: readonly
       include(...source, "content", field);
     include(...source, "content", "read_complete");
     const agendaPath = [...source, "content", "unfinished_delegations"];
-    if (canDeferManagerAgenda(_source, requiredInboxReads)) {
-      for (const field of ["incoming_count", "outgoing_count", "scan_complete"]) include(...agendaPath, field);
+    const agendaFields = managerAgendaOpeningFields(_source, requiredInboxReads);
+    if (agendaFields) {
+      for (const field of agendaFields) include(...agendaPath, field);
     } else include(...agendaPath);
     // The preference owner already supplies the complete current scoped view,
     // including corrections, retirement, provenance and advisory authority.

tests/control_plane/test_context_first_read.py

@@ -1017,21 +1017,26 @@ def test_initial_read_preserves_unfinished_requests_and_incomplete_observation(t
 
 
 @pytest.mark.parametrize("route", [
-    "matching", "different", "missing", "incomplete", "omitted", "unknown", "bad_counts", "small", "held",
+    "matching", "bounded", "different", "missing", "incomplete", "omitted", "unknown",
+    "bad_counts", "small", "held", "missing_instruction",
 ])
 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,
+    bounded = route in {"bounded", "missing_instruction"}
+    agenda = {"count_unit": "request", "incoming_count": 3 if bounded or 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",
+        "omitted_count": int(bounded or 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."
+    if route == "missing_instruction":
+        del agenda["instruction"]
     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)
@@ -1047,11 +1052,15 @@ def test_initial_read_leaves_large_agenda_rows_to_the_matching_required_read(tmp
     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":
+    if route in {"matching", "bounded"}:
         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
+        if bounded:
+            assert values[agenda_path + "/omitted_count"] == 1
+            assert values[agenda_path + "/instruction"] == agenda["instruction"]
+        else:
+            assert agenda_path + "/instruction" not in values
         assert read in values["/interaction_contract/agent_channel/required_reads"]
     else:
         assert values[agenda_path] == agenda

tests/control_plane/test_inbox_read_recovery.py

@@ -45,7 +45,7 @@ def lost_reply(runtime, query, **kwargs):
 
     monkeypatch.setattr(inbox, "read_inbox_transaction", lost_reply)
     route = ["--registry", str(registry), "--runtime-root", str(root), "--format", "json", "manager-inbox"]
-    assert main([*route, "read", "--goal-id", "goal-a", "--agent-id", "worker"]) == 1
+    assert main([*route, "read", "--format", "markdown", "--goal-id", "goal-a", "--agent-id", "worker"]) == 1
     error = json.loads(capsys.readouterr().out)
     assert error["ok"] is False and error["error_code"] == "runtime_response_ambiguous"
     assert error["read_complete"] is False

tests/test_delegation_turn_context.py

@@ -238,12 +238,45 @@ def run(*arguments):
         assert rows[queued["request_id"]]["step"] == "await_return_delivery"
         assert rows[queued["request_id"]]["pending_return_count"] == 1
         assert rows[peer["request_id"]]["step"] == "consume_peer_result"
-        assert any("manager-inbox read" in item["command"] for item in required_reads)
+        manager_read = next(item for item in required_reads if item["hook_id"] == "manager.context_inbox")
+        assert "manager-inbox read" in manager_read["command"]
+        assert ("--format markdown" in manager_read["command"]) == (delivery == "native_files")
         assert "Continue the linked work" not in json.dumps(content)
         assert os.environ["LOOPX_CONTEXT_DELIVERY"] == parent_context
+    if delivery == "native_files":
+        result = subprocess.run(shlex.split(manager_read["command"]), cwd=tmp_path, env=environment,
+            capture_output=True, text=True, encoding="utf-8", timeout=90)
+        assert result.returncode == 0, result.stderr
+        body = result.stdout.split("\n```json\n", 1)[1].removesuffix("\n```\n")
+        rendered = json.loads(body)
+        common = rendered.pop("shared_item_guidance", {})
+        rendered["items"] = [{**common, **row} for row in rendered["items"]]
+        # Both commands perform the same scoped read. Each reports its own
+        # observation clock; all supplied values and cursor facts must agree.
+        for view in (inbox, rendered):
+            view["unfinished_delegations"].pop("observed_at")
+        assert rendered == inbox
     assert {p.relative_to(root): p.read_bytes() for p in _root(root).rglob("*.json")} == before
 
 
+def test_inbox_markdown_keeps_different_or_missing_guidance_on_each_item():
+    from loopx.cli_commands.manager_inbox import _read_markdown
+
+    items = [
+        {"instruction": "Read the complete source.", "next_action": "Continue this request.", "message": "Original."},
+        {"instruction": "Read the complete source.", "next_action": "Wait for the original return.", "revision": 2},
+        {"instruction": "Read the complete source.", "unavailable": True},
+    ]
+    source = {"ok": True, "items": items, "has_more": True, "next_cursor": "scoped-next"}
+    body = _read_markdown(source).split("\n```json\n", 1)[1].removesuffix("\n```")
+    view = json.loads(body)
+    common = view.pop("shared_item_guidance")
+    assert common == {"instruction": "Read the complete source."}
+    assert all("instruction" not in item for item in view["items"])
+    view["items"] = [{**common, **item} for item in view["items"]]
+    assert view == source
+
+
 def test_real_overflow_keeps_total_and_full_read_route(flow):
     root, registry, _store, create = flow
     for _ in range(26):