From 78126294438af163b4a85e71fd73c879f7d592b3 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Mon, 9 Nov 2026 06:24:40 +0800 Subject: [PATCH 1/2] Use the existing composed settlement for ready native heartbeat Turns Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- docs/reference/required-work-context.md | 12 ++ loopx/cli_commands/quota.py | 25 ++-- loopx/cli_commands/quota_presentation.py | 24 ++++ .../control_plane/effect_runtime_handlers.ts | 2 + .../quota/progressive_context.ts | 12 +- loopx/control_plane/quota/settlement_plan.ts | 59 ++++++++- .../control_plane/test_refresh_settle_turn.py | 115 +++++++++++++++++- .../quota_settlement_plan.test.ts | 29 ++++- 8 files changed, 262 insertions(+), 16 deletions(-) diff --git a/docs/reference/required-work-context.md b/docs/reference/required-work-context.md index 7ac6fe66e1..001a71776e 100644 --- a/docs/reference/required-work-context.md +++ b/docs/reference/required-work-context.md @@ -1373,6 +1373,13 @@ pointer 从原始摘要绑定的 capture 批量读取详情。提供已有 `--ca ### Bounded settlement results / 有界结算结果 +A ready native Codex App heartbeat can publish a combined settlement plan for +its admitted Todo. Execute the displayed writeback command, which runs the owed +quota spend and returns the exact receipts. After a partial failure, use the +returned `settlement_owed` command for the original Turn. Read the result's +supporting snapshot only when needed. Raw delivery, manual plans and other +profiles retain their original settlement commands. + Generated Turn-bound JSON `quota spend-slot` commands include `--max-page-output-bytes 64000`. This output option does not enable Context Delivery or create a native continuity baseline. Small results remain complete; @@ -1398,6 +1405,11 @@ output option; a delivery failure does not prove that the effect failed to commi Never rerun admission or change the original Turn/effect identity to recover output. This transport change alone does not prove model retention or savings. +原生 Codex App heartbeat 的当前 Todo 已准入且就绪时，可能收到合并结算计划。 +执行计划给出的写回命令。该命令会执行尚欠的扣额，并返回原回执。部分失败时， +沿返回的 `settlement_owed` 命令恢复原 Turn。结果中的支持快照按需读取。 +原始投递、手动计划和其他运行模式继续使用原结算命令。 + 生成的 Turn 绑定 JSON `quota spend-slot` 命令携带 `--max-page-output-bytes 64000`。此输出选项不启用 Context Delivery，也不建立原生 连续性基线。小结果保持完整；超出预算时，将 `before` / `after` 快照存入私有、 diff --git a/loopx/cli_commands/quota.py b/loopx/cli_commands/quota.py index 7ce08cb724..05cc137ccc 100644 --- a/loopx/cli_commands/quota.py +++ b/loopx/cli_commands/quota.py @@ -9,6 +9,7 @@ from typing import TYPE_CHECKING, Any from .quota_presentation import ( + _project_native_cli_settlement, _project_quota_cli_payload as _project_quota_cli_payload, _quota_renderer as _quota_renderer, ) @@ -23,6 +24,7 @@ from ..usage_goal import observe_quota_result from ..control_plane.coordination.local_authority import CanonicalAuthorityRequired +from ..control_plane.quota.effective_action import EffectiveAction from ..control_plane.quota.effect_program import SettlementIdentity from ..control_plane.quota.error_codes import ( CloseoutQueryUnavailableError, @@ -339,6 +341,19 @@ def _emit_quota_result( from ..command_invocation import current_loopx_cli_argv reader_argv = current_loopx_cli_argv() + activation = None + if args.quota_command == "should-run": + if payload.get("ok") is True: + activation = native_context_activation(args) + if (args.format == "json" and capture_directory is not None and activation["progressive_enabled"] + and activation.get("scenario") == "codex_app_heartbeat" + and (getattr(args, "codex_context_dir", None) or getattr(args, "codex_context_root", None)) + and not getattr(args, "turn_envelope", False) and not getattr(args, "defer_context_presentation", False) + and payload.get("effective_action") == EffectiveAction.NORMAL_RUN.value + and payload.get("normal_delivery_allowed") is True): + payload = _project_native_cli_settlement(payload) + attach_reward_memory_recall_after_should_run(payload, quota_command=args.quota_command, + heartbeat_turn_id=heartbeat_turn_id, registry_path=registry_path, goal_id=args.goal_id, agent_id=args.agent_id) # Compose optional capability guidance after recovery has chosen the plan, # before capture/compaction. The quota decision owner stays capability-neutral. cli = (payload.get("interaction_contract") or {}).get("cli_channel") @@ -360,7 +375,7 @@ def _emit_quota_result( return _emit_bounded_settlement(payload, args, runtime_root=context.runtime_root, reader_argv=reader_argv, print_payload=print_payload) try: - activation = native_context_activation(args) + activation = activation or native_context_activation(args) context_delivery_enabled = activation["enabled"] except QuotaCommandValidationError: if payload.get("ok") is True: @@ -968,14 +983,6 @@ def handle_quota_command( spend_turn_instance_id=spend_turn_instance_id, rollout_details=rollout_details, append_cli_rollout_event=append_cli_rollout_event, ) - attach_reward_memory_recall_after_should_run( - payload, - quota_command=args.quota_command, - heartbeat_turn_id=heartbeat_turn_id, - registry_path=registry_path, - goal_id=args.goal_id, - agent_id=args.agent_id, - ) return _emit_quota_result( payload, args, diff --git a/loopx/cli_commands/quota_presentation.py b/loopx/cli_commands/quota_presentation.py index 421bde7e31..0a9f3ea2a7 100644 --- a/loopx/cli_commands/quota_presentation.py +++ b/loopx/cli_commands/quota_presentation.py @@ -8,6 +8,30 @@ from ..control_plane.quota.turn_envelope import build_turn_envelope, project_captured_replan_cli_view +def _project_native_cli_settlement(payload: dict[str, object]) -> dict[str, object]: + """Publish one projected plan before capability binding and capture.""" + from ..control_plane.effect_runtime import EffectRuntimeRemoteError, EffectRuntimeStartupError, effect_runtime_result + from ..control_plane.quota.turn_envelope import _apply_context_projection_changes + + interaction = payload.get("interaction_contract") + cli = interaction.get("cli_channel") if isinstance(interaction, Mapping) else None + todo = payload.get("selected_todo") + if not isinstance(cli, Mapping) or not isinstance(todo, Mapping): + return payload + try: + projection = effect_runtime_result("settlement.native_cli_projection", { + "cli": cli, "receipt": payload.get("heartbeat_receipt"), "todo_id": todo.get("todo_id"), + }) + except (EffectRuntimeRemoteError, EffectRuntimeStartupError): + # The already admitted Turn still has its canonical manual plan. + return payload + if not projection: + return payload + return {**payload, "interaction_contract": { + **interaction, "cli_channel": _apply_context_projection_changes(dict(cli), projection), + }} + + def _quota_renderer( args: argparse.Namespace, *, diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index d8653e76d5..b093286a8a 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -782,6 +782,8 @@ export function createEffectRuntimeHandlers( ], ["settlement.turn_scoped_cli_plan", lazyHandler(() => import("./quota/settlement_plan.ts"), ({turnScopedCliSettlementPlan}) => (params) => settlementPlanPayload(turnScopedCliSettlementPlan(params)))], + ["settlement.native_cli_projection", lazyHandler(() => import("./quota/settlement_plan.ts"), + ({nativeCliSettlementProjection}) => nativeCliSettlementProjection)], [ "settlement.result_payload", (params) => settlementResultPayload( diff --git a/loopx/control_plane/quota/progressive_context.ts b/loopx/control_plane/quota/progressive_context.ts index 1593589897..bb3db2972f 100644 --- a/loopx/control_plane/quota/progressive_context.ts +++ b/loopx/control_plane/quota/progressive_context.ts @@ -692,6 +692,12 @@ function initialContextReadGroups(context: JsonObject, decisionFields: readonly const prepared = object(cli?.execution_preparation); const identity = object(plan?.identity); const settlementSteps = steps?.map(object).filter(step => step?.kind === "durable_writeback" || step?.kind === "quota_spend"); + const [writeback, spend] = settlementSteps ?? []; + const composedSettlement = plan?.schema_version === "quota_settlement_plan_v2" && writeback?.owner === "agent" + && spend?.owner === "runtime" && !Object.hasOwn(spend, "command_template"); + const templates = plan?.schema_version === "quota_settlement_plan_v1" + ? [writeback?.command_template, spend?.command_template] + : composedSettlement ? [writeback?.command_template] : undefined; // A ready Turn already shows its ordered writeback and spend with their // conditions. Hide only the exact duplicate fallback in this first read; // recovery, legacy readers and the original capture keep both forms. @@ -701,9 +707,9 @@ function initialContextReadGroups(context: JsonObject, decisionFields: readonly && typeof identity.todo_id === "string" && prepared.todo_id === identity.todo_id && object(context.selected_todo)?.todo_id === identity.todo_id && isDeepStrictEqual(identity, object(object(context.heartbeat_receipt)?.settlement_identity)) - && settlementSteps?.length === 2 && settlementSteps[0]?.kind === "durable_writeback" && settlementSteps[1]?.kind === "quota_spend" - && settlementSteps.every(step => typeof step?.command_template === "string" && step.command_template.length > 0) - && isDeepStrictEqual(cli?.next_cli_actions, settlementSteps.map(step => step?.command_template)); + && settlementSteps?.length === 2 && writeback?.kind === "durable_writeback" && spend?.kind === "quota_spend" + && templates?.every(command => typeof command === "string" && command.length > 0) + && isDeepStrictEqual(cli?.next_cli_actions, templates); if (!turnBindingRequired) for (const field of ["execution_preparation", "next_cli_actions", "spend_allowed_now", "spend_after_validation", "spend_policy", "quota_spend_source", "delivery_workspace_causality", "vision_continuation_audit"]) diff --git a/loopx/control_plane/quota/settlement_plan.ts b/loopx/control_plane/quota/settlement_plan.ts index 291154a717..8b85ed6a63 100644 --- a/loopx/control_plane/quota/settlement_plan.ts +++ b/loopx/control_plane/quota/settlement_plan.ts @@ -2,15 +2,70 @@ * This is a projection, not a validation, lease or receipt authorization. */ import { settlementIdentity, type JsonObject, type SettlementIdentityInput, - type SettlementPlan, type SettlementStep, + SETTLEMENT_PLAN_SCHEMA_VERSION, type SettlementPlan, type SettlementStep, } from "../effect_program.ts"; +import {isDeepStrictEqual} from "node:util"; import {VISION_MATERIAL_CLOSEOUT_HINT, visionAuthoringContract} from "../goals/vision_checkpoint.ts"; import {parseExactGoalRef} from "../goals/goal_instance_identity.ts"; -import {requireBoolean, requireJsonObject, requireNonEmptyString, requireStringArray, requireStringLiteral} from "../runtime_decode.ts"; +import {jsonObject, requireBoolean, requireJsonObject, requireNonEmptyString, requireStringArray, requireStringLiteral} from "../runtime_decode.ts"; import {MATERIAL_DELIVERY_OUTCOMES} from "../work_items/delivery_outcome.ts"; import {DELIVERY_BOUNDARIES} from "../turn_driver/delivery_continuity.ts"; import {PROGRESS_RESULT_CLASSES} from "../work_items/replan_history.ts"; import {HANDOFF_MODES} from "../coordination/handoff_mode_vocabulary.ts"; +import {EXECUTION_PREPARATION_SCHEMA_VERSION} from "../work_items/execution_preparation.ts"; +import {contextProjectionChanges} from "./diagnostic_projection.ts"; + +export const COMPOSED_CLI_SETTLEMENT_PLAN_SCHEMA_VERSION = "quota_settlement_plan_v2"; + +/** A transport projection over the existing plan, never a new settlement identity. */ +export function composedCliSettlementChannel(cli: JsonObject): JsonObject { + const plan = jsonObject(cli.settlement_plan), actions = cli.next_cli_actions; + if (plan?.schema_version !== SETTLEMENT_PLAN_SCHEMA_VERSION || !Array.isArray(plan.ordered_steps) + || !Array.isArray(actions)) return cli; + const steps = plan.ordered_steps.map(jsonObject); + if (steps.some(step => !step)) return cli; + const writebacks = steps.filter(step => step?.kind === "durable_writeback"); + const spends = steps.filter(step => step?.kind === "quota_spend"); + const [writeback] = writebacks, [spend] = spends; + if (writebacks.length !== 1 || spends.length !== 1 || !writeback || !spend + || steps.indexOf(writeback) >= steps.indexOf(spend) + || writeback.owner !== "agent" || spend.owner !== "agent" + || typeof writeback.command_template !== "string" || typeof spend.command_template !== "string" + || typeof writeback.precondition !== "string" || typeof spend.precondition !== "string") return cli; + const writebackCommand = writeback.command_template, spendCommand = spend.command_template; + if (actions.filter(command => command === writebackCommand).length !== 1 + || actions.filter(command => command === spendCommand).length !== 1 + || actions.indexOf(writebackCommand) >= actions.indexOf(spendCommand)) return cli; + const composed = (command: string) => command + " --settle-turn --settlement-receipt"; + const newWriteback = {...writeback, command_template: composed(writebackCommand), + ...(typeof writeback.semantic_closeout_command_template === "string" + ? {semantic_closeout_command_template: composed(writeback.semantic_closeout_command_template)} : {})}; + const {command_template: _manualCommand, ...receiptStep} = spend; + const newSpend = {...receiptStep, owner: "runtime", precondition: spend.precondition + + ". The writeback command runs this step and returns its exact receipt. " + + "After a partial failure follow only settlement_owed for this Turn. After settlement do not run another spend."}; + return {...cli, settlement_plan: {...plan, schema_version: COMPOSED_CLI_SETTLEMENT_PLAN_SCHEMA_VERSION, + ordered_steps: steps.map(step => step === writeback ? newWriteback : step === spend ? newSpend : step)}, + next_cli_actions: actions.filter(command => command !== spendCommand) + .map(command => command === writebackCommand ? newWriteback.command_template : command)}; +} + +/** The heartbeat publisher uses only its current admitted Todo and exact receipt. */ +export function nativeCliSettlementProjection(params: JsonObject): JsonObject { + const cli = jsonObject(params.cli), receipt = jsonObject(params.receipt); + const preparation = jsonObject(cli?.execution_preparation); + const identity = jsonObject(jsonObject(cli?.settlement_plan)?.identity); + if (!cli || cli.selection_required === true || cli.spend_after_validation !== true + || preparation?.schema_version !== EXECUTION_PREPARATION_SCHEMA_VERSION || preparation.state !== "ready" + || preparation.delivery_allowed !== true || receipt?.closeout_required !== true + || (receipt.status !== "committed" && receipt.status !== "replayed") + || !identity || typeof params.todo_id !== "string" || !params.todo_id || params.todo_id !== identity.todo_id + || typeof identity.turn_instance_id !== "string" || receipt.turn_instance_id !== identity.turn_instance_id + || preparation.todo_id !== identity.todo_id || preparation.agent_id !== identity.agent_id + || !isDeepStrictEqual(identity, receipt.settlement_identity)) return {}; + const composed = composedCliSettlementChannel(cli); + return composed === cli ? {} : contextProjectionChanges(cli, composed); +} export function turnScopedCliSettlementPlan(params: JsonObject): SettlementPlan { const identity = settlementIdentity( diff --git a/tests/control_plane/test_refresh_settle_turn.py b/tests/control_plane/test_refresh_settle_turn.py index 7c5c362c42..6da9df0ccf 100644 --- a/tests/control_plane/test_refresh_settle_turn.py +++ b/tests/control_plane/test_refresh_settle_turn.py @@ -3,6 +3,7 @@ import json import copy +from pathlib import Path import shlex import subprocess import sys @@ -11,11 +12,123 @@ from canonical_authority_fixture import isolate_sqlite_runtime from test_quota_settlement_cli import ( - AGENT_ID, GOAL_ID, TODO_ID, TURN_ID, _initialize_fixture_authority, + AGENT_ID, GOAL_ID, TODO_ID, TURN_ID, _heartbeat_receipt_count, _initialize_fixture_authority, _run_cli, _run_generated_cli, _spend_run_count, _write_fixture, ) +def _ready_settlement_guard(tmp_path, monkeypatch, project, runtime, registry, progressive): + from test_codex_context_lifecycle import setup + + journal, capture = tmp_path / "journal", tmp_path / "capture" + journal.mkdir(mode=0o700) + setup(journal, monkeypatch) + binding = ["--goal-id", GOAL_ID, "--agent-id", AGENT_ID, "--todo-id", TODO_ID] + rc, claimed = _run_cli(registry, runtime, "todo", "claim", *binding, "--claimed-by", AGENT_ID, cwd=project) + assert rc == 0, claimed + rc, guard = _run_cli(registry, runtime, "quota", "should-run", "--codex-app", *binding, + "--turn-instance-id", TURN_ID, "--context-delivery", progressive, "--decision-output-dir", str(capture), + "--codex-context-dir", str(journal), cwd=project) + assert rc == 0 and guard["ok"], guard + return guard, json.loads((capture / "decision.json").read_bytes()) + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("case", ["combined", "off", "interrupted"]) +def test_ready_native_guard_publishes_a_runnable_settlement_plan(tmp_path, monkeypatch, provider, case): + from loopx.control_plane.turn_driver.decision_capture import read_context_read_reference + + isolate_sqlite_runtime(tmp_path, monkeypatch) + project, runtime, registry = _write_fixture(tmp_path) + _initialize_fixture_authority(project, runtime, provider=provider, handoff_mode="hard_lease") + guard, original = _ready_settlement_guard(tmp_path, monkeypatch, project, runtime, registry, + "off" if case == "off" else "on") + cli = original["interaction_contract"]["cli_channel"] + plan = cli["settlement_plan"] + identity = original["heartbeat_receipt"]["settlement_identity"] + assert plan["identity"] == identity + assert cli["execution_preparation"]["state"] == "ready" + validate, writeback, spend, terminal = plan["ordered_steps"] + assert validate["command_condition"] == "todo_deliverable_complete" + assert terminal["conditional"] is True and terminal["owner"] == "agent" + if case != "off": + assert plan["schema_version"] == "quota_settlement_plan_v2" + assert "--settle-turn --settlement-receipt" in writeback["semantic_closeout_command_template"] + assert spend["owner"] == "runtime" and "command_template" not in spend + assert cli["next_cli_actions"] == [writeback["command_template"]] + _, recovery = read_context_read_reference(guard["recovery"]["reference_file"]) + projected = json.loads(Path(recovery["path"]).read_bytes()) + assert projected["interaction_contract"]["cli_channel"]["settlement_plan"] == plan + if case == "combined": + from tests.control_plane.codex_checkpoint_fixture import selected_context_values + + command, pages = guard["first_read_command"], [] + while command: + words = shlex.split(command) + rc, page = _run_cli(registry, runtime, *words[words.index("quota"):], cwd=project) + assert rc == 0 and page["ok"] and page["offset"] == sum(len(body.encode()) for body in pages) + pages.append(page["content"]) + command = page.get("next_command") + values = selected_context_values("".join(pages)) + cli_path = "/interaction_contract/cli_channel" + assert cli_path + "/next_cli_actions" not in values + assert values[cli_path + "/settlement_plan/ordered_steps/1/command_template"] == writeback["command_template"] + else: + assert plan["schema_version"] == "quota_settlement_plan_v1" + assert "--settle-turn" not in writeback["semantic_closeout_command_template"] + assert cli["next_cli_actions"] == [writeback["command_template"], spend["command_template"]] + assert spend["owner"] == "agent" + + vision = tmp_path / "vision.json" + vision.write_text(json.dumps({ + "schema_version": "goal_vision_replan_contract_v0", "state": "vision_patch_proposed", + "vision_patch": {"acceptance_summary": "Verified an intermediate result. The original work remains open."}, + "path_delta": {"outcome": "continue", "prior_assumption": "The scoped result needed verification.", + "observed_reality": "The independently checked result advances the current route.", + "retained": ["Original requirements"], "evidence_refs": ["evidence:validation"]}, + })) + command = writeback["semantic_closeout_command_template"] + for key, value in {"": "verified_slice", "": "implementation", + "": "outcome_progress", "": shlex.quote(str(vision))}.items(): + command = command.replace(key, value) + command += " --no-global-sync --suppress-external-sinks" + if case == "interrupted": + from contextlib import chdir, redirect_stdout + from io import StringIO + from types import SimpleNamespace + from loopx.cli import main + from loopx.cli_commands import refresh_settlement + + def fail_child(argv, **kwargs): + assert "spend-slot" in argv and identity["turn_instance_id"] in argv + raise OSError("fixture child unavailable before debit") + + stdout = StringIO() + with monkeypatch.context() as fault, chdir(project), redirect_stdout(stdout): + fault.setattr(refresh_settlement, "subprocess", SimpleNamespace( + run=fail_child, TimeoutExpired=subprocess.TimeoutExpired)) + assert main(shlex.split(command)[1:]) != 0 + failed = json.loads(stdout.getvalue()) + assert failed["appended"] and failed["settlement_identity"] == identity, failed + assert failed["settlement_progress"]["state"] == "spend_required" and _spend_run_count(runtime) == 0 + owed = failed["settlement_owed"] + assert owed["effect_id"] == identity["effect_id"] and owed["kind"] == "quota_spend" + rc, result = _run_generated_cli(owed["command"], registry_path=registry, cwd=project) + else: + rc, result = _run_generated_cli(command, registry_path=registry, cwd=project) + assert rc == 0 and result["ok"] and result["settlement_identity"] == identity, result + if case == "off": + assert result["settlement_owed"]["kind"] == "quota_spend" and _spend_run_count(runtime) == 0 + rc, result = _run_generated_cli(spend["command_template"], registry_path=registry, cwd=project) + assert rc == 0 and result["ok"], result + assert result["settlement_progress"]["state"] == "settled" + assert result["settlement_identity"] == identity + assert _heartbeat_receipt_count(runtime, TURN_ID) == 1 and _spend_run_count(runtime) == 1 + rc, replay = _run_generated_cli(command, registry_path=registry, cwd=project) + assert rc == 0 and replay["settlement_progress"]["state"] == "settled", replay + assert replay["idempotent_replay"] and _spend_run_count(runtime) == 1 + + @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) diff --git a/tests/control_plane_ts/quota_settlement_plan.test.ts b/tests/control_plane_ts/quota_settlement_plan.test.ts index b5dfcc8f7d..0f635b17a0 100644 --- a/tests/control_plane_ts/quota_settlement_plan.test.ts +++ b/tests/control_plane_ts/quota_settlement_plan.test.ts @@ -3,7 +3,9 @@ import test from "node:test"; import {settlementPlanPayload} from "../../loopx/control_plane/effect_program.ts"; import {visionAuthoringContract} from "../../loopx/control_plane/goals/vision_checkpoint.ts"; import {projectMcpInteraction} from "../../loopx/control_plane/turn_driver/host_interaction.ts"; -import {turnScopedCliSettlementPlan} from "../../loopx/control_plane/quota/settlement_plan.ts"; +import {composedCliSettlementChannel, nativeCliSettlementProjection, turnScopedCliSettlementPlan} + from "../../loopx/control_plane/quota/settlement_plan.ts"; +import {EXECUTION_PREPARATION_SCHEMA_VERSION} from "../../loopx/control_plane/work_items/execution_preparation.ts"; import {DELIVERY_OUTCOMES, decodeOptionalDeliveryOutcome, isTurnScopedSettlementOutcome} from "../../loopx/control_plane/work_items/delivery_outcome.ts"; @@ -17,6 +19,31 @@ const input = { }, }; +test("native settlement only replaces matching executable steps on the original ready Turn", () => { + const plan = settlementPlanPayload(turnScopedCliSettlementPlan(input)); + const preparation = {schema_version: EXECUTION_PREPARATION_SCHEMA_VERSION, state: "ready", + todo_id: "todo", agent_id: "agent", delivery_allowed: true}; + const cli = {settlement_plan: plan, execution_preparation: preparation, spend_after_validation: true, + next_cli_actions: ["supporting action", "writeback", "spend"]}; + const receipt = {status: "committed", closeout_required: true, turn_instance_id: "turn", + settlement_identity: plan.identity}; + const request = {cli, receipt, todo_id: "todo"}; + assert.ok(Array.isArray(nativeCliSettlementProjection(request).changes)); + const projected = composedCliSettlementChannel(cli); + assert.deepEqual(projected.next_cli_actions, ["supporting action", "writeback --settle-turn --settlement-receipt"]); + assert.equal((projected.settlement_plan as typeof plan).identity, plan.identity); + for (const changed of [ + {todo_id: "other"}, + {receipt: {...receipt, turn_instance_id: "other"}}, + {receipt: {...receipt, settlement_identity: {...plan.identity as object, effect_id: "changed"}}}, + {cli: {...cli, execution_preparation: {...preparation, state: "claim_required"}}}, + {cli: {...cli, spend_after_validation: false}}, + {cli: {...cli, next_cli_actions: ["spend", "writeback"]}}, + {cli: {...cli, next_cli_actions: ["writeback", "spend", "spend"]}}, + ]) assert.deepEqual(nativeCliSettlementProjection({...request, ...changed}), {}); + assert.deepEqual(cli.next_cli_actions, ["supporting action", "writeback", "spend"]); +}); + for (const lane of ["ordinary", "in_flight", "autonomous_replan"]) { test(`${lane} settlement offers only outcomes that its writer can qualify`, () => { const identity = lane === "autonomous_replan" From 8857c70683499180d0c6451218bb38984b1adabf Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Mon, 9 Nov 2026 06:30:35 +0800 Subject: [PATCH 2/2] Keep composed settlement on live native sessions Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- docs/reference/required-work-context.md | 6 +-- loopx/cli_commands/quota.py | 60 +++++++++++++++---------- 2 files changed, 40 insertions(+), 26 deletions(-) diff --git a/docs/reference/required-work-context.md b/docs/reference/required-work-context.md index 001a71776e..de8e8a1a6b 100644 --- a/docs/reference/required-work-context.md +++ b/docs/reference/required-work-context.md @@ -1373,8 +1373,8 @@ pointer 从原始摘要绑定的 capture 批量读取详情。提供已有 `--ca ### Bounded settlement results / 有界结算结果 -A ready native Codex App heartbeat can publish a combined settlement plan for -its admitted Todo. Execute the displayed writeback command, which runs the owed +A ready Codex App heartbeat with a native session can publish a combined settlement +plan for its admitted Todo. Execute the displayed writeback command, which runs the owed quota spend and returns the exact receipts. After a partial failure, use the returned `settlement_owed` command for the original Turn. Read the result's supporting snapshot only when needed. Raw delivery, manual plans and other @@ -1405,7 +1405,7 @@ output option; a delivery failure does not prove that the effect failed to commi Never rerun admission or change the original Turn/effect identity to recover output. This transport change alone does not prove model retention or savings. -原生 Codex App heartbeat 的当前 Todo 已准入且就绪时，可能收到合并结算计划。 +Codex App heartbeat 有原生会话，且当前 Todo 已准入并就绪时，可能收到合并结算计划。 执行计划给出的写回命令。该命令会执行尚欠的扣额，并返回原回执。部分失败时， 沿返回的 `settlement_owed` 命令恢复原 Turn。结果中的支持快照按需读取。 原始投递、手动计划和其他运行模式继续使用原结算命令。 diff --git a/loopx/cli_commands/quota.py b/loopx/cli_commands/quota.py index 05cc137ccc..e0615d2aee 100644 --- a/loopx/cli_commands/quota.py +++ b/loopx/cli_commands/quota.py @@ -5,6 +5,7 @@ import time from collections.abc import Callable, Mapping from importlib import import_module +from os import getenv from pathlib import Path from typing import TYPE_CHECKING, Any @@ -325,6 +326,31 @@ def _emit_bounded_settlement(payload, args, *, runtime_root, reader_argv, print_ return 0 if payload.get("ok") is True else 1 +def _prepare_quota_result_context( + payload: dict[str, object], args: argparse.Namespace, *, + registry_path: Path, heartbeat_turn_id: str | None, capture_directory: Path | None, +) -> tuple[dict[str, object], dict[str, Any]]: + """Prepare current settlement and capability binding before publishing it.""" + try: + activation = native_context_activation(args) + except QuotaCommandValidationError: + if payload.get("ok") is True: + raise + # Keep the original diagnostic on a refused guard. + activation = {"progressive_enabled": False, "enabled": False} + if args.quota_command == "should-run": + if (args.format == "json" and capture_directory is not None and activation["progressive_enabled"] + and activation.get("scenario") == "codex_app_heartbeat" and getenv("CODEX_THREAD_ID") + and (getattr(args, "codex_context_dir", None) or getattr(args, "codex_context_root", None)) + and not getattr(args, "turn_envelope", False) and not getattr(args, "defer_context_presentation", False) + and payload.get("ok") is True and payload.get("effective_action") == EffectiveAction.NORMAL_RUN.value + and payload.get("normal_delivery_allowed") is True): + payload = _project_native_cli_settlement(payload) + attach_reward_memory_recall_after_should_run(payload, quota_command=args.quota_command, + heartbeat_turn_id=heartbeat_turn_id, registry_path=registry_path, goal_id=args.goal_id, agent_id=args.agent_id) + return payload, activation + + def _emit_quota_result( payload: dict[str, object], args: argparse.Namespace, @@ -341,19 +367,14 @@ def _emit_quota_result( from ..command_invocation import current_loopx_cli_argv reader_argv = current_loopx_cli_argv() + budgeted_spend = (args.quota_command == "spend-slot" and args.format == "json" and context is not None + and getattr(args, "max_page_output_bytes", None) is not None) activation = None - if args.quota_command == "should-run": - if payload.get("ok") is True: - activation = native_context_activation(args) - if (args.format == "json" and capture_directory is not None and activation["progressive_enabled"] - and activation.get("scenario") == "codex_app_heartbeat" - and (getattr(args, "codex_context_dir", None) or getattr(args, "codex_context_root", None)) - and not getattr(args, "turn_envelope", False) and not getattr(args, "defer_context_presentation", False) - and payload.get("effective_action") == EffectiveAction.NORMAL_RUN.value - and payload.get("normal_delivery_allowed") is True): - payload = _project_native_cli_settlement(payload) - attach_reward_memory_recall_after_should_run(payload, quota_command=args.quota_command, - heartbeat_turn_id=heartbeat_turn_id, registry_path=registry_path, goal_id=args.goal_id, agent_id=args.agent_id) + if not budgeted_spend: + payload, activation = _prepare_quota_result_context( + payload, args, registry_path=registry_path, heartbeat_turn_id=heartbeat_turn_id, + capture_directory=capture_directory, + ) # Compose optional capability guidance after recovery has chosen the plan, # before capture/compaction. The quota decision owner stays capability-neutral. cli = (payload.get("interaction_contract") or {}).get("cli_channel") @@ -370,19 +391,12 @@ def _emit_quota_result( turn_id=_effective_spend_turn_instance_id(payload, heartbeat_turn_id=heartbeat_turn_id), started_at=usage_quota_started, ) - if (args.quota_command == "spend-slot" and args.format == "json" and context is not None - and getattr(args, "max_page_output_bytes", None) is not None): + if budgeted_spend: + assert context is not None return _emit_bounded_settlement(payload, args, runtime_root=context.runtime_root, reader_argv=reader_argv, print_payload=print_payload) - try: - activation = activation or native_context_activation(args) - context_delivery_enabled = activation["enabled"] - except QuotaCommandValidationError: - if payload.get("ok") is True: - raise - # Preserve the original validation error when activation itself failed. - context_delivery_enabled = False - activation = {"progressive_enabled": False} + assert activation is not None + context_delivery_enabled = activation["enabled"] bind_capture_selection_transport(payload, args, capture_directory=capture_directory) delivery_request = None decision_identity = ("", "")