Show dated Todo continuation before settlement detail

docs/reference/required-work-context.md

@@ -277,6 +277,16 @@ evidence and the complete operation/settlement contract before depending on them
 The original SHA-bound full recovery remains available; a selected page's EOF
 does not prove complete source delivery, model retention or fresh authority.
 
+For an exact complete Turn bound to a Todo whose dated resume condition has
+fired, the starting selection puts the existing reopen and same-Turn guard
+commands and the current plan's obligations before the longer work context.
+Required and unknown step fields stay visible. After reopening, run the
+continuation guard for a fresh execution decision. For an alternative that
+settles before reopening,
+read the complete original plan at `/interaction_contract/cli_channel/settlement_plan`
+from the original recovery reference before acting. Delivery and spend require
+their existing separate authorization.
+
 The generated `first_read_command` and its paging cursor default to a 12,000-byte
 serialized UTF-8 output limit when the caller supplied no tool budget. This
 applies after explicit Todo selection as well. An explicit output limit still

loopx/control_plane/quota/progressive_context.ts

@@ -1,4 +1,5 @@
 import {EffectiveAction} from "./effective_action.generated.ts";
+import {AgentScopeFrontierAction} from "../agents/agent_scope_frontier.generated.ts";
 /** Model presentation: requirements stay inline, supporting material is
  * addressed in the immutable source. This plan changes no canonical state. */
 import {isDeepStrictEqual} from "node:util";
@@ -50,6 +51,42 @@ function readySettlementIdentity(context: JsonObject, interaction: JsonObject,
     ? identity : undefined;
 }
 
