From 9ca510d660e7e5d1ac07007a995d8f0f2032f70c Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 5 Nov 2026 12:58:32 +0800 Subject: [PATCH] feat(quota): return recoverable receipts for settled refreshes Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- docs/reference/protocols/turn-envelope-v0.md | 10 + .../project_lifecycle_refresh_state.py | 19 +- loopx/cli_commands/quota_capture.py | 17 +- loopx/cli_commands/refresh_settlement.py | 197 ++++++++++++++---- loopx/control_plane/quota/settlement.py | 3 + .../quota/settlement_readback.ts | 40 ++-- .../turn_driver/decision_capture.py | 15 +- .../control_plane/test_refresh_settle_turn.py | 58 ++++++ .../quota_settlement_readback.test.ts | 15 ++ 9 files changed, 301 insertions(+), 73 deletions(-) diff --git a/docs/reference/protocols/turn-envelope-v0.md b/docs/reference/protocols/turn-envelope-v0.md index 65a3485e0d..35275f91e3 100644 --- a/docs/reference/protocols/turn-envelope-v0.md +++ b/docs/reference/protocols/turn-envelope-v0.md @@ -270,6 +270,12 @@ operation also runs the owed command there, using the originating CLI's implementation. The normal workspace and settlement guards still apply; a missing or incompatible workspace leaves the original owed command recoverable. +Callers that need a compact result can add `--settlement-receipt` to +`refresh-state --settle-turn`. A successful settled result provides its exact +identity and receipts inline and a `result_details.command` for the full saved +result. Inspect unresolved controls before continuing. If saving the view fails +after commitment, retry the same Turn and check its receipts. + An in-flight recommendation does not require keeping a completed Todo open. Validate its original declaration, use ordinary completion when satisfied, and write semantic closeout with an evidence-linked vision decision on the original @@ -292,6 +298,10 @@ Todo 或 obligation/Turn 的 canonical 待结算命令并验证回执；不重 继续使用原调用方的 CLI 实现。原工作区和结算校验保持；目录缺失或不兼容时保留 原待结算命令以便恢复。 +需要紧凑回执时，在 `refresh-state --settle-turn` 后加 `--settlement-receipt`。 +结算成功后可直接核对本次身份和回执，用 `result_details.command` 读取完整结果。 +先处理未解决的控制项。若结算已提交而视图保存失败，按原 Turn 重试并核对回执。 + 执行中续跑建议不要求把已完成的 Todo 保持开放。原声明验收满足后可正常完成，并 在原 Turn 上按语义收尾写入有证据的 vision；未完成的阶段进展仍可保留开放状态。 新 Turn 尊重上轮明确收尾或结构化 `path_delta.outcome=replan`，放开旧续跑偏好， diff --git a/loopx/cli_commands/project_lifecycle_refresh_state.py b/loopx/cli_commands/project_lifecycle_refresh_state.py index c5057ff3ec..e34037aa89 100644 --- a/loopx/cli_commands/project_lifecycle_refresh_state.py +++ b/loopx/cli_commands/project_lifecycle_refresh_state.py @@ -410,6 +410,10 @@ def register_refresh_state_command( "--settle-turn", action="store_true", help="After verified writeback, execute the owed quota step once for this exact Turn; replay never creates another debit.", ) + refresh_state_parser.add_argument( + "--settlement-receipt", action="store_true", + help="With --settle-turn, retain successful supporting snapshots in an immutable result and print controls plus exact receipts.", + ) refresh_state_parser.add_argument( "--no-global-sync", action="store_true", @@ -484,6 +488,8 @@ def handle_refresh_state_command( outcome_update_result = None goal_acceptance_preview = None try: + if getattr(args, "settlement_receipt", False) and (not args.settle_turn or fmt != "json"): + raise ValueError("--settlement-receipt requires --settle-turn and --format json") if getattr(args, "settle_turn", False) and not ( args.agent_id and args.turn_instance_id and bool(args.todo_id) != bool(args.replan_obligation_id) ): @@ -975,7 +981,7 @@ def handle_refresh_state_command( def _settle_requested_refresh(payload, args, registry_path, goal_ref): if not getattr(args, "settle_turn", False): return - from .refresh_settlement import settle_refreshed_turn + from .refresh_settlement import present_settled_refresh, settle_refreshed_turn settle_refreshed_turn(payload, registry_path=registry_path, runtime_root=resolve_runtime_root(load_registry(registry_path), args.runtime_root), @@ -986,3 +992,14 @@ def _settle_requested_refresh(payload, args, registry_path, goal_ref): Path(args.delivery_workspace_path) if args.delivery_workspace_path else None )) + if getattr(args, "settlement_receipt", False): + from ..control_plane.quota.error_codes import QuotaDecisionTransportError + try: + view = present_settled_refresh(payload, runtime_root=resolve_runtime_root( + load_registry(registry_path), args.runtime_root)) + except QuotaDecisionTransportError as exc: + payload.update(ok=False, error=str(exc)) + return # Exact committed receipts stay visible; never re-admit. + if view is not payload: + payload.clear() + payload.update(view) diff --git a/loopx/cli_commands/quota_capture.py b/loopx/cli_commands/quota_capture.py index 9c598223b0..bfc41b3a7a 100644 --- a/loopx/cli_commands/quota_capture.py +++ b/loopx/cli_commands/quota_capture.py @@ -851,6 +851,7 @@ def capture_decision(directory: Path | None, payload: dict[str, object]) -> None def budgeted_settlement_result( payload: dict[str, object], *, runtime_root: Path, output_bytes: int, reader_argv: list[str] | None = None, + supporting_fields: tuple[str, ...] | None = None, ) -> dict[str, object]: """Present one evaluated result with its complete immutable diagnostic. @@ -864,7 +865,7 @@ def fits(value: dict[str, object]) -> bool: # Match the JSON CLI serializer, including print's newline. return len(json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= output_bytes - if fits(payload): + if supporting_fields is None and fits(payload): return payload try: uid = private_capture_uid() @@ -876,20 +877,21 @@ def fits(value: dict[str, object]) -> bool: raise ValueError("result capture requires a private physical directory owned by this user") directory = Path(tempfile.mkdtemp(prefix="settlement-", dir=root)) saved = directory / "result.json" - write_decision_capture(saved, payload) - raw = saved.read_bytes() - if len(raw) > MAX_REQUEST_BYTES: + digest = write_decision_capture(saved, payload) + if saved.stat().st_size > MAX_REQUEST_BYTES: raise ValueError(f"complete result exceeds the existing reader limit: {saved}") - digest = hashlib.sha256(raw).hexdigest() reference = write_context_read_reference(saved, digest) details = {"reference_file": str(reference), "command": context_capture_read_command(saved, digest, reference_file=reference, page_bytes=min(NATIVE_CONTEXT_PAGE_BYTES, output_bytes), max_page_output_bytes=output_bytes, reader_argv=reader_argv)} - deferred = [key for key in ("before", "after") if key in payload] + deferred = [key for key in (supporting_fields if supporting_fields is not None else ("before", "after")) if key in payload] view = {key: value for key, value in payload.items() if key not in deferred} view["result_details"] = {**details, "deferred_fields": deferred} view["instruction"] = ( + "The exact Turn is settled. Supporting snapshots are complete in result_details; read only needed details. " + "Keep every returned warning and unresolved control. No additional spend or admission is required for this result." + ) if supporting_fields is not None else ( "The operation was already evaluated. before/after are complete in the immutable result; " "read only needed details with context-read --select. Preserve the original Turn and effect " "identity during recovery; do not repeat admission or infer authority from this presentation." @@ -905,7 +907,8 @@ def fits(value: dict[str, object]) -> bool: max_page_output_bytes=output_bytes, reader_argv=reader_argv), "instruction": "Read the complete selected result controls before acting. The operation was " "already evaluated; preserve its original Turn/effect identity. No execution authority " - "or context baseline is granted. before/after remain available in the same immutable result."} + "or context baseline is granted. " + ("Supporting snapshots remain available in the same immutable result." + if supporting_fields is not None else "before/after remain available in the same immutable result.")} if fits(view): return view raise ValueError(f"result reader metadata exceeds the output budget; complete result: {saved}") diff --git a/loopx/cli_commands/refresh_settlement.py b/loopx/cli_commands/refresh_settlement.py index 845bf1c4fb..53fa3e33cf 100644 --- a/loopx/cli_commands/refresh_settlement.py +++ b/loopx/cli_commands/refresh_settlement.py @@ -33,63 +33,69 @@ def settle_refreshed_turn( if payload.get("ok") is not True or payload.get("dry_run"): return owed = payload.get("settlement_owed") - if not isinstance(owed, dict): - return # Includes settled replay, missing checkpoints and no-spend closeout. + settled_replay = not isinstance(owed, dict) and payload.get("settlement_progress", {}).get("state") == "settled" + if not isinstance(owed, dict) and not settled_replay: + return # Missing checkpoints and typed no-spend closeout. identity = payload.get("settlement_identity") expected = SettlementIdentity(goal_id=goal_id, agent_id=agent_id, todo_id=todo_id, replan_obligation_id=replan_obligation_id, turn_instance_id=turn_instance_id).as_dict() - if (owed.get("schema_version") != "turn_settlement_owed_v0" or owed.get("kind") != "quota_spend" - or not isinstance(identity, dict) - or any(identity.get(k) != v or owed.get(k) != v for k, v in expected.items() if k != "schema_version")): + if (not isinstance(identity, dict) + or any(identity.get(k) != v for k, v in expected.items() if k != "schema_version") + or isinstance(owed, dict) and (owed.get("schema_version") != "turn_settlement_owed_v0" + or owed.get("kind") != "quota_spend" + or any(owed.get(k) != v for k, v in expected.items() if k != "schema_version"))): payload.update(ok=False, error="refresh settlement command does not match its original identity") return error = None spent: dict[str, Any] = {} - try: - workspace = delivery_workspace_path.expanduser().resolve() if delivery_workspace_path else None - command = shlex.split(owed["command"]) - # Validate the owner-rendered argv through the actual CLI parser too: - # a projector bug must not debit a different valid Turn before readback. - from ..cli import build_parser - args = build_parser().parse_args(["--format", "json", *command[1:]]) - if (args.command != "quota" or args.quota_command != "spend-slot" - or not args.execute or args.slots != 1 - or Path(args.registry).expanduser().resolve() != registry_path.resolve() - or Path(args.runtime_root).expanduser().resolve() != runtime_root.resolve() - or args.source != payload["settlement_progress"]["quota_spend_source"] - or getattr(args, "goal_instance_id", None) != (goal_ref or {}).get("goal_instance_id") - or any(getattr(args, key, None) != value for key, value in ( - ("goal_id", goal_id), ("agent_id", agent_id), ("todo_id", todo_id), - ("replan_obligation_id", replan_obligation_id), ("turn_instance_id", turn_instance_id)))): - raise ValueError("owed quota command does not preserve the exact original binding") - # Keep the validated producing cwd for accounting, but pin imports to - # this CLI's package: the delivered worktree may contain another LoopX. - # -P excludes the child cwd from Python's initial module search path. - # Retain the complete CLI guards, hooks and rollout receipts; neither - # ambient PATH nor a lower-level debit may replace that adapter. - entry = ( - "import sys; sys.path.insert(0, sys.argv.pop(1)); " - "import runpy; runpy.run_module('loopx.cli', run_name='__main__')" - ) - result = subprocess.run( - [sys.executable, "-P", "-c", entry, str(Path(__file__).resolve().parents[2]), - "--format", "json", *command[1:]], - cwd=workspace, - capture_output=True, text=True, encoding="utf-8", timeout=60, check=False, - ) - value = json.loads(result.stdout) - if not isinstance(value, dict): - raise ValueError("quota settlement CLI returned a non-object result") - spent = value - if result.returncode or spent.get("ok") is not True: - error = str(spent.get("error") or spent.get("reason") or "quota settlement CLI failed") - except (OSError, RuntimeError, ValueError, TypeError, KeyError, subprocess.TimeoutExpired, SystemExit) as exc: - error = str(exc) + # A settled refresh replay still needs all three receipts below, but no child. + if not settled_replay: + try: + workspace = delivery_workspace_path.expanduser().resolve() if delivery_workspace_path else None + command = shlex.split(owed["command"]) + # Validate the owner-rendered argv through the actual CLI parser too: + # a projector bug must not debit a different valid Turn before readback. + from ..cli import build_parser + args = build_parser().parse_args(["--format", "json", *command[1:]]) + if (args.command != "quota" or args.quota_command != "spend-slot" + or not args.execute or args.slots != 1 + or Path(args.registry).expanduser().resolve() != registry_path.resolve() + or Path(args.runtime_root).expanduser().resolve() != runtime_root.resolve() + or args.source != payload["settlement_progress"]["quota_spend_source"] + or getattr(args, "goal_instance_id", None) != (goal_ref or {}).get("goal_instance_id") + or any(getattr(args, key, None) != value for key, value in ( + ("goal_id", goal_id), ("agent_id", agent_id), ("todo_id", todo_id), + ("replan_obligation_id", replan_obligation_id), ("turn_instance_id", turn_instance_id)))): + raise ValueError("owed quota command does not preserve the exact original binding") + # Keep the validated producing cwd for accounting, but pin imports to + # this CLI's package: the delivered worktree may contain another LoopX. + # -P excludes the child cwd from Python's initial module search path. + # Retain the complete CLI guards, hooks and rollout receipts; neither + # ambient PATH nor a lower-level debit may replace that adapter. + entry = ( + "import sys; sys.path.insert(0, sys.argv.pop(1)); " + "import runpy; runpy.run_module('loopx.cli', run_name='__main__')" + ) + result = subprocess.run( + [sys.executable, "-P", "-c", entry, str(Path(__file__).resolve().parents[2]), + "--format", "json", *command[1:]], + cwd=workspace, + capture_output=True, text=True, encoding="utf-8", timeout=60, check=False, + ) + value = json.loads(result.stdout) + if not isinstance(value, dict): + raise ValueError("quota settlement CLI returned a non-object result") + spent = value + if result.returncode or spent.get("ok") is not True: + error = str(spent.get("error") or spent.get("reason") or "quota settlement CLI failed") + except (OSError, RuntimeError, ValueError, TypeError, KeyError, subprocess.TimeoutExpired, SystemExit) as exc: + error = str(exc) try: readback = read_heartbeat_settlement( runtime_root, registry_path=registry_path, goal_ref=goal_ref, goal_id=goal_id, agent_id=agent_id, todo_id=todo_id, replan_obligation_id=replan_obligation_id, turn_instance_id=turn_instance_id, + include_record_bodies=False, ) except Exception as exc: payload["ok"] = False @@ -106,9 +112,108 @@ def settle_refreshed_turn( "reward_memory_ingest", "error", "error_code", ) if key in spent} payload["settlement_result"] = settlement_result_payload(readback.settlement) - if readback.settlement.failure is not None or spent and spent.get("ok") is not True: + if payload["settlement_progress"]["state"] == "spend_required": + payload.update(ok=False, error=error or "exact quota receipt is uncommitted; recover the returned original settlement_owed") + elif readback.settlement.failure is not None or spent and spent.get("ok") is not True: payload["ok"] = False payload["error"] = error or readback.settlement.failure.reason elif error: # Receipt readback, not child transport success, proves commitment. payload["quota_spend"]["transport_warning"] = error + + +def present_settled_refresh(payload: dict[str, Any], *, runtime_root: Path) -> dict[str, Any]: + """Keep controls/receipts inline; retain successful diagnostic snapshots. + + This opt-in output adapter runs after the original owners and exact receipt + readback. Errors, pending recovery and unknown fields remain fully visible. + It neither changes a result nor certifies model-context retention. + """ + progress = payload.get("settlement_progress") + if (payload.get("ok") is not True or payload.get("dry_run") + or not isinstance(progress, dict) or progress.get("state") != "settled" + or payload.get("settlement_owed") is not None): + return payload + from .quota_capture import budgeted_settlement_result + from ..command_invocation import current_loopx_cli_argv + from ..control_plane.turn_driver.native_context_profile import NATIVE_CONTEXT_PAGE_OUTPUT_BYTES + + supporting = ["registry", "runtime_root", "project", "json_path", "markdown_path", "index_path", + "health_check", "state", "registry_goal", "todo_source"] + # Only fully delivered projection diagnostics are supporting material. + # Failed, pending, unknown or warning-bearing owners stay inline unchanged. + outbox = payload.get("projection_outbox") + if (isinstance(outbox, dict) and outbox.get("schema_version") == "loopx_todo_projection_delivery_v0" + and set(outbox) <= {"schema_version", "provider_revision", "observed_provider_revision", "status", + "next_action", "delivery_attempts", "source", "goal_id", "state_file", "project", "source_authority", + "todo_count", "changed", "executed", "source_sha256", "rendered_sha256", "narrative_sha256", + "section_record_sha256", "active_fields_omitted", "omitted_active_records", "parse_render_parity", + "narrative_preserved", "legacy_fallback_used"} + and outbox.get("status") in {"delivered", "current"} and outbox.get("next_action") == "finish" + and outbox.get("parse_render_parity") is True and outbox.get("narrative_preserved") is True + and outbox.get("legacy_fallback_used") is False): + supporting.append("projection_outbox") + # These are completed IO snapshots from the existing typed owners. Keep + # their complete source in the same result capture, not another RPC or + # state/permission decision. Unknown keys and nonterminal results stay inline. + route = payload.get("runtime_projection_route") + if (isinstance(route, dict) and set(route) <= {"schema_version", "route_id", "status", + "projection_required", "declaration_source", "match_count", "source_mirror_match_count", + "conflict_count", "unreadable_target_count", "source_registry_sha256_16", + "source_runtime_sha256_16", "target_runtime_sha256_16", "projection_enabled", + "projection_marker_field"} + and route.get("schema_version") == "runtime_projection_route_v0" + and route.get("status") == "single_runtime" and route.get("declaration_source") == "source_runtime_fallback" + and route.get("projection_required") is False + and route.get("projection_enabled") is False + and all(type(route.get(key)) is int and route[key] == 0 for key in ( + "match_count", "source_mirror_match_count", "conflict_count", "unreadable_target_count"))): + supporting.append("runtime_projection_route") + shared = payload.get("shared_runtime_projection") + if (isinstance(shared, dict) and set(shared) <= {"ok", "status", "dry_run", + "raw_artifacts_copied", "recommended_action_copied"} + and shared.get("ok") is True and shared.get("status") == "disabled" + and shared.get("dry_run") is False and shared.get("raw_artifacts_copied") is False + and shared.get("recommended_action_copied") is False): + supporting.append("shared_runtime_projection") + explore = payload.get("explore_graph_sync") + if (isinstance(explore, dict) and set(explore) <= {"ok", "schema_version", "goal_id", "enabled", + "trigger", "external_sink_delivery_authorized", "status", "delivery_postcondition"} + and explore.get("schema_version") == "loopx_explore_graph_activation_v0" + and explore.get("ok") is True and explore.get("enabled") is False + and explore.get("status") == "disabled" and explore.get("external_sink_delivery_authorized") is False + and explore.get("delivery_postcondition") == {"schema_version": "loopx_explore_graph_delivery_postcondition_v0", + "required": False, "satisfied": True, "disposition": "not_required", "retry_required": False, + "blocks_delivery": False, "required_action": None}): + supporting.append("explore_graph_sync") + recommendation = payload.get("recommended_action_resolution") + if (isinstance(recommendation, dict) + and set(recommendation) <= {"schema_version", "recommended_action", "recommended_action_source", + "authority", "settlement_alignment", "todo_id", "selection_binding", "claim_required_before_work"} + and recommendation.get("schema_version") == "refresh_recommendation_v0" + and recommendation.get("settlement_alignment") == "exact" + and recommendation.get("recommended_action") == payload.get("recommended_action") + and recommendation.get("recommended_action_source") == payload.get("recommended_action_source")): + # This completed Turn's recommendation is an audit snapshot; continuing + # work still needs the fresh guard already required by the plan. + supporting.extend(("recommended_action", "recommended_action_source", "recommended_action_resolution")) + sync = payload.get("global_sync") + if (isinstance(sync, dict) and set(sync) <= {"enabled", "global_registry", "synced_goal_ids", "wrote"} + and sync.get("enabled") is False and sync.get("wrote") is False + and sync.get("synced_goal_ids") == []): + supporting.append("global_sync") + for field, schema in ( + ("goal_channel_blocked_notice_sync", "loopx_goal_channel_blocked_notice_delivery_v0"), + ("goal_channel_gate_sync", "loopx_goal_channel_gate_auto_delivery_v0"), + ): + sink = payload.get(field) + if (isinstance(sink, dict) and set(sink) <= {"schema_version", "ok", "enabled", "status", + "external_write_performed", "readback_verified", "delivery_postcondition"} + and sink.get("schema_version") == schema and sink.get("ok") is True + and sink.get("enabled") is False and sink.get("status") == "not_configured" + and sink.get("external_write_performed") is False + and sink.get("delivery_postcondition") == {"satisfied": True, "blocks_delivery": False}): + supporting.append(field) + return budgeted_settlement_result(payload, runtime_root=runtime_root, + output_bytes=NATIVE_CONTEXT_PAGE_OUTPUT_BYTES, supporting_fields=tuple(supporting), + reader_argv=current_loopx_cli_argv()) diff --git a/loopx/control_plane/quota/settlement.py b/loopx/control_plane/quota/settlement.py index 116d85f176..8fb3983721 100644 --- a/loopx/control_plane/quota/settlement.py +++ b/loopx/control_plane/quota/settlement.py @@ -340,6 +340,7 @@ def read_heartbeat_settlement( source_admission: Mapping[str, Any] | None = None, borrow_source_admission: bool = False, heartbeat_reentry_guard: dict[str, Any] | None = None, + include_record_bodies: bool = True, ) -> QuotaSettlementReadback | None: """Read one complete heartbeat settlement through the TS domain owner.""" @@ -357,6 +358,8 @@ def read(admission: Mapping[str, Any] | None) -> Any: "replan_obligation_id": replan_obligation_id, "infer_turn_instance_id": infer_turn_instance_id, "allow_unbound_binding": allow_unbound_binding, + **({"include_record_bodies": include_record_bodies} + if include_record_bodies is not True else {}), **( {"resolve_original_binding": True} if resolve_original_binding diff --git a/loopx/control_plane/quota/settlement_readback.ts b/loopx/control_plane/quota/settlement_readback.ts index 4bae8b67b2..9dbcc27cc4 100644 --- a/loopx/control_plane/quota/settlement_readback.ts +++ b/loopx/control_plane/quota/settlement_readback.ts @@ -109,6 +109,7 @@ interface ReadbackRequest { allow_unbound_binding: boolean; resolve_original_binding: boolean; borrow_source_admission: boolean; + include_record_bodies: boolean; refresh_retry: RefreshRetryRequest | null; heartbeat_reentry_guard: JsonObject | null; owner: QuotaAccountingOwner; @@ -220,6 +221,9 @@ function decodeRequest( typeof request.resolve_original_binding !== "boolean") { throw new EffectRuntimeRequestError("resolve_original_binding must be a boolean"); } + if (request.include_record_bodies !== undefined && typeof request.include_record_bodies !== "boolean") { + throw new EffectRuntimeRequestError("include_record_bodies must be a boolean"); + } if (request.resolve_original_binding === true && request.infer_turn_instance_id) { throw new EffectRuntimeRequestError("original binding requires an explicit Turn identity"); } @@ -278,6 +282,7 @@ function decodeRequest( allow_unbound_binding: request.allow_unbound_binding, resolve_original_binding: request.resolve_original_binding === true, borrow_source_admission: request.borrow_source_admission === true, + include_record_bodies: request.include_record_bodies !== false, refresh_retry: decodeRefreshRetry(request.refresh_retry), heartbeat_reentry_guard: request.heartbeat_reentry_guard == null ? null : requireJsonObject(request.heartbeat_reentry_guard, "heartbeat_reentry_guard"), @@ -1033,8 +1038,9 @@ function resolveIdentity( ); } -function bundle(result: SettlementResult): ResultBundle { - return { result, payload: settlementResultPayload(result) }; +function bundle(result: SettlementResult, includeRecordBody = true): ResultBundle { + return {result: includeRecordBody ? result : {...result, value: null}, + payload: settlementResultPayload(result)}; } function failedAfterIdentity( @@ -1335,31 +1341,37 @@ function readQuotaSettlementFromRequest( ? {...refreshDecision, decision: "reject", reason: "committed_writeback_payload_conflict"} : refreshDecision; + const progress = settlementProgress(identityResult, writeback, spend, writebackRun, spendRun, + receiptDetails.quota_spend_source ?? spendRun?.source, noSpendKind); + // The combined caller consumes controls and receipts after verification. + // Keep original bodies by default and throughout any pending recovery, + // including the same Runs carried by the four result bundles. Identity, + // controls and receipts remain complete; projection cannot settle a Turn. + const includeRecords = request.include_record_bodies || progress.state !== "settled"; return { schema_version: QUOTA_SETTLEMENT_READBACK_RESULT_SCHEMA, found: true, identity: bundle(identityResult), - writeback: bundle(writeback), - spend: bundle(spend), - delivery: bundle(withWriteback), - settlement: bundle(settled), + writeback: bundle(writeback, includeRecords), + spend: bundle(spend, includeRecords), + delivery: bundle(withWriteback, includeRecords), + settlement: bundle(settled, includeRecords), terminal_closeout: bundle(terminalCloseout), terminal_settlement: bundle(terminalSettlement), - progress: settlementProgress(identityResult, writeback, spend, writebackRun, spendRun, - receiptDetails.quota_spend_source ?? spendRun?.source, noSpendKind), + progress, workspace_causality: workspaceCausality, semantic_replan_guard: semanticReplanGuard, - writeback_run: writebackRun, + writeback_run: includeRecords ? writebackRun : null, refresh_recovery: recovery, external_delivery: request.refresh_retry === null ? null : refreshExternalDelivery( request.refresh_retry.external_delivery ?? null, identity, events, recovery?.decision !== "reject" && recovery?.decision !== "repair_receipt", ), - spend_run: spendRun, - heartbeat_receipt: heartbeatReceipt, - writeback_event: writebackEvent, - spend_event: spendEvent, - completion_event: completionEvent, + spend_run: includeRecords ? spendRun : null, + heartbeat_receipt: includeRecords ? heartbeatReceipt : null, + writeback_event: includeRecords ? writebackEvent : null, + spend_event: includeRecords ? spendEvent : null, + completion_event: includeRecords ? completionEvent : null, monitor_phase: monitorPhase, replay_phase: replayPhase, successor_replan_pending: replayPhase === "open" && diff --git a/loopx/control_plane/turn_driver/decision_capture.py b/loopx/control_plane/turn_driver/decision_capture.py index 91f3ea8e21..0f2453836c 100644 --- a/loopx/control_plane/turn_driver/decision_capture.py +++ b/loopx/control_plane/turn_driver/decision_capture.py @@ -46,16 +46,21 @@ def private_capture_uid() -> int: def write_decision_capture( path: Path, payload: Mapping[str, Any], *, publish: Callable[[str, Path], None] = os.link, -) -> None: - """Publish complete private JSON once; never replace an earlier observation.""" +) -> str: + """Publish complete private JSON once and return its exact byte digest. + + Digest the bytes we publish, rather than reopening the file solely to hash + it. Independent readers still verify the original bytes on every read. + """ + raw = (json.dumps(dict(payload), ensure_ascii=False, separators=(",", ":"), allow_nan=False) + "\n").encode("utf-8") descriptor, temporary = tempfile.mkstemp(prefix=".decision-", dir=path.parent) try: - with os.fdopen(descriptor, "w", encoding="utf-8") as stream: - json.dump(dict(payload), stream, ensure_ascii=False, separators=(",", ":"), allow_nan=False) - stream.write("\n") + with os.fdopen(descriptor, "wb") as stream: + stream.write(raw) stream.flush() os.fsync(stream.fileno()) publish(temporary, path) + return hashlib.sha256(raw).hexdigest() finally: Path(temporary).unlink(missing_ok=True) diff --git a/tests/control_plane/test_refresh_settle_turn.py b/tests/control_plane/test_refresh_settle_turn.py index a7113125ed..787b91aa1b 100644 --- a/tests/control_plane/test_refresh_settle_turn.py +++ b/tests/control_plane/test_refresh_settle_turn.py @@ -15,6 +15,64 @@ ) +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_receipt_capture_failure_keeps_committed_receipts_and_exact_retry(tmp_path, monkeypatch, provider): + 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, + "--todo-id", TODO_ID, "--turn-instance-id", TURN_ID] + rc, guard = _run_cli(registry, runtime, "quota", "should-run", "--codex-app", *binding, cwd=project) + assert rc == 0, guard + private = runtime / "context-delivery" + private.mkdir(mode=0o700, exist_ok=True) + private.chmod(0o755) + command = ["refresh-state", *binding, "--classification", "verified_receipt_recovery", + "--delivery-batch-scale", "implementation", "--delivery-outcome", "outcome_progress", + "--delivery-boundary", "in_flight_continuation", "--no-global-sync", "--suppress-external-sinks", + "--settle-turn", "--settlement-receipt"] + rc, failed = _run_cli(registry, runtime, *command, cwd=project) + assert rc != 0 and failed["ok"] is False + assert failed["settlement_progress"]["state"] == "settled" + assert len(failed["settlement_result"]["receipts"]) == 3 + assert _spend_run_count(runtime) == 1 + private.chmod(0o700) + rc, replay = _run_cli(registry, runtime, *command, cwd=project) + assert rc == 0, replay + assert replay["settlement_identity"] == failed["settlement_identity"] + assert len(replay["settlement_result"]["receipts"]) == 3 + assert _spend_run_count(runtime) == 1 + + +def test_receipt_keeps_unknown_controls_and_never_hides_partial_failure(tmp_path): + from loopx.cli_commands.refresh_settlement import present_settled_refresh + + payload = {"ok": True, "settlement_progress": {"state": "settled"}, + "state": {"large_supporting_snapshot": "x" * 3000}, + "future_control": {"required_resolution": "preserve exactly"}, + "projection_outbox": {"status": "pending", "next_action": "retry"}} + receipt = present_settled_refresh(payload, runtime_root=tmp_path) + assert receipt["future_control"] == payload["future_control"] + assert receipt["projection_outbox"] == payload["projection_outbox"] + for update in ({"ok": False}, {"dry_run": True}, {"settlement_owed": {"command": "original"}}, + {"settlement_progress": {"state": "spend_required"}}): + original = {**payload, **update} + assert present_settled_refresh(original, runtime_root=tmp_path) is original + large = {**payload, "future_control": {"required_resolution": "x" * 100000}} + bounded = present_settled_refresh(large, runtime_root=tmp_path) + assert "Read the complete selected result controls before acting" in bounded["instruction"] + assert "Supporting snapshots" in bounded["instruction"] + command, bodies = bounded["command"], [] + while command: + result = subprocess.run(shlex.split(command), capture_output=True, text=True, check=False) + page = json.loads(result.stdout) + assert result.returncode == 0 and len(result.stdout.encode()) <= 64000 + bodies.append(page["content"]) + command = page.get("next_command") + selected = {row["pointer"]: row["value"] for row in json.loads("".join(bodies))["selections"]} + assert selected["/future_control"] == large["future_control"] + + def _repository_delivery(tmp_path, provider): project, runtime, registry = _write_fixture(tmp_path) state = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md" diff --git a/tests/control_plane_ts/quota_settlement_readback.test.ts b/tests/control_plane_ts/quota_settlement_readback.test.ts index eeabe3dc20..8d743616ec 100644 --- a/tests/control_plane_ts/quota_settlement_readback.test.ts +++ b/tests/control_plane_ts/quota_settlement_readback.test.ts @@ -580,6 +580,21 @@ test("settlement progress requires both effects and exact receipts", async t => schema_version: "quota_settlement_progress_v0", state: entry.state, next_step: entry.next, quota_spend_source: "heartbeat", }); + assert.deepEqual(await readQuotaSettlement(request(root, {include_record_bodies: true})), result); + const controls = await readQuotaSettlement(request(root, {include_record_bodies: false})); + const expected = {...result}; + if (entry.state === "settled") { + assert.ok(result.writeback_run && result.spend_run && result.heartbeat_receipt); + for (const field of ["writeback_run", "spend_run", "heartbeat_receipt", + "writeback_event", "spend_event", "completion_event"]) expected[field] = null; + for (const field of ["writeback", "spend", "delivery", "settlement"]) { + const original = result[field] as {result: {value: unknown}}; + assert.ok(original.result.value, "Complete reads retain their original Run body"); + expected[field] = {...original, result: {...original.result, value: null}}; + } + } + assert.deepEqual(controls, expected, "Pending recovery and all controls/receipts retain complete parity"); + await assert.rejects(readQuotaSettlement(request(root, {include_record_bodies: "false"})), /must be a boolean/); } finally { await rm(root, { recursive: true, force: true }); } }); });