fix(context): keep unfinished delegation agenda in the first read

docs/reference/required-work-context.md

@@ -159,6 +159,12 @@ An enabled native **full** reference for a recognized decision can include
 `first_read_command`. It batches current admission/stop facts, the selected task,
 whole Goal/Todo narrative, source availability/revisions and read routes. Detailed
 history and optional authoring recipes remain expandable from the same capture.
+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.
 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
@@ -186,6 +192,9 @@ does not prove complete source delivery, model retention or fresh authority.
 首读也给出现有验收材料地址和当前结算义务。仅明确 `required=false` 的附件延后完整
 说明与模板，使用前须读原附件；未知计划保留全文。首读分页不重复正文中的指针，
 选读摘要仍阻止将它记作完整输入。
+已有的有界 `unfinished_delegations` 投影和 `read_complete` 标记也直接呈现；采纳和
+回传入队不等于完成。待对端结果、计数、遗漏和不可用观测保留原来源语义。
+完整请求按需读取，评估、执行或回传前仍须读其全文；首读不消费结果，也不授予权限。
 
 Checkpoint Host preflight uses the hidden `--defer-context-presentation` option
 with an explicitly enabled native JSON `should-run` capture. It collects the

loopx/control_plane/quota/turn_envelope.ts

@@ -1446,6 +1446,10 @@ function initialReadPointers(context: JsonObject): string[] {
     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");
     include(...source, "content", "current", "revision");
     include(...source, "content", "current", "instruction");
     include(...source, "content", "outcome_context");

tests/control_plane/test_context_first_read.py

@@ -119,6 +119,42 @@ def test_initial_read_keeps_settlement_obligations_and_optional_routes(tmp_path,
     assert exact["selected_pointers"] == ["/goal_boundary"]
 
 
+@pytest.mark.parametrize("unavailable", [False, True])
+def test_initial_read_preserves_unfinished_requests_and_incomplete_observation(tmp_path, decision, unavailable):
+    summary = {
+        "count_unit": "request", "incoming_count": 2, "outgoing_count": 1,
+        "items": [] if unavailable else [
+            {"request_id": "incoming", "direction": "incoming", "decision": "adopt",
+             "todo_ids": ["todo_finished"], "step": "review_request_work"},
+            {"request_id": "outgoing", "direction": "outgoing", "step": "await_peer_result"},
+            {"request_id": "return", "direction": "incoming", "step": "await_return_delivery",
+             "returns": [{"status": "queued", "consumed": False}]},
+        ],
+        "omitted_count": 3 if unavailable else 0,
+        "scan_complete": not unavailable, "unavailable_count": int(unavailable),
+        "instruction": "A done Todo or a queued return does not complete the request.",
+        "read_command": "read the full current request before acting",
+    }
+    sources = decision["interaction_contract"]["agent_channel"]["work_context"]["sources"]
+    sources.append({"source": "turn_start_capability_hook", "kind": "operator_inbox",
+        "content": {"ok": not unavailable, "read_complete": False, "unfinished_delegations": summary,
+            "items": [{"brief": {"context": "FULL-REQUEST-DETAIL-ON-DEMAND"}}]}})
+    before = json.loads(json.dumps(decision))
+    view = native_context_capture_view(decision, tmp_path)
+    page, values = read_selection(view["first_read_command"])
+    prefix = "/interaction_contract/agent_channel/work_context/sources/3/content/"
+    assert values[prefix + "unfinished_delegations"] == summary
+    assert values[prefix + "read_complete"] is False
+    assert values[prefix + "ok"] is not unavailable
+    assert "FULL-REQUEST-DETAIL-ON-DEMAND" not in page["content"]
+    # The starting view does not consume, complete or rewrite any request.
+    assert decision == before == json.loads(Path(view["recovery"]["path"]).read_bytes())
+    detail = read_context_capture(Namespace(capture_file=view["recovery"]["path"],
+        capture_sha256=view["recovery"]["sha256"], format="json", select=[prefix + "items"],
+        offset=0, page_bytes=64000, max_page_output_bytes=64000))
+    assert detail["ok"] and "FULL-REQUEST-DETAIL-ON-DEMAND" in detail["content"]
+
+
 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

tests/test_delegation_turn_context.py

@@ -1,5 +1,7 @@
 """Real receipt/store journeys: a Turn agenda does not complete or replay work."""
 import json
+import os
+import shlex
 
 import pytest
 
@@ -92,7 +94,9 @@ def test_separate_agent_has_no_access_to_another_receivers_agenda(flow):
 
 
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
-def test_actual_cli_guards_and_host_envelopes_preserve_the_agenda(tmp_path, monkeypatch, provider):
+@pytest.mark.parametrize("delivery", ["host_envelope", pytest.param("native_files",
+    marks=pytest.mark.skipif(os.name != "posix", reason="native context delivery requires POSIX"))])
+def test_actual_cli_guards_and_host_envelopes_preserve_the_agenda(tmp_path, monkeypatch, provider, delivery):
     from control_plane.canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime
     from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
     from loopx.control_plane.testing.canary_harness import run_json_cli_result, write_fixture_registry
@@ -121,17 +125,52 @@ def test_actual_cli_guards_and_host_envelopes_preserve_the_agenda(tmp_path, monk
     acknowledge(root, "research", "worker", receipt["request_id"], "adopt", "Continue the linked work", registry=registry)
     from loopx.control_plane.collaboration.links import link
     link(root, registry, "research", "worker", receipt["request_id"], ["todo_artifact"], [])
+    before = {p.relative_to(root): p.read_bytes() for p in _root(root).rglob("*.json")}
+    if delivery == "native_files":
+        from loopx.control_plane.turn_driver.codex_context import codex_context_directory, codex_context_hook
+        (tmp_path / "captures").mkdir()
+        (tmp_path / "context").mkdir(mode=0o700)
+        monkeypatch.setenv("CODEX_THREAD_ID", "agenda-session")
+        directory = codex_context_directory(root=str(tmp_path / "context"), create=True)
+        codex_context_hook(directory, {"session_id": "agenda-session", "cwd": str(tmp_path),
+            "hook_event_name": "SessionStart", "source": "startup"})
     for turn_id in ("agenda-turn-one", "agenda-turn-two"):
+        presentation = ["--turn-envelope"] if delivery == "host_envelope" else [
+            "--codex-app", "--context-delivery", "on", "--decision-output-root", str(tmp_path / "captures"),
+            "--codex-context-root", str(tmp_path / "context")]
         code, packet = run_json_cli_result("quota", "should-run", "--goal-id", "research",
             "--agent-id", "worker",  "--turn-instance-id", turn_id,
-            "--turn-envelope", registry_path=registry, runtime_root=root)
-        assert code == 0, packet
-        source = next(row for row in packet["work_context"]["sources"] if row.get("hook_id") == "manager.context_inbox")
-        assert source["content"]["unfinished_delegations"]["items"][0]["request_id"] == receipt["request_id"]
-        assert any("manager-inbox read" in row["command"] for row in packet["required_reads"])
-        assert packet["work_context"]["complete"] is True
-        assert "Continue the linked work" not in json.dumps(source)
-        assert source["content"]["unfinished_delegations"]["items"][0]["todo_ids"] == ["todo_artifact"]
+            *presentation, registry_path=registry, runtime_root=root)
+        assert code == 0, (packet.get("error_code"), packet.get("reason"), packet.get("recovery"))
+        if delivery == "native_files":
+            if turn_id == "agenda-turn-one":
+                assert packet["mode"] == "full" and "first_read_command" in packet
+                command = packet["first_read_command"]
+            else:
+                # Warm delivery retains its delta reader; recover this Turn's
+                # complete current capture, never prefetch an older baseline.
+                assert packet["mode"] == "delta"
+                command = packet["recovery"]["command"] + " --initial-read"
+            words = shlex.split(command)
+            assert words[0] == "loopx"
+            code, page = run_json_cli_result(*words[1:], registry_path=registry, runtime_root=root)
+            assert code == 0 and page["ok"] and page["eof"], page
+            values = {row["pointer"]: row["value"] for row in json.loads(page["content"])["selections"]}
+            pointer = next(key for key in values if key.endswith("/content/unfinished_delegations"))
+            content = {"unfinished_delegations": values[pointer],
+                "read_complete": values[pointer.removesuffix("unfinished_delegations") + "read_complete"]}
+            required_reads = values["/interaction_contract/agent_channel/required_reads"]
+        else:
+            source = next(row for row in packet["work_context"]["sources"] if row.get("hook_id") == "manager.context_inbox")
+            content, required_reads = source["content"], packet["required_reads"]
+            assert packet["work_context"]["complete"] is True
+        assert content["read_complete"] is False
+        row = content["unfinished_delegations"]["items"][0]
+        assert row["request_id"] == receipt["request_id"]
+        assert row["todo_ids"] == ["todo_artifact"] and row["step"] == "review_request_work"
+        assert any("manager-inbox read" in item["command"] for item in required_reads)
+        assert "Continue the linked work" not in json.dumps(content)
+    assert {p.relative_to(root): p.read_bytes() for p in _root(root).rglob("*.json")} == before
 
 
 def test_real_overflow_keeps_total_and_full_read_route(flow):