+/** Reorder the opening for an exact dated Todo reentry. The saved capture
+ * still supplies its original settlement obligations. */
+function hasBoundDatedTodoReentry(context: JsonObject, interaction: JsonObject, plan: JsonObject | undefined): boolean {
+  const action = AgentScopeFrontierAction.SUCCESSOR_REPLAN_REQUIRED;
+  const cli = jsonObject(interaction.cli_channel), channel = jsonObject(interaction.agent_channel);
+  const frontier = jsonObject(context.agent_scope_frontier), selected = jsonObject(context.selected_todo);
+  const condition = jsonObject(selected?.resume_condition), resume = jsonObject(condition?.resume_receipt);
+  const receipt = jsonObject(context.heartbeat_receipt), identity = jsonObject(plan?.identity);
+  const scheduled = jsonObject(context.contract_binding)?.scheduler_execution_args;
+  const [first] = Array.isArray(frontier?.deferred_resume_candidates) ? frontier.deferred_resume_candidates : [];
+  const alternateRoute = ["dependency_cycle_candidates", "monitor_blocked_resume_candidates",
+    "route_continuation_replan_candidates"].some(key => {
+      const candidates = frontier?.[key];
+      return candidates !== undefined && (!Array.isArray(candidates) || candidates.length > 0);
+    });
+  return context.schema_version === "loopx_turn_envelope_v1" && context.ok === true && context.should_run === true
+    && context.effective_action === action && interaction.mode === action && frontier?.action === action && !alternateRoute
+    && context.normal_delivery_allowed === false && channel?.delivery_allowed === false && channel.must_attempt === true
+    && jsonObject(interaction.user_channel)?.action_required === false && cli?.turn_binding_required !== true
+    && cli?.spend_allowed_now === false && Array.isArray(cli.next_cli_actions) && cli.next_cli_actions.length === 3
+    && cli.next_cli_actions.every(value => typeof value === "string" && value.length > 0)
+    && typeof scheduled === "string" && scheduled.length > 0
+    && typeof selected?.todo_id === "string" && selected.todo_id === jsonObject(first)?.todo_id
+    && selected.status === "deferred" && selected.resume_ready === true
+    && condition?.schema_version === "todo_resume_condition_v0" && condition.kind === "resume_at"
+    && condition.satisfied === true && condition.resume_when === selected.resume_when
+    && resume?.schema_version === "todo_resume_receipt_v0" && resume.condition_kind === condition.kind
+    && resume.todo_id === selected.todo_id && resume.resume_when === selected.resume_when
+    && receipt?.status === "committed" && orderedSettlementSteps(plan) !== undefined
+    && identity?.schema_version === "quota_settlement_identity_v0" && identity.todo_id === selected.todo_id
+    && identity.goal_id === context.goal_id && identity.agent_id === jsonObject(context.agent_identity)?.agent_id
+    && typeof identity.effect_id === "string" && typeof receipt.turn_instance_id === "string"
+    && identity.turn_instance_id === receipt.turn_instance_id
+    && isDeepStrictEqual(identity, receipt.settlement_identity);
+}
+
 function reusableWritebackPrecondition(context: JsonObject): ReusableReadField | undefined {
   const interaction = jsonObject(context.interaction_contract);
   if (interaction?.schema_version !== "loopx_interaction_contract_v0") return undefined;
@@ -888,13 +925,21 @@ function initialContextReadGroups(context: JsonObject, decisionFields: readonly
   include("interaction_contract", "user_channel");
   for (const field of ["primary_action", "delivery_allowed", "must_attempt", "quiet_noop_allowed", "next_task_action"])
     include("interaction_contract", "agent_channel", field);
-  const turnBindingRequired = object(interaction.cli_channel)?.turn_binding_required === true;
+  const cli = object(interaction.cli_channel), planPath = SETTLEMENT_PLAN_PATH;
+  const plan = object(cli?.settlement_plan);
+  const deferredReentry = hasBoundDatedTodoReentry(context, interaction, plan);
+  const turnBindingRequired = cli?.turn_binding_required === true;
+  const admissionFields = ["next_cli_actions", "spend_allowed_now", "spend_after_validation", "spend_policy"];
   if (turnBindingRequired)
-    for (const field of ["turn_binding_required", "next_cli_actions", "spend_allowed_now", "spend_after_validation", "spend_policy"])
+    for (const field of ["turn_binding_required", ...admissionFields])
       include("interaction_contract", "cli_channel", field);
   include("interaction_contract", "agent_channel", "required_reads");
   const work = ["interaction_contract", "agent_channel", "work_context"];
   include(...work, "instruction");
+  if (deferredReentry) {
+    // Surface the owner's reopen and same-Turn guard before the long read.
+    for (const field of admissionFields) include("interaction_contract", "cli_channel", field);
+  }
   const channel = object(interaction.agent_channel);
   const requiredInboxReads = context.effective_action === EffectiveAction.NORMAL_RUN && context.normal_delivery_allowed === true
     && channel?.delivery_allowed === true && channel.must_attempt === true && Array.isArray(channel.required_reads)
@@ -1016,9 +1061,6 @@ function initialContextReadGroups(context: JsonObject, decisionFields: readonly
     include(...source, "content", "harness", "guidance");
     include(...source, "content", "harness", "orchestration_gate");
   });
-  const cli = object(interaction.cli_channel);
-  const planPath = SETTLEMENT_PLAN_PATH;
-  const plan = object(cli?.settlement_plan);
   const steps = orderedSettlementSteps(plan);
   const settlementSteps = steps?.map(object).filter(step => step?.kind === "durable_writeback" || step?.kind === "quota_spend");
   const [writeback, spend] = settlementSteps ?? [];
@@ -1037,7 +1079,7 @@ function initialContextReadGroups(context: JsonObject, decisionFields: readonly
   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"])
-      if (field !== "next_cli_actions" || !repeatsSettlement) {
+      if ((field !== "next_cli_actions" || !repeatsSettlement) && !(deferredReentry && admissionFields.includes(field))) {
         const before = pointers.length;
         include("interaction_contract", "cli_channel", field);
         // A task-facing operation and its declared continuation belong in the
@@ -1053,6 +1095,7 @@ function initialContextReadGroups(context: JsonObject, decisionFields: readonly
     include("action_portfolio", "selection_policy");
     include("action_portfolio", "suggested_actions");
   }
+  const planStart = pointers.length;
   if (plan && steps) {
     for (const field of Object.keys(plan)) if (field !== "ordered_steps") include(...planPath, field);
     steps.forEach((value, stepIndex) => {
@@ -1081,6 +1124,7 @@ function initialContextReadGroups(context: JsonObject, decisionFields: readonly
       }
     });
   } else include(...planPath); // Unknown plan versions keep their complete obligations.
+  if (deferredReentry) opening.push(...pointers.splice(planStart));
   // Keep the pinned supporting-read route and its obligations inline.
   // The complete index is available at the same captured address.
   if (inlineSupportingDetails) {

tests/control_plane/test_deferred_resume_generated_command.py

@@ -17,11 +17,15 @@
 from loopx.control_plane.coordination.runtime_shadow import (
     build_todo_runtime_shadow_projection,
 )
+from loopx.control_plane.quota.turn_envelope import compose_turn_decision
+from loopx.control_plane.scheduler.execution_context import scheduler_execution_context_for_runtime_profile
 from loopx.control_plane.todos.active_state_todo_parser import parse_active_state_todos
 from loopx.control_plane.todos.todo_summary import (
     canonical_todo_read_record,
     structured_todo_item,
 )
+from loopx.control_plane.turn_driver.decision_capture import write_decision_capture
+from tests.control_plane.codex_checkpoint_fixture import selected_context_values
 from test_quota_settlement_cli import (
     AGENT_ID,
     GOAL_ID,
@@ -127,7 +131,61 @@ def cli(*args: str) -> tuple[int, dict]:
     )
     assert rc == 0 and guard["effective_action"] == "successor_replan_required", guard
     assert guard["selected_todo"]["todo_id"] == TODO_ID
-    generated = guard["interaction_contract"]["cli_channel"]["next_cli_actions"][0]
+    # Replay the signed capture through the same saved-project entrypoint used
+    # by native hosts. It must expose the real action before evidence paging.
+    profile = "codex_app_heartbeat" if host_args == ("--codex-app",) else "generic_cli"
+    complete = compose_turn_decision(guard,
+        scheduler_execution_context=scheduler_execution_context_for_runtime_profile(profile))
+    capture = tmp_path / "resume-decision.json"
+    digest = write_decision_capture(capture, complete)
+    before_context = state.read_bytes()
+    capture_args = ("quota", "context-project", *host_args, "--goal-id", GOAL_ID,
+        "--agent-id", AGENT_ID, "--capture-file", str(capture), "--capture-sha256", digest)
+    bound = (*capture_args, "--turn-instance-id", "resume-command-turn")
+    native = ("--context-delivery", "on", "--codex-context-dir", str(tmp_path / "context"),
+        "--initial-read", "--max-page-output-bytes", "64000")
+    with monkeypatch.context() as context:
+        context.setenv("LOOPX_CONTEXT_DELIVERY", "on")
+        context.setenv("LOOPX_EXECUTION_MODE", "heartbeat")
+        rc, opening = cli(*bound, *native)
+        wrong_rc, wrong = cli(*capture_args, "--turn-instance-id", "other-turn", *native)
+    assert wrong_rc != 0 and wrong["ok"] is False and "exact Goal, Agent and Turn" in wrong["reason"]
+    off_rc, off = cli(*bound, "--context-delivery", "off")
+    assert off_rc == 0 and off["mode"] == "full" and off["recovery"]["sha256"] == digest
+    assert rc == 0 and opening["eof"], opening
+    selected = selected_context_values(opening["content"])
+    pointers = list(selected)
+    cli_path = "/interaction_contract/cli_channel"
+    actions_path = cli_path + "/next_cli_actions"
+    settlement_path = cli_path + "/settlement_plan"
+    work_path = "/interaction_contract/agent_channel/work_context/user_todos"
+    assert pointers.index(actions_path) < pointers.index(work_path)
+    actions = selected[actions_path]
+    assert actions == guard["interaction_contract"]["cli_channel"]["next_cli_actions"]
+    rc, full = cli("quota", "context-read", "--capture-file", str(capture),
+        "--capture-sha256", digest, "--select", settlement_path)
+    assert rc == 0 and full["eof"], full
+    plan = complete["interaction_contract"]["cli_channel"]["settlement_plan"]
+    assert json.loads(full["content"])["selections"][0]["value"] == plan
+
+    # The literal reader also serves historical captures with compatible step
+    # fields. Keep this example separate from the validated current Turn.
+    validation_index = next(
+        index for index, step in enumerate(plan["ordered_steps"]) if step["kind"] == "validation"
+    )
+    required_control = {"required": True, "instruction": "Check the recorded owner stop before reopening."}
+    plan["ordered_steps"][validation_index]["required_control"] = required_control
+    compatibility_capture = tmp_path / "resume-compatible-decision.json"
+    compatibility_digest = write_decision_capture(compatibility_capture, complete)
+    rc, compatible = cli("quota", "context-read", "--capture-file", str(compatibility_capture),
+        "--capture-sha256", compatibility_digest, "--initial-read", "--max-page-output-bytes", "64000")
+    assert rc == 0 and compatible["eof"], compatible
+    compatible_values = selected_context_values(compatible["content"])
+    control_path = f"{settlement_path}/ordered_steps/{validation_index}/required_control"
+    assert compatible_values[control_path] == required_control
+    assert list(compatible_values).index(control_path) < list(compatible_values).index(work_path)
+    assert state.read_bytes() == before_context and _spend_run_count(runtime) == 0
+    generated = actions[0]
     assert generated.startswith("loopx ")
     resume = shlex.split(
         generated.replace(
@@ -201,7 +259,6 @@ def cli(*args: str) -> tuple[int, dict]:
         "task-lease", "inspect", "--goal-id", GOAL_ID, "--todo-id", TODO_ID
     )
     assert rc == 0 and absent["active"] is False, absent
-    actions = guard["interaction_contract"]["cli_channel"]["next_cli_actions"]
     # Execute the delivered continuation instead of constructing a successful
     # guard that the consumer was never shown. No new Turn or premature spend.
     reentry_args = shlex.split(actions[1])[1:]