Combine writeback and quota spend for ready native heartbeat Turns

docs/reference/required-work-context.md

@@ -1373,6 +1373,13 @@ pointer 从原始摘要绑定的 capture 批量读取详情。提供已有 `--ca
 
 ### Bounded settlement results / 有界结算结果
 
+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
+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` 快照存入私有、

loopx/cli_commands/quota.py

@@ -5,10 +5,12 @@
 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
 
 from .quota_presentation import (
+    _project_native_cli_settlement,
     _project_quota_cli_payload as _project_quota_cli_payload,
     _quota_renderer as _quota_renderer,
 )
@@ -23,6 +25,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,
@@ -323,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,
@@ -339,6 +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 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")
@@ -355,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 = 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 = ("", "")
@@ -968,14 +997,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,

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,
     *,

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(

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"])

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(