diff --git a/docs/quota-allocation.md b/docs/quota-allocation.md index 47b5804644..0e1cc6717d 100644 --- a/docs/quota-allocation.md +++ b/docs/quota-allocation.md @@ -868,6 +868,21 @@ including `generic_cli`, `outer_controller`, and native CLI profiles. An explici asking for the selection it just accepted. This does not enable `--begin-turn` for those hosts or change legacy calls without a Turn identity. +Hosts that already support `--begin-turn` may combine it with `--todo-id` to +qualify a known choice on the initial guard. Reuse the returned Turn identity +for recovery; beginning a Turn does not claim a lease or bypass admission. +With an open scoped user gate, an explicitly selected eligible action is checked +against that gate directly, including choices outside the compact display. +The default fallback ranking is not an authority boundary. Covered actions, +budget/health stops, capability requirements and committed receipt identities +retain their existing checks. + +已支持 `--begin-turn` 的宿主可以同时传入 `--todo-id`，在第一次 guard 中校验 +已选工作，减少一次选择调用。恢复时复用返回的 Turn 身份；开始 Turn 不等于 +取得租约或绕过准入。存在限定范围的用户审批时，显式选择的合格工作直接按该 +审批范围校验，即使它不在精简展示中。默认推荐顺序不构成权限边界；受审批 +约束的工作、预算和健康停止条件、能力要求及已绑定回执仍保留原有校验。 + A single already-claimed candidate keeps the direct execution path without an extra selection round trip. A sole unclaimed candidate instead projects the existing explicit selection contract: recommendation alone must not create a diff --git a/loopx/cli_commands/quota_request.py b/loopx/cli_commands/quota_request.py index 4d9ec8e011..127be33127 100644 --- a/loopx/cli_commands/quota_request.py +++ b/loopx/cli_commands/quota_request.py @@ -161,9 +161,9 @@ def validate_quota_command_request(args: argparse.Namespace) -> None: raise QuotaCommandValidationError( "`loopx quota void-slot` requires --void-generated-at" ) - if command == "should-run" and args.todo_id and not args.turn_instance_id: + if command == "should-run" and args.todo_id and not (args.turn_instance_id or begin_turn): raise QuotaCommandValidationError( - "`loopx quota should-run --todo-id` requires --turn-instance-id" + "`loopx quota should-run --todo-id` requires --turn-instance-id or --begin-turn" ) if begin_turn and command != "should-run": raise QuotaCommandValidationError( @@ -173,11 +173,6 @@ def validate_quota_command_request(args: argparse.Namespace) -> None: raise QuotaCommandValidationError( "--begin-turn cannot be combined with --turn-instance-id" ) - if begin_turn and args.todo_id: - raise QuotaCommandValidationError( - "--begin-turn cannot select --todo-id; reuse the exact " - "--turn-instance-id returned by the initial guard" - ) if begin_turn and not args.agent_id: raise QuotaCommandValidationError( "--begin-turn requires exact --agent-id identity" diff --git a/loopx/control_plane/agents/agent_scope.py b/loopx/control_plane/agents/agent_scope.py index fa6f6c2d18..13466b3991 100644 --- a/loopx/control_plane/agents/agent_scope.py +++ b/loopx/control_plane/agents/agent_scope.py @@ -217,6 +217,8 @@ def _scoped_user_gate_fallback( capability_gate: dict[str, Any] | None = None, allow_unrelated_gate: bool = False, monitor_debt_backoff_active: bool = False, + requested_todo_id: str | None = None, + requested_candidate: dict[str, Any] | None = None, ) -> dict[str, Any] | None: gates = _open_user_gate_todo_items(user_todo_summary) if not gates or not isinstance(agent_todo_summary, dict): @@ -258,6 +260,10 @@ def _scoped_user_gate_fallback( *ready_deferred_candidates, ] executable_items = [item for item in executable_items if isinstance(item, dict)] + # Exact selection was qualified from the complete inventory. Display lanes + # may omit it; transport the original row so gate metadata is not lost. + if requested_candidate is not None: + executable_items = [requested_candidate, *executable_items] claim_scope = ( agent_todo_summary.get("claim_scope") if isinstance(agent_todo_summary.get("claim_scope"), dict) @@ -266,6 +272,7 @@ def _scoped_user_gate_fallback( selection = select_scoped_gate_fallback( gates, executable_items, agent_id=claim_scope.get("agent_id") if claim_scope else None, allow_unrelated_gate=allow_unrelated_gate, monitor_debt_backoff_active=monitor_debt_backoff_active, + requested_todo_id=requested_todo_id, ) if selection is None: return None diff --git a/loopx/control_plane/quota/should_run_prepare.py b/loopx/control_plane/quota/should_run_prepare.py index fdb6894116..2b5355ef5e 100644 --- a/loopx/control_plane/quota/should_run_prepare.py +++ b/loopx/control_plane/quota/should_run_prepare.py @@ -740,12 +740,30 @@ def _prepare_quota_should_run_item( if task_orchestration_contract_is_actionable(task_orchestration_contract): capability_monitor_contract = capability_monitor_fallback = None work_lane_contract = capability_monitor_contract or work_lane_contract + # Resolve the exact choice before selecting a gate-independent route. The + # ranked fallback is advice, not an additional permission boundary. + requested_action_candidate = ( + build_explicit_advancement_next_action( + agent_identity=agent_identity, + agent_todo_items=agent_todo_planning_source_items, + available_capabilities=effective_available_capabilities, + todo_id=requested_action_todo_id, + selection_binding="pending_action_selection", + ) + if requested_action_todo_id + else None + ) scoped_user_gate_fallback = _scoped_user_gate_fallback( user_todo_summary, agent_todo_summary, capability_gate=capability_gate, allow_unrelated_gate=bool(quota.get("safe_bypass_allowed")), monitor_debt_backoff_active=bool(monitor_debt_arbitration.get("active")), + requested_todo_id=requested_action_todo_id, + requested_candidate=next(( + item for item in agent_todo_planning_source_items + if item.get("todo_id") == requested_action_todo_id + ), None) if requested_action_candidate is not None else None, ) work_lane_contract = ( scoped_user_gate_due_monitor_contract( @@ -912,19 +930,6 @@ def _prepare_quota_should_run_item( recovery_allowed = False reason = str(projection_gap_repair.get("reason") or reason) boundary_projection_repair = None - # Resolve exact identity before Agent/display compaction. Live callers supply - # the complete source; pure status callers use only their supplied snapshot. - requested_action_candidate = ( - build_explicit_advancement_next_action( - agent_identity=agent_identity, - agent_todo_items=agent_todo_planning_source_items, - available_capabilities=effective_available_capabilities, - todo_id=requested_action_todo_id, - selection_binding="pending_action_selection", - ) - if requested_action_todo_id - else None - ) state_path = goal_state_path(registry_goal) return _QuotaDecisionPreparation( status_payload=status_payload, diff --git a/loopx/control_plane/todos/decision_scope.py b/loopx/control_plane/todos/decision_scope.py index bf419d1ca1..3b4ec46c31 100644 --- a/loopx/control_plane/todos/decision_scope.py +++ b/loopx/control_plane/todos/decision_scope.py @@ -317,7 +317,8 @@ def todo_gate_relations(gates: list[dict[str, Any]], items: list[dict[str, Any]] def select_scoped_gate_fallback(gates: list[dict[str, Any]], items: list[dict[str, Any]], *, agent_id: str | None, allow_unrelated_gate: bool, - monitor_debt_backoff_active: bool) -> dict[str, Any] | None: + monitor_debt_backoff_active: bool, + requested_todo_id: str | None = None) -> dict[str, Any] | None: """Decode legacy facts; the typed owner returns positions, never display rows.""" if not gates or not items: return None @@ -336,6 +337,7 @@ def facts(item: dict[str, Any]) -> dict[str, Any]: "fallback", _evaluate("fallback", gates=[facts(gate) for gate in gates], candidates=[facts(item) for item in items], agent_id=normalize_todo_claimed_by(agent_id), allow_unrelated_gate=allow_unrelated_gate, + requested_todo_id=normalize_todo_id(requested_todo_id), monitor_debt_backoff_active=monitor_debt_backoff_active), schema_versions=frozenset({"scoped_gate_fallback_selection_v0"}), nullable=True, diff --git a/loopx/control_plane/todos/decision_scope.ts b/loopx/control_plane/todos/decision_scope.ts index 085a5cd588..da0c8faf11 100644 --- a/loopx/control_plane/todos/decision_scope.ts +++ b/loopx/control_plane/todos/decision_scope.ts @@ -97,6 +97,7 @@ export function fallbackGateRelation(gate: JsonObject, item: JsonObject): JsonOb * Return source positions, leaving display compaction and wording to adapters. */ export function selectScopedGateFallback(request: JsonObject): JsonObject | null { const agent = optionalNonEmptyString(request.agent_id, "agent_id"); + const requested = optionalNonEmptyString(request.requested_todo_id, "requested_todo_id"); const debt = requireBoolean(request.monitor_debt_backoff_active, "monitor_debt_backoff_active"); const allowUnrelated = requireBoolean(request.allow_unrelated_gate, "allow_unrelated_gate"); const gates = rows(request.gates).map((gate, index) => ({gate, index})).filter(({gate}) => @@ -129,7 +130,7 @@ export function selectScopedGateFallback(request: JsonObject): JsonObject | null if (match) { blockingGate ??= match; blocked.push({candidate_index: candidate.index, gate_index: match.index, relation: match.relation}); - } else selected ??= candidate; + } else if (!requested || candidate.item.todo_id === requested) selected ??= candidate; } if (!selected || (!blockingGate && !allowUnrelated)) return null; const surface = blockingGate ?? gates[0]!; diff --git a/tests/control_plane/test_quota_settlement_cli.py b/tests/control_plane/test_quota_settlement_cli.py index bfe0fbba30..682d905621 100644 --- a/tests/control_plane/test_quota_settlement_cli.py +++ b/tests/control_plane/test_quota_settlement_cli.py @@ -6313,6 +6313,7 @@ def test_pending_action_selection_does_not_commit_after_new_user_gate( "--priority", "P0", "--task-class", "user_gate", "--action-kind", "approve_delivery", "--blocks-agent", AGENT_ID, "--agent-id", AGENT_ID, "--decision-scope", "direction:action:fixture", + "--unblocks-todo-id", ALTERNATIVE_TODO_ID, ) assert added_rc == 0, added selected_rc, selected = _run_cli( diff --git a/tests/control_plane/test_scoped_gate_explicit_choice.py b/tests/control_plane/test_scoped_gate_explicit_choice.py new file mode 100644 index 0000000000..76e2eaf75f --- /dev/null +++ b/tests/control_plane/test_scoped_gate_explicit_choice.py @@ -0,0 +1,257 @@ +"""An unrelated approval does not turn the fallback ranking into authority.""" + +import json +import shlex +import subprocess +import sys + +import pytest + +from loopx.control_plane.quota.should_run import build_quota_should_run +from loopx.control_plane.testing.quota_fixtures import ( + quota_status_payload, + quota_todo_item, +) + + +AGENT = "scoped-choice-agent" +GOAL = "scoped-choice-goal" +SCOPE = {"kind": "production", "granularity": "action", "scope_key": "release"} + + +def choice_status(*, blocked=False, state="operator_gate"): + items = [quota_todo_item( + todo_id=name, index=index, priority=priority, claimed_by=AGENT, + title="Validate the selected change.", action_kind="validate", + required_decision_scopes=[SCOPE] if blocked and index == 2 else [], + ) for index, (name, priority) in enumerate([ + ("todo_recommended", "P0"), ("todo_alternative", "P1") + ], 1)] + gate = quota_todo_item( + todo_id="todo_release_gate", role="user", task_class="user_gate", + title="Approve the release.", blocks_agent=AGENT, decision_scope=SCOPE, + ) + return quota_status_payload( + goal_id=GOAL, status="active", quota_state=state, safe_bypass=True, + recommended_action="Validate an independent change.", + agent_todo_items=items, user_todo_items=[gate], + claim_scope_agent_id=AGENT, + coordination={"agent_model": "peer_v1", "registered_agents": [AGENT]}, + ) + + +def test_explicit_independent_choice_is_not_restricted_to_ranked_fallback(): + packet = build_quota_should_run( + choice_status(), goal_id=GOAL, agent_id=AGENT, + requested_action_todo_id="todo_alternative", + ) + assert packet["action_selection_qualification"]["state"] == "qualified" + assert packet["selected_todo"]["todo_id"] == "todo_alternative" + assert packet["interaction_contract"]["agent_channel"]["delivery_allowed"] + assert packet["requires_user_action"] # The release approval remains open. + + +@pytest.mark.parametrize("blocked,state,selected", [ + (True, "operator_gate", "todo_alternative"), + (False, "budget_exhausted", "todo_alternative"), + (False, "operator_gate", "todo_missing"), +]) +def test_explicit_choice_preserves_scope_budget_and_eligibility(blocked, state, selected): + packet = build_quota_should_run( + choice_status(blocked=blocked, state=state), goal_id=GOAL, agent_id=AGENT, + requested_action_todo_id=selected, + ) + assert packet["action_selection_qualification"]["state"] != "qualified" + assert not packet["interaction_contract"]["agent_channel"]["delivery_allowed"] + + +def test_default_fallback_is_unchanged(): + packet = build_quota_should_run(choice_status(), goal_id=GOAL, agent_id=AGENT) + assert packet["selected_todo"]["todo_id"] == "todo_recommended" + + +@pytest.mark.parametrize("provider", ["sqlite", "file"]) +def test_cli_begin_selects_once_and_replay_keeps_original_identity(tmp_path, provider): + from tests.control_plane.test_quota_settlement_cli import ( + AGENT_ID, ALTERNATIVE_TODO_ID, GOAL_ID, TODO_ID, + _configure_selectable_alternative, _initialize_fixture_authority, + _run_cli, _write_fixture, + ) + + project, runtime, registry = _write_fixture(tmp_path) + _configure_selectable_alternative(project) + state = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md" + state.write_text(state.read_text() + ( + "\n## User Todo\n\n- [ ] [P0] Approve the release.\n" + f" \n" + )) + _initialize_fixture_authority(project, runtime, provider=provider) + guard = ("quota", "should-run", "--runtime-profile", "codex_app_ssh_goal", + "--goal-id", GOAL_ID, "--agent-id", AGENT_ID) + rc, first = _run_cli(registry, runtime, *guard, "--begin-turn", + "--todo-id", ALTERNATIVE_TODO_ID) + assert rc == 0, first.get("error") + assert first["selected_todo"]["todo_id"] == ALTERNATIVE_TODO_ID + assert first["interaction_contract"]["agent_channel"]["delivery_allowed"] + turn = first["heartbeat_receipt"]["turn_instance_id"] + rc, replay = _run_cli(registry, runtime, *guard, "--turn-instance-id", turn, + "--todo-id", ALTERNATIVE_TODO_ID) + assert rc == 0, replay.get("error") + assert replay["selected_todo"]["todo_id"] == ALTERNATIVE_TODO_ID + rc, refused = _run_cli(registry, runtime, *guard, "--turn-instance-id", turn, + "--todo-id", TODO_ID) + assert rc != 0 + assert refused["error_code"] == "heartbeat_receipt_identity_conflict" + + +@pytest.mark.parametrize("provider", ["sqlite", "file"]) +def test_rejected_turn_recovers_to_validated_work_and_one_spend(tmp_path, provider): + """A real scoped rejection must recover progress, not just an admission packet.""" + from tests.control_plane.test_quota_settlement_cli import ( + AGENT_ID, ALTERNATIVE_TODO_ID, GOAL_ID, SECOND_ALTERNATIVE_TODO_ID, + TODO_ID, TURN_ID, _configure_selectable_alternative, + _heartbeat_receipt_count, _initialize_fixture_authority, + _run_cli, _run_generated_cli, _spend_run_count, _write_fixture, + ) + + project, runtime, registry = _write_fixture(tmp_path) + _configure_selectable_alternative(project) + state = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md" + state.write_text(state.read_text().replace( + "Advance the independent alternative delivery.", + "Compute and validate the input count and total.", + ).replace( + f"todo_id={TODO_ID} status=open", + f"todo_id={TODO_ID} required_decision_scopes=production:action:release status=open", + ) + ( + "\n## User Todo\n\n- [ ] [P0] Approve the release.\n" + f" \n" + )) + initial = _initialize_fixture_authority(project, runtime, provider=provider) + + def cli(*args): + return _run_cli(registry, runtime, *args, cwd=project) + + def projected(command): + return _run_generated_cli(command, registry_path=registry, cwd=project) + + def read_todos(): + rc, result = cli("todo", "list", "--goal-id", GOAL_ID) + assert rc == 0, result + return result + + # The validator independently derives acceptance from input, not a supplied + # result hash. Its arguments and all state are confined to this fixture. + (project / "counts.json").write_text("[3, 5, 7]\n") + artifact = project / "summary.json" + validator = [sys.executable, "-c", ( + "import json,sys; from pathlib import Path; " + "x=json.loads(Path(sys.argv[1]).read_text()); " + "y=json.loads(Path(sys.argv[2]).read_text()); " + "assert y == {'count':len(x),'total':sum(x)}" + ), str(project / "counts.json"), str(artifact)] + rc, configured = cli( + "todo", "update", "--goal-id", GOAL_ID, "--todo-id", ALTERNATIVE_TODO_ID, + "--agent-id", AGENT_ID, + "--validation-command-json", json.dumps(validator), + "--update-operation-id", "configure-independent-validation", + "--update-expected-provider-revision", initial["provider_revision"], + ) + assert rc == 0, json.dumps(configured) + before = read_todos() + guard = ("quota", "should-run", "--runtime-profile", "codex_app_heartbeat", + "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, + "--turn-instance-id", TURN_ID) + rc, rejected = cli(*guard, "--todo-id", TODO_ID) + assert rc == 1, rejected + assert rejected["error_code"] == "quota_action_selection_deferred" + assert rejected["action_selection_qualification"]["reason"] == "delivery_not_allowed" + assert rejected["heartbeat_receipt"]["status"] == "not_committed" + assert _heartbeat_receipt_count(runtime, TURN_ID) == 0 + assert _spend_run_count(runtime) == 0 + reentry = rejected["interaction_contract"]["cli_channel"]["next_cli_actions"][0] + args = shlex.split(reentry) + assert args[args.index("--turn-instance-id") + 1] == TURN_ID + assert "--todo-id" not in args and "--begin-turn" not in args + rc, recovered = projected(reentry) + assert rc == 0, recovered + # Scoped fallback selects independent work on reentry. Explicit replay must + # retain that same selection, rather than create or silently rebind a Turn. + assert recovered["selected_todo"]["todo_id"] == ALTERNATIVE_TODO_ID + rc, selected = cli(*guard, "--todo-id", ALTERNATIVE_TODO_ID) + assert rc == 0, selected + identity = selected["heartbeat_receipt"]["settlement_identity"] + assert identity == recovered["heartbeat_receipt"]["settlement_identity"] + assert identity["turn_instance_id"] == TURN_ID + assert selected["interaction_contract"]["agent_channel"]["delivery_allowed"] + assert selected["selected_todo"]["claim_required_before_work"] + rc, claim = cli("todo", "claim", "--goal-id", GOAL_ID, + "--todo-id", ALTERNATIVE_TODO_ID, + "--agent-id", AGENT_ID, "--claimed-by", AGENT_ID) + assert rc == 0 and claim["claimed_by"] == AGENT_ID, claim + + artifact.write_text('{"count":3,"total":14}\n') + assert subprocess.run(validator, cwd=project, capture_output=True).returncode != 0 + artifact.write_text('{"count":3,"total":15}\n') + verified = subprocess.run(validator, cwd=project, capture_output=True) + assert verified.returncode == 0, verified.stderr + actions = selected["interaction_contract"]["cli_channel"]["next_cli_actions"] + refresh_command = next(command for command in actions if " refresh-state " in command) + refresh_command = (refresh_command.replace("", "validated_progress") + .replace("", "implementation") + .replace("", "outcome_progress")) + rc, refresh = projected(refresh_command + ( + " --delivery-boundary in_flight_continuation --delivery-workspace-path " + + shlex.quote(str(project)) + " --no-global-sync --suppress-external-sinks" + )) + assert rc == 0, refresh + assert refresh["settlement_identity"] == identity + assert refresh["settlement_progress"]["state"] == "spend_required" + spend_command = refresh["settlement_owed"]["command"] + args = shlex.split(spend_command) + assert args[args.index("--source") + 1] == "heartbeat" + assert "--execute" in args + rc, spend = projected(spend_command) + assert rc == 0, spend + assert spend["settlement_identity"] == identity + assert spend["delivery_workspace_validated"] + assert spend["settlement_progress"]["state"] == "settled" + rc, retried = projected(spend_command) + assert rc == 0 and retried["appended"] is False, retried + assert _spend_run_count(runtime) == 1 + + rc, completed = cli( + "todo", "complete", "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, + "--todo-id", ALTERNATIVE_TODO_ID, "--turn-instance-id", TURN_ID, + "--evidence", "Independent validator checked summary.json against counts.json.", + "--no-follow-up", + ) + assert rc == 0 and completed["completed"], json.dumps(completed) + assert completed["validation_receipt"]["passed"] is True + after = read_todos() + # These independent CLI reads include the canonical records, not Markdown + # written by the work producer. The original gate and covered Todo survive. + def rows(payload): + return {row["todo_id"]: row for row in payload["todos"]} + + old_rows, new_rows = rows(before), rows(after) + for todo_id in (TODO_ID, "todo_release_gate"): + assert new_rows[todo_id] == old_rows[todo_id] + assert new_rows[ALTERNATIVE_TODO_ID]["status"] == "done" + assert subprocess.run(validator, cwd=project, capture_output=True).returncode == 0 + rc, historical = cli(*guard) + assert rc == 0, historical + assert historical["heartbeat_receipt"]["settlement_identity"] == identity + assert not historical["interaction_contract"]["agent_channel"]["delivery_allowed"] + assert historical["interaction_contract"]["mode"] == "heartbeat_settled_skip" + new_guard = tuple("turn-next-independent" if arg == TURN_ID else arg for arg in guard) + rc, fresh = cli(*new_guard, "--todo-id", ALTERNATIVE_TODO_ID) + assert rc == 1 and fresh["error_code"] == "quota_action_selection_rejected", fresh + rc, continuation = projected( + fresh["interaction_contract"]["cli_channel"]["next_cli_actions"][0]) + assert rc == 0, continuation + assert continuation["selected_todo"]["todo_id"] == SECOND_ALTERNATIVE_TODO_ID + assert continuation["interaction_contract"]["agent_channel"]["delivery_allowed"] + assert _spend_run_count(runtime) == 1