fix(selection): share canonical admission across quota readers

Signed-off-by: song <liusongstep@gmail.com>

Author: songoow

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:

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<JsonObject, string> {
   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);

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),

loopx/control_plane/quota/scoped_override.ts

@@ -1,13 +1,17 @@
 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.
  */
 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");

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
 

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);
+}

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");
+  }
+}

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(