diff --git a/docs/architecture/rfcs/controller-executor-unified-contract-v0.md b/docs/architecture/rfcs/controller-executor-unified-contract-v0.md index 8ecbe9525f..60a03fa168 100644 --- a/docs/architecture/rfcs/controller-executor-unified-contract-v0.md +++ b/docs/architecture/rfcs/controller-executor-unified-contract-v0.md @@ -130,6 +130,19 @@ the existing context-delivery owner. C1 must converge the remaining composition and consumer assumptions, rather than introduce another capture service or rewrite already shared delta/native-resume logic. +A separate native-presentation slice reuses the committed full recovery as its +reader target, avoiding an unused second wire publication. Delta and unwrapped +inputs retain their own immutable files and the same reader/lifecycle owners. +On a matched 107,289-byte synthetic source, capture writes fall from 216,754 to +107,289 bytes; full reader bytes and the 3,937-byte warm delta are unchanged. +This qualifies only duplicate capture IO removal, not model-input reduction, +complete cost, accepted outcomes or C1 writer/consumer cutover. +For a small explicit output ceiling, native presentation can also omit repeated +directory/selection commands while retaining the complete recovery command: +append `--list` or `--select` to discover/read current facts without a baseline. +The original 4,000-byte cold/warm/compaction cases retain their requirements, +Stop, original identity and full recovery; this is transport qualification. + Journal persistence is an independent source-fidelity prerequisite. The proposed Python adapter reuses context delivery's sparse number-token collector; both TypeScript consumers reuse one source-number validator. Journals remain JSON diff --git a/docs/architecture/rfcs/controller-executor-unified-contract-v0.zh-CN.md b/docs/architecture/rfcs/controller-executor-unified-contract-v0.zh-CN.md index e3ecac51e4..f1d605a147 100644 --- a/docs/architecture/rfcs/controller-executor-unified-contract-v0.zh-CN.md +++ b/docs/architecture/rfcs/controller-executor-unified-contract-v0.zh-CN.md @@ -104,6 +104,16 @@ Quota capture 与 Turn journal capture 都调用 context-delivery owner。C1 应收敛剩余的组合逻辑与消费者假设，不另建 capture service， 也不重写已经共享的差分与 native-resume 逻辑。 +独立的原生展示切片复用已提交的完整恢复原件作为 reader 目标，避免再次发布未使用的 +wire 副本。Delta 与未包装输入仍保留各自的不可变文件，reader/lifecycle owner 不变。 +同一 107,289 字节合成来源的 capture 写入从 216,754 降为 107,289 字节；完整 reader +字节及 3,937 字节的 warm delta 均不变。这只验证重复 capture IO 的消除，不证明模型 +输入减少、完整成本收益、成果验收或 C1 writer/consumer 切换。 +显式小预算下，原生展示还可省略重复的目录/选读快捷命令，保留完整恢复命令； +在其后追加 `--list` 或 `--select`，无须基线即可发现或读取当前事实。 +原有 4,000 字节冷态、温态及压缩恢复用例保留完整要求、Stop、原身份和完整恢复； +这属于传输验证。 + Journal 持久化是独立的来源保真前置。本次提议由 Python adapter 复用 context delivery 的稀疏数值 token 收集器；两个 TypeScript 消费者复用同一来源数值校验器。 Journal 继续传递 JSON 对象，仅为整值浮点、带符号零和非安全整数增加元数据。 diff --git a/docs/reference/required-work-context.md b/docs/reference/required-work-context.md index d60cc87e8f..056d3c6a04 100644 --- a/docs/reference/required-work-context.md +++ b/docs/reference/required-work-context.md @@ -443,6 +443,9 @@ capture and return a small file reference, not an eager first page. The response names the delta/full `path`, digest and reader `command`; `recovery` names the current complete file. A file delta additionally carries `baseline_reference`. `index_command` lists current fields and `select_command` batches exact pointers. +Under a smaller explicit ceiling, these shortcut commands may be omitted; +append `--list` or `--select ` to `recovery.command` instead. Both +operations address the current complete capture, including during delta delivery. The reference stays below 7900 serialized bytes (or the explicit smaller output ceiling); if metadata cannot fit, recovery retains the current capture. Reader pages remain independently bounded. A complete raw tool log is not proof that @@ -732,7 +735,9 @@ SSH Goal 使用 `--runtime-profile codex_app_ssh_goal` 生成 context-config 和 不内联第一页。`path`、摘要与 `command` 指向当前差分或全量；`recovery` 指向当前 完整文件；文件差分另带 `baseline_reference`。`index_command` 展示当前字段目录， `select_command` 批量选读精确路径。引用输出不超过 7900 序列化字节或调用者显式更小 -的上限；元数据放不下时保留当前捕获恢复入口。读取页另受输出预算约束。 +的上限；较小预算下可省略这两个快捷命令，改在 `recovery.command` 后追加 +`--list` 或 `--select `，差分投递时也直接读取当前完整捕获。 +元数据仍放不下时保留当前捕获恢复入口。读取页另受输出预算约束。 原始工具日志完整，不证明宿主保留给模型的字节也完整。 原生 `captured_file` 基线来自校验后的前一完整文件，可差分复用所有未变字段，包括 diff --git a/examples/cli-control-plane-command-modularization-smoke.py b/examples/cli-control-plane-command-modularization-smoke.py index c750d30bde..31cd0cb91a 100644 --- a/examples/cli-control-plane-command-modularization-smoke.py +++ b/examples/cli-control-plane-command-modularization-smoke.py @@ -16,6 +16,7 @@ INIT_PATH = REPO_ROOT / "loopx" / "cli_commands" / "__init__.py" TODO_MODULE = REPO_ROOT / "loopx" / "cli_commands" / "todo.py" QUOTA_MODULE = REPO_ROOT / "loopx" / "cli_commands" / "quota.py" +QUOTA_PRESENTATION_MODULE = REPO_ROOT / "loopx" / "cli_commands" / "quota_presentation.py" def run_cli(*args: str) -> subprocess.CompletedProcess[str]: @@ -41,6 +42,7 @@ def assert_source_shape() -> None: init_text = INIT_PATH.read_text(encoding="utf-8") todo_text = TODO_MODULE.read_text(encoding="utf-8") quota_text = QUOTA_MODULE.read_text(encoding="utf-8") + quota_presentation_text = QUOTA_PRESENTATION_MODULE.read_text(encoding="utf-8") assert_contains(init_text, "register_todo_command", "__init__") assert_contains(init_text, "handle_todo_command", "__init__") @@ -61,7 +63,8 @@ def assert_source_shape() -> None: assert_contains(runtime_text, "handle_quota_command(", "cli_runtime.py") assert_contains(todo_text, "render_todo_markdown", "todo module") assert_contains(todo_text, "append_cli_rollout_event", "todo module") - assert_contains(quota_text, "render_quota_should_run_markdown", "quota module") + assert_contains(quota_text, "from .quota_presentation import", "quota module") + assert_contains(quota_presentation_text, "render_quota_should_run_markdown", "quota presentation") assert_contains(quota_text, "append_cli_rollout_event", "quota module") diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py index d82a5896e8..57a6880f8b 100644 --- a/loopx/cli_commands/quota_capture.py +++ b/loopx/cli_commands/quota_capture.py @@ -209,21 +209,25 @@ def native_context_capture_view(projected: dict[str, Any], directory: Path, *, max_page_output_bytes: int | None = None) -> dict[str, Any]: """Publish file references for an already-decided native capture. - Persist the complete wire and expose navigation without eagerly importing - content into model history. Reading never collects or admits a decision. + Reuse full recovery; persist delta or unwrapped wire before exposing it. + Reading never collects or admits a decision. """ from ..control_plane.turn_driver.decision_capture import write_decision_capture if max_page_output_bytes is not None and max_page_output_bytes <= 0: raise QuotaDecisionTransportError("Native context output limit must be positive.") - wire = directory / "delivery.json" - write_decision_capture(wire, projected) - reference = {"path": str(wire), "sha256": hashlib.sha256(wire.read_bytes()).hexdigest()} + reference = None mode, target = "full", None if projected.get("schema_version") == "context_delivery_v1": mode, target = projected["mode"], projected["target"] if mode == "full": reference = {key: projected["recovery"][key] for key in ("path", "sha256")} + if reference is None: + # Full delivery already reads its committed recovery. Delta and + # unwrapped inputs still need their own immutable reader target. + wire = directory / "delivery.json" + write_decision_capture(wire, projected) + reference = {"path": str(wire), "sha256": hashlib.sha256(wire.read_bytes()).hexdigest()} page_bytes, output_bytes = native_context_page_budget() if max_page_output_bytes is not None: output_bytes = min(output_bytes, max_page_output_bytes) @@ -265,7 +269,7 @@ def native_context_capture_view(projected: dict[str, Any], directory: Path, *, result["instruction"] = CONTEXT_CAPTURE_NAVIGATION_GUIDANCE if mode != "full": result["instruction"] += (" command reads the delta; baseline_reference identifies its exact saved base when file-backed. " - "index_command/select_command read current facts directly without that baseline.") + "recovery.command reads current facts directly without that baseline.") # The reference must fit even under an explicit small caller ceiling. def fits(): return len(json.dumps(result, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= min(output_bytes, 7900) @@ -275,6 +279,9 @@ def fits(): result.pop("select_command") # --select also works on the current recovery command. if not fits() and "baseline_reference" in result: result["baseline_reference"].pop("command", None) # Exact path/hash remain usable. + if not fits(): + result.pop("index_command") + result["instruction"] += " Append --list or --select to recovery.command to discover or read current fields." if not fits(): raise QuotaDecisionTransportError("Native context reference exceeds the output limit; " "read the saved original decision and receipt without rerunning admission.") diff --git a/tests/control_plane/test_progressive_native_context.py b/tests/control_plane/test_progressive_native_context.py index a605e10323..0112c05932 100644 --- a/tests/control_plane/test_progressive_native_context.py +++ b/tests/control_plane/test_progressive_native_context.py @@ -33,6 +33,29 @@ def outcome(document): return document["planning_horizon"]["core_goal"]["acceptance_contract"]["outcome_requirements"]["outcomes"][0] +def test_full_native_reference_reuses_the_committed_recovery(tmp_path, monkeypatch): + from loopx.cli_commands.quota_capture import native_context_capture_view + from loopx.control_plane.turn_driver import decision_capture + + setup(tmp_path, monkeypatch) + original, full = project(tmp_path, payload=payload()) + recovery = full["recovery"] + saved = Path(recovery["path"]).read_bytes() + + def unexpected_publication(*args, **kwargs): + pytest.fail("Full presentation must reuse its already committed recovery") + + monkeypatch.setattr(decision_capture, "write_decision_capture", unexpected_publication) + view = native_context_capture_view(full, Path(recovery["path"]).parent, + context_directory=str(tmp_path)) + assert (view["path"], view["sha256"]) == (recovery["path"], recovery["sha256"]) + assert Path(view["path"]).read_bytes() == saved + pages = read_pages(tmp_path, full) + assert json.loads("".join(page["content"] for page in pages)) == full["context"] + details = full["context"]["context_details"] + assert json.loads(Path(details["path"]).read_bytes()) == original + + def test_cold_warm_and_repeated_compaction_keep_requirements_without_replaying_material(tmp_path, monkeypatch): event = setup(tmp_path, monkeypatch) original = payload() @@ -124,7 +147,17 @@ def check_recovery(view): key: journal[key] for key in ("path", "sha256")} argv = shlex.split(view["recovery"]["command"]) if output_limit is not None: + assert len((json.dumps(view, ensure_ascii=False, separators=(",", ":")) + "\n").encode()) <= output_limit assert int(argv[argv.index("--max-page-output-bytes") + 1]) <= output_limit + # Discovery and selection remain usable from the complete current + # recovery, independently of optional shortcut command fields. + code, index = _run_cli(registry, runtime, *argv[argv.index("quota"):], "--list", cwd=tmp_path) + assert code == 0 and index["ok"], index + fields = json.loads(index["content"])["selections"][0]["fields"] + assert {"/stop", "/heartbeat_receipt"} <= {field["pointer"] for field in fields} + code, selected = _run_cli(registry, runtime, *argv[argv.index("quota"):], "--select", "/stop", cwd=tmp_path) + assert code == 0 and selected["ok"], selected + assert json.loads(selected["content"])["selections"][0]["value"] == original["stop"] pages, combined = [], b"" while True: code, recovered = _run_cli(registry, runtime, *argv[argv.index("quota"):], cwd=tmp_path) diff --git a/tests/control_plane/test_saved_context_project.py b/tests/control_plane/test_saved_context_project.py index 8fc3358b22..870ee6aa00 100644 --- a/tests/control_plane/test_saved_context_project.py +++ b/tests/control_plane/test_saved_context_project.py @@ -490,6 +490,7 @@ def test_native_saved_preflight_uses_lifecycle_without_guard_replay(tmp_path, mo from loopx.control_plane.quota.turn_envelope import prepare_context_delivery, restore_context_delivery from tests.control_plane.codex_checkpoint_fixture import local_inference_provider from tests.control_plane.test_codex_context_boundary import _app_server, _configure_provider, _native_fixture, _input_text + from tests.control_plane.codex_checkpoint_fixture import last_context_page as _last_page from tests.control_plane.test_codex_context_capture import _delivered from tests.control_plane.test_quota_settlement_cli import ( AGENT_ID, GOAL_ID, TODO_ID, _configure_selectable_alternative, @@ -575,18 +576,16 @@ def response(request, number, message): assert descriptor["ok"] and descriptor["mode"] == ("full" if name == "cold" else "delta") if benchmark: assert len(json.dumps(descriptor, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= 32000 - wire = json.loads((Path(descriptor["path"]).parent / "delivery.json").read_text()) - assert descriptor["target"] == wire["target"] - observed[name] = (wire, descriptor) data = Path(descriptor["path"]).read_bytes() - # The existing page protocol also admits the projector's first - # page. Old installed projectors still start with command. - offset = descriptor["next_offset"] if "content" in descriptor else 0 - for offset in range(len(data) if offset is None else offset, len(data), 4096): - # Synthetic captures are ASCII JSON; public reader owns every byte. - queue.append(shlex.join([*cli, "--format", "json", "quota", "context-read", - "--capture-file", descriptor["path"], "--capture-sha256", descriptor["sha256"], - "--offset", str(offset)])) + assert hashlib.sha256(data).hexdigest() == descriptor["sha256"] + published = json.loads(data) + if descriptor["mode"] == "delta": + assert descriptor["target"] == published["target"] + observed[name] = (published, descriptor) + # Follow the public reader's actual page cursor rather than + # guessing offsets from the capture size. This keeps each + # request's real page budget and avoids overlapping reads. + queue.append(descriptor["command"]) if progressive: # Exercise the actual small reference and selective public # reader in native model input; never consume either file in full. @@ -614,11 +613,17 @@ def response(request, number, message): native_context = home / "loopx-context" if benchmark else context assert "candidate" not in json.loads((native_context / "session.json").read_bytes()) else: + page = _last_page(request) + assert page is not None and page["sha256"] == descriptor["sha256"] + if not page["eof"]: + assert page["next_offset"] > page["offset"] and page["next_command"] + queue.append(page["next_command"]) + return response(request, number, message) assert _delivered(request, descriptor["sha256"]) == Path(descriptor["path"]).read_text().rstrip("\n") if benchmark and name in ("cold", "warm"): # Consume the command actually delivered to the native # worker, rather than manually supplying missing flags. - presented = json.loads(Path(observed[name][0]["recovery"]["path"]).read_text()) + presented = json.loads(Path(descriptor["recovery"]["path"]).read_text()) selection = presented["interaction_contract"]["cli_channel"]["selection_command"] command = selection["route_prefix"] + " " + selection["command_args_template"].replace("{todo_id}", TODO_ID) argv = shlex.split(command) @@ -705,12 +710,16 @@ def response(request, number, message): assert _heartbeat_receipt_count(runtime, "saved-native-turn") == receipts and _spend_run_count(runtime) == 0 cold = observed["cold"][0] baseline = None - for wire, _ in observed.values(): - projected = json.loads(Path(wire["recovery"]["path"]).read_text()) + for published, descriptor in observed.values(): + projected = json.loads(Path(descriptor["recovery"]["path"]).read_text()) current = prepare_context_delivery(projected, - scope={key: wire["target"][key] for key in - ("goal_id", "agent_id", "session_id", "context_epoch", "continuity_generation") if key in wire["target"]}, - capture_id=wire["target"]["capture_id"], surface="quota_packet") + scope={key: descriptor["target"][key] for key in + ("goal_id", "agent_id", "session_id", "context_epoch", "continuity_generation") if key in descriptor["target"]}, + capture_id=descriptor["target"]["capture_id"], surface="quota_packet") + # Full presentation publishes the original document; only deltas have + # a separate wire target. Both retain the same typed recovery check. + wire = current if descriptor["mode"] == "full" else published + assert current["target"] == descriptor["target"] restored = restore_context_delivery(wire, baseline=baseline, current=current) assert restored["context"] == projected if wire["mode"] == "delta": @@ -720,7 +729,7 @@ def response(request, number, message): baseline = {**current, "ancestry": 0 if wire.get("continuity_basis") == "captured_file" else wire["ancestry"]} if benchmark: for name, (_, raw) in zip(("cold", "warm"), sources): - presentation = json.loads(Path(observed[name][0]["recovery"]["path"]).read_text()) + presentation = json.loads(Path(observed[name][1]["recovery"]["path"]).read_text()) original = json.loads(raw) selected = presentation["interaction_contract"]["cli_channel"]["selection_command"] original_command = original["interaction_contract"]["cli_channel"]["selection_command"]["command_args_template"] @@ -749,7 +758,7 @@ def response(request, number, message): presentation.pop("context_reference_instruction", None) assert presentation == original # only presentation transport differs else: - assert cold["context"] == json.loads(sources[0][1]) and projected == json.loads(sources[1][1]) + assert cold == json.loads(sources[0][1]) and projected == json.loads(sources[1][1]) assert all(path.read_bytes() == raw for path, raw in sources)