diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py index 4bf1e7449a..bef9bd97fe 100644 --- a/loopx/cli_commands/quota_capture.py +++ b/loopx/cli_commands/quota_capture.py @@ -805,7 +805,8 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str 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 _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.reference_file or args.selection_sha256 + or (args.page_bytes is not None and not activation["enabled"]) or (args.initial_read and args.select)): 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.") diff --git a/loopx/cli_commands/quota_presentation.py b/loopx/cli_commands/quota_presentation.py index c2248602a7..421bde7e31 100644 --- a/loopx/cli_commands/quota_presentation.py +++ b/loopx/cli_commands/quota_presentation.py @@ -5,7 +5,7 @@ from collections.abc import Callable, Mapping from ..control_plane.effect_runtime import EffectRuntimeRejected -from ..control_plane.quota.turn_envelope import build_turn_envelope +from ..control_plane.quota.turn_envelope import build_turn_envelope, project_captured_replan_cli_view def _quota_renderer( @@ -84,8 +84,7 @@ def _project_quota_cli_payload( return compact_quota_plan_cli_payload(payload, detail_sections=detail_sections) if args.quota_command == "should-run": if captured_decision_path and "vision" not in detail_sections: - from ..control_plane.quota.turn_envelope import _turn_envelope_result - payload = _turn_envelope_result("replan_cli_view", payload=payload) + payload = project_captured_replan_cli_view(payload) return compact_quota_should_run_cli_payload( payload, include_todo_summary_detail="agent-todos" in detail_sections, diff --git a/loopx/control_plane/goals/vision_checkpoint.ts b/loopx/control_plane/goals/vision_checkpoint.ts index 4f15ede891..3968dfc40a 100644 --- a/loopx/control_plane/goals/vision_checkpoint.ts +++ b/loopx/control_plane/goals/vision_checkpoint.ts @@ -104,10 +104,10 @@ export function visionAuthoringContract(autonomousReplan = false): JsonObject { "Replace todo_ with a verified Todo ID after normal authoring/readback. Entries record plans, not Todo mutations, claims or execution authority. " + "For ordinary CLI writeback, pass the packet with --agent-vision-json . " + (autonomousReplan - ? "JSON replaces the complete packet; the sparse minimal_example is not an update to an existing Vision. " + - "Start from the current Agent's complete packet and full vision_patch, then replace only deliberately reviewed fields and the fresh path_delta. " + - "Preserve every other field and the current state; path_delta.retained text does not merge omitted fields. " + - "Reuse the known current packet, or read it from canonical status/history if unavailable; never reconstruct missing intent from summaries. " + ? "JSON replaces the complete packet. For an existing Vision, the sparse minimal_example shows syntax. " + + "Use the current Agent's complete packet and full vision_patch. Read canonical status/history if unavailable. Never reconstruct intent from summaries. " + + "Edit only deliberately reviewed fields and supply a fresh path_delta. Keep the current state and all other fields. " + + "path_delta.retained text does not merge omitted fields. " : "") + "checkpoint-context is only for recovery after the original committed Turn writeback; " + "follow its returned same-Turn recovery action, not a fresh-turn preflight.", diff --git a/loopx/control_plane/quota/turn_envelope.py b/loopx/control_plane/quota/turn_envelope.py index 421ac1872f..581ad845c1 100644 --- a/loopx/control_plane/quota/turn_envelope.py +++ b/loopx/control_plane/quota/turn_envelope.py @@ -181,6 +181,17 @@ def bind_turn_decision_capture( ) +def project_captured_replan_cli_view(payload: dict[str, Any]) -> dict[str, Any]: + """Stage replan reads without round-tripping unchanged source values.""" + if not isinstance(payload.get("replan_read_path"), str): + return payload + from .cli_projection import _apply_context_projection_changes + + result = _turn_envelope_result("replan_cli_view", payload=payload, source_edits=True, + source_number_tokens=source_number_tokens(payload)) + return _apply_context_projection_changes(payload, result) + + def prepare_context_delivery( context: Mapping[str, Any], *, scope: Mapping[str, str], capture_id: str, baseline: Mapping[str, Any] | None = None, diff --git a/loopx/control_plane/quota/turn_envelope.ts b/loopx/control_plane/quota/turn_envelope.ts index 363dad69d5..372ec4c466 100644 --- a/loopx/control_plane/quota/turn_envelope.ts +++ b/loopx/control_plane/quota/turn_envelope.ts @@ -1737,7 +1737,12 @@ export function evaluateTurnEnvelope(value: unknown): JsonObject { const operation = request.operation; if (operation === "replan_view") return replanView(request); if (operation === "context_select") return capturedDecisionSelection(request); - if (operation === "replan_cli_view") return replanCliView(requireJsonObject(request.payload, "replan CLI payload")); + if (operation === "replan_cli_view") { + const source = requireJsonObject(request.payload, "replan CLI payload"); + if (request.source_edits !== true) return replanCliView(source); + return contextProjectionChanges(source, + replanCliView(source, contextSourceIdentity(source, request.source_number_tokens))); + } if (operation === "selected_work_todo") { const source = requireJsonObject(request.payload, "Turn work selection source"); const historical = scalarString(request.historical_primary_text, "historical primary Todo"); diff --git a/loopx/control_plane/turn_driver/decision_capture.py b/loopx/control_plane/turn_driver/decision_capture.py index 75bac0a572..08f18d2b1c 100644 --- a/loopx/control_plane/turn_driver/decision_capture.py +++ b/loopx/control_plane/turn_driver/decision_capture.py @@ -656,7 +656,8 @@ def context_selection_view(payload: dict[str, Any], *, select: list[str] | None, if plan.get("rejected_selection"): return plan, None, None presentation = initial_read if present_goal_narrative is None else present_goal_narrative - def render(plan: dict[str, Any], present: bool): + + def render(plan: dict[str, Any], present: bool) -> tuple[bytes, str, list[str]]: raw, displayed = _context_selection_bytes(payload, plan["selections"], list_fields=list_fields, list_depth=list_depth, present_goal_narrative=present) return raw, hashlib.sha256(raw).hexdigest(), displayed diff --git a/tests/control_plane/test_replan_staged_read.py b/tests/control_plane/test_replan_staged_read.py index 05df95313d..e56dc2a11e 100644 --- a/tests/control_plane/test_replan_staged_read.py +++ b/tests/control_plane/test_replan_staged_read.py @@ -231,19 +231,32 @@ def observe(command, **kwargs): try: body = checkpoint.capture_before_replan(env, receipt, "original heartbeat instructions") assert len(commands) == 1 and "--turn-envelope" not in commands[0] - assert "--decision-output-root" in commands[0] - source_path, view_path = wake / "checkpoint-start-guard.json", wake / "checkpoint-start-view.json" - if context_delivery: - argv = shlex.split(body.split("```sh\n", 1)[1].split("\n```", 1)[0]) - view_path = Path(argv[argv.index("--capture-file") + 1]) - assert view_path.parent != wake and view_path.is_relative_to(wake) - source, view = json.loads(source_path.read_bytes()), json.loads(view_path.read_bytes()) - assert str(view_path) in body and str(source_path) not in body + source_path = wake / "checkpoint-start-guard.json" + raw = source_path.read_bytes() + source = json.loads(raw) assert body.endswith("original heartbeat instructions") if context_delivery: - assert "--defer-context-presentation" in commands[0] - assert "--context-delivery" in commands[0] + from loopx.control_plane.turn_driver.codex_context import codex_context_hook + + assert "--prepared-capture-reference" in commands[0] and "--defer-context-presentation" in commands[0] + assert "--decision-output-root" not in commands[0] + captured_dir = Path(commands[0][commands[0].index("--decision-output-dir") + 1]) + argv = shlex.split(body.split("```sh\n", 1)[1].split("\n```", 1)[0]) + original = Path(argv[argv.index("--capture-file") + 1]) + assert original == captured_dir / "decision.json" and original.is_relative_to(wake) + assert original.read_bytes() == raw and str(source_path) not in body assert not list(native.iterdir()), "Host preflight must not alter native continuity" + session_id = "checkpoint-preflight-session" + codex_context_hook(str(native), {"hook_event_name": "SessionStart", "source": "startup", + "session_id": session_id, "cwd": str(fixture["project"])}) + env["CODEX_THREAD_ID"] = session_id + projected = run(argv, cwd=fixture["project"], env=env, capture_output=True, text=True, timeout=60) + assert projected.returncode == 0, projected.stdout + presentation = reference_metadata_oracle(json.loads(projected.stdout)) + assert presentation["ok"] and presentation["mode"] == "full" + assert presentation["target"]["capture_id"] == argv[argv.index("--capture-sha256") + 1] + view = read_cli_policy_payload(projected.stdout, returncode=projected.returncode) + assert original.read_bytes() == raw and source_path.read_bytes() == raw assert not any(read.get("source") == "goal_state" for read in view["interaction_contract"]["agent_channel"]["required_reads"]) delivered = [row["content"] for row in view["interaction_contract"]["agent_channel"] @@ -256,18 +269,14 @@ def observe(command, **kwargs): else: assert len(delivered) == 1 and delivered[0]["full_source"] is True assert delivered[0]["text"] == fixture["state"].read_text() - # The optional full/delta transport must present the staged CLI - # view too. Its full recovery must not restore old inline hints. - command = body.split("```sh\n", 1)[1].split("\n```", 1)[0] - projected = run(shlex.split(command), cwd=fixture["project"], env=env, - capture_output=True, text=True, timeout=60) - assert projected.returncode == 0, projected.stdout - presentation = reference_metadata_oracle(json.loads(projected.stdout)) - assert presentation["ok"] and presentation["mode"] == "full" - assert reference_metadata_oracle(presentation)["recovery"]["path"] == str(view_path) - assert read_cli_policy_payload(projected.stdout, returncode=projected.returncode) == view + else: + assert "--decision-output-root" in commands[0] + view_path = wake / "checkpoint-start-view.json" + assert str(view_path) in body and str(source_path) not in body + view = json.loads(view_path.read_bytes()) assert source["heartbeat_receipt"]["turn_instance_id"] == TURN assert source["heartbeat_receipt"] == view["heartbeat_receipt"] + assert json.dumps(source["quota"], sort_keys=True) == json.dumps(view["quota"], sort_keys=True) assert receipt["checkpoint"]["reason"] == "synthetic_no_native_capture" selection = view["interaction_contract"]["cli_channel"].get("selection_command") if selection: @@ -627,10 +636,13 @@ def test_large_acceptance_is_not_repeated_in_cli_or_envelope_work_context(tmp_pa def test_staged_entry_replays_goal_and_changed_horizon_with_existing_delta_owner(tmp_path): + from loopx.control_plane.quota.turn_envelope import project_captured_replan_cli_view from replan_packet_fixture import populated_replan_source source = populated_replan_source(tmp_path, acceptance=True) - entry = _turn_envelope_result("replan_cli_view", payload=source) + source["measurements"] = [1.0, -0.0, 9007199254740993] + entry = project_captured_replan_cli_view(source) + assert json.dumps(entry["measurements"]) == json.dumps(source["measurements"]) assert entry["action_portfolio"] == source["action_portfolio"] scope = {"goal_id": source["goal_id"], "agent_id": source["agent_identity"]["agent_id"], "session_id": "synthetic-session", "context_epoch": "one"} @@ -639,7 +651,7 @@ def test_staged_entry_replays_goal_and_changed_horizon_with_existing_delta_owner current = copy.deepcopy(source) if changed: current["planning_horizon"]["work_items"][0].update(priority="P0", successor_todo_ids=["todo-new-dependency"]) - value = _turn_envelope_result("replan_cli_view", payload=current) + value = project_captured_replan_cli_view(current) candidate = prepare_context_delivery(value, scope=scope, capture_id="entry-b", surface="quota_packet", baseline=full, retained={"complete": True, "target": full["target"]}) # Even without native delta, the entry has one immutable reader rather @@ -663,7 +675,7 @@ def test_staged_entry_replays_goal_and_changed_horizon_with_existing_delta_owner {"effective_action": "governed_capability_intent"}, {"replan_action_packet": {**source["replan_action_packet"], "settlement_only": True}}): legacy = {**source, **patch} - assert _turn_envelope_result("replan_cli_view", payload=legacy) == legacy + assert project_captured_replan_cli_view(legacy) == legacy def test_native_reader_keeps_its_installation_across_acceptance_and_writeback(tmp_path, monkeypatch): diff --git a/tests/control_plane/test_saved_context_project.py b/tests/control_plane/test_saved_context_project.py index 6ccdcf8b83..6b58b9d70f 100644 --- a/tests/control_plane/test_saved_context_project.py +++ b/tests/control_plane/test_saved_context_project.py @@ -520,6 +520,10 @@ def consume(wire): if enabled: assert bound["mode"] == ("full" if compacted else "delta"), bound assert len(json.dumps(bound, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= 32000 + current = Path(machine_reference_oracle(bound)["recovery"]["path"]) + captured = json.loads(current.with_name("decision.json").read_bytes()) + required = json.loads(current.read_bytes()) + assert json.dumps(required["quota"], sort_keys=True) == json.dumps(captured["quota"], sort_keys=True) delivered = consume(bound) if not compacted: assert delivered["base"]["capture_id"] == machine_reference_oracle(cold["navigation"])["target"]["capture_id"] @@ -949,11 +953,18 @@ def response(request, number, message): original_command = original["interaction_contract"]["cli_channel"]["selection_command"]["command_args_template"] words = shlex.split(selected["command_args_template"]) original_words = shlex.split(original_command) - for option in ("--context-delivery", "--codex-context-dir", "--max-page-output-bytes"): - for command_words in (words, original_words): - if option in command_words: - index = command_words.index(option) - del command_words[index:index + 2] + transport = { + "--decision-output-root": str(home / "loopx-context"), + "--context-delivery": "on", + "--delta-delivery": "auto", + "--codex-context-dir": str(home / "loopx-context"), + "--max-page-output-bytes": "32000", + } + for option, value in transport.items(): + assert option not in original_words and words.count(option) == 1 + index = words.index(option) + assert words[index + 1] == value + del words[index:index + 2] assert words == original_words selected["command_args_template"] = original_command # Main already deduplicates identical Todo rows within this exact @@ -1423,31 +1434,46 @@ 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"]): + ["--context-delivery", "off", "--initial-read"], ["--goal-id", "foreign", "--initial-read"], + ["--delta-delivery", "off", "--initial-read", "--page-bytes", "3200"]): code, error = cli(*project, *invalid) assert code and not error["ok"] and "content" not in error + code, error = cli(*project, "--select", "/instructions", "--page-bytes", "255") + assert code and not error["ok"] and "--page-bytes must be between 256 and" in error["reason"] assert set(context.rglob("*")) == initial_files - code, first = cli(*project, "--select", "/instructions") + window = 3200 + code, first = cli(*project, "--select", "/instructions", "--page-bytes", str(window)) 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"] + def read_selection(*, resume): result = bytearray() - args = ["quota", "context-read", "--reference-file", ref, "--max-page-output-bytes", "6400"] + if resume: + result.extend(first["content"].encode()) + assert len(result) <= window + words = shlex.split(first["next_command"]) + args = words[words.index("quota"):] + else: + args = ["quota", "context-read", "--reference-file", first["reference_file"], + "--max-page-output-bytes", "6400"] while True: code, page = cli(*args) assert code == 0 and page["offset"] == len(result) - result.extend(page["content"].encode()) + content = page["content"].encode() + if resume: + assert len(content) <= window + result.extend(content) 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] + replay = pool.submit(read_selection, resume=False) + continuation = pool.submit(read_selection, resume=True) + assert replay.result() == continuation.result() == 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()