diff --git a/benchmark/runtime/context_delivery.py b/benchmark/runtime/context_delivery.py index 1a4c82b645..f05a6e1051 100644 --- a/benchmark/runtime/context_delivery.py +++ b/benchmark/runtime/context_delivery.py @@ -27,17 +27,18 @@ def preflight_context_instruction(env: dict[str, str], path: Path, digest: str, project = [*prefix, "context-project", *source, "--goal-id", env["LOOPX_GOAL_ID"], "--agent-id", env["LOOPX_AGENT_ID"], "--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"] + receipt_reader = [*prefix, "context-read", *source, "--select", "/heartbeat_receipt"] return ("Run this command directly and deliver its complete bounded stdout to the model. " "Keep top-level content and next_command when present; navigation alone omits the returned page. " "Preserve any failure response.\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. " + + "\n```\n" + ("" if guide_in_host else "After projection succeeds: " + CONTEXT_CAPTURE_NAVIGATION_GUIDANCE) + + "For known fields, replace --initial-read with repeated --select pointers in the first command. " + "After a successful projection, continue the returned page or use navigation.recovery.command to reread. " + "If projection fails, follow its recovery instructions and preserve the original receipt. " + "If it provides no verified context or usable instructions, restore the intact capture and retry the same " + "context-project with its original digest and Turn. Use this command for diagnostic receipt readback:\n```sh\n" + + shlex.join(receipt_reader) + "\n```\n" + "Use verified context before work. Do not rerun admission to recover presentation. " "Neither presentation nor recovery renews this decision's authority or freshness. ") diff --git a/benchmark/tests/test_context_delivery.py b/benchmark/tests/test_context_delivery.py index 11b42fb975..215c53c069 100644 --- a/benchmark/tests/test_context_delivery.py +++ b/benchmark/tests/test_context_delivery.py @@ -68,6 +68,7 @@ def test_projection_failure_keeps_original_capture_selectively_readable(tmp_path for command in re.findall(r"```sh\n(.*?)\n```", instruction, re.S) ] assert len(commands) == 2 + assert commands[1][commands[1].index("--select") + 1:] == ["/heartbeat_receipt"] def invoke(argv): return subprocess.run( @@ -95,6 +96,14 @@ def invoke(argv): assert json.loads(index["content"])["schema_version"] == "context_capture_index_v1" assert len(discovery.stdout.encode()) < len(raw) // 5 + diagnostic = invoke(commands[1]) + assert diagnostic.returncode == 0, diagnostic.stderr + receipt_page = json.loads(diagnostic.stdout) + assert receipt_page["sha256"] == digest and receipt_page["eof"] + assert json.loads(receipt_page["content"])["selections"] == [ + {"pointer": "/heartbeat_receipt", "value": source["heartbeat_receipt"]}, + ] + reader = recovery[:recovery.index("context-read") + 1] branch = invoke([ *reader, "--capture-file", str(capture), "--capture-sha256", digest, diff --git a/docs/reference/protocols/turn-envelope-v0.md b/docs/reference/protocols/turn-envelope-v0.md index fa23f0a0e3..9474a88164 100644 --- a/docs/reference/protocols/turn-envelope-v0.md +++ b/docs/reference/protocols/turn-envelope-v0.md @@ -56,6 +56,20 @@ An older CLI rejects the new flag before guard execution. Omit it to roll back; existing complete captures remain readable through their original digest-bound `quota context-read`/`context-project` recovery. +`quota context-project` checks a saved complete decision before presenting it +as current context, including with Delta alone or with context delivery off. +If the saved source cannot be verified, the returned digest-bound +`context-read` command selects only its recorded `heartbeat_receipt` for +diagnosis. Restore the intact, supported source and retry `context-project` +with the original Goal, Agent, Turn and digest before using it for work. +Native recovery publishes the current session source after the same typed +check succeeds. Rejection retires the implicit lookup for the same native +session, Goal and Agent. The original captures remain available by their +explicit digest-bound references. For a rejected ready capture, the native +hook can still read back its recorded claim and lease references. A +diagnostic read cannot acknowledge delivery of ready work context. Optional +presentation failures after verification can still refer to the verified capture. + For eligible original Todo plans, the same typed settlement owner projects one `refresh-state ... --settle-turn --settlement-receipt` command. It retains the original effect identity, validation and terminal conditions. A non-ready @@ -80,6 +94,15 @@ model context. Actual live adoption and efficiency still require separate proof. 可执行的原 Todo 结算计划复用同一 typed owner，合并为一条写回并保留原 effect id， 不把未 ready 的准备或错误回执转成授权。部分失败只沿原 `settlement_owed` 恢复。 旧 CLI 在 guard 前拒绝新选项，回滚时移除该选项；旧完整 capture 仍用原 reader。 + +`quota context-project` 先校验已保存的完整决定，再将它呈现为当前上下文。仅开启差分 +或关闭上下文交付时同样校验。来源校验失败或中断时，返回的摘要绑定 +`context-read` 命令只选取原记录的 `heartbeat_receipt` 用于诊断。恢复完整且受支持 +的来源后，以原 Goal、Agent、Turn 和摘要重试 `context-project`，通过后再作为工作 +上下文。原生会话通过同一校验后发布当前来源。拒绝来源时撤下同一会话、Goal 和 Agent +的隐式读回入口。原捕获仍可通过显式摘要引用读取。准备就绪的捕获被拒绝后，原生 +Hook 仍能读回已记录的领取与租约引用。诊断读取不会确认已交付就绪的工作上下文。 +校验成功后如遇可选呈现失败，仍可读取已验证的捕获。 这只是 Host 开发协议入口，不新增设置、模型/子 worker、预算或权限，也不证明 已运行的 Host 采用或效率收益。 diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py index 7fd10f5cb8..093a594738 100644 --- a/loopx/cli_commands/quota_capture.py +++ b/loopx/cli_commands/quota_capture.py @@ -278,6 +278,39 @@ def progressive_context_capture_view(payload: dict[str, Any], args: argparse.Nam return readable if page_attached or len(_native_output_bytes(readable)) < len(_native_output_bytes(required)) else required +def _saved_context_transport_failure( + error: Exception, *, args: argparse.Namespace, receipt: dict[str, Any], + recovery: dict[str, Any], reader_argv: list[str], source_unverified: bool = False, +) -> dict[str, Any]: + page_bytes, output_bytes = native_context_page_budget() + limit = getattr(args, "max_page_output_bytes", None) + if isinstance(limit, int) and limit > 0: + output_bytes = min(output_bytes, limit) + page_bytes = min(page_bytes, output_bytes) + if source_unverified: + instruction = ( + "The saved source could not be verified. Use recovery.command for diagnostic receipt readback. " + "Restore an intact, supported capture and retry context-project with its original Goal, Agent, Turn " + "and digest before using it as work context. Preserve the original receipt and do not rerun admission." + ) + else: + instruction = ( + "Context presentation was not published. Discover current fields with recovery.command, then select " + "only the facts the current decision needs from the original capture; do not drain unrelated optional " + "material or rerun admission. Preserve its Turn receipt. If a selected value spans pages, follow " + "next_command only for that selection and verify contiguous offsets, total_bytes and sha256. " + "Authority and freshness are unchanged. " + CONTEXT_CAPTURE_NAVIGATION_GUIDANCE + ) + return {"ok": False, "error_code": QuotaDecisionTransportError.error_code, + "reason": " ".join(str(error).split())[:500], + "heartbeat_receipt": receipt, "capture_sha256": args.capture_sha256, + "recovery": {**recovery, "command": context_capture_read_command( + Path(recovery["path"]), recovery["sha256"], page_bytes=page_bytes, + max_page_output_bytes=output_bytes, list_fields=not source_unverified, + select=["/heartbeat_receipt"] if source_unverified else None, reader_argv=reader_argv)}, + "instruction": instruction} + + 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. @@ -289,6 +322,9 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str Invalid run-mode opt-ins fail before that optional transport path. """ from ..command_invocation import current_loopx_cli_argv + from ..control_plane.quota.turn_envelope import ( + COMPLETE_TURN_DECISION_VERSION, assert_complete_turn_decision, is_legacy_saved_quota_decision, + ) reader_argv = current_loopx_cli_argv() try: @@ -327,6 +363,23 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str reference = {"path": str(path), "sha256": args.capture_sha256, "command": context_capture_read_command(path, args.capture_sha256, reader_argv=reader_argv)} full_reference = dict(reference) + version = payload.get("schema_version") + complete = version == COMPLETE_TURN_DECISION_VERSION + try: + if not complete and not is_legacy_saved_quota_decision(payload): + if activation["enabled"]: + from ..control_plane.turn_driver.codex_context import retire_rejected_saved_context + + retire_rejected_saved_context(directory=args.codex_context_dir, context_root=args.codex_context_root, + goal_id=args.goal_id, agent_id=args.agent_id, capture_sha256=args.capture_sha256, + hook_session=hook_session) + raise QuotaDecisionTransportError("Unsupported saved quota decision version or mixed Turn binding.") + if complete and not activation["enabled"] and not activation["progressive_enabled"]: + assert_complete_turn_decision(payload) + except (QuotaDecisionTransportError, EffectRuntimeRemoteError, EffectRuntimeStartupError) as exc: + return _saved_context_transport_failure(exc, args=args, receipt=receipt, + recovery=full_reference, reader_argv=reader_argv, source_unverified=True) + mode = "full" target = None native_prepared = False @@ -342,15 +395,13 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str 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) + initial_read_plan=None, output_limit=output_limit, reader_argv=reader_argv) return navigation - except EffectRuntimeRejected as exc: - if payload.get("schema_version") == "loopx_turn_envelope_v1": - failure = QuotaDecisionTransportError(str(exc)) - return {"ok": False, "error_code": failure.error_code, "reason": str(failure), - "capture_sha256": args.capture_sha256, "recovery": full_reference} - except (QuotaDecisionTransportError, OSError, ValueError, EffectRuntimeRemoteError, EffectRuntimeStartupError): - pass # The complete original reference remains the fallback below. + except (QuotaDecisionTransportError, OSError, ValueError, EffectRuntimeRemoteError, EffectRuntimeStartupError) as exc: + if complete: + return _saved_context_transport_failure(exc, args=args, receipt=receipt, + recovery=full_reference, reader_argv=reader_argv, source_unverified=True) + # Historical readers keep their original fallback below. try: from ..control_plane.turn_driver.codex_context import codex_context_directory, prepare_codex_context @@ -385,7 +436,6 @@ 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. - complete = payload.get("schema_version") == "loopx_turn_envelope_v1" transport = _capture_selection_transport_template(payload, args, capture_root=root) presentation = payload if args.initial_read and activation["progressive_enabled"] and not complete: @@ -430,6 +480,11 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str 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 (EffectRuntimeRemoteError, EffectRuntimeStartupError) as exc: + if not complete: + raise + return _saved_context_transport_failure(exc, args=args, receipt=receipt, + recovery=full_reference, reader_argv=reader_argv, source_unverified=not native_prepared) 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 @@ -438,28 +493,9 @@ def project_saved_context_capture(args: argparse.Namespace, *, hook_session: str 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 " - "original decision and receipt; do not rerun admission." - ) - page_bytes, output_bytes = native_context_page_budget() - if isinstance(caller_output_limit, int) and caller_output_limit > 0: - output_bytes = min(output_bytes, caller_output_limit) - page_bytes = min(page_bytes, output_bytes) - recovery_index = {**full_reference, "command": context_capture_read_command( - Path(full_reference["path"]), full_reference["sha256"], - page_bytes=page_bytes, max_page_output_bytes=output_bytes, list_fields=True, - reader_argv=reader_argv)} - return {"ok": False, "error_code": exc.error_code, - "reason": " ".join(str(exc).split())[:500], - "heartbeat_receipt": receipt, "capture_sha256": args.capture_sha256, - "recovery": recovery_index, - "instruction": "Context presentation was not published. Discover current fields with " - "recovery.command, then select only the facts the current decision needs from the original " - "capture; do not drain unrelated optional material or rerun admission. Preserve its Turn " - "receipt. If a selected value spans pages, follow next_command only for that selection and " - "verify contiguous offsets, total_bytes and sha256. Authority and freshness are unchanged. " - + CONTEXT_CAPTURE_NAVIGATION_GUIDANCE} + exc = QuotaDecisionTransportError("Current context recovery could not be prepared.") + return _saved_context_transport_failure(exc, args=args, receipt=receipt, + recovery=full_reference, reader_argv=reader_argv, source_unverified=complete and not native_prepared) # Off is inert; after native preparation an optional wire write can fall # back to the committed required view without leaving an older selector live. result: dict[str, Any] = {"ok": True, "mode": mode, **reference, **({"target": target} if target else {}), @@ -915,11 +951,16 @@ def handle_native_context_command(args: argparse.Namespace, *, print_payload: Ca "goal_id": request["goal_id"], "agent_id": request["agent_id"], "turn_instance_id": request["turn_instance_id"], }), hook_session=event["session_id"]) - if receipt["context"].get("ok") is False: - from ..command_invocation import current_loopx_cli_argv - - receipt["context"]["recovery_command"] = context_capture_read_command( - source, digest, reader_argv=current_loopx_cli_argv()) + context = receipt["context"] + if context.get("ok") is False: + recovery = context.get("recovery") + if isinstance(recovery, dict) and isinstance(recovery.get("command"), str): + context["recovery_command"] = recovery["command"] + else: + from ..command_invocation import current_loopx_cli_argv + + context["recovery_command"] = context_capture_read_command( + source, digest, reader_argv=current_loopx_cli_argv()) preparation_context = codex_execution_preparation_context(receipt) except LegacyNativePreparationCapture as error: preparation_context = str(error) diff --git a/loopx/control_plane/quota/turn_envelope.py b/loopx/control_plane/quota/turn_envelope.py index eadd76862c..f59e380016 100644 --- a/loopx/control_plane/quota/turn_envelope.py +++ b/loopx/control_plane/quota/turn_envelope.py @@ -302,6 +302,14 @@ def compose_turn_decision( return _apply_context_projection_changes(source, result) +def is_legacy_saved_quota_decision(payload: Mapping[str, Any]) -> bool: + """Recognize saved raw/v0 carriers that predate complete-source checking.""" + return ( + "contract_binding" not in payload + and payload.get("schema_version", TURN_ENVELOPE_SCHEMA_VERSION) == TURN_ENVELOPE_SCHEMA_VERSION + ) + + def assert_complete_turn_decision(payload: Mapping[str, Any]) -> None: """Validate the original contract; keep its host binding and caller-owned bytes.""" _turn_envelope_result("complete_consumer", payload=dict(payload)) diff --git a/loopx/control_plane/quota/turn_envelope.ts b/loopx/control_plane/quota/turn_envelope.ts index adc2fef63e..5e6ec826e3 100644 --- a/loopx/control_plane/quota/turn_envelope.ts +++ b/loopx/control_plane/quota/turn_envelope.ts @@ -1891,16 +1891,22 @@ function completeHostDecision(request: JsonObject): JsonObject { return completeHostSource(request).decision; } -/** Reuse the existing presentation plan only after its complete source passes - * the versioned decision owner. The unsigned root is used for path selection; - * edits and recovery still address the unchanged, signed original capture. */ -function checkedContextPresentationPlan(request: JsonObject): JsonObject { +function checkedContextPresentationSource(request: JsonObject): JsonObject { const context = requireJsonObject(request.context, "current projection"); + if (request.surface === "complete_turn") { + const numbers = sourceNumberTokens(request, request.source_number_tokens) + .filter(({path}) => path[0] === "context").map(({path, token}) => [path.slice(1), token]); + requireCompleteTurnDecision(context, numbers); + } + return context; +} + +/** The unsigned root is used for path selection; edits and recovery still + * address the unchanged, signed original capture. */ +function checkedContextPresentationPlan(request: JsonObject): JsonObject { + const context = checkedContextPresentationSource(request); if (request.surface !== "complete_turn") return nativeReplanPresentation(request) ?? contextPresentationPlan(request); - const numbers = sourceNumberTokens(request, request.source_number_tokens) - .filter(({path}) => path[0] === "context").map(({path, token}) => [path.slice(1), token]); - requireCompleteTurnDecision(context, numbers); const source = {...context}; delete source.schema_version; delete source.contract_binding; @@ -2613,10 +2619,14 @@ export function evaluateTurnEnvelope(value: unknown): JsonObject { } if (operation === "progressive_context_delivery") return prepareProgressiveContext(request, checkedContextPresentationPlan(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, + if (operation === "context_delivery") { + if (request.native_projection === undefined) { + if (request.surface === "complete_turn") checkedContextPresentationSource(request); + return prepareContextDelivery(request, LEGACY_QUOTA_CONTEXT_FIELDS); + } + return prepareProgressiveContextDelivery(request, LEGACY_QUOTA_CONTEXT_FIELDS, INITIAL_CONTEXT_DECISION_FIELDS, checkedContextPresentationPlan(request), readyWriteScope(requireJsonObject(request.context, "current projection"))); + } 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 cd9595e833..8b31614600 100644 --- a/loopx/control_plane/turn_driver/codex_context.py +++ b/loopx/control_plane/turn_driver/codex_context.py @@ -413,7 +413,7 @@ def codex_context_hook(directory: str | None, event: dict[str, Any]) -> dict[str previous = updated.get("host_execution_preparation") ready = previous.get("ready_capture") if isinstance(previous, dict) else None ready_digest = ready.get("sha256") if isinstance(ready, dict) else None - if isinstance(ready_digest, str) and any( + if isinstance(ready_digest, str) and not previous.get("retired_context_readback") and any( page.get("ok") is True and ready_digest in (page.get("sha256"), page.get("source_sha256")) for page in pages @@ -453,6 +453,34 @@ def codex_context_hook(directory: str | None, event: dict[str, Any]) -> dict[str return {} +def _preparation_readback(action: dict[str, Any], previous: dict[str, Any], *, + event_name: str) -> dict[str, Any] | None: + if event_name != "SessionStart" and previous.get("ready_read_observed") is True: + return None + readback = {**action, "readback_only": True} + if isinstance(previous.get("claim_operation_id"), str) and previous["claim_operation_id"]: + readback["claim_operation_id"] = previous["claim_operation_id"] + if isinstance(previous.get("ready_capture"), dict): + readback["ready_capture"] = dict(previous["ready_capture"]) + return readback + + +def _retired_preparation_readback(root: Path, state: dict[str, Any], *, + session: str, event_name: str, + readback_for: str | None) -> dict[str, Any] | None: + previous = state.get("host_execution_preparation") + if not isinstance(previous, dict) or previous.get("state") != "ready": + return None + action = previous.get("retired_context_readback") + source = previous.get("source_sha256") + claim_source = previous.get("claim_source_sha256") + if (not isinstance(action, dict) or (readback_for is not None and readback_for != source) + or not _original_native_claim_source(root, session, claim_source)): + return None + return _preparation_readback({**action, "source_sha256": source, + "claim_source_sha256": claim_source}, previous, event_name=event_name) + + def native_execution_preparation_request( directory: str | None, event: dict[str, Any], *, readback_for: str | None = None, ) -> dict[str, Any] | None: @@ -471,7 +499,8 @@ def native_execution_preparation_request( return None recovery = state.get("recovery") if not isinstance(recovery, dict): - return None + return _retired_preparation_readback(root, state, session=session, + event_name=event["hook_event_name"], readback_for=readback_for) request = recovery.get("execution_preparation") source_sha256 = recovery.get("sha256") if (not isinstance(request, dict) or not isinstance(source_sha256, str) @@ -495,17 +524,7 @@ def native_execution_preparation_request( ) ): if previous.get("state") in {"attempted", "ready"}: - if ( - event.get("hook_event_name") != "SessionStart" - and previous.get("ready_read_observed") is True - ): - return None - readback = {**action, "readback_only": True} - if isinstance(previous.get("claim_operation_id"), str) and previous["claim_operation_id"]: - readback["claim_operation_id"] = previous["claim_operation_id"] - if isinstance(previous.get("ready_capture"), dict): - readback["ready_capture"] = dict(previous["ready_capture"]) - return readback + return _preparation_readback(action, previous, event_name=event["hook_event_name"]) if readback_for is not None: return None if request.get("turn_instance_id"): @@ -749,6 +768,134 @@ class NativeContextPreparation: initial_read_plan: dict[str, Any] | None = None +def _preserve_preparation_readback(state: dict[str, Any], recovery: dict[str, Any], *, capture_sha256: str) -> None: + """Keep only the existing claim/lease readback for this exact ready capture.""" + previous = state.get("host_execution_preparation") + if (not isinstance(previous, dict) or previous.get("state") != "ready" + or previous.get("source_sha256") != recovery.get("sha256")): + return + ready = previous.get("ready_capture") + request = recovery.get("execution_preparation") + if (not isinstance(ready, dict) or ready.get("sha256") != capture_sha256 + or not isinstance(request, dict) or request.get("turn_instance_id") != previous.get("turn_instance_id")): + return + previous["retired_context_readback"] = {key: request[key] + for key in ("goal_id", "agent_id", "todo_id", "turn_instance_id", "authority_route") if key in request} + + +def _retire_native_recovery_lookup(root: Path, state: dict[str, Any], *, + goal_id: str, agent_id: str, capture_sha256: str) -> None: + """Retire the current lookup for this work scope while its journal is locked.""" + recovery = state.get("recovery") + if (not isinstance(recovery, dict) or recovery.get("goal_id") != goal_id + or recovery.get("agent_id") != agent_id): + return + _preserve_preparation_readback(state, recovery, capture_sha256=capture_sha256) + for key in ("recovery", "source_recovery", "pending", "candidate"): + state.pop(key, None) + try: + _save(root, state) + except (OSError, ValueError, TypeError, RecursionError): + _discard_recovery(root) + raise + + +def retire_rejected_saved_context(*, directory: str | None, context_root: str | None, + goal_id: str, agent_id: str, capture_sha256: str, + hook_session: str | None) -> None: + """Retire the lookup after the saved file and identity have been checked.""" + from ..quota.error_codes import QuotaDecisionTransportError + + session = hook_session or os.environ.get("CODEX_THREAD_ID") + if not session: + return + try: + location = codex_context_directory(directory=directory, root=context_root, session=session) + if location is None: + return + with _locked(location) as root: + state = _state(root) + if state.get("session_id") == session and isinstance(state.get("cwd"), str) and state["cwd"]: + _retire_native_recovery_lookup(root, state, goal_id=goal_id, agent_id=agent_id, + capture_sha256=capture_sha256) + except FileNotFoundError: + return # There is no initialized journal at this location. + except (OSError, ValueError, TypeError, RecursionError) as exc: + raise QuotaDecisionTransportError( + "The saved source was rejected and its native recovery lookup could not be safely retired." + ) from exc + + +def _commit_native_context_recovery( + root: Path, state: dict[str, Any], reference: dict[str, Any], *, + scope: Mapping[str, str], source: dict[str, Any], + payload: dict[str, Any], capture_id: str, saved_capture: bool, + page_budget: tuple[int, int], +) -> None: + """Publish the original source and the active recovery view together.""" + from ...command_invocation import current_loopx_cli_argv + + preparation = compact_execution_preparation_request(payload) + if preparation: + session = scope["session_id"] + claim_source = None + if _original_native_claim_source(root, session, capture_id, bind_unattached=True): + claim_source = capture_id + elif not saved_capture and preparation.get("turn_instance_id"): + claim_source = capture_id if BARE_SHA256_PATTERN.fullmatch(capture_id) else reference["sha256"] + write_decision_capture(root / "claim-source.json", + {"session_id": session, "sha256": claim_source}, publish=os.replace) + if claim_source is not None: + preparation = compact_execution_preparation_request(payload, claim_source_sha256=claim_source) + page_bytes, output_bytes = page_budget + reference["command"] = context_capture_read_command(Path(reference["path"]), reference["sha256"], + page_bytes=page_bytes, max_page_output_bytes=output_bytes, reader_argv=current_loopx_cli_argv()) + state["source_recovery"] = source + state["recovery"] = {**{key: reference[key] for key in ("path", "sha256", "command")}, + "goal_id": scope["goal_id"], "agent_id": scope["agent_id"], + "total_bytes": Path(reference["path"]).stat().st_size, + **({"execution_preparation": preparation} if preparation else {})} + state["pending"] = {"path": reference["path"], "sha256": reference["sha256"], + "total_bytes": state["recovery"]["total_bytes"], "ranges": {}} + if isinstance(prepared := state.get("host_execution_preparation"), dict): + prepared.pop("retired_context_readback", None) + _save(root, state) + + +def _record_native_delivery(root: Path, state: dict[str, Any], result: dict[str, Any], + transaction: dict[str, Any], request: dict[str, Any], *, + directory: str, progressive: bool) -> dict[str, Any] | None: + """Record the validated delivery and its associated page observations.""" + recovery = result["recovery"] + read_plan = transaction.get("initial_read_plan") + captured_delta = result.get("continuity_basis") == "captured_file" and result.get("mode") == "delta" + if captured_delta: + result["baseline_reference"] = request["baseline_reference"] + if result.get("schema_version") == "context_delivery_v1": + state["recovery"].update(target=result["target"], continuity_basis="captured_file") + if captured_delta: + base = request["baseline_reference"] + state["recovery"]["comparison_base"] = { + "path": base["path"], "sha256": base["sha256"], "target": result["base"]} + read_sections = transaction["read_sections"] if progressive else {} + state["pending"].update(target=result["target"], ancestry=result["ancestry"], + full_read_sections=read_sections, read_sections=read_sections if result["mode"] == "full" else {}) + if plan := transaction.get("observed_first_read"): + from .native_first_read import observed_first_read_reference + + pinned = observed_first_read_reference(directory, result["target"], recovery, + transaction["context"], plan, read_plan) + if pinned is not None: + reference, opening = pinned + state["pending"]["observed_first_read"] = reference + if opening is not None: + read_plan = {**read_plan, "observed_opening": opening} + _save(root, state) + if read_plan is None: + return None + return {**read_plan, "capture": {key: recovery[key] for key in ("path", "sha256")}} + + def prepare_codex_context(payload: dict[str, Any], *, directory: str | None, capture_directory: Path, goal_id: str, agent_id: str, surface: str, capture_id: str, continuity_generation: str | None = None, @@ -764,13 +911,15 @@ def prepare_codex_context(payload: dict[str, Any], *, directory: str | None, cap Full fallback must recover this capture or refuse lookup, never an older one. Saved-file comparison is independent of any complete model-read observation. """ - from ...command_invocation import current_loopx_cli_argv from ..quota.error_codes import QuotaDecisionTransportError - from ..quota.turn_envelope import prepare_context_delivery - from ..effect_runtime import EffectRuntimeRemoteError, EffectRuntimeStartupError + from ..quota.turn_envelope import assert_complete_turn_decision, prepare_context_delivery + from ..effect_runtime import EffectRuntimeRejected, EffectRuntimeRemoteError, EffectRuntimeStartupError + check_saved_source = saved_capture and surface == "complete_turn" if directory is None: if not progressive_enabled: + if check_saved_source: + assert_complete_turn_decision(payload) return NativeContextPreparation(payload) from .decision_capture import capture_progressive_context @@ -785,6 +934,8 @@ def prepare_codex_context(payload: dict[str, Any], *, directory: str | None, cap native_session = hook_session or os.environ.get("CODEX_THREAD_ID") if (not isinstance(session, str) or not session or (native_session and session != native_session)): + if check_saved_source: + assert_complete_turn_decision(payload) return NativeContextPreparation(payload) # Do not replace another session's recovery. # Native lifecycle events own workspace identity and epoch. # A shell tool may run in another checkout without changing @@ -804,55 +955,28 @@ def prepare_codex_context(payload: dict[str, Any], *, directory: str | None, cap state.pop(key, None) state["read_scope"] = read_scope request = _native_context_request(state, scope, capture_id) + if check_saved_source: + # Snapshot a file baseline, then retire implicit recovery + # before any operation can fail or publish the new source. + _retire_native_recovery_lookup(root, state, goal_id=goal_id, agent_id=agent_id, + capture_sha256=capture_id) recovery = capture_context_projection( payload, directory=capture_directory, read_guidance=CONTEXT_CAPTURE_PAGE_GUIDANCE) - state["source_recovery"] = {key: recovery[key] for key in ("path", "sha256")} + source_recovery = {key: recovery[key] for key in ("path", "sha256")} state.pop("section_reads", None) # partial reads of an older source are not current coverage page_bytes, output_bytes = native_context_page_budget() - preparation = compact_execution_preparation_request(payload) - bound_claim_source = ( - capture_id - if preparation and _original_native_claim_source(root, session, capture_id, bind_unattached=True) - else None - ) def publish_recovery(reference: dict[str, Any]) -> None: - claim_source_sha256 = bound_claim_source - if ( - preparation - and claim_source_sha256 is None - and not saved_capture - and preparation.get("turn_instance_id") - ): - claim_source_sha256 = ( - capture_id - if BARE_SHA256_PATTERN.fullmatch(capture_id) is not None - else reference["sha256"] - ) - write_decision_capture( - root / "claim-source.json", - {"session_id": session, "sha256": claim_source_sha256}, - publish=os.replace, - ) - published_preparation = ( - compact_execution_preparation_request(payload, claim_source_sha256=claim_source_sha256) - if preparation and claim_source_sha256 is not None - else preparation - ) - reference["command"] = context_capture_read_command(Path(reference["path"]), reference["sha256"], - page_bytes=page_bytes, max_page_output_bytes=output_bytes, reader_argv=current_loopx_cli_argv()) - state["recovery"] = {**{key: reference[key] for key in ("path", "sha256", "command")}, - "goal_id": goal_id, "agent_id": agent_id, - "total_bytes": Path(reference["path"]).stat().st_size, - **({"execution_preparation": published_preparation} if published_preparation else {})} - state["pending"] = {"path": reference["path"], "sha256": reference["sha256"], - "total_bytes": state["recovery"]["total_bytes"], "ranges": {}} - _save(root, state) - - # Publish the current complete source before optional planning - # or encoding can be interrupted. Never recover its predecessor. - publish_recovery(recovery) - recovery_committed = True + nonlocal recovery_committed + _commit_native_context_recovery(root, state, reference, scope=scope, source=source_recovery, + payload=payload, capture_id=capture_id, saved_capture=saved_capture, + page_budget=(page_bytes, output_bytes)) + recovery_committed = True + + if not check_saved_source: + # Live and historical callers keep their original + # recovery checkpoint before optional planning. + publish_recovery(recovery) # Legacy direct native callers omitted the presentation option. progressive = progressive_enabled is not False presentation = progressive or selection_command_args_template is not None @@ -874,46 +998,32 @@ def publish_recovery(reference: dict[str, Any]) -> None: } if presentation else None, consumed_pointers=state.get("consumed_pointers"), observe_first_read=progressive) except (EffectRuntimeRemoteError, EffectRuntimeStartupError) as exc: - if captured_decision_path is not None and isinstance(payload.get("replan_read_path"), str): + if check_saved_source and isinstance(exc, EffectRuntimeRejected): + raise QuotaDecisionTransportError(str(exc)) from exc + if check_saved_source or (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 presentation and transaction["projected"]: + projected = presentation and transaction["projected"] + if projected: recovery = capture_context_projection(transaction["context"], directory=capture_directory, filename="required-context.json", read_guidance=CONTEXT_CAPTURE_PAGE_GUIDANCE) + if check_saved_source: + state.pop("candidate", None) + if check_saved_source or projected: publish_recovery(recovery) 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": - result["baseline_reference"] = request["baseline_reference"] - read_plan = transaction.get("initial_read_plan") - if result.get("schema_version") == "context_delivery_v1": - state["recovery"]["target"] = result["target"] - state["recovery"]["continuity_basis"] = "captured_file" - if result.get("continuity_basis") == "captured_file" and result.get("mode") == "delta": - base = request["baseline_reference"] - state["recovery"]["comparison_base"] = { - "path": base["path"], "sha256": base["sha256"], "target": result["base"]} - state["pending"].update(target=result["target"], ancestry=result["ancestry"], - full_read_sections=read_sections, - read_sections=read_sections if result["mode"] == "full" else {}) - if plan := transaction.get("observed_first_read"): - from .native_first_read import observed_first_read_reference - - pinned = observed_first_read_reference(directory, result["target"], recovery, - transaction["context"], plan, read_plan) - if pinned is not None: - reference, opening = pinned - state["pending"]["observed_first_read"] = reference - if opening is not None: - read_plan = {**read_plan, "observed_opening": opening} - _save(root, state) - if read_plan is not None: - read_plan = {**read_plan, "capture": {key: recovery[key] for key in ("path", "sha256")}} + read_plan = _record_native_delivery(root, state, result, transaction, request, + directory=directory, progressive=progressive) return NativeContextPreparation(result, read_plan) - except (OSError, ValueError, KeyError, TypeError, RecursionError): + except (OSError, ValueError, KeyError, TypeError, RecursionError) as exc: + if check_saved_source: + raise QuotaDecisionTransportError( + "Saved context presentation could not be prepared. Read the original capture and do not rerun admission." + ) from exc if not recovery_committed: _discard_recovery(root) return NativeContextPreparation(payload) diff --git a/loopx/control_plane/turn_driver/codex_execution_preparation.py b/loopx/control_plane/turn_driver/codex_execution_preparation.py index c57735fe99..7c109f832e 100644 --- a/loopx/control_plane/turn_driver/codex_execution_preparation.py +++ b/loopx/control_plane/turn_driver/codex_execution_preparation.py @@ -193,16 +193,26 @@ def codex_execution_preparation_readback_context(request: Mapping[str, Any]) -> and isinstance(ready_sha256, str) and BARE_SHA256_PATTERN.fullmatch(ready_sha256) is not None ): + from ..quota.turn_envelope import is_legacy_saved_quota_decision from .decision_capture import context_capture_read_command, read_context_capture_file try: - read_context_capture_file(ready_path, ready_sha256, private=True) + path, _, saved = read_context_capture_file(ready_path, ready_sha256, private=True) except (OSError, ValueError, KeyError, TypeError, RecursionError): pass else: - recovery_command = context_capture_read_command( - Path(ready_path), ready_sha256, reader_argv=current_loopx_cli_argv(), - ) + if is_legacy_saved_quota_decision(saved): + recovery_command = context_capture_read_command( + path, ready_sha256, reader_argv=current_loopx_cli_argv(), + ) + else: + # Re-open the exact ready source through its checked path. + # An outer digest alone does not validate a complete Turn. + recovery_command = shlex.join([ + *base, "quota", "context-project", "--capture-file", str(path), + "--capture-sha256", ready_sha256, "--goal-id", goal, "--agent-id", agent, + "--turn-instance-id", turn, "--context-delivery", "off", "--delta-delivery", "off", + ]) facts["capture_sha256"] = ready_sha256 facts["context"] = { "capture_sha256": ready_sha256, @@ -212,6 +222,7 @@ def codex_execution_preparation_readback_context(request: Mapping[str, Any]) -> "Automatic preparation started for this captured Turn and may still be finishing. " "Read the original claim receipt and current lease before deciding how to recover. " "With the matching active lease, follow the current captured same-Todo and same-Turn readiness check before work. " + "If the recovery command returns a projection, follow its reader after the projection succeeds. " "For an authorized retry of this claim, pass the original --claim-operation-id. " "A retired lease requires a fresh eligible capture. " "Receipt: " + json.dumps(facts, ensure_ascii=False, separators=(",", ":")) @@ -360,8 +371,8 @@ def codex_execution_preparation_context(receipt: Mapping[str, Any]) -> str: guidance = "" if isinstance(context, Mapping): if context.get("ok") is False: - guidance = ("Ready context delivery failed. Restore the private original capture and use its " - "recovery command before work. Retain the original claim reference. ") + guidance = ("Ready context delivery failed. Follow the recovery instructions in this receipt. " + "Use verified context for this Turn before work. Retain the original claim reference. ") else: guidance = ("Read the ready context in this receipt for current work and settlement, " "replacing the earlier pre-claim view. Retain its original claim reference " diff --git a/skills/loopx-self-repair/references/repair-patterns.md b/skills/loopx-self-repair/references/repair-patterns.md index 2a21221211..5753a1733b 100644 --- a/skills/loopx-self-repair/references/repair-patterns.md +++ b/skills/loopx-self-repair/references/repair-patterns.md @@ -14,6 +14,7 @@ teaches a reusable control-plane lesson. | `collaboration_return_format_projection_gap` | A complete answer that fits the result budget is rejected, or a published versioned return becomes unreadable or cannot replay through the real CLI. | Selected renderer output and Unicode count; persisted reply, compact publication/Inbox observations, original-route delivery and legacy replay at the same boundary. | Validation counted a different renderer before choosing the immutable format, or a compact codec omitted the persisted format marker and silently treated it as legacy. | Apply the unchanged budget to the actual selected format, carry its existing marker through every compact observation, and retain old bytes on replay. Validate BMP/non-BMP exact-limit and one-character-over cases, unsupported versions, real CLI publication/retry, original-audience return and File/SQLite packaged readback. Do not raise the budget, truncate the answer, rewrite historical receipts or add a second model. | | `typed_adapter_validation_scope_mismatch` | A local adapter type check passes while the same CI entry reports a large legacy dependency error set. | Exact source revisions, configured roots, cold same-environment runs, installation/import paths, and original-root plus adapter counterfactuals. | Adding runtime adapters to a kernel root list expands its import closure; an extra source path can mask that traversal locally. | Preserve the existing kernel gate and its import behavior. Make the additive adapter boundary explicit, retain imported-code debt as unqualified, and prove strict root checking with original diagnostics and core/adapter mutations in an unmasked source installation. Do not globally disable import checks, remove original roots or rewrite historical failures as passes. | | `saved_projection_goal_route_coupling` | A saved context projection fails on conflicting default runtime roots even with native transport disabled, while the same capture remains readable. | Run the shipped projection and returned reader against one digest-bound synthetic capture, with implicit, missing and global registry selectors; retain digest/identity refusals and unchanged runtime snapshots. | Shared CLI registry discovery runs before the saved-view owner validates its source. | Classify the saved presentation at the existing CLI route owner; preserve machine-configuration routing, live authority conflict checks and source/session validation. Reuse the registry-independence integration tests rather than adding a second recovery path or repairing live state. | +| `unverified_saved_context_recovery` | A saved complete Turn fails projection validation, but recovery guidance directs the Host to its body or the preceding Turn. | With a matching outer file digest, test each rejected version or inner binding after publishing a preceding Turn. Check implicit lookup and the compact hook before the next attempt, then follow the receipt-only diagnostic and compare the quota ledger. | A pre-verification rejection uses generic recovery or returns before retiring the native lookup. | Retire the same session, Goal and Agent lookup under the native journal lock. Direct the Host to the original receipt and preserve explicit capture reads, verified presentation fallback and legacy behavior. | | `native_reader_program_route_loss` | A native producer or read uses a selected CLI or interpreter, but returned navigation, continuation or recovery invokes another runtime. | Actual lifecycle capture, generated first-read and recovery instructions, unchanged execution of emitted commands with an isolated competing program, complete source/digest readback and installed-package provenance. | A presentation layer used its generic command-renderer default after the caller had already selected the runtime. | Pass the current program argv through existing CLI, navigation, bounded-result and reader renderers; retain constructor defaults and shell quoting. Include emitted command metadata in native byte-cost estimates. Cover cold guard and saved-context projection before direct first reads, raw/selected/index pagination, quoted aliases, refusal details and source-state immutability through real CLI and isolated wheel entrypoints. Routing preserves the selected program; it does not pin an installation version or prove Host adoption. | | `host_bridge_reader_orphan` | A rejected bridge observation leaves its reader thread and output pipe alive after the child has exited. | Original framing error, cleanup result, child exit, bounded-queue backpressure and reader/pipe state across repeated failures. | Cancellation rejects another malformed frame; the sole consumer leaves while the producer blocks on a full queue, and joining alone cannot release it. | Stop publication, release the waiting producer, join and close through the existing transport helper. Retain cleanup-unconfirmed results and original Host drain ownership; reclaiming a reader does not prove Host exit. Cover malformed JSON and invalid typed observations with real subprocesses, plus existing real Host cancellation and success paths. Do not increase timeouts or fabricate a successful result. | | `codex_strict_result_schema_gap` | A Codex Turn fails with invalid_json_schema before execution, or a strict-schema stop result fails internal evidence validation. | Generated Codex schema, required properties, unsupported conditional keywords, and both CLI/app-server result boundaries. | Optional evidence and conditional schema requirements conflict with strict Structured Outputs; internal stop results prohibit evidence fields. | Require nullable evidence fields on the Codex wire and remove unsupported conditionals. Normalize only stop null placeholders in both Codex transports; reject missing/null profile material evidence and preserve non-null evidence for the existing validator. Cover legacy absence, invalid arrays and an isolated schema-acceptance probe without changing other hosts. | diff --git a/tests/control_plane/test_quota_execution_preparation.py b/tests/control_plane/test_quota_execution_preparation.py index b3804d7e17..a394b4748e 100644 --- a/tests/control_plane/test_quota_execution_preparation.py +++ b/tests/control_plane/test_quota_execution_preparation.py @@ -473,7 +473,7 @@ def test_native_hook_returns_readable_ready_source_and_original_claim( @pytest.mark.parametrize("provider", ["file", "sqlite"]) -@pytest.mark.parametrize("replace_stage", ["guard", "record"]) +@pytest.mark.parametrize("replace_stage", ["guard", "record", "version"]) def test_prepared_hook_refuses_a_replaced_ready_capture_without_reclaim( tmp_path, monkeypatch, capsys, provider, replace_stage, ): @@ -504,11 +504,22 @@ def replace_capture(): source.write_bytes(verified["bytes"] + b" ") def observe_guard(command, **kwargs): + prepared = "--prepared-capture-reference" in command + if prepared and replace_stage == "version": + command = [*command, "--complete-turn"] result = run(command, **kwargs) - if "--prepared-capture-reference" in command: + if prepared: verified["writer"] = json.loads(result.stdout) if replace_stage == "guard": replace_capture() + elif replace_stage == "version": + source, = native.glob("prepared-*/ready/decision.json") + payload = json.loads(source.read_bytes()) + assert payload["schema_version"] == "loopx_turn_envelope_v1" + payload["schema_version"] = "loopx_turn_envelope_v999" + source.write_text(json.dumps(payload)) + verified["writer"]["capture_sha256"] = hashlib.sha256(source.read_bytes()).hexdigest() + result.stdout = json.dumps(verified["writer"]) return result monkeypatch.setattr(codex_execution_preparation, "_default_runner", observe_guard) @@ -535,8 +546,41 @@ def hook(tool_response=None): assert receipt["claim_operation_id"].startswith("todo-claim:") context = receipt["context"] assert context["ok"] is False - assert "digest mismatch" in context["reason"] recovery = shlex.split(context["recovery_command"]) + if replace_stage == "version": + assert "Unsupported" in context["reason"] + assert context["recovery_command"] == context["recovery"]["command"] + assert [recovery[index + 1] for index, word in enumerate(recovery) if word == "--select"] == ["/heartbeat_receipt"] + log = tmp_path / "runtime" / "goals" / GOAL / "rollout-event-log.jsonl" + after_claim = log.read_bytes() + diagnostic = subprocess.run(recovery, cwd=tmp_path, check=True, text=True, capture_output=True) + diagnostic_page = json.loads(diagnostic.stdout) + assert json.loads(diagnostic_page["content"])["selections"] == [ + {"pointer": "/heartbeat_receipt", "value": context["heartbeat_receipt"]}, + ] + # Even reading the rejected file explicitly does not observe ready work context. + select_at = recovery.index("--select") + raw_reader = recovery[:select_at] + recovery[select_at + 2:] + raw = subprocess.run(raw_reader, cwd=tmp_path, check=True, text=True, capture_output=True) + raw_page = json.loads(raw.stdout) + assert raw_page["sha256"] == context["capture_sha256"] + for page in (diagnostic_page, raw_page): + response = hook(page) + readback = json.loads(response["hookSpecificOutput"]["additionalContext"].rsplit("Receipt: ", 1)[1]) + assert readback["event"] == "host_execution_preparation_readback_required" + assert readback["claim_operation_id"] == receipt["claim_operation_id"] + implicit = subprocess.run([*raw_reader[:raw_reader.index("quota")], "quota", "context-read", + "--codex-context-dir", str(native), "--goal-id", GOAL, "--agent-id", AGENT], + cwd=tmp_path, text=True, capture_output=True) + assert implicit.returncode == 1 and json.loads(implicit.stdout)["ok"] is False + retry = shlex.split(readback["context"]["recovery_command"]) + assert "context-project" in retry and "--turn-instance-id" in retry + refusal = subprocess.run(retry, cwd=tmp_path, text=True, capture_output=True) + error = json.loads(refusal.stdout) + assert refusal.returncode == 1 and error["heartbeat_receipt"] == context["heartbeat_receipt"] + assert "content" not in error and log.read_bytes() == after_claim + return + assert "digest mismatch" in context["reason"] digest = hashlib.sha256(verified["bytes"]).hexdigest() writer = verified["writer"] assert writer["capture_sha256"] == digest and writer["execution_preparation"] == receipt["execution_preparation"] diff --git a/tests/control_plane/test_saved_context_project.py b/tests/control_plane/test_saved_context_project.py index 300c0fcdbc..2569d6c73f 100644 --- a/tests/control_plane/test_saved_context_project.py +++ b/tests/control_plane/test_saved_context_project.py @@ -62,6 +62,134 @@ def test_saved_projection_checks_original_before_any_publication(tmp_path, failu assert hashlib.sha256(path.read_bytes()).hexdigest() == digest +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("route", ["stateless", "native", "delta-only", "raw", "pre-session"]) +def test_saved_complete_quota_projection_rejects_invalid_source_without_legacy_fallback( + tmp_path, monkeypatch, provider, route): + from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime + from tests.control_plane.test_quota_settlement_cli import ( + AGENT_ID, GOAL_ID, TODO_ID, TURN_ID, _heartbeat_receipt_count, + _initialize_fixture_authority, _spend_run_count, _write_fixture, + ) + from loopx.control_plane.turn_driver.codex_context import codex_context_hook + + isolate_sqlite_runtime(tmp_path, monkeypatch) + project, runtime, registry = _write_fixture(tmp_path) + _initialize_fixture_authority(project, runtime, provider=provider) + binding = ["--goal-id", GOAL_ID, "--agent-id", AGENT_ID, "--turn-instance-id", TURN_ID] + directory = tmp_path / "capture" + code, result = _run_cli(registry, runtime, "quota", "should-run", *binding, "--todo-id", TODO_ID, + "--complete-turn", "--context-delivery", "off", "--decision-output-dir", str(directory), cwd=project) + assert code == 0, result + original = directory / "decision.json" + raw = original.read_bytes() + signed = json.loads(raw) + assert signed["schema_version"] == "loopx_turn_envelope_v1" + receipt = signed["heartbeat_receipt"] + + options = ["--context-delivery", "off" if route in {"raw", "delta-only"} else "auto"] + if route in {"stateless", "native", "pre-session"}: + options += ["--initial-read", "--max-page-output-bytes", "12000"] + journal = tmp_path / "native" + session = "source-validation" + event = {"hook_event_name": "SessionStart", "session_id": session, "cwd": str(project)} + active_native = route in {"native", "delta-only"} + if route in {"native", "delta-only", "pre-session"}: + journal.mkdir(mode=0o700) + monkeypatch.setenv("CODEX_THREAD_ID", session) + if active_native: + codex_context_hook(str(journal), {**event, "source": "startup"}) + options += ["--codex-app", "--codex-context-dir", str(journal)] + if route == "delta-only": + options += ["--delta-delivery", "auto"] + + def project_capture(path, *, turn=TURN_ID, goal=GOAL_ID, agent=AGENT_ID, digest=None, delivery=options): + return _run_cli(registry, runtime, "quota", "context-project", "--goal-id", goal, "--agent-id", agent, + "--turn-instance-id", turn, *delivery, "--capture-file", str(path), + "--capture-sha256", digest or hashlib.sha256(path.read_bytes()).hexdigest(), cwd=project) + + lookup = ("quota", "context-read", "--codex-context-dir", str(journal), + "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, "--select", "/heartbeat_receipt") + prior_turn = "preceding-turn" + prior_receipt = {"turn_instance_id": prior_turn} + prior = directory / "preceding.json" + prior_bytes = json.dumps({"ok": True, "goal_id": GOAL_ID, "agent_identity": {"agent_id": AGENT_ID}, + "heartbeat_receipt": prior_receipt, "requirement": "This requirement belongs to the preceding Turn."}).encode() + if active_native: + prior.write_bytes(prior_bytes) + + def assert_preceding_recovery(): + code, page = _run_cli(registry, runtime, *lookup, cwd=project) + assert code == 0 and page["eof"], page + assert json.loads(page["content"])["selections"] == [ + {"pointer": "/heartbeat_receipt", "value": prior_receipt}, + ] + + code, valid = project_capture(original) + assert code == 0 and valid["ok"], valid + for version, reason in ( + ("loopx_turn_envelope_v2", "Unsupported"), ("loopx_turn_envelope_v0", "mixed"), + (None, "Unsupported"), ("loopx_turn_envelope_v1", "integrity mismatch"), + ): + bad = {**signed, "schema_version": version} + if version is None: + del bad["schema_version"] # The retained binding cannot become an unversioned packet. + elif version == "loopx_turn_envelope_v1": + bad["contract_binding"] = {**signed["contract_binding"], "content_hash": "sha256:" + "0" * 64} + capture = directory / "rejected.json" + capture.write_text(json.dumps(bad)) + digest = hashlib.sha256(capture.read_bytes()).hexdigest() + if active_native: + code, preceding = project_capture(prior, turn=prior_turn, + delivery=[arg for arg in options if arg != "--initial-read"]) + assert code == 0 and preceding["ok"], preceding + assert_preceding_recovery() + if version == "loopx_turn_envelope_v2": + assert project_capture(capture, digest="0" * 64)[0] == 1 + assert project_capture(capture, turn=prior_turn)[0] == 1 + disabled = ["--context-delivery", "off", "--delta-delivery", "off", + "--codex-app", "--codex-context-dir", str(journal)] + assert project_capture(capture, delivery=disabled)[0] == 1 + with monkeypatch.context() as other: + other.setenv("CODEX_THREAD_ID", "different-session") + assert project_capture(capture)[0] == 1 + if version in {"loopx_turn_envelope_v2", "loopx_turn_envelope_v1"}: + foreign = directory / "foreign.json" + foreign.write_text(json.dumps({**bad, "goal_id": "different-goal"})) + assert project_capture(foreign, goal="different-goal")[0] == 1 + foreign.write_text(json.dumps({**bad, "agent_identity": {"agent_id": "different-agent"}})) + assert project_capture(foreign, agent="different-agent")[0] == 1 + assert_preceding_recovery() + code, rejected = project_capture(capture) + assert code == 1 and rejected["ok"] is False, rejected + assert rejected["error_code"] == "quota_decision_transport_failed" and reason in rejected["reason"] + assert "content" not in rejected and "first_read_command" not in rejected + assert rejected["heartbeat_receipt"] == receipt + assert rejected["recovery"]["path"] == str(capture) and rejected["recovery"]["sha256"] == digest + recovery = shlex.split(rejected["recovery"]["command"]) + assert "--list" not in recovery and "--initial-read" not in recovery + assert "diagnostic" in rejected["instruction"] and "retry context-project" in rejected["instruction"] + rc, page = _run_cli(registry, runtime, *recovery[recovery.index("quota"):], cwd=project) + assert rc == 0 and page["eof"] + assert json.loads(page["content"])["selections"] == [{"pointer": "/heartbeat_receipt", "value": receipt}] + if active_native: + code, implicit = _run_cli(registry, runtime, *lookup, cwd=project) + assert code == 1 and implicit["ok"] is False, implicit + assert not codex_context_hook(str(journal), {**event, "source": "compact"}) + journal_state = journal / "session.json" + if active_native: + state = json.loads(journal_state.read_text()) + assert "recovery" not in state and "pending" not in state + code, explicit = _run_cli(registry, runtime, "quota", "context-read", + "--capture-file", str(prior), "--capture-sha256", hashlib.sha256(prior_bytes).hexdigest(), cwd=project) + assert code == 0 and explicit["content"].encode() == prior_bytes + assert project_capture(original)[0] == 0 + else: + assert not journal_state.exists() + assert original.read_bytes() == raw + assert _heartbeat_receipt_count(runtime, TURN_ID) == 1 and _spend_run_count(runtime) == 0 + + @pytest.mark.parametrize("root_kind", ["unsafe", "symlink", "missing_parent"]) @pytest.mark.parametrize(("opening_args", "recovery_limit"), [ ([], 64000), (["--initial-read"], 64000), @@ -396,7 +524,7 @@ def refuse_recovery(payload, **kwargs): assert not registry.exists() and not runtime.exists() -def test_preflight_instruction_pins_both_projection_and_full_fallback(tmp_path): +def test_preflight_instruction_pins_projection_and_diagnostic_receipt(tmp_path): from benchmark.runtime.context_delivery import preflight_context_instruction from loopx.control_plane.turn_driver.decision_capture import CONTEXT_CAPTURE_NAVIGATION_GUIDANCE @@ -418,6 +546,9 @@ def test_preflight_instruction_pins_both_projection_and_full_fallback(tmp_path): 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] + assert commands[1][commands[1].index("--select") + 1:] == ["/heartbeat_receipt"] + assert "--list" not in commands[1] + assert "retry the same context-project" in body @pytest.mark.parametrize("provider", ["file", "sqlite"]) @@ -1327,6 +1458,44 @@ def finish(page, *, limit=None, content_limit=None): assert not registry.exists() and not runtime.exists() +def test_stateless_saved_initial_read_bounds_the_default_cli_output(tmp_path): + import subprocess + + requirement = "Preserve the original acceptance and stop. 保留原要求。\n" * 850 + receipt = {"event_id": "original-receipt", "turn_instance_id": "original-turn"} + source = (tmp_path / "original.json").resolve() + raw = json.dumps({"ok": True, "goal_id": "g", "agent_identity": {"agent_id": "a"}, + "heartbeat_receipt": receipt, "effective_action": "normal_run", + "interaction_contract": {"schema_version": "loopx_interaction_contract_v0", "agent_channel": { + "work_context": {"sources": [{"source": "goal_state", "ordering": "before_work", + "content": {"text": requirement, "state_file": "goal.md"}}]}}}}, ensure_ascii=False).encode() + source.write_bytes(raw) + registry, runtime = tmp_path / "absent-registry.json", tmp_path / "runtime" + base = [sys.executable, "-m", "loopx.entrypoint", "--registry", str(registry), + "--runtime-root", str(runtime), "--format", "json"] + argv = ["quota", "context-project", "--context-delivery", "auto", "--delta-delivery", "off", + "--initial-read", "--capture-file", str(source), "--capture-sha256", hashlib.sha256(raw).hexdigest(), + "--goal-id", "g", "--agent-id", "a", "--turn-instance-id", receipt["turn_instance_id"]] + body = bytearray() + for _ in range(16): + result = subprocess.run([*base, *argv], cwd=tmp_path, capture_output=True, check=True) + assert len(result.stdout) <= 12000 + page = json.loads(result.stdout) + assert page["ok"] and page["offset"] == len(body) + body.extend(page["content"].encode()) + if page["eof"]: + assert page["total_bytes"] == len(body) + break + reader = shlex.split(page["next_command"]) + argv = reader[reader.index("quota"):] + else: + pytest.fail("stateless first read did not complete within its bounded fixture") + chosen = {row["pointer"]: row["value"] for row in json.loads(body)["selections"]} + assert chosen["/heartbeat_receipt"] == receipt + assert chosen["/interaction_contract/agent_channel/work_context/sources/0/content/text"] == requirement + assert source.read_bytes() == raw and not registry.exists() and not (runtime / "goals").exists() + + def test_attached_selected_page_uses_the_same_window_as_its_reader(tmp_path, monkeypatch): from loopx.control_plane.turn_driver.codex_context import codex_context_hook