diff --git a/docs/reference/required-work-context.md b/docs/reference/required-work-context.md index 76bfceef69..973970e951 100644 --- a/docs/reference/required-work-context.md +++ b/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 diff --git a/loopx/control_plane/quota/progressive_context.ts b/loopx/control_plane/quota/progressive_context.ts index abd3033e2c..2997fdd69f 100644 --- a/loopx/control_plane/quota/progressive_context.ts +++ b/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) { diff --git a/tests/control_plane/test_deferred_resume_generated_command.py b/tests/control_plane/test_deferred_resume_generated_command.py index a6736f0156..f90dd19195 100644 --- a/tests/control_plane/test_deferred_resume_generated_command.py +++ b/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:]