diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py index 2350050a5e..4e77e76faf 100644 --- a/loopx/cli_commands/quota_capture.py +++ b/loopx/cli_commands/quota_capture.py @@ -390,6 +390,7 @@ def _native_context_reference(projected: dict[str, Any], reference: dict[str, An page_bytes: int, output_bytes: int, first_read_budget: tuple[int, int], page_attached: bool = False, + first_read_requested: bool = False, reader_argv: list[str] | None = None) -> dict[str, Any]: """Build the bounded native reference without publishing or registering it.""" target = projected.get("target") if projected.get("schema_version") == "context_delivery_v1" else None @@ -425,10 +426,13 @@ def _native_context_reference(projected: dict[str, Any], reference: dict[str, An # The reference must fit even under an explicit small caller ceiling. def fits() -> bool: return len(json.dumps(result, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= min(output_bytes, 7900) - if initial_read_pointers: + if first_read_requested or initial_read_pointers: overview = result.pop("overview") instruction = result["instruction"] - if any(pointer.endswith("/vision_authoring/rule") for pointer in initial_read_pointers): + if first_read_requested: + instruction = ("If the initial selection names a Vision authoring step, select the complete vision_authoring object " + "at that step before composing a Vision. " + instruction) + elif any(pointer.endswith("/vision_authoring/rule") for pointer in (initial_read_pointers or ())): instruction = ("Before composing a Vision, select the complete vision_authoring object " "at the displayed step in this capture. " + instruction) first_read = ("Read the top-level content and follow next_command if eof is false. " @@ -563,6 +567,9 @@ def native_context_capture_view(projected: dict[str, Any], directory: Path, *, field_index = None initial_read_pointers = None initial_read_reference = None + # A stateless attached page uses the ordinary verified selector below. + # Its recovery recipe does not require an extra root index to be planned. + stateless_first_read = progressive_enabled and page_attached and captured_recovery is not None def pin_initial_selection(plan: dict[str, Any]) -> None: nonlocal initial_read_pointers, initial_read_reference @@ -586,7 +593,7 @@ def pin_initial_selection(plan: dict[str, Any]) -> None: except (EffectRuntimeRemoteError, EffectRuntimeStartupError): # Optional discovery cannot invalidate the immutable file reference. pass - elif progressive_enabled: + elif progressive_enabled and not stateless_first_read: try: plan = _turn_envelope_result("context_select", context=current_document, select=[""], list_fields=True) except (EffectRuntimeRemoteError, EffectRuntimeStartupError): @@ -602,7 +609,7 @@ def view(kind: str, ref: dict[str, Any], body: bytes) -> dict[str, Any]: progressive_enabled=progressive_enabled, initial_read_reference=initial_read_reference, page_bytes=page_bytes, output_bytes=output_bytes, first_read_budget=first_read_budget, - page_attached=page_attached, reader_argv=reader_argv) + page_attached=page_attached, first_read_requested=stateless_first_read, reader_argv=reader_argv) if typed and mode == "delta": result = None @@ -690,7 +697,9 @@ def progressive_context_or_current(payload: dict[str, Any], args: argparse.Names def progressive_context_capture_view(payload: dict[str, Any], args: argparse.Namespace, *, runtime_root: Path, directory: Path | None, surface: str = "quota_packet", - reader_argv: list[str] | None = None) -> dict[str, Any]: + reader_argv: list[str] | None = None, + captured_decision_path: str | None = None, + page_attached: bool = False) -> dict[str, Any]: """Same-capture presentation for every caller, without a native journal. A plain read-only should-run remains read-only. Private transport files do @@ -702,16 +711,17 @@ def progressive_context_capture_view(payload: dict[str, Any], args: argparse.Nam if directory is None: root = Path(_private_transport_root(runtime_root)) directory = Path(tempfile.mkdtemp(prefix="context-", dir=root)) - bind_capture_selection_transport(payload, args, capture_root=directory.parent) - required, reference = capture_progressive_context(payload, directory=directory, surface=surface) + transport = _capture_selection_transport_template(payload, args, capture_root=directory.parent) + required, reference = capture_progressive_context(payload, directory=directory, surface=surface, + selection_command_args_template=transport, captured_decision_path=captured_decision_path) view = native_context_capture_view(required, directory, captured_recovery=reference, metadata={"context_delivery": native_context_activation(args), **({"capture_sha256": args.capture_sha256} if getattr(args, "capture_sha256", None) else {})}, - max_page_output_bytes=getattr(args, "max_page_output_bytes", None), reader_argv=reader_argv) + max_page_output_bytes=getattr(args, "max_page_output_bytes", None), page_attached=page_attached, reader_argv=reader_argv) readable = model_context_reference(view, reader_argv=reader_argv) # Small results need no extra navigation. The complete original and required # captures remain available; never shrink by dropping current obligations. - return readable if len(_native_output_bytes(readable)) < len(_native_output_bytes(required)) else required + return readable if page_attached or len(_native_output_bytes(readable)) < len(_native_output_bytes(required)) else required def _bounded_initial_page(navigation: dict[str, Any], plan: dict[str, Any], *, @@ -772,6 +782,12 @@ def _saved_context_read_page(args: argparse.Namespace, navigation: dict[str, Any return page +def _context_page_output_enabled(args: argparse.Namespace, activation: dict[str, Any]) -> bool: + return activation["enabled"] or ( + args.quota_command == "context-project" and getattr(args, "initial_read", False) + and activation["progressive_enabled"]) + + 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. @@ -788,11 +804,11 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str try: activation = native_context_activation(args) 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 + if saved_page and (args.format != "json" or not _context_page_output_enabled(args, activation) 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.") + raise ValueError("Use JSON context-project at offset zero with native delivery or progressive --initial-read. " + "Choose either --initial-read or --select. Use context-read for replay or continuation.") caller_output_limit = getattr(args, "max_page_output_bytes", None) output_limit = caller_output_limit if args.initial_read: @@ -832,9 +848,13 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str # Recovery is presentation of this exact saved decision, including its # original Turn. It does not need a supported host or a native session. try: - return progressive_context_capture_view(payload, args, + navigation = progressive_context_capture_view(payload, args, runtime_root=Path(getattr(args, "_context_delivery_machine_runtime", path.parent)), - directory=None, reader_argv=reader_argv) + directory=None, reader_argv=reader_argv, captured_decision_path=str(path), page_attached=saved_page) + if saved_page: + return _saved_context_read_page(args, navigation, initial_reference=None, + initial_read_plan=None, output_limit=caller_output_limit, reader_argv=reader_argv) + return navigation except (QuotaDecisionTransportError, OSError, ValueError, EffectRuntimeRemoteError, EffectRuntimeStartupError): pass # The complete original reference remains the fallback below. try: @@ -1243,7 +1263,8 @@ def handle_native_context_command(args: argparse.Namespace, *, print_payload: Ca 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 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) + print_payload({"ok": False, "reason": "--initial-read is only supported by context-read or 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) @@ -1257,7 +1278,8 @@ def handle_native_context_command(args: argparse.Namespace, *, print_payload: Ca settlement_output = (args.quota_command == "spend-slot" and args.format == "json" and bool(getattr(args, "turn_instance_id", None)) and bool(getattr(args, "agent_id", None))) budget_scope_error = ("max-page-output-bytes is only supported by JSON context-read " - "or enabled native should-run/context-project, or Turn-bound JSON spend-slot") + "or enabled native should-run/context-project, progressive context-project --initial-read, " + "or Turn-bound JSON spend-slot") if has_output_limit and args.quota_command != "context-read" and not settlement_output and ( args.quota_command not in ("should-run", "context-project") or args.format != "json" or (getattr(args, "context_delivery", None) == "off" and getattr(args, "delta_delivery", None) != "auto")): @@ -1285,7 +1307,7 @@ def handle_native_context_command(args: argparse.Namespace, *, print_payload: Ca return 1 if has_output_limit and args.quota_command != "context-read" and not settlement_output: try: - if not native_context_activation(args)["enabled"]: + if not _context_page_output_enabled(args, native_context_activation(args)): raise QuotaCommandValidationError(budget_scope_error) if args.max_page_output_bytes <= 0: raise QuotaCommandValidationError("max-page-output-bytes must be positive") diff --git a/loopx/cli_commands/quota_registration.py b/loopx/cli_commands/quota_registration.py index 02f68a6b4e..86df857ebb 100644 --- a/loopx/cli_commands/quota_registration.py +++ b/loopx/cli_commands/quota_registration.py @@ -239,7 +239,7 @@ 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 or enabled native context-project: read current controls, task text and source navigation from a recognized capture. " + help="For JSON context-read or enabled 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 or enabled native JSON context-project: repeat an exact RFC 6901 JSON pointer to read a batch from the original capture. " @@ -260,7 +260,8 @@ def register_quota_command( "Larger pages require a sufficient --max-page-output-bytes within the capture limit; " "use only with known sufficient direct and retained-history output budgets.") quota_parser.add_argument("--max-page-output-bytes", type=int, - help="For JSON context-read, enabled native should-run/context-project, or Turn-bound JSON spend-slot: optional positive output byte limit. " + help="For JSON context-read, native should-run/context-project, progressive context-project --initial-read, " + "or Turn-bound JSON spend-slot: optional positive output byte limit. " "Reader pages shrink at UTF-8 boundaries; oversized settlement diagnostics remain complete in a private immutable result with reader references. " "An insufficient metadata budget returns an error; this option does not enable native context continuity.") quota_parser.add_argument( diff --git a/loopx/control_plane/quota/turn_envelope.ts b/loopx/control_plane/quota/turn_envelope.ts index c8961823a7..363dad69d5 100644 --- a/loopx/control_plane/quota/turn_envelope.ts +++ b/loopx/control_plane/quota/turn_envelope.ts @@ -1791,7 +1791,8 @@ export function evaluateTurnEnvelope(value: unknown): JsonObject { return {sections: contextReadSectionDigests(request, (selected.selections as {path: (string | number)[]}[]).map(row => row.path))}; } - if (operation === "progressive_context_delivery") return prepareProgressiveContext(request, contextPresentationPlan(request)); + if (operation === "progressive_context_delivery") + return prepareProgressiveContext(request, nativeReplanPresentation(request) ?? contextPresentationPlan(request)); if (operation === "context_delivery") return request.native_projection === undefined ? prepareContextDelivery(request, LEGACY_QUOTA_CONTEXT_FIELDS) : prepareProgressiveContextDelivery(request, LEGACY_QUOTA_CONTEXT_FIELDS, INITIAL_CONTEXT_DECISION_FIELDS, diff --git a/loopx/control_plane/turn_driver/decision_capture.py b/loopx/control_plane/turn_driver/decision_capture.py index 3c5800f3c4..75bac0a572 100644 --- a/loopx/control_plane/turn_driver/decision_capture.py +++ b/loopx/control_plane/turn_driver/decision_capture.py @@ -425,17 +425,23 @@ def capture_context_projection(payload: dict[str, Any], *, directory: Path, def capture_progressive_context(payload: dict[str, Any], *, directory: Path, - surface: str = "quota_packet") -> tuple[dict[str, Any], dict[str, Any]]: + surface: str = "quota_packet", + selection_command_args_template: str | None = None, + captured_decision_path: str | None = None) -> tuple[dict[str, Any], dict[str, Any]]: """Publish the complete current source before optional stateless presentation. The typed owner applies the shrink guard. This neither creates a native journal nor requires/adopts an execution Turn. All recovery is same-capture. + A caller may supply the exact decision it just verified for staged reads. """ from ..quota.turn_envelope import prepare_progressive_context source = capture_context_projection(payload, directory=directory) transaction = prepare_progressive_context(payload, surface=surface, source_reference={ **{key: source[key] for key in ("path", "sha256")}, + **({"selection_command_args_template": selection_command_args_template} + if selection_command_args_template is not None else {}), + **({"captured_decision_path": captured_decision_path} if captured_decision_path is not None else {}), "source_path": source["path"], "command_template": source["command"] + " --select ''", }) diff --git a/tests/control_plane/test_replan_progress_native_cli.py b/tests/control_plane/test_replan_progress_native_cli.py index d7d580f187..b861fdd23d 100644 --- a/tests/control_plane/test_replan_progress_native_cli.py +++ b/tests/control_plane/test_replan_progress_native_cli.py @@ -4,6 +4,7 @@ import hashlib import json import os +import shlex import subprocess import sys from pathlib import Path @@ -167,6 +168,81 @@ def test_progress_cli_reads_policy_with_context_delivery( restart_effect_runtime(wait_for_exit=True) +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_stateless_progressive_replan_uses_the_saved_decision(tmp_path, monkeypatch, provider): + isolate_sqlite_runtime(tmp_path, monkeypatch) + fixture = make_progress_cli_fixture(tmp_path, provider, "advanced") + capture = tmp_path / "capture" + scope = ("--codex-app", "--goal-id", GOAL, "--agent-id", AGENT, "--turn-instance-id", TURN) + delivery = ("--context-delivery", "on", "--delta-delivery", "off") + + def read_stage(command, status=0): + result = subprocess.run(command, capture_output=True, text=True, timeout=30) + assert result.returncode == status, result.stdout + result.stderr + return json.loads(result.stdout) + + def read_pages(page, limit=64_000): + body = bytearray() + while True: + assert page["ok"] and page["offset"] == len(body) + assert len(json.dumps(page, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= limit + body.extend(page["content"].encode()) + if page["eof"]: + assert len(body) == page["total_bytes"] + return bytes(body) + assert page["next_offset"] == len(body) + page = read_stage(shlex.split(page["next_command"])) + + def review_for(body): + fields = {row["pointer"]: row["value"] for row in json.loads(body)["selections"]} + assert fields["/heartbeat_receipt"] == source["heartbeat_receipt"] + assert "/replan_action_packet/planning_guidance" not in fields + return shlex.split(fields["/replan_action_packet/review"]) + + try: + initial = call_progress_cli(fixture, "quota", "should-run", *scope, "--todo-id", TODO, + "--decision-output-dir", str(capture), *delivery) + original = capture / "decision.json" + raw = original.read_bytes() + source = json.loads(raw) + receipt = source["heartbeat_receipt"] + authority = _authority(fixture) + review_command = review_for(read_pages(read_stage(shlex.split(initial["first_read_command"])))) + assert review_command[review_command.index("--decision-file") + 1] == str(original) + review = read_stage(review_command) + assert review["planning_guidance"] == source["replan_action_packet"]["planning_guidance"] + writer = read_stage(shlex.split(review["next"]["writeback"])) + assert writer["writeback"]["settlement_plan"]["identity"] == receipt["settlement_identity"] + assert writer["required_outcome"] == source["replan_action_packet"]["required_outcome"] + + recovered_path = original.with_name("recovered.json") + original.rename(recovered_path) + saved = ("quota", "context-project", *scope, "--capture-file", str(recovered_path), "--initial-read") + digest = hashlib.sha256(raw).hexdigest() + first = call_progress_cli(fixture, *saved, *delivery, "--capture-sha256", digest, + "--max-page-output-bytes", "8192") + assert not first["eof"] and list(first).index("content") < list(first).index("navigation") + nav = first["navigation"] + assert nav["mode"] == "full" and "top-level content" in nav["instruction"] + body = read_pages(first, 8192) + assert read_pages(read_stage(shlex.split(nav["first_read_command"])), 8192) == body + recovered_review = review_for(body) + assert recovered_review[recovered_review.index("--decision-file") + 1] == str(recovered_path) + assert read_stage(recovered_review)["source_hash"] == review["source_hash"] + recovered_review[recovered_review.index("--source-hash") + 1] = "sha256:" + "0" * 64 + refused = read_stage(recovered_review, status=1) + assert refused["ok"] is False and "matching source hash" in refused["reason"] + wrong_capture = call_progress_cli(fixture, *saved, *delivery, "--capture-sha256", "0" * 64, expected_code=1) + assert not wrong_capture["ok"] and "content" not in wrong_capture + disabled = call_progress_cli(fixture, *saved, "--context-delivery", "off", "--delta-delivery", "off", + "--capture-sha256", digest, expected_code=1) + assert not disabled["ok"] and "content" not in disabled + assert recovered_path.read_bytes() == raw + assert _authority(fixture) == authority + finally: + restart_effect_runtime(wait_for_exit=True) + + @pytest.mark.parametrize("provider", ["file", "sqlite"]) @pytest.mark.parametrize("context_args,navigation", [ ((), False),