diff --git a/loopx/control_plane/quota/settlement_precedence.py b/loopx/control_plane/quota/settlement_precedence.py index d3605e0449..3fa42b9dcb 100644 --- a/loopx/control_plane/quota/settlement_precedence.py +++ b/loopx/control_plane/quota/settlement_precedence.py @@ -9,6 +9,10 @@ "the receipt-bound work binding and required settlement receipts " "are complete for this heartbeat turn; defer successor selection to a new turn" ) +RECEIPT_BOUND_DEFERRED_REASON = ( + "the Todo bound to this heartbeat receipt is deferred; do not " + "select or spend an independent successor in the same Turn" +) _ACTION_PROJECTION_KEYS = ( "agent_command", @@ -89,3 +93,24 @@ def settled_replay_fields() -> dict[str, Any]: "spend_policy": "no quota spend for an already-settled heartbeat turn", }, } + + +def deferred_receipt_bound_skip_fields( + quota: dict[str, Any], + heartbeat_recommendation: dict[str, Any], +) -> tuple[str, dict[str, Any], dict[str, Any]]: + """Project a deferred receipt without borrowing a successor's authority.""" + + reason = RECEIPT_BOUND_DEFERRED_REASON + return ( + reason, + {**quota, "safe_bypass_allowed": False}, + { + **heartbeat_recommendation, + "recommended_mode": EffectiveAction.QUOTA_SKIP.value, + "notify": "DONT_NOTIFY", + "reason": reason, + "spend_policy": "no quota spend for a deferred receipt-bound Todo", + "stop_if_unchanged": True, + }, + ) diff --git a/loopx/control_plane/quota/should_run_packet.py b/loopx/control_plane/quota/should_run_packet.py index 270b9510ba..41ede57316 100644 --- a/loopx/control_plane/quota/should_run_packet.py +++ b/loopx/control_plane/quota/should_run_packet.py @@ -85,6 +85,7 @@ ) from ..todos.contract import ( normalize_todo_claimed_by, + normalize_todo_id, ) from ..todos.todo_semantics import ( todo_item_is_actionable_open as projection_todo_item_is_actionable_open, @@ -123,10 +124,12 @@ user_action_owns_empty_agent_lane_from_summaries as _user_action_owns_empty_agent_lane, ) from ..work_items.work_lane import ( + WORK_LANE_RECEIPT_BOUND_DEFERRED_OBLIGATION, work_lane_contract_is_due_monitor_attempt, work_lane_contract_is_receipt_bound_monitor_settled, ) from .settlement_precedence import ( + deferred_receipt_bound_skip_fields, settled_replay_fields, HEARTBEAT_SETTLED_REPLAY_REASON, clear_quota_action_projections, @@ -533,6 +536,15 @@ def _resolve_agent_lane_delivery_route( # decision. A newly runnable Todo remains visible in summaries, but it # cannot become the selected settlement target in the same packet. fallback = None + if ( + prepared.receipt_bound_todo_id + and isinstance(fallback, dict) + and normalize_todo_id(fallback.get("todo_id")) + != prepared.receipt_bound_todo_id + ): + # Feed only the committed identity into the TS delivery router. The + # independent successor remains discoverable on a fresh Turn. + fallback = None delivery_agent_id = normalize_todo_claimed_by( (prepared.agent_identity or {}).get("agent_id") @@ -617,6 +629,17 @@ def _resolve_agent_lane_delivery_route( else: selected_action = None + if ( + prepared.receipt_bound_todo_id + and isinstance(selected_action, dict) + and normalize_todo_id(selected_action.get("todo_id")) + != prepared.receipt_bound_todo_id + ): + # The router may find an independent successor after the bound Todo + # becomes unavailable. That successor cannot replace an already + # committed settlement identity inside the same heartbeat Turn. + return None + boundary = delivery_route.get("boundary") if ( isinstance(selected_action, dict) @@ -697,6 +720,37 @@ def _planning_projections( ) +def _resolve_external_evidence_observation( + prepared: _QuotaDecisionPreparation, + *, + state: str, +) -> tuple[dict[str, Any] | None, dict[str, Any] | None]: + """Suppress unchanged or premature polls before choosing a delivery route.""" + + external_evidence_observation = build_external_evidence_observation_obligation( + prepared.item, + state=state, + agent_todo_summary=prepared.agent_todo_summary, + work_lane_contract=prepared.work_lane_contract, + ) + external_evidence_observation_recent = None + if external_evidence_observation: + external_evidence_observation_recent = _recent_external_monitor_observation_unchanged( + prepared.status_payload, + goal_id=prepared.safe_goal_id, + agent_id=( + normalize_todo_claimed_by(prepared.agent_identity.get("agent_id")) + if isinstance(prepared.agent_identity, dict) + else None + ), + ) + if external_evidence_observation_recent or ( + external_evidence_observation.get("poll_window_status") == "before_next_due" + ): + external_evidence_observation = None + return external_evidence_observation, external_evidence_observation_recent + + def _resolve_quota_should_run_route( prepared: _QuotaDecisionPreparation, ) -> _QuotaDecisionRoute: @@ -794,27 +848,10 @@ def _resolve_quota_should_run_route( automation_prompt_upgrade_required=prepared.automation_prompt_upgrade_required, blocked_priority_fallback=prepared.blocked_priority_fallback, ) - external_evidence_observation = build_external_evidence_observation_obligation( - item, - state=state, - agent_todo_summary=prepared.agent_todo_summary, - work_lane_contract=prepared.work_lane_contract, - ) - external_evidence_observation_recent = None - if external_evidence_observation: - external_evidence_observation_recent = _recent_external_monitor_observation_unchanged( - prepared.status_payload, - goal_id=prepared.safe_goal_id, - agent_id=( - normalize_todo_claimed_by(prepared.agent_identity.get("agent_id")) - if isinstance(prepared.agent_identity, dict) - else None - ), - ) - if external_evidence_observation_recent or ( - external_evidence_observation.get("poll_window_status") == "before_next_due" - ): - external_evidence_observation = None + ( + external_evidence_observation, + external_evidence_observation_recent, + ) = _resolve_external_evidence_observation(prepared, state=state) ready_deferred_resume_candidates: list[dict[str, Any]] = [] if isinstance(prepared.agent_identity, dict) and isinstance( prepared.agent_todo_summary, dict @@ -867,6 +904,33 @@ def _resolve_quota_should_run_route( "reason": reason, "spend_policy": "no quota spend for an already-settled heartbeat turn", } + receipt_bound_deferred = ( + prepared.receipt_bound_todo_id + and isinstance(prepared.work_lane_contract, dict) + and prepared.work_lane_contract.get("obligation") + == WORK_LANE_RECEIPT_BOUND_DEFERRED_OBLIGATION + ) + if receipt_bound_deferred: + # Every action path is closed for this immutable, deferred receipt. + ( + normal_delivery_allowed, + recovery_allowed, + self_repair_allowed, + capability_repair_allowed, + workspace_repair_allowed, + replan_decision_allowed, + receipt_bound_replan_decision, + should_run, + ) = (False,) * 8 + ( + reason, + quota, + heartbeat_recommendation, + ) = deferred_receipt_bound_skip_fields( + quota, + heartbeat_recommendation, + ) + effective_action = EffectiveAction.QUOTA_SKIP.value monitor_quiet_skip = ( not replan_decision_allowed and normal_delivery_allowed @@ -934,7 +998,11 @@ def _resolve_quota_should_run_route( ) agent_scope_frontier = None agent_lane_frontier_hint = None - if not replan_decision_allowed and not receipt_bound_monitor_settled: + if receipt_bound_deferred: + # The no-spend route and its public next action must describe the + # same committed binding, even when summaries offer independent work. + selected_recommended_action = prepared.work_lane_contract["action"] + elif not replan_decision_allowed and not receipt_bound_monitor_settled: selected_recommended_action = selected_action_with_agent_lane( selected_recommended_action, agent_lane_next_action=agent_lane_next_action, diff --git a/loopx/control_plane/quota/should_run_prepare.py b/loopx/control_plane/quota/should_run_prepare.py index 430bbcdce4..902929bbd3 100644 --- a/loopx/control_plane/quota/should_run_prepare.py +++ b/loopx/control_plane/quota/should_run_prepare.py @@ -69,6 +69,7 @@ ) from ..todos.contract import ( TODO_STATUS_BLOCKED, + TODO_STATUS_DEFERRED, TODO_STATUS_OPEN, TODO_TASK_CLASS_ADVANCEMENT, TODO_TASK_CLASS_BLOCKER, @@ -104,6 +105,7 @@ lark_inbox_reply_due_work_lane_contract, operator_inbox_material_review_due_work_lane_contract, preserve_heartbeat_receipt_bound_work_lane, + receipt_bound_deferred_work_lane, scoped_user_gate_due_monitor_contract, work_lane_contract_is_lark_inbox_reply_due, work_lane_contract_is_operator_inbox_material_review_due, @@ -450,6 +452,20 @@ def _build_agent_work_lane( return monitor_only, work_lane, task_orchestration +def _deferred_receipt_bound_work_lane( + *, todo_id: str, source_items: list[dict[str, Any]], +) -> dict[str, Any] | None: + """Keep a deferred Todo's old receipt visible without selecting new work.""" + + if any( + normalize_todo_id(source_item.get("todo_id")) == todo_id + and normalize_todo_status(source_item.get("status")) == TODO_STATUS_DEFERRED + for source_item in source_items + ): + return receipt_bound_deferred_work_lane(todo_id=todo_id) + return None + + def _prepare_quota_should_run_item( status_payload: dict[str, Any], *, @@ -736,12 +752,25 @@ def _prepare_quota_should_run_item( and candidate.get("selection_binding") == "heartbeat_receipt" ): receipt_bound_agent_next_action = candidate - preserved_work_lane = preserve_heartbeat_receipt_bound_work_lane( - work_lane_contract, - selected_todo=candidate, + work_lane_contract = ( + preserve_heartbeat_receipt_bound_work_lane( + work_lane_contract, + selected_todo=candidate, + ) + or work_lane_contract + ) + else: + # The old Turn still owns its committed settlement identity, but a + # deferred Todo is not an executable candidate. A successor may be + # selected only by a fresh Turn; do not leak it through work-lane + # fallback on this replay. + work_lane_contract = ( + _deferred_receipt_bound_work_lane( + todo_id=receipt_bound_todo_id, + source_items=agent_todo_planning_source_items, + ) + or work_lane_contract ) - if isinstance(preserved_work_lane, dict): - work_lane_contract = preserved_work_lane if inbox_priority_due: task_orchestration_contract = capability_gate = capability_monitor_contract = None capability_monitor_fallback = scoped_user_gate_fallback = workspace_guard = None diff --git a/loopx/control_plane/work_items/work_lane.py b/loopx/control_plane/work_items/work_lane.py index a214748c7f..b1647de469 100644 --- a/loopx/control_plane/work_items/work_lane.py +++ b/loopx/control_plane/work_items/work_lane.py @@ -39,6 +39,9 @@ def observe_work_lane( WORK_LANE_RECEIPT_BOUND_MONITOR_SETTLED_OBLIGATION = ( "finish_settled_receipt_bound_monitor_turn" ) +WORK_LANE_RECEIPT_BOUND_DEFERRED_OBLIGATION = ( + "wait_for_receipt_bound_deferred_todo" +) WORK_LANE_CURRENT_AGENT_MONITOR_REPAIR_OBLIGATIONS = { "attempt_due_monitor", "repair_monitor_schedule_metadata", @@ -278,6 +281,33 @@ def preserve_heartbeat_receipt_bound_work_lane( } +def receipt_bound_deferred_work_lane( + *, todo_id: str, +) -> dict[str, Any]: + """Keep an immutable Turn binding visible without executing a deferred Todo.""" + + normalized = normalize_todo_id(todo_id) + if not normalized: + raise ValueError("receipt-bound deferred work lane requires a Todo id") + return { + "schema_version": WORK_LANE_CONTRACT_SCHEMA_VERSION, + "lane": "advancement_task", + "obligation": WORK_LANE_RECEIPT_BOUND_DEFERRED_OBLIGATION, + "must_attempt_work": False, + "selection_binding": "heartbeat_receipt", + "selected_todo_id": normalized, + "reason_codes": [ + "heartbeat_receipt_bound_replay", + "receipt_bound_todo_deferred", + "successor_requires_fresh_turn", + ], + "action": ( + "the Todo bound to this heartbeat turn is deferred; do not execute " + "or spend this turn, and select independent work under a fresh turn" + ), + } + + def work_lane_contract_is_lark_inbox_reply_due( contract: dict[str, Any] | None, ) -> bool: diff --git a/tests/control_plane/test_quota_settlement_cli.py b/tests/control_plane/test_quota_settlement_cli.py index c8226c3eac..214d522873 100644 --- a/tests/control_plane/test_quota_settlement_cli.py +++ b/tests/control_plane/test_quota_settlement_cli.py @@ -3528,6 +3528,119 @@ def test_pending_deferred_p0_allows_independent_p1_selection( assert payload["action_selection_qualification"]["state"] == "qualified" +@pytest.mark.parametrize("provider", ["legacy", "file", "sqlite"]) +def test_same_turn_bound_p0_does_not_project_p1_after_p0_becomes_deferred( + tmp_path: Path, + provider: str, + monkeypatch: pytest.MonkeyPatch, +) -> None: + from canonical_authority_fixture import ( + initialize_canonical_authority, + isolate_sqlite_runtime, + ) + from loopx.control_plane.coordination.runtime_shadow import ( + build_todo_runtime_shadow_projection, + ) + + if provider == "sqlite": + isolate_sqlite_runtime(tmp_path, monkeypatch) + project, runtime, registry_path = _write_fixture(tmp_path) + _configure_selectable_alternative(project) + state_path = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md" + state_path.write_text( + state_path.read_text(encoding="utf-8").replace( + "[P1] Validate and settle the selected delivery.", + "[P0] Validate and settle the selected delivery.", + ), + encoding="utf-8", + ) + if provider != "legacy": + listed_rc, listed = _run_cli( + registry_path, runtime, "todo", "list", "--goal-id", GOAL_ID, + ) + assert listed_rc == 0, listed + projection = build_todo_runtime_shadow_projection( + goal_id=GOAL_ID, + handoff_mode="soft_claim", + todos=listed["todos"], + ) + initialize_canonical_authority( + runtime, GOAL_ID, projection, state_path=state_path, provider=provider, + ) + turn_id = "turn-bound-p0-then-deferred" + guard = ( + "quota", "should-run", "--codex-app", "--goal-id", GOAL_ID, + "--agent-id", AGENT_ID, "--turn-instance-id", turn_id, + "--scan-path", str(project), + ) + first_rc, first = _run_cli(registry_path, runtime, *guard, "--todo-id", TODO_ID) + assert first_rc == 0, first + assert first["heartbeat_receipt"]["settlement_identity"]["todo_id"] == TODO_ID + if provider == "legacy": + state_path.write_text( + state_path.read_text(encoding="utf-8").replace( + f"todo_id={TODO_ID} status=open", + f"todo_id={TODO_ID} status=deferred " + "resume_when=resume_at:2099-01-01T00:00:00Z", + ), + encoding="utf-8", + ) + else: + update_rc, update = _run_cli( + registry_path, runtime, + "todo", "update", "--goal-id", GOAL_ID, + "--agent-id", AGENT_ID, "--todo-id", TODO_ID, + "--status", "deferred", + "--resume-when", "resume_at:2099-01-01T00:00:00Z", + "--reason", "Wait for the future resume condition.", + ) + assert update_rc == 0, update + replay_rc, replay = _run_cli(registry_path, runtime, *guard) + assert replay_rc == 0, replay + assert replay["effective_action"] == "quota_skip" + assert replay["should_run"] is False + assert replay["heartbeat_receipt"]["status"] == "replayed" + assert replay["heartbeat_receipt"]["settlement_identity"]["todo_id"] == TODO_ID + assert replay["selected_todo"]["todo_id"] == TODO_ID + assert replay["work_lane_contract"]["obligation"] == ( + "wait_for_receipt_bound_deferred_todo" + ) + assert replay["work_lane_contract"]["must_attempt_work"] is False + assert replay["recommended_action"] == replay["work_lane_contract"]["action"] + assert "independent alternative delivery" not in replay["recommended_action"] + assert replay.get("agent_lane_next_action") is None + interaction = replay["interaction_contract"] + assert interaction["agent_channel"]["must_attempt"] is False + assert interaction["cli_channel"]["spend_after_validation"] is False + assert ALTERNATIVE_TODO_ID not in json.dumps( + interaction["cli_channel"].get("next_cli_actions", []) + ) + plan = interaction["cli_channel"].get("settlement_plan") + assert plan is None or plan["identity"]["todo_id"] == TODO_ID + assert _heartbeat_receipt_count(runtime, turn_id) == 1 + assert _spend_run_count(runtime) == 0 + + conflict_rc, conflict = _run_cli( + registry_path, runtime, *guard, "--todo-id", ALTERNATIVE_TODO_ID, + ) + assert conflict_rc != 0, conflict + assert conflict["error_code"] in { + "heartbeat_receipt_identity_conflict", + "quota_action_selection_rejected", + } + assert _heartbeat_receipt_count(runtime, turn_id) == 1 + + next_rc, next_turn = _run_cli( + registry_path, runtime, + "quota", "should-run", "--codex-app", + "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, + "--turn-instance-id", "turn-after-bound-p0-deferred", + "--scan-path", str(project), "--todo-id", ALTERNATIVE_TODO_ID, + ) + assert next_rc == 0, next_turn + assert next_turn["selected_todo"]["todo_id"] == ALTERNATIVE_TODO_ID + + def test_pending_selection_preserves_workspace_repair_then_reenters_same_turn( tmp_path: Path, ) -> None: