diff --git a/benchmark/runtime/context_delivery.py b/benchmark/runtime/context_delivery.py index 277ec2631e..90b08240c8 100644 --- a/benchmark/runtime/context_delivery.py +++ b/benchmark/runtime/context_delivery.py @@ -26,13 +26,15 @@ def preflight_context_instruction(env: dict[str, str], path: Path, digest: str, source = ["--capture-file", str(path), "--capture-sha256", digest] project = [*prefix, "context-project", *source, "--goal-id", env["LOOPX_GOAL_ID"], "--agent-id", env["LOOPX_AGENT_ID"], - "--turn-instance-id", turn_id, "--context-delivery", "on", "--runtime-profile", "generic_cli", "--codex-context-dir", env["LOOPX_CODEX_CONTEXT_DIR"]] + "--turn-instance-id", turn_id, "--initial-read", "--context-delivery", "on", "--runtime-profile", "generic_cli", "--codex-context-dir", env["LOOPX_CODEX_CONTEXT_DIR"]] recovery_index = [*prefix, "context-read", *source, "--list"] return ("Present that same immutable packet with:\n```sh\n" + shlex.join(project) + "\n```\n" + ("" if guide_in_host else CONTEXT_CAPTURE_NAVIGATION_GUIDANCE) + "If the projection command itself fails, recover necessary facts from the same original capture. " "When pointers are unknown, discover them with:\n```sh\n" + shlex.join(recovery_index) + "\n```\nUse the same capture's generic reader and selectors for the facts the current decision needs. " + "For known fields, replace --initial-read with repeated --select pointers in the first command. " + "Continue the returned page or use navigation.recovery.command to reread; do not project again. " "Discovery does not require draining the packet. " "Neither presentation nor recovery renews this decision's authority or freshness. ") diff --git a/docs/reference/required-work-context.md b/docs/reference/required-work-context.md index 9becb300cc..76415c99bb 100644 --- a/docs/reference/required-work-context.md +++ b/docs/reference/required-work-context.md @@ -252,6 +252,19 @@ evidence and the complete operation/settlement contract before depending on them The original SHA-bound full recovery remains available; a selected page's EOF does not prove complete source delivery, model retention or fresh authority. +For a saved benchmark preflight, the ordinary wake prompt carries the command for +the first page: native JSON `quota context-project --initial-read` publishes the +saved decision and reads it in one call. Replace `--initial-read` with repeated +`--select` for known fields. The output is a normal bounded reader page with its +navigation attached. Follow `next_command` for remaining bytes or use +`navigation.recovery.command` for other fields. If no output arrived, read the +original digest-bound capture identified in the wake prompt. Neither retry +requires another projection, guard, notification or model acknowledgement. + +中文：对于已保存的 benchmark 准入结果，唤醒正文直接给出首读命令。一次调用即可 +取得首屏；已知字段可改用多个 `--select`。续页和补读都沿原文件读取。如果没拿到 +任何输出，就用唤醒正文中的原文件及其摘要恢复，无须重做投影或准入。 + When the existing acceptance owner supplies complete `outcome_context` navigation matching this capture's Goal, Agent, selected Todo, outcome revision and provider revision, the initial read expands its suggested outcomes in full diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py index b67a92a059..3bdffc9463 100644 --- a/loopx/cli_commands/quota_capture.py +++ b/loopx/cli_commands/quota_capture.py @@ -661,6 +661,22 @@ def progressive_context_capture_view(payload: dict[str, Any], args: argparse.Nam return readable if len(_native_output_bytes(readable)) < len(_native_output_bytes(required)) else required +def _saved_context_read_page(args: argparse.Namespace, navigation: dict[str, Any], *, + initial_reference: str | None, reader_argv: list[str]) -> dict[str, Any]: + """Use the ordinary verified reader; the first page and navigation share its output budget.""" + recovery = navigation["recovery"] + pinned = initial_reference if args.initial_read else None + reference = pinned or recovery.get("reference_file") + read_args = argparse.Namespace(**{**vars(args), "quota_command": "context-read", + "reference_file": reference, "capture_file": None if reference else recovery["path"], + "capture_sha256": None if reference else recovery["sha256"], + "codex_context_dir": None, "codex_context_root": None, "initial_read": bool(args.initial_read and not pinned)}) + page = read_context_capture(read_args, presentation={"navigation": navigation}, reader_argv=reader_argv) + if not page["ok"] and "recovery_command" not in page: + page["recovery_command"] = recovery["command"] + return page + + def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str | None = None) -> dict[str, Any]: """Present an immutable already-decided quota capture, without authority IO. @@ -676,7 +692,14 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str reader_argv = current_loopx_cli_argv() try: activation = native_context_activation(args) - path, raw, payload = _read_capture(args.capture_file, args.capture_sha256, private=hook_session is not None) + saved_page = bool(getattr(args, "initial_read", False) or getattr(args, "select", None)) + if saved_page and (args.format != "json" or not activation["enabled"] or args.offset != 0 + or args.reference_file or args.selection_sha256 or args.page_bytes is not None + or (args.initial_read and args.select)): + raise ValueError("Use an enabled native JSON context-project with either --initial-read or --select " + "at offset zero. Use context-read for replay or continuation.") + path, raw, payload = _read_capture(args.capture_file, args.capture_sha256, + private=hook_session is not None) identity = payload.get("agent_identity") receipt = payload.get("heartbeat_receipt") if (payload.get("ok") is not True or not isinstance(identity, dict) or not isinstance(receipt, dict) @@ -699,6 +722,7 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str target = None native_prepared = False native_view = None + initial_reference = None if not activation["enabled"] and activation["progressive_enabled"]: # Recovery is presentation of this exact saved decision, including its # original Turn. It does not need a supported host or a native session. @@ -765,14 +789,15 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str max_page_output_bytes=getattr(args, "max_page_output_bytes", None), reader_argv=reader_argv) reference = {key: native_view[key] for key in ("path", "sha256", "command")} mode, target = native_view["mode"], native_view.get("target") + initial_reference = native_view.get("first_read_reference") native_view = model_context_reference(native_view, reader_argv=reader_argv) except (QuotaDecisionTransportError, OSError, ValueError, TypeError, RecursionError) as exc: if native_view is not None and not isinstance(exc, QuotaDecisionTransportError): # The native view and journal are already valid. Publishing the # shorter reader metadata is optional; retain their exact legacy # recovery rather than losing the original capture or admitting again. - return native_view - if isinstance(exc, QuotaDecisionTransportError) or (activation["enabled"] and not native_prepared): + pass + elif isinstance(exc, QuotaDecisionTransportError) or (activation["enabled"] and not native_prepared): if not isinstance(exc, QuotaDecisionTransportError): exc = QuotaDecisionTransportError( "Current context recovery could not be prepared. Read the saved " @@ -805,9 +830,10 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str **({"context_delivery": activation} if activation["enabled"] else {}), "instruction": "Presentation of the original decision only; authority and freshness are unchanged. " + (CONTEXT_CAPTURE_NAVIGATION_GUIDANCE if activation["enabled"] else CONTEXT_CAPTURE_READ_GUIDANCE)} - if native_view is not None: - return native_view - return result + navigation = native_view or result + if saved_page: + return _saved_context_read_page(args, navigation, initial_reference=initial_reference, reader_argv=reader_argv) + return navigation def prepare_quota_capture(args: argparse.Namespace, directory: Path | None) -> Path | None: @@ -1089,8 +1115,8 @@ def handoff_decision(directory: Path | None, socket_path: str | None, def handle_native_context_command(args: argparse.Namespace, *, print_payload: Callable[..., None], registry_path: Path | None = None, runtime_root_arg: str | None = None) -> int | None: """Dispatch native presentation/lifecycle IO before any quota authority read.""" - if getattr(args, "initial_read", False) and args.quota_command != "context-read": - print_payload({"ok": False, "reason": "--initial-read is only supported by context-read"}, args.format, str) + if getattr(args, "initial_read", False) and args.quota_command not in ("context-read", "context-project"): + print_payload({"ok": False, "reason": "--initial-read is only supported by context-read or enabled native JSON context-project"}, args.format, str) return 1 list_depth = getattr(args, "list_depth", None) if list_depth is not None and (args.quota_command != "context-read" or not getattr(args, "list_fields", False) diff --git a/loopx/cli_commands/quota_registration.py b/loopx/cli_commands/quota_registration.py index ccdb29de0b..8cced17797 100644 --- a/loopx/cli_commands/quota_registration.py +++ b/loopx/cli_commands/quota_registration.py @@ -239,10 +239,10 @@ def register_quota_command( "context-config renders native configuration for the exact Goal/Agent; no host configuration, trust, guard or spend are changed.") quota_parser.add_argument("--capture-sha256", help="For context-read/context-project: exact SHA-256 supplied with the capture reference.") quota_parser.add_argument("--initial-read", action="store_true", - help="For JSON context-read: read current controls, task text and source navigation from a recognized capture. " + help="For JSON context-read or enabled native context-project: read current controls, task text and source navigation from a recognized capture. " "Detailed criteria and operation contracts remain required before dependent work. Start at offset zero; follow next_command for continuation.") quota_parser.add_argument("--select", action="append", - help="For context-read: repeat an exact RFC 6901 JSON pointer to read a batch from the original capture. " + help="For context-read or enabled native JSON context-project: repeat an exact RFC 6901 JSON pointer to read a batch from the original capture. " "Selected detail pages never acknowledge full-context retention; omit for unchanged full capture reads.") quota_parser.add_argument("--list", dest="list_fields", action="store_true", help="For context-read: list immediate fields at the selected pointers (root by default), " diff --git a/tests/control_plane/test_saved_context_project.py b/tests/control_plane/test_saved_context_project.py index ba3f003a9e..d996496542 100644 --- a/tests/control_plane/test_saved_context_project.py +++ b/tests/control_plane/test_saved_context_project.py @@ -407,6 +407,7 @@ def test_preflight_instruction_pins_both_projection_and_full_fallback(tmp_path): assert argv[argv.index("--capture-sha256") + 1] == "a" * 64 assert "should-run" not in argv and "spend-slot" not in argv assert commands[0][commands[0].index("--turn-instance-id") + 1] == "original-turn" + assert "--initial-read" in commands[0] assert "--decision-output-root" not in commands[0] @@ -490,9 +491,9 @@ def consume(wire): if enabled: command.extend(("--max-page-output-bytes", "32000")) rc, cold = _run_cli(registry, runtime, *command[command.index("quota"):], cwd=project) - assert rc == 0 and cold["mode"] == "full", cold + assert rc == 0 and cold["navigation"]["mode"] == "full" and cold["offset"] == 0, cold assert len(json.dumps(cold, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= 32000 - presented = consume(cold) + presented = consume(cold["navigation"]) if compacted: codex_context_hook(str(native), {**event, "hook_event_name": "PreCompact"}) else: @@ -512,7 +513,7 @@ def consume(wire): assert len(json.dumps(bound, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= 32000 delivered = consume(bound) if not compacted: - assert delivered["base"]["capture_id"] == machine_reference_oracle(cold)["target"]["capture_id"] + assert delivered["base"]["capture_id"] == machine_reference_oracle(cold["navigation"])["target"]["capture_id"] else: assert bound["mode"] == "should-run" and not list(native.iterdir()) assert _heartbeat_receipt_count(runtime, turn) == 2 and _spend_run_count(runtime) == 0 @@ -670,6 +671,10 @@ def response(request, number, message): sources.append((source, source.read_bytes())) receipts = _heartbeat_receipt_count(runtime, actual_turn) assert hashlib.sha256(source.read_bytes()).hexdigest() == argv[argv.index("--capture-sha256") + 1] + assert "--initial-read" in argv + if progressive: + command = shlex.join([word for word in argv if word != "--initial-read"] + [ + "--select", "/goal_id", "--select", "/heartbeat_receipt"]) command += " --max-page-output-bytes 32000" else: source, raw = sources[0 if phase == "cold" else 1] @@ -686,6 +691,15 @@ def response(request, number, message): if isinstance(output, list): output = [part["text"] for part in output if part.get("type") == "input_text"][-1] navigation = json.loads(output.split("\nOutput:\n", 1)[-1]) + if benchmark and name in ("cold", "warm"): + page = navigation + assert page["ok"] and len(json.dumps(page, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= 32000 + assert page["offset"] == 0 and page["content"] + navigation = page["navigation"] + if progressive: + assert page["eof"] + values = {row["pointer"]: row["value"] for row in json.loads(page["content"])["selections"]} + assert values["/goal_id"] == GOAL_ID and values["/heartbeat_receipt"]["turn_instance_id"] == actual_turn descriptor = machine_reference_oracle(navigation) # A selectively read large capture cannot become a retained # full baseline; its next projection must remain full. @@ -739,6 +753,8 @@ def response(request, number, message): argv[index] = str(damaged) queue.insert(0, shlex.join(argv)) rejected_read = True + if benchmark and progressive and name in ("cold", "warm"): + queue.clear() # The first tool result already contains the requested facts. phase = name + "_read" if not queue: return response(request, number, message) @@ -1065,3 +1081,55 @@ def present(turn): codex_context_hook(str(journal), {**event, "source": "compact"}) assert present("warm")["mode"] == "full" assert not registry.exists() and not runtime.exists() + + +def test_saved_page_continues_and_replays_after_compaction_without_republication(tmp_path, monkeypatch): + from concurrent.futures import ThreadPoolExecutor + from loopx.control_plane.turn_driver.codex_context import codex_context_hook + + runtime, registry, context = tmp_path / "runtime", tmp_path / "registry.json", tmp_path / "native" + context.mkdir(mode=0o700) + monkeypatch.setenv("CODEX_THREAD_ID", "session") + monkeypatch.chdir(tmp_path) + event = {"source": "startup", "session_id": "session", "cwd": str(tmp_path)} + codex_context_hook(str(context), {**event, "hook_event_name": "SessionStart"}) + value = "scope and stops: 保留原要求。" * 1200 + original = tmp_path / "original.json" + original.write_text(json.dumps({"ok": True, "goal_id": "g", "agent_identity": {"agent_id": "a"}, + "heartbeat_receipt": {"turn_instance_id": "t"}, "instructions": value}, ensure_ascii=False)) + digest = hashlib.sha256(original.read_bytes()).hexdigest() + project = ["quota", "context-project", "--capture-file", str(original), "--capture-sha256", digest, + "--goal-id", "g", "--agent-id", "a", "--turn-instance-id", "t", "--codex-app", + "--context-delivery", "on", "--codex-context-dir", str(context), "--max-page-output-bytes", "6400"] + def cli(*args): + return _run_cli(registry, runtime, *args, cwd=tmp_path) + initial_files = set(context.rglob("*")) + for invalid in (["--initial-read", "--select", "/instructions"], ["--offset", "1", "--select", "/instructions"], + ["--context-delivery", "off", "--initial-read"], ["--goal-id", "foreign", "--initial-read"]): + code, error = cli(*project, *invalid) + assert code and not error["ok"] and "content" not in error + assert set(context.rglob("*")) == initial_files + code, first = cli(*project, "--select", "/instructions") + assert code == 0 and first["offset"] == 0 and not first["eof"] + assert first["navigation"]["mode"] == "full" + codex_context_hook(str(context), {**event, "hook_event_name": "PreCompact"}) + journal = (context / "session.json").read_bytes() + files = set(context.rglob("*")) + def read_reference(): + ref = first["reference_file"] + result = bytearray() + args = ["quota", "context-read", "--reference-file", ref, "--max-page-output-bytes", "6400"] + while True: + code, page = cli(*args) + assert code == 0 and page["offset"] == len(result) + result.extend(page["content"].encode()) + if page["eof"]: + assert len(result) == page["total_bytes"] + return json.loads(result)["selections"][0]["value"] + words = shlex.split(page["next_command"]) + args = words[words.index("quota"):] + with ThreadPoolExecutor(2) as pool: + assert list(pool.map(lambda _: read_reference(), range(2))) == [value, value] + assert (context / "session.json").read_bytes() == journal and set(context.rglob("*")) == files + assert hashlib.sha256(original.read_bytes()).hexdigest() == digest + assert not registry.exists() and not runtime.exists()