diff --git a/docs/quota-allocation.md b/docs/quota-allocation.md index 860b1fc4b7..3fef6d339a 100644 --- a/docs/quota-allocation.md +++ b/docs/quota-allocation.md @@ -642,6 +642,50 @@ empty results; missing/stale display and provider failure never authorize a Markdown fallback. The guard does not append a second Markdown candidate list. Historical receipt-bound recovery remains separate from new work admission. +The final current-selection admission and recovery predicate has one TypeScript +owner. CLI preflight and interaction construction use that projection; the Effect +interpreter, new TurnEnvelope construction and scoped-override projection also +reject an unadmitted deferred/rejected packet that still carries execution, +spend, safe-bypass or executable settlement fields. A missing typed recovery action +fails explicitly. The recovery command itself remains valid: these checks do not +classify command prose or remove the same-Turn re-entry route. + +Admission reads the canonical agent channel: pending delivery requires both +`must_attempt=true` and `delivery_allowed=true`; pending workspace repair requires +its explicit repair route and `delivery_allowed=false`. Missing delivery permission +is unknown, not a grant. Compatibility `execution_obligation` annotations cannot +independently grant or veto final admission. Its producer rejects non-boolean +permission values before constructing the canonical channel; null/missing values +do not grant permission. This tightens malformed/contradictory selected-Todo inputs +without changing valid receipt-bound or current-work behavior. + +Receipt-bound work stays with its reconciliation owner. Existing admitted current +work and explicitly qualified workspace repair retain their exceptions; an +unrelated pending selection cannot borrow those facts. These are consistency and +preflight projections over the current guard, not lease, Goal-identity or provider +authorization. Calls without a qualification keep their existing path. Python +retains command/structural adaptation and shares one recovery decision across the +two interaction channels; it does not keep a parallel admission predicate. + +准入读取正式 agent channel:待选交付要求 `must_attempt=true` 且 +`delivery_allowed=true`;待选工作区修复要求明确的修复路径及 +`delivery_allowed=false`。缺失交付权限表示未知,不能当作授权。兼容字段 +`execution_obligation` 不能独立授予或否决最终准入;构造正式通道前会拒绝 +非布尔权限值,null/缺失值不授予权限。这会收紧非法或矛盾的选择输入,合法的 +回执绑定和当前工作路径保持不变。 + +最终的当前工作准入和选择恢复判定统一由 TypeScript 持有,CLI preflight 与交互 +构造复用该投影。Effect 解释器、新 TurnEnvelope 构造和 scoped-override 投影也 +拒绝未获准的 deferred/rejected packet 中残留的执行、扣费、safe-bypass 或可执行 +结算字段;缺失类型化 recovery action 会明确失败。恢复命令本身仍有效,这些检查 +不按命令文案分类,也不删除同 Turn 的重新准入路径。 + +已绑定回执的工作仍由回执协调 owner 处理;既有已获准工作和明确允许的工作区 +修复保留各自例外,不相关的待选择工作不能借用这些事实。此处只做当前 guard 的 +一致性和 preflight 投影,不验证租约、Goal 身份或 provider 权限。没有 qualification +的调用保留原路径。Python 保留命令与结构适配,两条交互 channel 复用一次恢复 +判定,不再维护平行的准入规则。 + Hosts supplying their own `--turn-instance-id` use this same selection contract, including `generic_cli`, `outer_controller`, and native CLI profiles. An explicit `--todo-id` is qualified before receipt binding; it must not leave the response diff --git a/loopx/cli_commands/quota_action_selection.py b/loopx/cli_commands/quota_action_selection.py index 64ab8197cd..52aed66051 100644 --- a/loopx/cli_commands/quota_action_selection.py +++ b/loopx/cli_commands/quota_action_selection.py @@ -161,12 +161,13 @@ def _requested_quota_action_selection_preflight( # which side of the delivery boundary refused it without re-deriving it from # prose. agent_must_attempt = agent_channel.get("must_attempt") is True - agent_delivery_allowed = agent_channel.get("delivery_allowed") is not False + delivery = agent_channel.get("delivery_allowed") + agent_delivery_allowed = delivery if isinstance(delivery, bool) else None selected_todo_id, admitted = current_action_selection_admission( payload, requested_todo_id=requested_todo_id, agent_must_attempt=agent_must_attempt, - agent_delivery_refused=not agent_delivery_allowed, + agent_delivery_allowed=agent_delivery_allowed, ) if receipt_bound_replan_obligation_id: if not receipt_identity_upgraded: diff --git a/loopx/control_plane/effect_program.ts b/loopx/control_plane/effect_program.ts index 1c5cfe4402..f7eb4fd118 100644 --- a/loopx/control_plane/effect_program.ts +++ b/loopx/control_plane/effect_program.ts @@ -1,3 +1,4 @@ +import {assertActionSelectionConsistency} from "./work_items/action_selection_admission.ts"; import { EffectRuntimeRequestError } from "./effect_runtime_errors.ts"; import { isStringLiteral } from "./runtime_decode.ts"; import { decodeInteractionContract } from "./work_items/interaction_contract.ts"; @@ -305,6 +306,7 @@ export function interpretQuotaShouldRunPacket( ): EffectTurn { const packet = asObject(packetValue); const interaction = decodeInteractionContract(packet.interaction_contract); + assertActionSelectionConsistency(packet, interaction); const lane = asObject(packet.work_lane_contract); const scheduler = asObject(packet.scheduler_hint); const codexApp = asObject(scheduler.codex_app); diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index 9558769848..b4db14fc2d 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -337,6 +337,8 @@ export function createEffectRuntimeHandlers( ["turn.delivery_route.evaluate", lazyHandler(() => import("./turn_driver/delivery_continuity.ts"), ({evaluateDeliveryRoute}) => evaluateDeliveryRoute)], ["work_item.action_portfolio.project", lazyHandler(() => import("./work_items/action_portfolio.ts"), ({projectQuotaActionPortfolio}) => projectQuotaActionPortfolio)], ["work_item.action_selection.qualify", lazyHandler(() => import("./work_items/action_portfolio.ts"), ({qualifyActionSelection}) => qualifyActionSelection)], + ["work_item.action_selection.admission.project", lazyHandler(() => import("./work_items/action_selection_admission.ts"), ({projectCurrentActionSelectionAdmission}) => projectCurrentActionSelectionAdmission)], + ["work_item.action_selection.recovery.project", lazyHandler(() => import("./work_items/action_selection_admission.ts"), ({projectActionSelectionRecovery}) => projectActionSelectionRecovery)], [ "work_item.action_selection.reconcile_retained", lazyHandler(() => import("./work_items/action_portfolio.ts"), ({reconcileRetainedActionSelection}) => reconcileRetainedActionSelection), diff --git a/loopx/control_plane/quota/scoped_override.ts b/loopx/control_plane/quota/scoped_override.ts index 7e0c887df5..9fd445ac58 100644 --- a/loopx/control_plane/quota/scoped_override.ts +++ b/loopx/control_plane/quota/scoped_override.ts @@ -1,6 +1,7 @@ import type { JsonObject } from "../effect_program.ts"; import { requireJsonObject, requireNonEmptyString } from "../runtime_decode.ts"; import { decodeInteractionContract } from "../work_items/interaction_contract.ts"; +import {assertActionSelectionConsistency} from "../work_items/action_selection_admission.ts"; /** Project diagnostics after selection, guards, settlement and notification overlays. * Gate admission metadata is historical context; it cannot select another Todo. @@ -8,6 +9,9 @@ import { decodeInteractionContract } from "../work_items/interaction_contract.ts export function projectScopedOverride(params: JsonObject): JsonObject { const override = { ...requireJsonObject(params.override, "override") }; const interaction = decodeInteractionContract(params.interaction_contract); + if (params.selection_context !== undefined) { + assertActionSelectionConsistency(requireJsonObject(params.selection_context, "selection_context"), interaction); + } delete override.selected_action; if (interaction.agent_channel.delivery_allowed && params.selected_todo != null) { const selected = requireJsonObject(params.selected_todo, "selected_todo"); diff --git a/loopx/control_plane/quota/should_run_packet.py b/loopx/control_plane/quota/should_run_packet.py index 293fc529ec..44788e78bf 100644 --- a/loopx/control_plane/quota/should_run_packet.py +++ b/loopx/control_plane/quota/should_run_packet.py @@ -112,6 +112,7 @@ ) from ..work_items.execution_obligation import build_execution_obligation from ..work_items.action_selection_contract import ( + action_selection_context, apply_action_selection_recovery_projection, ) from ..work_items.goal_route_hint import build_goal_route_hint @@ -1643,6 +1644,7 @@ def _build_quota_should_run_payload( "override": payload[kind], "selected_todo": payload.get("selected_todo"), "interaction_contract": payload["interaction_contract"], + "selection_context": action_selection_context(payload), }) return payload diff --git a/loopx/control_plane/todos/identity.ts b/loopx/control_plane/todos/identity.ts new file mode 100644 index 0000000000..4a69382966 --- /dev/null +++ b/loopx/control_plane/todos/identity.ts @@ -0,0 +1,4 @@ +/** Canonical grammar only: existing callers retain normalization and error policy. */ +export function isCanonicalTodoId(value: string): boolean { + return /^todo_[a-z0-9_-]{3,64}$/u.test(value); +} diff --git a/loopx/control_plane/work_items/action_selection_admission.ts b/loopx/control_plane/work_items/action_selection_admission.ts new file mode 100644 index 0000000000..c8f64a8409 --- /dev/null +++ b/loopx/control_plane/work_items/action_selection_admission.ts @@ -0,0 +1,99 @@ +/** Final selected-work admission, shared by CLI recovery and packet readers. + * Receipt labels are reconciliation facts, never proof of current lease authority. */ +import type {JsonObject} from "../effect_program.ts"; +import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts"; +import {jsonObject, requireBoolean, requireJsonObject} from "../runtime_decode.ts"; +import {EffectiveAction} from "../quota/effective_action.generated.ts"; +import {isCanonicalTodoId} from "../todos/identity.ts"; + +interface AgentAdmissionFacts {must_attempt: boolean; delivery_allowed: boolean | null} +interface SelectionAdmission {selected_todo_id: string | null; admitted: boolean} + +function selectionTodoId(value: unknown): string | null { + if (typeof value !== "string") return null; + // Preserve the Python selection reader's strip, not JS trim's BOM policy. + const whitespace = /[\p{White_Space}\u001c-\u001f]/u; + let start = 0, end = value.length; + while (start < end && whitespace.test(value[start]!)) start++; + while (end > start && whitespace.test(value[end - 1]!)) end--; + const candidate = value.slice(start, end).toLowerCase(); + return isCanonicalTodoId(candidate) ? candidate : null; +} + +export function currentActionSelectionAdmission( + packet: JsonObject, requestedTodoId: string, agent: AgentAdmissionFacts, +): SelectionAdmission { + const selected = jsonObject(packet.selected_todo) ?? {}; + const qualification = jsonObject(packet.action_selection_qualification) ?? {}; + let selectedId = selectionTodoId(selected.todo_id); + if (selectedId === null && qualification.state === "qualified") { + selectedId = selectionTodoId(jsonObject(qualification.selected_todo)?.todo_id); + } + const pending = selected.selection_binding === "pending_action_selection"; + const delivery = pending && packet.normal_delivery_allowed === true + && agent.must_attempt && agent.delivery_allowed === true; + const workspaceRepair = pending && packet.workspace_repair_allowed === true + && packet.effective_action === EffectiveAction.AGENT_WORKSPACE_REPAIR + && agent.must_attempt && agent.delivery_allowed === false; + const currentWork = !pending && agent.must_attempt; + return {selected_todo_id: selectedId, + admitted: selectedId !== null && selectedId === requestedTodoId + && packet.ok === true && packet.should_run === true && (delivery || workspaceRepair || currentWork)}; +} + +export function actionSelectionNeedsRecovery(packet: JsonObject, agent: AgentAdmissionFacts): boolean { + const qualification = jsonObject(packet.action_selection_qualification); + if (qualification === null) return false; + if (qualification.state !== null && typeof qualification.state === "object") { + throw new EffectRuntimeRequestError("action selection qualification state must be scalar"); + } + if (qualification.state !== "deferred" && qualification.state !== "rejected") return false; + const selected = jsonObject(packet.selected_todo), replan = jsonObject(packet.autonomous_replan_obligation); + if (selected?.selection_binding === "heartbeat_receipt" || replan?.selection_binding === "heartbeat_receipt") return false; + const requested = typeof qualification.requested_todo_id === "string" ? qualification.requested_todo_id : ""; + if (currentActionSelectionAdmission(packet, requested, agent).admitted) return false; + if (qualification.recovery_action !== "reenter_guard_without_selection") { + throw new EffectRuntimeRequestError("rejected action selection omitted its typed recovery action"); + } + return true; +} + +function admissionRequest(request: JsonObject) { + return {packet: requireJsonObject(request.packet, "selection admission packet"), agent: { + must_attempt: requireBoolean(request.agent_must_attempt, "agent_must_attempt"), + delivery_allowed: request.agent_delivery_allowed === null ? null + : requireBoolean(request.agent_delivery_allowed, "agent_delivery_allowed"), + }}; +} + +export function projectCurrentActionSelectionAdmission(request: JsonObject): JsonObject { + const {packet, agent} = admissionRequest(request); + if (typeof request.requested_todo_id !== "string") throw new EffectRuntimeRequestError("requested_todo_id must be a string"); + return {...currentActionSelectionAdmission(packet, request.requested_todo_id, agent)}; +} + +export function projectActionSelectionRecovery(request: JsonObject): JsonObject { + const {packet, agent} = admissionRequest(request); + return {required: actionSelectionNeedsRecovery(packet, agent)}; +} + +const ROOT_EXECUTION_FLAGS = ["should_run", "normal_delivery_allowed", "recovery_delivery_allowed", + "self_repair_allowed", "capability_repair_allowed", "workspace_repair_allowed", "safe_bypass_allowed", + "actionable_by_codex", "spend_allowed_now", "spend_after_validation"] as const; + +/** Reject stale grants without inspecting command prose or consuming receipts. */ +export function assertActionSelectionConsistency(packet: JsonObject, interaction: JsonObject): void { + const agent = jsonObject(interaction.agent_channel) ?? {}, cli = jsonObject(interaction.cli_channel) ?? {}; + if (!actionSelectionNeedsRecovery(packet, { + must_attempt: agent.must_attempt === true, + delivery_allowed: typeof agent.delivery_allowed === "boolean" ? agent.delivery_allowed : null, + })) return; + const granted = (value: unknown) => value !== undefined && value !== null && value !== false; + if (ROOT_EXECUTION_FLAGS.some(field => granted(packet[field])) + || granted(agent.must_attempt) || granted(agent.delivery_allowed) + || granted(cli.spend_allowed_now) || granted(cli.spend_after_validation) || granted(cli.selection_required) + || ["settlement_plan", "replan_settlement_contract", "selection_command", "selection_policy_ref"] + .some(field => cli[field] !== undefined && cli[field] !== null)) { + throw new EffectRuntimeRequestError("unadmitted action selection cannot carry execution or settlement authority"); + } +} diff --git a/loopx/control_plane/work_items/action_selection_contract.py b/loopx/control_plane/work_items/action_selection_contract.py index 3bb3d00fc8..9f2041f049 100644 --- a/loopx/control_plane/work_items/action_selection_contract.py +++ b/loopx/control_plane/work_items/action_selection_contract.py @@ -157,102 +157,68 @@ def delivery_spend_allowed( ) +def action_selection_context(payload: Mapping[str, Any]) -> dict[str, Any]: + """Transport bounded current facts; TypeScript owns every admission decision.""" + fields = ("ok", "should_run", "effective_action", "normal_delivery_allowed", "recovery_delivery_allowed", + "self_repair_allowed", "capability_repair_allowed", "workspace_repair_allowed", "safe_bypass_allowed", + "actionable_by_codex", "spend_allowed_now", "spend_after_validation") + result = {key: payload[key] for key in fields if key in payload} + for field, keys in ( + ("selected_todo", ("todo_id", "selection_binding")), + ("autonomous_replan_obligation", ("selection_binding",)), + ("action_selection_qualification", ("state", "requested_todo_id", "recovery_action")), + ): + raw = payload.get(field) + if isinstance(raw, Mapping): + result[field] = {key: raw[key] for key in keys if key in raw} + if field == "selected_todo": + result[field]["todo_id"] = normalize_todo_id(raw.get("todo_id")) + if field == "action_selection_qualification" and isinstance(raw.get("selected_todo"), Mapping): + result[field]["selected_todo"] = {"todo_id": normalize_todo_id(raw["selected_todo"].get("todo_id"))} + return result + + +def _selection_projection(method: str, payload: Mapping[str, Any], *, + agent_must_attempt: bool, agent_delivery_allowed: bool | None, + requested_todo_id: str | None = None) -> dict[str, Any]: + from ..effect_runtime import effect_runtime_result + + request = {"packet": action_selection_context(payload), "agent_must_attempt": agent_must_attempt, + "agent_delivery_allowed": agent_delivery_allowed} + if requested_todo_id is not None: + request["requested_todo_id"] = requested_todo_id + result = effect_runtime_result(method, request) + if not isinstance(result, dict): + raise TypeError("typed action selection returned an invalid projection") + return result + + def current_action_selection_admission( payload: Mapping[str, Any], *, requested_todo_id: str, - agent_must_attempt: bool, agent_delivery_refused: bool, + agent_must_attempt: bool, agent_delivery_allowed: bool | None, ) -> tuple[str | None, bool]: - """Read current obligation admission, including repair and monitor positives.""" - selected_todo = payload.get("selected_todo") - selected_todo_id = ( - normalize_todo_id(selected_todo.get("todo_id")) - if isinstance(selected_todo, Mapping) - else None - ) - qualification_value = payload.get("action_selection_qualification") - qualification: Mapping[str, object] = ( - qualification_value if isinstance(qualification_value, Mapping) else {} - ) - if selected_todo_id is None and str(qualification.get("state") or "") == ( - "qualified" - ): - # An unsettled-host-turn recovery decision carries no top-level - # `selected_todo`: its qualification names the Todo that prior Turn - # has to settle, and binding the guard to that Todo is the documented - # closeout path rather than a conflict with the projection. - qualification_selected = qualification.get("selected_todo") - selected_todo_id = ( - normalize_todo_id(qualification_selected.get("todo_id")) - if isinstance(qualification_selected, Mapping) - else None - ) - selection_binding = ( - selected_todo.get("selection_binding") - if isinstance(selected_todo, Mapping) - else None - ) - execution_obligation_value = payload.get("execution_obligation") - execution_obligation: Mapping[str, object] = ( - execution_obligation_value - if isinstance(execution_obligation_value, Mapping) - else {} - ) - pending_selection_delivery_qualified = ( - selection_binding == "pending_action_selection" - and payload.get("normal_delivery_allowed") is True - ) - pending_selection_workspace_repair_qualified = ( - selection_binding == "pending_action_selection" - and payload.get("workspace_repair_allowed") is True - and payload.get("effective_action") == EffectiveAction.AGENT_WORKSPACE_REPAIR.value - and execution_obligation.get("kind") == "agent_workspace_repair" - and execution_obligation.get("must_attempt_work") is True - and agent_must_attempt - and agent_delivery_refused - ) - exact_current_obligation_qualified = ( - selection_binding != "pending_action_selection" - and execution_obligation.get("must_attempt_work") is True - and agent_must_attempt - ) - if ( - selected_todo_id == requested_todo_id - and payload.get("ok") is True - and payload.get("should_run") is True - and ( - pending_selection_delivery_qualified - or pending_selection_workspace_repair_qualified - or exact_current_obligation_qualified - ) - ): - return selected_todo_id, True - - return selected_todo_id, False + result = _selection_projection("work_item.action_selection.admission.project", payload, + requested_todo_id=requested_todo_id, agent_must_attempt=agent_must_attempt, + agent_delivery_allowed=agent_delivery_allowed) + selected, admitted = result.get("selected_todo_id"), result.get("admitted") + if (selected is not None and not isinstance(selected, str)) or type(admitted) is not bool: + raise TypeError("typed action selection returned invalid admission facts") + return selected, admitted def action_selection_needs_recovery( payload: Mapping[str, Any], *, agent_must_attempt: bool = False, - agent_delivery_refused: bool = False, + agent_delivery_allowed: bool | None = None, ) -> bool: - """Read the typed qualifier's result without reimplementing admission.""" - qualification = payload.get("action_selection_qualification") - if not isinstance(qualification, Mapping) or qualification.get("state") not in {"deferred", "rejected"}: + # No selected-request input means there is no recovery projection to read. + if not isinstance(payload.get("action_selection_qualification"), Mapping): return False - # A committed binding is reconciled by the receipt owner, not by pending - # selection recovery. Preserve the same exemption as CLI preflight. - selected = payload.get("selected_todo") - replan = payload.get("autonomous_replan_obligation") - if any(isinstance(value, Mapping) and value.get("selection_binding") == "heartbeat_receipt" - for value in (selected, replan)): - return False - _, admitted = current_action_selection_admission( - payload, requested_todo_id=str(qualification.get("requested_todo_id") or ""), - agent_must_attempt=agent_must_attempt, agent_delivery_refused=agent_delivery_refused, - ) - if admitted: - return False - if qualification.get("recovery_action") != "reenter_guard_without_selection": - raise RuntimeError("rejected action selection omitted its typed recovery action") - return True + result = _selection_projection("work_item.action_selection.recovery.project", payload, + agent_must_attempt=agent_must_attempt, agent_delivery_allowed=agent_delivery_allowed) + required = result.get("required") + if type(required) is not bool: + raise TypeError("typed action selection returned an invalid recovery decision") + return required def action_selection_recovery_fields( diff --git a/loopx/control_plane/work_items/interaction_contract.py b/loopx/control_plane/work_items/interaction_contract.py index 0d583bde44..acf9215976 100644 --- a/loopx/control_plane/work_items/interaction_contract.py +++ b/loopx/control_plane/work_items/interaction_contract.py @@ -80,7 +80,7 @@ class InteractionContractPacket(_InteractionContractRequired, total=False): fallback_policy: dict[str, Any] -def _blocked_successor_wait_observation_required(payload: dict[str, Any]) -> bool: +def _blocked_successor_wait_observation_required(payload: Mapping[str, object]) -> bool: return bool( payload.get("effective_action") == AgentScopeFrontierAction.AGENT_SCOPE_WAIT.value @@ -170,12 +170,12 @@ def user_channel_action_required(payload: dict[str, Any]) -> bool: ) or user_action_owns_empty_agent_lane(payload) -def _user_gate_notification_suppressed(payload: dict[str, Any]) -> bool: +def _user_gate_notification_suppressed(payload: Mapping[str, object]) -> bool: cooldown = payload.get("user_gate_notification_cooldown") return isinstance(cooldown, dict) and cooldown.get("notification_suppressed") is True -def _user_gate_scope_projection_repair_active(payload: dict[str, Any]) -> bool: +def _user_gate_scope_projection_repair_active(payload: Mapping[str, object]) -> bool: repair = payload.get("stall_self_repair") return bool( isinstance(repair, dict) @@ -763,6 +763,29 @@ def interaction_next_cli_actions( turn_instance_id=turn_instance_id, runtime_root=runtime_root, registry_path=registry_path, )] + return _admitted_interaction_next_cli_actions( + payload, mode=mode, available_capabilities=available_capabilities, + scheduler_execution_context=scheduler_execution_context, capability_reentry=capability_reentry, + capability_reentry_resolved=capability_reentry_resolved, settlement_plan=settlement_plan, + turn_instance_id=turn_instance_id, runtime_root=runtime_root, registry_path=registry_path, + ) + + +def _admitted_interaction_next_cli_actions( + payload: dict[str, Any], + *, + mode: str, + available_capabilities: Any = None, + scheduler_execution_context: ( + Mapping[str, Any] | SchedulerExecutionContextResolution | None + ) = None, + capability_reentry: dict[str, Any] | None = None, + capability_reentry_resolved: bool = False, + settlement_plan: Mapping[str, Any] | None = None, + turn_instance_id: str | None = None, + runtime_root: str | None = None, + registry_path: str | None = None, +) -> list[str]: goal_id = str(payload.get("goal_id") or "") command_prefix = selection.render_cli_command_prefix(runtime_root=runtime_root, registry_path=registry_path) agent_identity = payload.get("agent_identity") if isinstance(payload.get("agent_identity"), dict) else {} @@ -1351,6 +1374,7 @@ def _build_interaction_cli_channel( heartbeat_recommendation: dict[str, Any], *, mode: str, + selection_recovery_required: bool, spend_after_validation: bool, available_capabilities: Any = None, scheduler_execution_context: ( @@ -1361,7 +1385,7 @@ def _build_interaction_cli_channel( runtime_root: str | None = None, registry_path: str | None = None, ) -> dict[str, Any]: - if unadmitted_action_selection(payload): + if selection_recovery_required: return selection.action_selection_recovery_cli_channel(_selection_recovery_command( payload, available_capabilities=available_capabilities, scheduler_execution_context=scheduler_execution_context, @@ -1380,7 +1404,7 @@ def _build_interaction_cli_channel( ) ) channel = { - "next_cli_actions": interaction_next_cli_actions( + "next_cli_actions": _admitted_interaction_next_cli_actions( payload, mode=mode, available_capabilities=available_capabilities, @@ -1639,7 +1663,7 @@ def _attach_interaction_vision_wait_state( } -def _interaction_fallback_policy_required(payload: dict[str, Any], *, mode: str) -> bool: +def _interaction_fallback_policy_required(payload: Mapping[str, object], *, mode: str) -> bool: return mode in { "user_gate", "user_todo_blocker_push", @@ -1654,7 +1678,7 @@ def _interaction_fallback_policy_required(payload: dict[str, Any], *, mode: str) def unadmitted_action_selection(payload: dict[str, Any]) -> bool: """Use the same current-obligation facts as interaction and CLI preflight.""" qualification = payload.get("action_selection_qualification") - if not isinstance(qualification, Mapping) or qualification.get("state") not in {"deferred", "rejected"}: + if not isinstance(qualification, Mapping): return False mode = _interaction_mode(payload) user_required = False if payload.get("agent_work_mode") == "monitor_only" else user_channel_action_required(payload) @@ -1663,7 +1687,7 @@ def unadmitted_action_selection(payload: dict[str, Any]) -> bool: blocked_successor_wait_observation=_blocked_successor_wait_observation_required(payload), ) return selection.action_selection_needs_recovery( - payload, agent_must_attempt=must_attempt, agent_delivery_refused=delivery_allowed is False, + payload, agent_must_attempt=must_attempt, agent_delivery_allowed=delivery_allowed, ) @@ -1695,6 +1719,9 @@ def build_interaction_contract( payload, mode=mode, user_required=user_required, blocked_successor_wait_observation=_blocked_successor_wait_observation_required(payload), ) + selection_recovery_required = selection.action_selection_needs_recovery( + payload, agent_must_attempt=must_attempt, agent_delivery_allowed=delivery_allowed, + ) quiet_noop_allowed = interaction_quiet_noop_allowed( mode=mode, user_required=user_required, @@ -1723,7 +1750,7 @@ def build_interaction_contract( "notify": "DONT_NOTIFY", "reason": payload.get("reason"), } - if unadmitted_action_selection(payload): + if selection_recovery_required: agent_channel = { "must_attempt": False, "delivery_allowed": False, "quiet_noop_allowed": quiet_noop_allowed, @@ -1759,6 +1786,7 @@ def build_interaction_contract( heartbeat_recommendation, mode=mode, spend_after_validation=spend_after_validation, + selection_recovery_required=selection_recovery_required, available_capabilities=available_capabilities, scheduler_execution_context=scheduler_execution_context, capability_reentry=capability_reentry, diff --git a/loopx/control_plane/work_items/primary_action.py b/loopx/control_plane/work_items/primary_action.py index c91efaa0e1..70f5a2be6c 100644 --- a/loopx/control_plane/work_items/primary_action.py +++ b/loopx/control_plane/work_items/primary_action.py @@ -347,6 +347,12 @@ def build_primary_action_projection(payload: dict[str, Any], *, mode: str) -> di return projection +def _validate_permission_flag(value: object, field: str) -> None: + """Legacy producer inputs must not turn truthy strings into permission.""" + if value is not None and not isinstance(value, bool): + raise ValueError(f"{field} must be a boolean") + + def _interaction_must_attempt( execution_obligation: dict[str, Any], *, @@ -356,12 +362,12 @@ def _interaction_must_attempt( bounded_delivery_with_user_notice: bool, ) -> bool: if mode == "governed_capability_intent": - return bool(execution_obligation.get("must_attempt_work")) + return execution_obligation.get("must_attempt_work") is True if user_required and not ( scoped_user_gate_fallback or bounded_delivery_with_user_notice ): return False - return bool(execution_obligation.get("must_attempt_work")) + return execution_obligation.get("must_attempt_work") is True def _interaction_delivery_allowed( @@ -374,7 +380,7 @@ def _interaction_delivery_allowed( bounded_delivery_with_user_notice: bool, ) -> bool: if mode == "governed_capability_intent": - return bool(execution_obligation.get("must_attempt_work")) + return execution_obligation.get("must_attempt_work") is True if mode == "mapped_noop_if_unchanged": return False if user_required and not ( @@ -417,7 +423,12 @@ def interaction_execution_flags( payload: dict[str, Any], *, mode: str, user_required: bool, blocked_successor_wait_observation: bool, ) -> tuple[bool, bool]: - execution_obligation = payload.get("execution_obligation") if isinstance(payload.get("execution_obligation"), dict) else {} + obligation = payload.get("execution_obligation") + execution_obligation = obligation if isinstance(obligation, dict) else {} + for field in ("must_attempt_work", "delivery_allowed"): + _validate_permission_flag(execution_obligation.get(field), f"execution_obligation.{field}") + for field in ("normal_delivery_allowed", "recovery_delivery_allowed", "self_repair_allowed", "should_run"): + _validate_permission_flag(payload.get(field), field) scoped_user_gate_fallback = mode == "scoped_user_gate_fallback" bounded_delivery_with_user_notice = mode == "bounded_delivery_with_user_notice" must_attempt = _interaction_must_attempt( diff --git a/loopx/control_plane/work_items/task_lease_acquire.ts b/loopx/control_plane/work_items/task_lease_acquire.ts index 9bd184109b..d66904913a 100644 --- a/loopx/control_plane/work_items/task_lease_acquire.ts +++ b/loopx/control_plane/work_items/task_lease_acquire.ts @@ -1,4 +1,5 @@ import {observeLeaseWorktree, type LeaseWorkspace} from "./task_lease_workspace.ts"; +import {isCanonicalTodoId} from "../todos/identity.ts"; import {executeCanonicalTaskLeaseAcquire} from "../coordination/task_lease_acquire.ts"; import {withCanonicalTaskLeaseAuthority} from "./canonical_task_lease_lifecycle.ts"; import {evaluateTaskLeaseWriteScopesOverlap, evaluateTaskLeaseAcquireDecision, materializeTaskLeaseAcquire, type AcquireDecisionOtherLease} from "./task_lease_acquire_decision.ts"; @@ -204,7 +205,7 @@ export function normalizeGoalId(value: unknown): string { export function normalizeTodoId(value: unknown, label = "todo_id"): string { const todoId = stringValue(value, label).trim().toLowerCase(); - if (!/^todo_[a-z0-9_-]{3,64}$/u.test(todoId)) { + if (!isCanonicalTodoId(todoId)) { throw new TaskLeaseAcquireError( "todo id must use the todo_ shape", "invalid_todo_id", diff --git a/tests/control_plane/test_quota_action_selection_conflict.py b/tests/control_plane/test_quota_action_selection_conflict.py index 4dfde67741..5927836ee4 100644 --- a/tests/control_plane/test_quota_action_selection_conflict.py +++ b/tests/control_plane/test_quota_action_selection_conflict.py @@ -123,7 +123,7 @@ def test_failure_payload_reports_the_conflict_instead_of_collection_failure() -> } -def test_requested_todo_that_is_the_projection_selection_is_not_a_conflict() -> None: +def test_missing_delivery_permission_is_unknown_on_same_selection_refusal() -> None: """The projection already selects the requested Todo, so name the refusal.""" error = _raise( @@ -142,7 +142,7 @@ def test_requested_todo_that_is_the_projection_selection_is_not_a_conflict() -> assert "neither the projection's current selection" not in str(error) assert "not admitted to settle" in str(error) assert error.admission_must_attempt is True - assert error.admission_delivery_allowed is True + assert error.admission_delivery_allowed is None def test_unsettled_prior_turn_is_named_instead_of_a_selection_conflict() -> None: @@ -189,7 +189,7 @@ def test_unsettled_prior_turn_is_named_instead_of_a_selection_conflict() -> None "qualification_state": "qualified", "unsettled_prior_turn_instance_id": prior_turn_id, "unsettled_repair": "resume_prior_turn", - "admission": {"agent_must_attempt": True, "delivery_allowed": True}, + "admission": {"agent_must_attempt": True, "delivery_allowed": None}, } diff --git a/tests/control_plane/test_unadmitted_selection_construction.py b/tests/control_plane/test_unadmitted_selection_construction.py index 3c10774e42..8e461bf459 100644 --- a/tests/control_plane/test_unadmitted_selection_construction.py +++ b/tests/control_plane/test_unadmitted_selection_construction.py @@ -10,8 +10,12 @@ _requested_quota_action_selection_preflight, ) from loopx.control_plane.work_items import interaction_contract +from loopx.control_plane.effect_program import interpret_quota_should_run_packet +from loopx.control_plane.effect_runtime import EffectRuntimeRejected, effect_runtime_result +from loopx.control_plane.quota.turn_envelope import build_turn_envelope from loopx.control_plane.work_items.action_selection_contract import ( action_selection_needs_recovery, + action_selection_context, bind_action_selection_recovery_command, ) @@ -30,6 +34,37 @@ def _source(state: str) -> dict: } +@pytest.mark.parametrize("state", ["deferred", "rejected"]) +@pytest.mark.parametrize("grant", ["work", "spend", "bypass", "settlement_plan", "should_run", "actionable"]) +def test_unadmitted_packet_cannot_reintroduce_authority_at_typed_readers(state, grant): + source = _source(state) + source["interaction_contract"] = interaction_contract.build_interaction_contract(source) + source["ok"] = source["should_run"] = False + source["execution_obligation"] = {"must_attempt_work": False} + contract = source["interaction_contract"] + if grant == "work": + contract["agent_channel"].update(must_attempt=True, delivery_allowed=True, quiet_noop_allowed=False) + elif grant == "spend": + contract["cli_channel"]["spend_after_validation"] = True + elif grant == "bypass": + source["safe_bypass_allowed"] = True + elif grant == "should_run": + source["should_run"] = True + elif grant == "actionable": + source["actionable_by_codex"] = True + else: + contract["cli_channel"]["settlement_plan"] = {} + with pytest.raises(EffectRuntimeRejected, match="unadmitted action selection"): + interpret_quota_should_run_packet(source) + with pytest.raises(EffectRuntimeRejected, match="unadmitted action selection"): + build_turn_envelope(source) + with pytest.raises(EffectRuntimeRejected, match="unadmitted action selection"): + effect_runtime_result("quota.scoped_override.project", { + "override": {}, "interaction_contract": contract, + "selection_context": action_selection_context(source), + }) + + @pytest.mark.parametrize("state", ["deferred", "rejected"]) def test_recovery_never_constructs_settlement_or_executable_primary_action(monkeypatch, state): def unexpected(*args, **kwargs): @@ -96,6 +131,95 @@ def test_qualified_selection_is_not_recovery(): assert action_selection_needs_recovery(_source("qualified")) is False +@pytest.mark.parametrize("must_attempt,delivery", [(False, True), (True, False), (True, None)]) +def test_pending_delivery_requires_explicit_canonical_permission(must_attempt, delivery): + from loopx.control_plane.quota.error_codes import QuotaActionSelectionConflictError + + source = _source("qualified") + source.update(normal_delivery_allowed=True, selected_todo={ + "todo_id": "todo_pending", "selection_binding": "pending_action_selection", + }) + agent = {"must_attempt": must_attempt} + if delivery is not None: + agent["delivery_allowed"] = delivery + source["interaction_contract"] = {"agent_channel": agent} + with pytest.raises(QuotaActionSelectionConflictError): + _requested_quota_action_selection_preflight( + source, requested_todo_id="todo_pending", receipt_bound_todo_id=None, + receipt_bound_replan_obligation_id=None, + ) + + +@pytest.mark.parametrize("legacy", [None, {}, {"must_attempt_work": False}, {"must_attempt_work": True}]) +def test_compatibility_obligation_does_not_override_canonical_admission(legacy): + source = _source("qualified") + source["selected_todo"] = {"todo_id": "todo_pending"} + if legacy is None: + source.pop("execution_obligation") + else: + source["execution_obligation"] = legacy + source["interaction_contract"] = {"agent_channel": {"must_attempt": True, "delivery_allowed": True}} + assert _requested_quota_action_selection_preflight( + source, requested_todo_id="todo_pending", receipt_bound_todo_id=None, + receipt_bound_replan_obligation_id=None, + ) is None + + +@pytest.mark.parametrize("legacy", [None, {}, {"must_attempt_work": False}, {"must_attempt_work": True}]) +def test_closed_canonical_packet_ignores_compatibility_authority_at_readers(legacy): + source = _source("rejected") + source["interaction_contract"] = interaction_contract.build_interaction_contract(source) + source["ok"] = source["should_run"] = False + if legacy is None: + source.pop("execution_obligation") + else: + source["execution_obligation"] = legacy + interpret_quota_should_run_packet(source) + envelope = build_turn_envelope(source) + assert envelope["writeback"]["spend_allowed_now"] is False + assert envelope["writeback"]["spend_after_validation"] is False + effect_runtime_result("quota.scoped_override.project", { + "override": {}, "interaction_contract": source["interaction_contract"], + "selection_context": action_selection_context(source), + }) + + +@pytest.mark.parametrize("field", ["must_attempt_work", "delivery_allowed"]) +def test_null_legacy_permission_does_not_grant_work_or_delivery(field): + source = _source("qualified") + source["execution_obligation"][field] = None + contract = interaction_contract.build_interaction_contract(source) + canonical = "must_attempt" if field == "must_attempt_work" else "delivery_allowed" + assert contract["agent_channel"][canonical] is False + + +@pytest.mark.parametrize("field", ["must_attempt_work", "delivery_allowed"]) +@pytest.mark.parametrize("value", ["false", "true", 1, [], {}]) +def test_malformed_legacy_boolean_cannot_become_canonical_permission(field, value): + source = _source("qualified") + source["execution_obligation"][field] = value + with pytest.raises(ValueError, match="must be a boolean"): + interaction_contract.build_interaction_contract(source) + + +def test_one_recovery_decision_drives_both_interaction_channels(monkeypatch): + from loopx.control_plane.work_items import action_selection_contract as selection + + calls = [] + original = selection.action_selection_needs_recovery + + def observed(payload, **facts): + calls.append(facts) + return original(payload, **facts) + + monkeypatch.setattr(selection, "action_selection_needs_recovery", observed) + contract = interaction_contract.build_interaction_contract(_source("rejected")) + assert len(calls) == 1 + assert contract["agent_channel"]["must_attempt"] is False + assert contract["cli_channel"]["spend_after_validation"] is False + assert "settlement_plan" not in contract["cli_channel"] + + @pytest.mark.parametrize("delivery_allowed", [False, True, None]) def test_pending_workspace_repair_requires_explicit_delivery_refusal(delivery_allowed): from loopx.control_plane.quota.error_codes import QuotaActionSelectionConflictError diff --git a/tests/control_plane_ts/action_selection_admission.test.ts b/tests/control_plane_ts/action_selection_admission.test.ts new file mode 100644 index 0000000000..df86d32c27 --- /dev/null +++ b/tests/control_plane_ts/action_selection_admission.test.ts @@ -0,0 +1,96 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import {currentActionSelectionAdmission, actionSelectionNeedsRecovery, + assertActionSelectionConsistency} from "../../loopx/control_plane/work_items/action_selection_admission.ts"; +import {normalizeTodoId} from "../../loopx/control_plane/work_items/task_lease_acquire.ts"; +import type {JsonObject} from "../../loopx/control_plane/effect_program.ts"; + +const agent = {must_attempt: true, delivery_allowed: true}; +function source(state = "deferred"): JsonObject { + return {ok: true, should_run: true, effective_action: "autonomous_replan", + execution_obligation: {must_attempt_work: true}, + action_selection_qualification: {state, requested_todo_id: "todo_target", + recovery_action: "reenter_guard_without_selection"}}; +} + +test("current work and pending delivery use canonical channel permission", () => { + const absent = source(); + assert.deepEqual(currentActionSelectionAdmission(absent, "todo_target", agent), {selected_todo_id: null, admitted: false}); + assert.equal(actionSelectionNeedsRecovery(absent, agent), true); + for (const selected of [{todo_id: "todo_target"}, {todo_id: "todo_target", selection_binding: "pending_action_selection"}]) { + const packet = {...source(), selected_todo: selected, normal_delivery_allowed: true}; + assert.equal(currentActionSelectionAdmission(packet, "todo_target", agent).admitted, true); + assert.equal(actionSelectionNeedsRecovery(packet, agent), false); + assert.equal(currentActionSelectionAdmission(packet, "todo_other", agent).admitted, false); + assert.equal(currentActionSelectionAdmission({...packet, ok: false}, "todo_target", agent).admitted, false); + } +}); + +test("workspace repair needs its repair route and explicit delivery refusal", () => { + const packet = {...source(), workspace_repair_allowed: true, effective_action: "agent_workspace_repair", + execution_obligation: {kind: "agent_workspace_repair", must_attempt_work: true}, + selected_todo: {todo_id: "todo_target", selection_binding: "pending_action_selection"}}; + assert.equal(currentActionSelectionAdmission(packet, "todo_target", {...agent, delivery_allowed: false}).admitted, true); + assert.equal(currentActionSelectionAdmission(packet, "todo_target", agent).admitted, false); + assert.equal(currentActionSelectionAdmission(packet, "todo_target", {must_attempt: false, delivery_allowed: false}).admitted, false); + assert.equal(currentActionSelectionAdmission({...packet, workspace_repair_allowed: false}, "todo_target", + {...agent, delivery_allowed: false}).admitted, false); +}); + +test("pending delivery cannot borrow root permission from a denied or missing channel grant", () => { + const packet = {...source(), normal_delivery_allowed: true, + selected_todo: {todo_id: "todo_target", selection_binding: "pending_action_selection"}}; + for (const facts of [{must_attempt: false, delivery_allowed: true}, + {must_attempt: true, delivery_allowed: false}, {must_attempt: true, delivery_allowed: null}]) { + assert.equal(currentActionSelectionAdmission(packet, "todo_target", facts).admitted, false); + assert.equal(actionSelectionNeedsRecovery(packet, facts), true); + } +}); + +test("compatibility annotations neither grant nor veto canonical current-work admission", () => { + for (const legacy of [undefined, {}, {must_attempt_work: false}, {must_attempt_work: true}]) { + const packet = {...source(), selected_todo: {todo_id: "todo_target"}}; + if (legacy === undefined) delete packet.execution_obligation; + else packet.execution_obligation = legacy; + assert.equal(currentActionSelectionAdmission(packet, "todo_target", agent).admitted, true); + assert.equal(currentActionSelectionAdmission(packet, "todo_target", {...agent, must_attempt: false}).admitted, false); + } +}); + +test("receipt bindings are left to reconciliation and qualified nested work is retained", () => { + for (const field of ["selected_todo", "autonomous_replan_obligation"]) { + assert.equal(actionSelectionNeedsRecovery({...source(), [field]: {selection_binding: "heartbeat_receipt"}}, agent), false); + } + const packet = source("qualified"); + (packet.action_selection_qualification as JsonObject).selected_todo = {todo_id: "todo_target"}; + assert.equal(currentActionSelectionAdmission(packet, "todo_target", agent).admitted, true); + assert.equal(actionSelectionNeedsRecovery(packet, agent), false); +}); + +test("a missing recovery action is an error only on the recovery path", () => { + const packet = source("rejected"); + delete (packet.action_selection_qualification as JsonObject).recovery_action; + assert.equal(currentActionSelectionAdmission(packet, "todo_target", agent).admitted, false); + assert.throws(() => actionSelectionNeedsRecovery(packet, agent), /omitted its typed recovery action/); +}); + +test("Todo grammar is shared without changing the callers' whitespace policies", () => { + for (const [raw, expected] of [[" TODO_TARGET ", "todo_target"], ["\u001ctodo_target\u001c", "todo_target"], + ["\ufefftodo_target\ufeff", null], ["todo_x", null], ["../todo_target", null], [["todo_target"], null]] as const) { + assert.equal(currentActionSelectionAdmission({...source(), selected_todo: {todo_id: raw}}, "todo_target", agent).selected_todo_id, expected); + } + assert.equal(normalizeTodoId(" TODO_TARGET "), "todo_target"); + assert.equal(normalizeTodoId("\ufefftodo_target\ufeff"), "todo_target"); + assert.throws(() => normalizeTodoId("\u001ctodo_target\u001c"), /todo_/); + assert.throws(() => normalizeTodoId("../todo_target"), /todo_/); +}); + +test("refusal keeps re-entry guidance but rejects stale work and settlement grants", () => { + const packet = {...source("rejected"), ok: false, should_run: false, execution_obligation: {must_attempt_work: false}}; + const interaction = {agent_channel: {must_attempt: false, delivery_allowed: false}, + cli_channel: {spend_allowed_now: false, spend_after_validation: false, next_cli_actions: ["rerun the current guard"]}}; + assert.doesNotThrow(() => assertActionSelectionConsistency(packet, interaction)); + assert.throws(() => assertActionSelectionConsistency({...packet, safe_bypass_allowed: true}, interaction), /unadmitted/); + assert.throws(() => assertActionSelectionConsistency(packet, {...interaction, + cli_channel: {...interaction.cli_channel, selection_command: {}}}), /unadmitted/); +});