From 1a1ebd5e46c1204ff925ba47211987e70fd291e0 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Wed, 11 Nov 2026 16:37:09 +0800 Subject: [PATCH] fix(quota): resume verified replan settlement before stopping Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/control_plane/quota/decision_summary.py | 10 +++++++ loopx/control_plane/quota/live_decision.py | 26 ++++++++++++++++--- loopx/control_plane/quota/run_decision.ts | 12 +++++++++ .../control_plane/quota/should_run_packet.py | 5 ++++ .../work_items/interaction_contract.py | 24 ++++++++++++----- .../test_replan_successor_guard_reentry.py | 25 +++++++++++++++--- 6 files changed, 90 insertions(+), 12 deletions(-) diff --git a/loopx/control_plane/quota/decision_summary.py b/loopx/control_plane/quota/decision_summary.py index 3d9c5c2764..a6aa9afc71 100644 --- a/loopx/control_plane/quota/decision_summary.py +++ b/loopx/control_plane/quota/decision_summary.py @@ -12,6 +12,7 @@ goal_frontier_is_terminal_no_followup, ) from ..effect_runtime import effect_runtime_result +from .effect_program import ReceiptBoundReplayPhase from ..todos.contract import normalize_todo_claimed_by from ..work_items.work_lane import work_lane_contract_is_due_monitor_attempt @@ -263,10 +264,12 @@ def resolve_quota_run_decision( registered_agent_ids: list[str], goal_frontier_projection: dict[str, Any] | None, task_orchestration_contract: dict[str, Any] | None, + receipt_bound_replan_phase: ReceiptBoundReplayPhase | None = None, ) -> QuotaRunDecision: """Apply the ordered final guards for one quota run decision.""" replan_required = autonomous_replan_is_required(replan_obligation) + settlement_packet = (replan_obligation or {}).get("settlement_action_packet") # Scope evaluation can synthesize a legacy identity; preserve its lazy read boundary. scope = autonomous_replan_scope_decision( replan_obligation, agent_id=agent_frontier_id, @@ -294,6 +297,13 @@ def resolve_quota_run_decision( "automation_prompt_upgrade_required": automation_prompt_upgrade_required, "replan_required": replan_required, "replan_applies": bool(scope.get("applies")), + "replan_settlement_pending": ( + receipt_bound_replan_phase is ReceiptBoundReplayPhase.SETTLEMENT_PENDING + and (replan_obligation or {}).get("selection_binding") == "heartbeat_receipt" + and isinstance(settlement_packet, dict) + and settlement_packet.get("settlement_only") is True + and settlement_packet.get("decision") == receipt_bound_replan_phase.value + ), "goal_health_ok": goal_health_ok, "inbox_reply_due": inbox_reply_due, "inbox_material_review_due": inbox_material_review_due, diff --git a/loopx/control_plane/quota/live_decision.py b/loopx/control_plane/quota/live_decision.py index 4aa1728da3..c4b598c681 100644 --- a/loopx/control_plane/quota/live_decision.py +++ b/loopx/control_plane/quota/live_decision.py @@ -69,12 +69,32 @@ def _receipt_bound_guard_fields( def _historical_ack_writeback_required(payload: Mapping[str, Any]) -> bool: - """Dispatch the typed historical ACK without hiding pending current closeout.""" + """Recognize the exact historical ACK closeout before durable writeback.""" packet = payload.get("replan_action_packet") return (isinstance(packet, Mapping) and packet.get("settlement_only") is True and packet.get("decision") == "historical_ack_recorded") +def _original_replan_settlement_required( + payload: Mapping[str, Any], readback: QuotaSettlementReadback | None, +) -> bool: + if _historical_ack_writeback_required(payload): + return True + if readback is None or readback.replay_phase is not ReceiptBoundReplayPhase.SETTLEMENT_PENDING: + return False + packet = payload.get("replan_action_packet") + identity = readback.identity.value + return ( + isinstance(packet, Mapping) and packet.get("settlement_only") is True + and packet.get("decision") == readback.replay_phase.value + and identity is not None and identity.replan_obligation_id is not None + and identity.replan_obligation_id == packet.get("obligation_id") + and (readback.progress or {}).get("next_step") == "quota_spend" + and payload.get("should_run") is True + and payload.get("effective_action") == EffectiveAction.AUTONOMOUS_REPLAN_REQUIRED.value + ) + + def _project_receipt_replan_settlement( payload: dict[str, Any], readback: QuotaSettlementReadback | None, *, available_capabilities: list[str] | None, @@ -83,7 +103,7 @@ def _project_receipt_replan_settlement( goal_ref: Mapping[str, Any] | None, ) -> None: """Project closeout from the original readback, never the current selection.""" - if not readback or not _historical_ack_writeback_required(payload): + if not readback or not _original_replan_settlement_required(payload, readback): return attach_settlement_progress(payload, readback, registry_path=registry_path, runtime_root=runtime_root, goal_ref=goal_ref) @@ -871,7 +891,7 @@ def build_live_quota_should_run_decision( # unsettled Turn here can overwrite the settled-skip route with a recovery # obligation and then select a successor against the immutable receipt # identity. Leave prior-Turn recovery to the next fresh Turn instead. - original_replan_settlement = _historical_ack_writeback_required(payload) + original_replan_settlement = _original_replan_settlement_required(payload, settlement_readback) if receipt_bound_replay_phase is not ReceiptBoundReplayPhase.SETTLED and not original_replan_settlement: apply_unsettled_host_turn_recovery_if_required( payload, diff --git a/loopx/control_plane/quota/run_decision.ts b/loopx/control_plane/quota/run_decision.ts index 751c96b286..91c2031556 100644 --- a/loopx/control_plane/quota/run_decision.ts +++ b/loopx/control_plane/quota/run_decision.ts @@ -18,6 +18,7 @@ interface RunFacts { automation_prompt_upgrade_required: boolean; replan_required: boolean; replan_applies: boolean; + replan_settlement_pending: boolean; goal_health_ok: boolean; inbox_reply_due: boolean; inbox_material_review_due: boolean; @@ -69,6 +70,8 @@ function decodeRunFacts(params: JsonObject): RunFacts { automation_prompt_upgrade_required: requireBoolean(params.automation_prompt_upgrade_required, "automation_prompt_upgrade_required"), replan_required: requireBoolean(params.replan_required, "replan_required"), replan_applies: requireBoolean(params.replan_applies, "replan_applies"), + replan_settlement_pending: params.replan_settlement_pending === undefined ? false + : requireBoolean(params.replan_settlement_pending, "replan_settlement_pending"), goal_health_ok: requireBoolean(params.goal_health_ok, "goal_health_ok"), inbox_reply_due: requireBoolean(params.inbox_reply_due, "inbox_reply_due"), inbox_material_review_due: requireBoolean(params.inbox_material_review_due, "inbox_material_review_due"), @@ -139,6 +142,15 @@ function applyGoalOverrides(facts: RunFacts, decision: RunDecision): void { decision.replan_decision_allowed = !inboxPriority && facts.replan_required && facts.replan_applies && facts.goal_health_ok && !facts.workspace_guard && !facts.automation_prompt_upgrade_required; if (decision.replan_decision_allowed) { + if (facts.replan_settlement_pending) { + // A covered frontier can stop new work before its original receipt has + // settled. Finish that receipt before applying derived terminal stop. + clearExecutionLanes(decision); + decision.should_run = true; + decision.effective_action = EffectiveAction.AUTONOMOUS_REPLAN_REQUIRED; + decision.reason = "verified replan writeback requires settlement for the original Turn"; + return; + } decision.normal_delivery_allowed = false; decision.recovery_delivery_allowed = false; decision.should_run = true; diff --git a/loopx/control_plane/quota/should_run_packet.py b/loopx/control_plane/quota/should_run_packet.py index a7071d940c..188a422a11 100644 --- a/loopx/control_plane/quota/should_run_packet.py +++ b/loopx/control_plane/quota/should_run_packet.py @@ -935,6 +935,11 @@ def _resolve_quota_should_run_route( registered_agent_ids=prepared.registered_agent_ids, goal_frontier_projection=prepared.goal_frontier_projection, task_orchestration_contract=prepared.task_orchestration_contract, + receipt_bound_replan_phase=( + prepared.receipt_bound_replay_phase + if prepared.receipt_bound_todo_id is None + else None + ), ) if prepared.receipt_bound_replay_phase is ReceiptBoundReplayPhase.SETTLED: fields = settled_replay_fields() diff --git a/loopx/control_plane/work_items/interaction_contract.py b/loopx/control_plane/work_items/interaction_contract.py index 959ff01533..42c0e2ffdc 100644 --- a/loopx/control_plane/work_items/interaction_contract.py +++ b/loopx/control_plane/work_items/interaction_contract.py @@ -447,8 +447,10 @@ def _turn_scoped_cli_settlement_context( agent_id = str(agent_identity.get("agent_id") or "").strip() selected_todo_id = normalize_todo_id(selected_todo.get("todo_id")) receipt = payload.get("heartbeat_receipt") - if ((payload.get("replan_action_packet") or {}).get("settlement_only") is True - and isinstance(receipt, Mapping) and receipt.get("settlement_identity") is not None): + replan_packet = payload.get("replan_action_packet") + replan_settlement_only = isinstance(replan_packet, Mapping) and replan_packet.get("settlement_only") is True + if (replan_settlement_only and isinstance(receipt, Mapping) + and receipt.get("settlement_identity") is not None): identity = SettlementIdentity.from_runtime_payload(receipt["settlement_identity"]) if (identity.goal_id, identity.agent_id, identity.turn_instance_id) != (goal_id, agent_id, turn_instance_id): raise ValueError("receipt settlement projection does not match the original Goal/Agent/Turn") @@ -504,10 +506,20 @@ def _turn_scoped_cli_settlement_context( else None ), ) - return ( - plan.as_dict() if plan is not None else None, - replan_settlement_contract, - ) + if plan is None: + return None, replan_settlement_contract + cli_plan = plan.as_dict() + owed = payload.get("settlement_owed") + if (replan_settlement_only and isinstance(owed, Mapping) + and owed.get("effect_id") == plan.identity.effect_id + and owed.get("kind") == SettlementStepKind.QUOTA_SPEND.value): + # Verified receipt progress already owns the remaining step. Keep + # its original command and any conditional terminal closeout. + cli_plan["ordered_steps"] = [ + step for step in cli_plan["ordered_steps"] + if step["kind"] not in {SettlementStepKind.VALIDATION.value, SettlementStepKind.DURABLE_WRITEBACK.value} + ] + return cli_plan, replan_settlement_contract def _terminal_cli_actions( diff --git a/tests/control_plane/test_replan_successor_guard_reentry.py b/tests/control_plane/test_replan_successor_guard_reentry.py index 2b41f28a08..fd5586492a 100644 --- a/tests/control_plane/test_replan_successor_guard_reentry.py +++ b/tests/control_plane/test_replan_successor_guard_reentry.py @@ -306,7 +306,11 @@ def test_successor_guard_returns_original_settlement_not_repeated_planning( assert packet["obligation_id"] == obligation_id assert packet["successor_todo_id"] == added["todo_id"] assert packet["writeback_contract"]["preferred_input"] == "canonical_successor_transition" - actions = reread["interaction_contract"]["cli_channel"]["next_cli_actions"] + cli = reread["interaction_contract"]["cli_channel"] + assert [step["kind"] for step in cli["settlement_plan"]["ordered_steps"]] == [ + "validation", "durable_writeback", "quota_spend", + ] + actions = cli["next_cli_actions"] assert len(actions) == 2 assert all("todo add" not in command for command in actions) assert all("--replan-obligation-id " + obligation_id in command for command in actions) @@ -330,11 +334,18 @@ def test_successor_guard_returns_original_settlement_not_repeated_planning( pending = _guard(call) assert pending["heartbeat_receipt"]["settlement_identity"] == identity assert pending["settlement_progress"]["state"] == "spend_required" - spend_actions = pending["interaction_contract"]["cli_channel"]["next_cli_actions"] + assert pending["settlement_owed"]["effect_id"] == identity["effect_id"] + pending_cli = pending["interaction_contract"]["cli_channel"] + spend_actions = pending_cli["next_cli_actions"] assert spend_actions == [actions[1]] + assert pending["settlement_owed"]["command"] == actions[1] + pending_plan = pending_cli["settlement_plan"] + assert pending_plan["identity"] == identity + assert pending_plan["ordered_steps"] == [cli["settlement_plan"]["ordered_steps"][-1]] pending_envelope = call("quota", "should-run", "--codex-app", "--goal-id", GOAL, "--agent-id", AGENT, "--turn-instance-id", TURN, "--turn-envelope") assert pending_envelope["writeback"]["next_cli_actions"] == spend_actions + assert pending_envelope["writeback"]["settlement_plan"] == pending_plan assert pending_envelope["action_signature"]["matches"] is True spent = call(*shlex.split(spend_actions[0])[1:]) assert spent["settlement_progress"]["state"] == "settled" @@ -459,7 +470,15 @@ def test_later_vision_replan_can_close_covered_frontier_and_settle_once( assert pending["heartbeat_receipt"]["settlement_identity"] == identity assert pending["settlement_progress"]["state"] == "spend_required" assert pending.get("selected_todo") is None - spent = call("quota", "spend-slot", *binding, "--slots", "1", "--source", "heartbeat", "--execute") + assert pending["should_run"] is True + assert pending["normal_delivery_allowed"] is False + cli = pending["interaction_contract"]["cli_channel"] + plan = cli["settlement_plan"] + assert plan["identity"] == identity + assert [step["kind"] for step in plan["ordered_steps"]] == ["quota_spend"] + command = pending["settlement_owed"]["command"] + assert cli["next_cli_actions"] == [command] + spent = call(*shlex.split(command)[1:]) assert spent["appended"] is True assert spent["settlement_progress"]["state"] == "settled" assert call("quota", "spend-slot", *binding, "--slots", "1", "--source", "heartbeat", "--execute")["appended"] is False