fix(context): show one ready settlement recipe

loopx/control_plane/quota/progressive_context.ts

@@ -503,22 +503,40 @@ export function initialContextReadPointers(context: JsonObject, decisionFields:
     include(...source, "content", "harness", "guidance");
     include(...source, "content", "harness", "orchestration_gate");
   });
+  const cli = object(interaction.cli_channel);
+  const planPath = ["interaction_contract", "cli_channel", "settlement_plan"];
+  const plan = object(cli?.settlement_plan);
+  const steps = plan?.schema_version === "quota_settlement_plan_v1" && Array.isArray(plan.ordered_steps)
+    ? plan.ordered_steps : undefined;
+  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");
+  // 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.
+  const repeatsSettlement = context.effective_action === EffectiveAction.NORMAL_RUN && context.normal_delivery_allowed === true
+    && channel?.delivery_allowed === true && cli?.selection_required !== true && prepared?.state === "ready"
+    && typeof identity?.effect_id === "string" && identity.effect_id.length > 0
+    && 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));
   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"])
-    include("interaction_contract", "cli_channel", field);
+    if (field !== "next_cli_actions" || !repeatsSettlement) include("interaction_contract", "cli_channel", field);
   // Explicit selection clears next_cli_actions. Deliver the owner's discovery
   // and same-Turn binding recipe alongside its non-exhaustive suggestions.
-  if (object(interaction.cli_channel)?.selection_required === true) {
+  if (cli?.selection_required === true) {
     for (const field of ["selection_required", "selection_policy_ref", "selection_command"])
       include("interaction_contract", "cli_channel", field);
     include("action_portfolio", "selection_policy");
     include("action_portfolio", "suggested_actions");
   }
-  const planPath = ["interaction_contract", "cli_channel", "settlement_plan"];
-  const plan = object(object(interaction.cli_channel)?.settlement_plan);
-  if (plan?.schema_version === "quota_settlement_plan_v1" && Array.isArray(plan.ordered_steps)) {
+  if (plan && steps) {
     for (const field of Object.keys(plan)) if (field !== "ordered_steps") include(...planPath, field);
-    plan.ordered_steps.forEach((value, stepIndex) => {
+    steps.forEach((value, stepIndex) => {
       const step = object(value), stepPath = [...planPath, "ordered_steps", stepIndex];
       if (!step) { include(...stepPath); return; }
       for (const field of Object.keys(step)) {

tests/control_plane/test_context_first_read.py

@@ -448,8 +448,19 @@ def test_advertised_native_reader_keeps_the_complete_current_contract_across_pag
     expected_cli = selected_context_values(json.dumps({"selections": [
         {"pointer": cli_path, "value": source["interaction_contract"]["cli_channel"]}]}))
     vision_path = next(path for path in expected_cli if path.endswith("/vision_authoring"))
-    # Every CLI leaf outside the authoring reference must still be delivered.
+    actions_path = cli_path + "/next_cli_actions"
+    current_cli = source["interaction_contract"]["cli_channel"]
+    bound_steps = current_cli["settlement_plan"]["ordered_steps"]
+    actual_commands = [step["command_template"] for step in bound_steps
+                       if step["kind"] in ("durable_writeback", "quota_spend")]
+    assert len(actual_commands) == 2 and current_cli["next_cli_actions"] == actual_commands
+    # The ordered plan carries these exact templates and all their conditions.
+    # A ready first read should not present a second apparent route for them.
+    assert not any(path == actions_path or path.startswith(actions_path + "/") for path in values)
+    # Every distinct CLI leaf outside the authoring reference is still delivered.
     for path, item in expected_cli.items():
+        if path == actions_path or path.startswith(actions_path + "/"):
+            continue
         if path.startswith(vision_path + "/") or (isinstance(item, (dict, list)) and item):
             continue
         assert values[path] == item
@@ -459,12 +470,48 @@ def test_advertised_native_reader_keeps_the_complete_current_contract_across_pag
         "--max-page-output-bytes", "9000", cwd=project)
     assert code == 0 and detail["eof"]
     assert selected_context_values(detail["content"])[vision_path] == expected_cli[vision_path]
+    code, fallback = _run_cli(registry, runtime, "quota", "context-read", "--reference-file",
+        view["recovery"]["reference_file"], "--select", actions_path, cwd=project)
+    assert code == 0 and fallback["eof"]
+    assert selected_context_values(fallback["content"])[actions_path] == actual_commands
     assert values["/heartbeat_receipt"] == source["heartbeat_receipt"]
     assert values["/selected_todo"] == source["selected_todo"]
     assert _heartbeat_receipt_count(runtime, turn_id) == 1 and _spend_run_count(runtime) == 0
     assert Path(reference["path"]).read_bytes() == raw and state.read_bytes() == original_state
 
 
+@pytest.mark.parametrize("variant", ["different_argument", "different_order", "unready", "other_effect", "future_plan"])
+def test_initial_read_keeps_fallback_actions_when_the_bound_plan_is_not_equivalent(tmp_path, decision, variant):
+    interaction = decision["interaction_contract"]
+    cli = interaction["cli_channel"]
+    identity = {"schema_version": "quota_settlement_identity_v0", "effect_id": "original-effect",
+                "todo_id": decision["selected_todo"]["todo_id"], "turn_instance_id": "original"}
+    decision.update(effective_action="normal_run", normal_delivery_allowed=True,
+                    heartbeat_receipt={"settlement_identity": identity})
+    interaction["agent_channel"]["delivery_allowed"] = True
+    writeback = "loopx refresh-state --turn-instance-id original"
+    spend = "loopx quota spend-slot --turn-instance-id original --execute"
+    cli.update(execution_preparation={"state": "ready", "todo_id": identity["todo_id"]},
+               next_cli_actions=[writeback, spend],
+               settlement_plan={"schema_version": "quota_settlement_plan_v1", "identity": identity,
+                                "ordered_steps": [{"kind": "durable_writeback", "command_template": writeback},
+                                                  {"kind": "quota_spend", "command_template": spend}]})
+    if variant == "different_argument":
+        cli["next_cli_actions"][0] += " --suppress-external-sinks"
+    elif variant == "different_order":
+        cli["next_cli_actions"].reverse()
+    elif variant == "unready":
+        cli["execution_preparation"]["state"] = "claim_required"
+    elif variant == "other_effect":
+        decision["heartbeat_receipt"]["settlement_identity"] = {**identity, "effect_id": "another"}
+    else:
+        cli["settlement_plan"]["schema_version"] = "unknown_settlement_plan"
+
+    view = native_context_capture_view(decision, tmp_path)
+    _, values = read_selection(view["first_read_command"])
+    assert values["/interaction_contract/cli_channel/next_cli_actions"] == cli["next_cli_actions"]
+
+
 def test_initial_read_keeps_current_constraints_whole_and_history_reachable(tmp_path, decision):
     before = json.loads(json.dumps(decision))
     view = native_context_capture_view(decision, tmp_path)