diff --git a/benchmark/runtime/RUNTIME.md b/benchmark/runtime/RUNTIME.md index 788507184c..f8e0e543d8 100644 --- a/benchmark/runtime/RUNTIME.md +++ b/benchmark/runtime/RUNTIME.md @@ -175,9 +175,9 @@ Darwin native-checkpoint refusal remains in force. The host-side checkpoint preflight captures the complete original decision. With context delivery enabled, its prompt supplies `quota context-project` -with that original path/hash for ordinary work. An admitted replan with staged -review and writeback commands uses the existing replan projection in a private -presentation. Its readers retain the original guard's path and hash. The +with that original path and hash. The shared presentation owner prepares the +initial review and writeback routes for an admitted replan from the verified +capture. Their readers retain the original guard's path and hash. The projection command also carries the Goal, Agent, Turn and private cache location. The native tool presents the saved decision using the same native lifecycle projector; no guard, checkpoint, admission or debit is repeated. Cold delivery supplies the diff --git a/benchmark/runtime/checkpoint.py b/benchmark/runtime/checkpoint.py index 1a733a3e91..822052e0f5 100644 --- a/benchmark/runtime/checkpoint.py +++ b/benchmark/runtime/checkpoint.py @@ -265,16 +265,6 @@ def capture_before_replan(env: dict[str, str], receipt: dict[str, Any], body: st from .context_delivery import preflight_context_instruction assert guard_digest is not None - if guard.get("replan_read_path"): - from loopx.control_plane.quota.turn_envelope import _turn_envelope_result - from loopx.control_plane.turn_driver.decision_capture import write_decision_capture - - # Replan's staged commands are derived from the original packet. - # Keep their source hash and path bound to that immutable capture. - view = _turn_envelope_result("replan_cli_view", payload=guard) - if view != guard: - source_path = capture_root / "checkpoint-start-view.json" - guard_digest = write_decision_capture(source_path, view) reading = preflight_context_instruction(env, source_path, guard_digest, receipt["turn_id"], guide_in_host=guide_in_host) else: diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py index 57a075e001..95c9a2eb23 100644 --- a/loopx/cli_commands/quota_capture.py +++ b/loopx/cli_commands/quota_capture.py @@ -793,12 +793,14 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str # Carry the now-active transport into its selection presentation, not # into the immutable source or a second admission. Native lifecycle # still owns when the following full/delta view can be published. - bind_capture_selection_transport(payload, args, capture_root=root) + transport = _capture_selection_transport_template(payload, args, capture_root=root) prepared = prepare_codex_context(payload, directory=directory, capture_directory=cache, goal_id=args.goal_id, agent_id=args.agent_id, surface="quota_packet", capture_id=hashlib.sha256(raw).hexdigest(), continuity_generation=getattr(args, "_context_delivery_generation", None), - progressive_enabled=activation["progressive_enabled"], hook_session=hook_session) + progressive_enabled=activation["progressive_enabled"], selection_command_args_template=transport, + captured_decision_path=str(path) if args.initial_read and activation["progressive_enabled"] else None, + hook_session=hook_session) projected = prepared.delivery native_prepared = True if projected.get("schema_version") == "context_delivery_v1": @@ -1033,6 +1035,20 @@ def fits(value: dict[str, object]) -> bool: ) from exc +def _capture_selection_transport_template(payload: dict[str, Any], args: argparse.Namespace, *, + capture_root: Path) -> str | None: + """Derive display argv without editing the captured decision.""" + interaction = payload.get("interaction_contract") + cli = interaction.get("cli_channel") if isinstance(interaction, dict) else None + selection = cli.get("selection_command") if isinstance(cli, dict) else None + if not isinstance(selection, dict): + return None + presentation = {**payload, "interaction_contract": {**interaction, + "cli_channel": {**cli, "selection_command": dict(selection)}}} + bind_capture_selection_transport(presentation, args, capture_root=capture_root) + return presentation["interaction_contract"]["cli_channel"]["selection_command"].get("command_args_template") + + def bind_capture_selection_transport(payload: dict[str, Any], args: argparse.Namespace, *, capture_root: Path | None = None) -> None: """Retain caller display/capture options in the existing selection command. diff --git a/loopx/control_plane/quota/progressive_context.ts b/loopx/control_plane/quota/progressive_context.ts index 178b75f437..38a97d03dc 100644 --- a/loopx/control_plane/quota/progressive_context.ts +++ b/loopx/control_plane/quota/progressive_context.ts @@ -245,14 +245,17 @@ export function progressiveContextPlan(request: JsonObject): JsonObject { const pathStartsWith = (path: Path, start: Path) => start.every((key, i) => path[i] === key); -function requiredNativeContext(request: JsonObject, sourceValues: ProjectionSourceValues): +function requiredNativeContext(request: JsonObject, sourceValues: ProjectionSourceValues, + plan = progressiveContextPlan({payload: request.context, surface: request.surface})): {context: JsonObject; tokens: [Path, string][]; projected: boolean} { - const context = requireJsonObject(request.context, "current projection"); + const original = requireJsonObject(request.context, "current projection"); + const staged = jsonObject(plan.staged_context); + const context = staged ?? original; const details = requireJsonObject(request.native_projection, "supporting material reference"); const identity = contextSourceIdentity(request, request.source_number_tokens); - const plan = progressiveContextPlan({payload: context, surface: request.surface}); - const deferred = plan.deferred as JsonObject[], references = (plan.references ?? []) as JsonObject[]; - const instructions = plan.instructions as JsonObject[]; + const edits = details.progressive_enabled === false ? {deferred: [], instructions: []} : plan; + const deferred = edits.deferred as JsonObject[], references = (edits.references ?? []) as JsonObject[]; + const instructions = edits.instructions as JsonObject[]; const at = (root: unknown, path: Path): unknown => path.reduce((row, key) => row !== null && typeof row === "object" && Object.hasOwn(row, key) ? (row as Record)[key] : undefined, root); @@ -310,7 +313,8 @@ function requiredNativeContext(request: JsonObject, sourceValues: ProjectionSour }); const sourcePath: Path = ["interaction_contract", "agent_channel", "work_context", "sources"]; const sources = at(context, sourcePath); - required.context_details = {...details, source_path: details.path, + required.context_details = {path: details.path, sha256: details.sha256, + command_template: details.command_template, source_path: details.path, unread_context_index: { source: {path: details.path, sha256: details.sha256}, coverage_confirmed: false, @@ -342,7 +346,33 @@ function requiredNativeContext(request: JsonObject, sourceValues: ProjectionSour path[0] === "context" ? size + token.length - JSON.stringify(at(request, path)).length : size, 0); const projected = required !== context && Buffer.byteLength(JSON.stringify(required)) - Buffer.byteLength(JSON.stringify(context)) + adjustment(remaining) - adjustment(tokens) < 0; - return {context: projected ? required : context, tokens: projected ? remaining : tokens, projected}; + if (!projected) { required = context; copies.clear(); } + // The capture can predate the current native invocation. Only the view gets + // its current selection recipe. Replan commands still use the original hash. + const selectionPath = ["interaction_contract", "cli_channel", "selection_command", "command_args_template"]; + const transport = details.selection_command_args_template; + const bindTransport = typeof transport === "string" && typeof at(context, selectionPath) === "string" + && transport !== at(context, selectionPath); + if (bindTransport) parent(selectionPath)[selectionPath.at(-1)!] = transport; + if (staged) { + if (required.context_details === undefined) { + required = {...required, context_details: {path: details.path, sha256: details.sha256}}; + } + // Staging can omit whole subtrees, but the retained numeric facts still + // refer to the same source fields, including in the encoded full/delta. + const retained = tokens.filter(([path]) => { + if (path[0] !== "context") return true; + const target = path.slice(1), value = at(required, target); + if (typeof value !== "number" || !Object.is(value, at(original, target))) return false; + const owner = at(required, target.slice(0, -1)) as object; + const fields = sourceValues.get(owner) ?? new Map(); + fields.set(String(target.at(-1)), path); + sourceValues.set(owner, fields); + return true; + }); + return {context: required, tokens: retained, projected: true}; + } + return {context: required, tokens: projected ? remaining : tokens, projected: projected || bindTransport}; } function nativeContextContinuity(codec: JsonObject): JsonObject { @@ -364,11 +394,11 @@ function nativeContextContinuity(codec: JsonObject): JsonObject { /** Plan and encode one native capture without sending its intermediate view * back through Python. Capture publication and native locking remain Host IO. */ export function prepareProgressiveContextDelivery(request: JsonObject, legacyFields: readonly string[], - decisionFields: readonly string[]): JsonObject { + decisionFields: readonly string[], plan?: JsonObject): JsonObject { if (request.include_initial_read_plan !== undefined) requireBoolean(request.include_initial_read_plan, "include_initial_read_plan"); const sourceValues: ProjectionSourceValues = new WeakMap(); - const required = requiredNativeContext(request, sourceValues); + const required = requiredNativeContext(request, sourceValues, plan); const codec = nativeContextContinuity({...request, context: required.context, ...(request.source_number_tokens === undefined ? {} : {source_number_tokens: required.tokens})}); const delivery = encodeContextDelivery(codec, legacyFields, sourceValues, fileNavigationPaths(required.context)); diff --git a/loopx/control_plane/quota/turn_envelope.ts b/loopx/control_plane/quota/turn_envelope.ts index 52ec2676d1..a0f982cbf3 100644 --- a/loopx/control_plane/quota/turn_envelope.ts +++ b/loopx/control_plane/quota/turn_envelope.ts @@ -653,9 +653,12 @@ function replanGoalContext(payload: JsonObject, reads: JsonObject | null, coreRe /** Move only an identical acceptance copy behind the existing captured read. * Different revisions, other source fields and unavailable context stay intact. */ -function replanWorkContext(payload: JsonObject, context: JsonObject): JsonObject { +type ReplanValueIdentity = (value: unknown, path: (string | number)[]) => string; + +function replanWorkContext(payload: JsonObject, context: JsonObject, reads = replanReadCommands(payload), + identity: ReplanValueIdentity = canonicalHash): JsonObject { context = resolveExploreGuidance(payload, context); - const reads = replanReadCommands(payload); + const corePath = ["autonomous_replan_obligation", "replan_context", "core_goal"]; const core = object(object(object(payload.autonomous_replan_obligation).replan_context).core_goal); const acceptance = object(core.acceptance_contract); if (!reads || !Array.isArray(context.sources)) return context; @@ -663,12 +666,12 @@ function replanWorkContext(payload: JsonObject, context: JsonObject): JsonObject const steps = Array.isArray(plan.ordered_steps) ? plan.ordered_steps.map(object) : []; const attachments = object(steps.find(step => step.kind === "durable_writeback")).optional_attachments; const explore = Array.isArray(attachments) ? attachments.map(object).filter(row => row.capability_id === "explore") : []; - const identity = object(plan.identity); + const binding = object(plan.identity); const unboundReplan = payload.effective_action === EffectiveAction.AUTONOMOUS_REPLAN_REQUIRED && - selectedWorkTodoId(payload) === null && identity.binding_kind === "autonomous_replan" && - identity.replan_obligation_id === object(payload.replan_action_packet).obligation_id && - !identity.todo_id && explore.length === 0; - return {...context, sources: context.sources.map(value => { + selectedWorkTodoId(payload) === null && binding.binding_kind === "autonomous_replan" && + binding.replan_obligation_id === object(payload.replan_action_packet).obligation_id && + !binding.todo_id && explore.length === 0; + return {...context, sources: context.sources.map((value, index) => { const source = object(value), content = object(source.content); if (source.source === "turn_start_capability_hook" && source.hook_id === "explore.turn_context" && source.capability_id === "explore" && explore.length === 1 && typeof reads.writeback === "string") { @@ -679,7 +682,9 @@ function replanWorkContext(payload: JsonObject, context: JsonObject): JsonObject return {...source, content: omitUnboundExploreResultWriteback(content)}; } if (!Object.keys(acceptance).length || source.source !== "goal_acceptance" || content.status != null && content.status !== "loaded" || - canonicalHash(content.goal_acceptance_contract) !== canonicalHash(acceptance)) return value; + identity(content.goal_acceptance_contract, + ["interaction_contract", "agent_channel", "work_context", "sources", index, "content", "goal_acceptance_contract"]) + !== identity(acceptance, [...corePath, "acceptance_contract"])) return value; const {goal_acceptance_contract: _contract, ...metadata} = content; return {...source, content: {...metadata, context_read: reads.review}}; })}; @@ -1639,43 +1644,64 @@ function capturedDecisionSelection(request: JsonObject): JsonObject { return capturedContextSelection(request, initial ? initialContextReadPointers(requireJsonObject(request.context, "captured context"), INITIAL_CONTEXT_DECISION_FIELDS) : []); } +function replanCliView(payload: JsonObject, identity: ReplanValueIdentity = canonicalHash, + reads = replanReadCommands(payload)): JsonObject { + if (!reads) return payload; + const obligation = {...object(payload.autonomous_replan_obligation)}; + delete obligation.replan_context; + const interaction = object(payload.interaction_contract), agent = object(interaction.agent_channel); + const cli = object(interaction.cli_channel); + // Defer only authoring/settlement format, not admission or stop facts. + // This default captured path needs no optional native delta transport. + const goal = replanGoalContext(payload, reads); + const summary = object(payload.agent_todo_summary); + const originalCore = object(object(object(payload.autonomous_replan_obligation).replan_context).core_goal); + // Reuse the same captured acceptance index for an exact summary copy. + // Otherwise this legacy carrier silently restores the full deferred detail. + // A different revision or an existing summary index remains untouched. + const indexedSummary = goal.core.outcome_requirements_index && + !Object.hasOwn(summary, "outcome_requirements_index") && + identity(summary.goal_acceptance_contract ?? null, ["agent_todo_summary", "goal_acceptance_contract"]) + === identity(originalCore.acceptance_contract ?? null, + ["autonomous_replan_obligation", "replan_context", "core_goal", "acceptance_contract"]) + ? {...summary, goal_acceptance_contract: goal.core.acceptance_contract, + outcome_requirements_index: goal.core.outcome_requirements_index} : summary; + // Review owns the captured planning facts. The entry retains admission, + // current selection and stops; the mandatory reader supplies the complete + // Goal and horizon before the Agent chooses its next work. + const {core_goal: _core, planning_horizon: _horizon, ...entry} = payload; + return {...entry, + ...(indexedSummary !== summary ? {agent_todo_summary: indexedSummary} : {}), + interaction_contract: {...interaction, cli_channel: deferredReplanWriteback(cli), agent_channel: {...agent, + work_context: replanWorkContext(payload, object(agent.work_context), reads, identity)}}, + autonomous_replan_obligation: {...obligation, context_read: reads.review}, + replan_action_packet: replanActionPacket(payload, reads)}; +} + +function nativeReplanPresentation(request: JsonObject): JsonObject | undefined { + const path = object(request.native_projection).captured_decision_path; + if (typeof path !== "string") return undefined; + const source = requireJsonObject(request.context, "captured replan source"); + const packet = object(source.replan_action_packet); + const review = object(object(source.autonomous_replan_obligation).replan_context); + // Existing staged files are presentation inputs. Do not derive a new source + // hash from them or replace their already pinned reader commands. + if (typeof source.replan_read_path !== "string" || packet.schema_version !== "replan_action_packet_v0" + || typeof packet.obligation_id !== "string" + || packet.obligation_id !== review.obligation_id || packet.review !== undefined + || packet.writeback !== undefined) return undefined; + const identity = contextSourceIdentity(request, request.source_number_tokens); + const view = replanCliView(source, (value, field) => identity(value, ["context", ...field]), + replanReadCommands(source, path)); + return view === source ? undefined : {staged_context: view, deferred: [], instructions: []}; +} + export function evaluateTurnEnvelope(value: unknown): JsonObject { const request = requireJsonObject(value, "turn envelope request"); const operation = request.operation; if (operation === "replan_view") return replanView(request); if (operation === "context_select") return capturedDecisionSelection(request); - if (operation === "replan_cli_view") { - const payload = requireJsonObject(request.payload, "replan CLI payload"); - const reads = replanReadCommands(payload); - if (!reads) return payload; - const obligation = {...object(payload.autonomous_replan_obligation)}; - delete obligation.replan_context; - const interaction = object(payload.interaction_contract), agent = object(interaction.agent_channel); - const cli = object(interaction.cli_channel); - // Defer only authoring/settlement format, not admission or stop facts. - // This default captured path needs no optional native delta transport. - const goal = replanGoalContext(payload, reads); - const summary = object(payload.agent_todo_summary); - const originalCore = object(object(object(payload.autonomous_replan_obligation).replan_context).core_goal); - // Reuse the same captured acceptance index for an exact summary copy. - // Otherwise this legacy carrier silently restores the full deferred detail. - // A different revision or an existing summary index remains untouched. - const indexedSummary = goal.core.outcome_requirements_index && - !Object.hasOwn(summary, "outcome_requirements_index") && - canonicalHash(summary.goal_acceptance_contract ?? null) === canonicalHash(originalCore.acceptance_contract ?? null) - ? {...summary, goal_acceptance_contract: goal.core.acceptance_contract, - outcome_requirements_index: goal.core.outcome_requirements_index} : summary; - // Review owns the captured planning facts. The entry retains admission, - // current selection and stops; the mandatory reader supplies the complete - // Goal and horizon before the Agent chooses its next work. - const {core_goal: _core, planning_horizon: _horizon, ...entry} = payload; - return {...entry, - ...(indexedSummary !== summary ? {agent_todo_summary: indexedSummary} : {}), - interaction_contract: {...interaction, cli_channel: deferredReplanWriteback(cli), agent_channel: {...agent, - work_context: replanWorkContext(payload, object(agent.work_context))}}, - autonomous_replan_obligation: {...obligation, context_read: reads.review}, - replan_action_packet: replanActionPacket(payload)}; - } + if (operation === "replan_cli_view") return replanCliView(requireJsonObject(request.payload, "replan CLI payload")); if (operation === "selected_work_todo") { const source = requireJsonObject(request.payload, "Turn work selection source"); const historical = scalarString(request.historical_primary_text, "historical primary Todo"); @@ -1732,7 +1758,8 @@ export function evaluateTurnEnvelope(value: unknown): JsonObject { if (operation === "progressive_context_delivery") return prepareProgressiveContext(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); + : prepareProgressiveContextDelivery(request, LEGACY_QUOTA_CONTEXT_FIELDS, INITIAL_CONTEXT_DECISION_FIELDS, + nativeReplanPresentation(request)); if (operation === "native_context_lifecycle") return updateNativeContextLifecycle(request); if (operation === "context_recovery") return restoreContextDelivery(request, LEGACY_QUOTA_CONTEXT_FIELDS); if (operation === "bind_capture") return bindTurnDecisionCapture(request); diff --git a/loopx/control_plane/turn_driver/codex_context.py b/loopx/control_plane/turn_driver/codex_context.py index a8f0b5ebb4..5cb3f8f59b 100644 --- a/loopx/control_plane/turn_driver/codex_context.py +++ b/loopx/control_plane/turn_driver/codex_context.py @@ -620,6 +620,8 @@ def prepare_codex_context(payload: dict[str, Any], *, directory: str | None, cap continuity_generation: str | None = None, include_initial_read_plan: bool = True, progressive_enabled: bool | None = None, + selection_command_args_template: str | None = None, + captured_decision_path: str | None = None, hook_session: str | None = None) -> NativeContextPreparation: """Commit current recovery before optional projection or candidate updates. @@ -690,6 +692,7 @@ def publish_recovery(reference: dict[str, Any]) -> None: recovery_committed = True # Legacy direct native callers omitted the presentation option. progressive = progressive_enabled is not False + presentation = progressive or selection_command_args_template is not None try: transaction = prepare_context_delivery(payload, scope=scope, capture_id=capture_id, surface=surface, baseline=request.get("baseline"), @@ -698,18 +701,27 @@ def publish_recovery(reference: dict[str, Any]) -> None: native_projection={ "path": recovery["path"], "sha256": recovery["sha256"], "source_path": recovery["path"], + **({"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 {}), + **({"progressive_enabled": False} if not progressive else {}), "command_template": context_capture_read_command(Path(recovery["path"]), recovery["sha256"], page_bytes=page_bytes, max_page_output_bytes=output_bytes) + " --select '/sources'", - } if progressive else None, consumed_pointers=state.get("consumed_pointers"), + } if presentation else None, consumed_pointers=state.get("consumed_pointers"), observe_first_read=progressive) - except (EffectRuntimeRemoteError, EffectRuntimeStartupError): + except (EffectRuntimeRemoteError, EffectRuntimeStartupError) as exc: + if captured_decision_path is not None and isinstance(payload.get("replan_read_path"), str): + raise QuotaDecisionTransportError( + "Saved context presentation could not be prepared. Read the original capture and do not rerun admission." + ) from exc return NativeContextPreparation(payload) - if progressive and transaction["projected"]: + if presentation and transaction["projected"]: recovery = capture_context_projection(transaction["context"], directory=capture_directory, filename="required-context.json", read_guidance=CONTEXT_CAPTURE_PAGE_GUIDANCE) publish_recovery(recovery) - result: dict[str, Any] = transaction["delivery"] if progressive else transaction + result: dict[str, Any] = transaction["delivery"] if presentation else transaction read_sections = transaction["read_sections"] if progressive else {} result["recovery"] = recovery if result.get("continuity_basis") == "captured_file" and result.get("mode") == "delta": diff --git a/tests/control_plane/test_context_delivery_cli.py b/tests/control_plane/test_context_delivery_cli.py index de5c27dfce..af5808208c 100644 --- a/tests/control_plane/test_context_delivery_cli.py +++ b/tests/control_plane/test_context_delivery_cli.py @@ -181,10 +181,15 @@ def test_bounded_continuation_rechecks_original_capture_and_utf8_cursor(tmp_path capture = tmp_path / "original.json" raw = json.dumps({"requirements": "🙂" * 900}, ensure_ascii=False).encode() capture.write_bytes(raw) + # Complete recovery guidance and the pinned executable/path metadata already + # exceed 1600 bytes on source checkouts. Keep a bounded partial page without + # deleting guidance or changing any product/Host output limit. + limit = 2400 args = argparse.Namespace(capture_file=str(capture), capture_sha256=hashlib.sha256(raw).hexdigest(), - offset=0, page_bytes=8192, max_page_output_bytes=1600, format="json") + offset=0, page_bytes=8192, max_page_output_bytes=limit, format="json") page = read_context_capture(args) assert page["ok"] and not page["eof"] + assert len(json.dumps(page, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= limit args.offset = page["next_offset"] + 1 assert "UTF-8 boundary" in read_context_capture(args)["reason"] args.offset = page["next_offset"] diff --git a/tests/control_plane/test_replan_saved_context.py b/tests/control_plane/test_replan_saved_context.py new file mode 100644 index 0000000000..8a5f2b3766 --- /dev/null +++ b/tests/control_plane/test_replan_saved_context.py @@ -0,0 +1,198 @@ +"""Old preflight callers adopt staged reads through the installed CLI owner.""" +from __future__ import annotations + +import hashlib +import json +import os +from pathlib import Path +import subprocess +import sys +import shlex + +import pytest + +from canonical_authority_fixture import isolate_sqlite_runtime +from loopx.control_plane.effect_runtime import restart_effect_runtime +from loopx.control_plane.quota.turn_envelope import _turn_envelope_result +from loopx.control_plane.turn_driver.codex_context import codex_context_hook +from loopx.control_plane.turn_driver.decision_capture import write_decision_capture +from test_context_capture_registry_independence import run_cli +from test_replan_progress_native_cli import ( + AGENT, GOAL, TODO, TURN, call_progress_cli, make_progress_cli_fixture, +) + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("entry", [ + "original", "copied", "staged", "numeric", "selected", "off", "delta_only", "projection_failure", +]) +def test_saved_replan_reads_keep_original_authority(tmp_path, monkeypatch, provider, entry): + tmp_path = tmp_path.resolve() + isolate_sqlite_runtime(tmp_path, monkeypatch) + monkeypatch.setenv("CODEX_THREAD_ID", "saved-replan-session") + monkeypatch.delenv("CODEX_ROLLOUT_TRACE_ROOT", raising=False) + fixture = make_progress_cli_fixture(tmp_path, provider, "advanced") + native = tmp_path / "native" + native.mkdir(mode=0o700) + try: + prepared = call_progress_cli(fixture, "quota", "should-run", "--codex-app", + "--goal-id", GOAL, "--agent-id", AGENT, "--todo-id", TODO, + "--turn-instance-id", TURN, "--decision-output-dir", str(tmp_path / "capture"), + "--prepared-capture-reference", "--context-delivery", "on", + "--codex-context-dir", str(native), "--defer-context-presentation") + capture = tmp_path / "capture" / "decision.json" + original_path = capture + reader_path = capture + original = capture.read_bytes() + source = json.loads(original) + assert prepared["capture_sha256"] == hashlib.sha256(original).hexdigest() + assert "review" not in source["replan_action_packet"] + identity = source["heartbeat_receipt"]["settlement_identity"] + obligation = source["autonomous_replan_obligation"]["obligation_id"] + # SessionStart may follow the original Host guard. Neither projection + # nor its stage readers may replay admission or mutate durable work. + codex_context_hook(str(native), dict(session_id="saved-replan-session", + cwd=str(fixture["project"]), hook_event_name="SessionStart", source="startup")) + if entry == "copied": + capture = tmp_path / "copied.json" + capture.write_bytes(original) + reader_path = capture + original_path.unlink() + if entry == "numeric": + # These two acceptance carriers compare equal after JSON.parse. + # Both exact values must survive this synthetic source and view. + core = source["autonomous_replan_obligation"]["replan_context"]["core_goal"] + acceptance = {**(core.get("acceptance_contract") or {}), "precision_probe": 2**53, + "outcome_requirements": {"schema_version": "loopx_goal_outcomes_v0", + "outcomes": [{"id": "precision", "objective": "Keep distinct source values", + "revision": 1, "requirements": [], "sources": []}]}} + core["acceptance_contract"] = acceptance + distinct = {**acceptance, "precision_probe": 2**53 + 1} + source["agent_todo_summary"]["goal_acceptance_contract"] = distinct + source["interaction_contract"]["agent_channel"]["work_context"]["sources"].append({ + "source": "goal_acceptance", "content": {"status": "loaded", "goal_acceptance_contract": distinct}, + }) + source["precision_fixture"] = {"exact_integer": 2**53 + 1, "exact_float": 1.0} + capture = tmp_path / "precision-source.json" + reader_path = capture + write_decision_capture(capture, source) + if entry == "staged": + capture = tmp_path / "already-staged.json" + write_decision_capture(capture, _turn_envelope_result("replan_cli_view", payload=source)) + selected_raw = capture.read_bytes() + + def snapshot(): + return {str(p): p.read_bytes() for root in (fixture["project"], fixture["runtime"]) + for p in root.rglob("*") if p.is_file()} | {"registry": fixture["registry"].read_bytes()} + + before = snapshot() + options = ["--context-delivery", "off", "--delta-delivery", "off"] if entry == "off" else [ + "--context-delivery", "on", "--codex-app", "--codex-context-dir", str(native), + *(["--select", "/replan_action_packet"] if entry == "selected" else ["--initial-read"])] + if entry == "delta_only": + options = ["--context-delivery", "off", "--delta-delivery", "auto", "--codex-app", + "--codex-context-dir", str(native), "--initial-read"] + command = ["quota", "context-project", "--capture-file", str(capture), + "--capture-sha256", hashlib.sha256(selected_raw).hexdigest(), + "--goal-id", GOAL, "--agent-id", AGENT, "--turn-instance-id", TURN, *options] + if entry == "projection_failure": + # Only the presentation provider fails. The ordinary CLI must + # return this same admitted capture, not hide it or admit again. + marker = tmp_path / "presentation-failed" + script = "from pathlib import Path\nmarker = Path(" + repr(str(marker)) + ")\n" + """from unittest.mock import patch +from loopx.control_plane.quota import turn_envelope +from loopx.control_plane.effect_runtime import EffectRuntimeStartupError +from loopx.entrypoint import main +original = turn_envelope._turn_envelope_result +def fail(operation, **kwargs): + if operation == 'context_delivery' and (kwargs.get('native_projection') or {}).get('captured_decision_path'): + marker.write_text('context_delivery') + raise EffectRuntimeStartupError('synthetic presentation outage', diagnostic_code='synthetic_fixture') + return original(operation, **kwargs) +with patch.object(turn_envelope, '_turn_envelope_result', fail): + raise SystemExit(main()) +""" + result = subprocess.run([sys.executable, "-c", script, "--format", "json", *command], + cwd=fixture["project"], env={**os.environ, + "PYTHONPATH": str(Path(__file__).resolve().parents[2]), "LOOPX_USAGE_PING": "0"}, + capture_output=True, text=True, timeout=30) + response = json.loads(result.stdout) + assert result.returncode == 1 and response["ok"] is False + assert response["capture_sha256"] == hashlib.sha256(original).hexdigest() + assert response["heartbeat_receipt"] == source["heartbeat_receipt"] + assert response["recovery"]["path"] == str(capture) + assert response["recovery"]["sha256"] == hashlib.sha256(original).hexdigest() + assert marker.read_text() == "context_delivery" + assert "do not rerun admission" in response["reason"] + assert snapshot() == before and capture.read_bytes() == original + return + result = run_cli("loopx.entrypoint", fixture["project"], *command) + assert result.returncode == 0, result.stdout + result.stderr + response = json.loads(result.stdout) + assert response["ok"], response + first_review = None + staged = entry in {"original", "copied", "staged", "numeric"} + if staged: + first_review = next(row["value"] for row in json.loads(response["content"])["selections"] + if row["pointer"] == "/replan_action_packet/review") + navigation = response.get("navigation", response) + command, observed = navigation["recovery"]["command"], bytearray() + while True: + argv = shlex.split(command) + output = run_cli("loopx.entrypoint", fixture["project"], *argv[argv.index("quota"):]) + assert output.returncode == 0, output.stdout + output.stderr + page = json.loads(output.stdout) + assert page["ok"] and page["offset"] == len(observed) + observed.extend(page["content"].encode()) + if page["eof"]: + assert len(observed) == page["total_bytes"] + break + command = page["next_command"] + recovered = json.loads(observed) + assert recovered["heartbeat_receipt"]["settlement_identity"] == identity + assert recovered["autonomous_replan_obligation"]["obligation_id"] == obligation + if staged: + packet = recovered["replan_action_packet"] + assert "review" in packet and "writeback" in packet + assert packet["review"] == first_review + if entry == "staged": + assert packet == json.loads(selected_raw)["replan_action_packet"] + argv = shlex.split(packet["review"]) + assert argv[argv.index("--decision-file") + 1] == str(reader_path) + assert "replan_context" not in recovered["autonomous_replan_obligation"] + + def read(command): + argv = shlex.split(command) + output = run_cli("loopx.entrypoint", fixture["project"], *argv[argv.index("quota"):]) + assert output.returncode == 0, output.stdout + output.stderr + value = json.loads(output.stdout) + assert value["ok"] and value["obligation_id"] == obligation + return value + + review = read(packet["review"]) + writeback = read(packet["writeback"]) + assert review["source_hash"] == writeback["source_hash"] + if entry == "copied": + pinned = _turn_envelope_result("replan_cli_view", payload=source)["replan_action_packet"]["review"] + pinned = shlex.split(pinned) + assert review["source_hash"] == pinned[pinned.index("--source-hash") + 1] + assert writeback["writeback"]["settlement_plan"]["identity"] == identity + assert review["context_id"] == source["autonomous_replan_obligation"]["replan_context"]["context_id"] + if entry == "numeric": + assert recovered["precision_fixture"] == {"exact_integer": 2**53 + 1, "exact_float": 1.0} + assert type(recovered["precision_fixture"]["exact_float"]) is float + assert recovered["agent_todo_summary"]["goal_acceptance_contract"]["precision_probe"] == 2**53 + 1 + sources = recovered["interaction_contract"]["agent_channel"]["work_context"]["sources"] + assert sources[-1]["content"]["goal_acceptance_contract"]["precision_probe"] == 2**53 + 1 + else: + # Explicit raw selectors and disabled transport keep their existing + # content. They are not implicitly converted to a different view. + assert recovered["replan_action_packet"] == source["replan_action_packet"] + assert snapshot() == before + assert capture.read_bytes() == selected_raw + if entry == "copied": + assert not original_path.exists() + else: + assert original_path.read_bytes() == original + finally: + restart_effect_runtime(wait_for_exit=True) diff --git a/tests/control_plane/test_research_checkpoint_capture.py b/tests/control_plane/test_research_checkpoint_capture.py index f096351758..6eeb8012ae 100644 --- a/tests/control_plane/test_research_checkpoint_capture.py +++ b/tests/control_plane/test_research_checkpoint_capture.py @@ -86,26 +86,26 @@ def observe_guard(argv, **kwargs): digest = command[command.index("--capture-sha256") + 1] assert hashlib.sha256(source.read_bytes()).hexdigest() == digest assert json.loads(source.read_bytes())["heartbeat_receipt"]["turn_instance_id"] == turn + assert source.name == "decision.json" + assert digest == observed[0]["capture_sha256"] + assert not list(wake.glob("checkpoint-start-*/checkpoint-start-view.json")) if replan: - assert source.name == "checkpoint-start-view.json" - original = source.parent / "decision-original/decision.json" - assert hashlib.sha256(original.read_bytes()).hexdigest() == observed[0]["capture_sha256"] - command.remove("--initial-read") - command.extend(["--select", "/replan_action_packet/review", "--select", "/replan_action_packet/writeback"]) + from loopx.control_plane.turn_driver.codex_context import codex_context_hook + + session = "checkpoint-native-session" + codex_context_hook(str(context), dict(session_id=session, cwd=str(project), + hook_event_name="SessionStart", source="startup")) + env["CODEX_THREAD_ID"] = session selected = actual_run(command, cwd=project, env=env, capture_output=True, text=True, check=True) page = json.loads(selected.stdout) assert page["ok"] and page["eof"], page actions = {row["pointer"]: row["value"] for row in json.loads(page["content"])["selections"]} - assert "--replan-stage writeback" in actions["/replan_action_packet/writeback"] review = actual_run(shlex.split(actions["/replan_action_packet/review"]), cwd=project, env=env, capture_output=True, text=True, check=True) review_packet = json.loads(review.stdout) - original_packet = json.loads(original.read_bytes()) + original_packet = json.loads(source.read_bytes()) assert review_packet["obligation_id"] == original_packet["replan_action_packet"]["obligation_id"] - else: - assert source.name == "decision.json" - assert digest == observed[0]["capture_sha256"] - assert not list(wake.glob("checkpoint-start-*/checkpoint-start-view.json")) + assert hashlib.sha256(source.read_bytes()).hexdigest() == digest assert body.endswith("original work instructions") assert len(observed) == 1 and "interaction_contract" not in observed[0] assert cli._heartbeat_receipt_count(runtime, turn) == 1