fix(quota): make an unreconcilable action selection self-diagnosing

loopx/cli_commands/quota.py

@@ -22,6 +22,8 @@
 from ..control_plane.quota.effect_program import SettlementIdentity
 from ..control_plane.quota.error_codes import (
     QuotaCommandValidationError,
+    QuotaActionSelectionConflictError,
+    QuotaActionSelectionConflictKind,
 )
 from ..control_plane.quota.heartbeat_receipt import (
     HEARTBEAT_RECEIPT_SCHEMA_VERSION,
@@ -268,11 +270,18 @@ def _apply_requested_quota_action_selection_preflight(
         return False
 
     if not isinstance(qualification_value, Mapping):
-        raise RuntimeError("requested action selection lacks typed qualification")
+        raise QuotaActionSelectionConflictError(
+            QuotaActionSelectionConflictKind.UNQUALIFIED,
+            requested_todo_id=requested_todo_id,
+            selected_todo_id=selected_todo_id,
+        )
     qualification_state = str(qualification.get("state") or "")
     if qualification_state not in {"deferred", "rejected"}:
-        raise RuntimeError(
-            "requested action selection qualification conflicts with its projection"
+        raise QuotaActionSelectionConflictError(
+            QuotaActionSelectionConflictKind.CONFLICT,
+            requested_todo_id=requested_todo_id,
+            selected_todo_id=selected_todo_id,
+            qualification_state=qualification_state,
         )
     qualification_reason = str(
         qualification.get("reason") or "candidate_not_currently_eligible"

loopx/cli_commands/quota_failure_report.py

@@ -20,6 +20,7 @@
 )
 from ..control_plane.quota.error_codes import (
     HeartbeatReceiptIdentityConflictError,
+    QuotaActionSelectionConflictError,
     QuotaCommandValidationError,
     QuotaIdentityPreconditionError,
     quota_error_code,
@@ -127,6 +128,23 @@ def quota_failure_payload(
         **verbose_debug,
         **lock_timeout_fields,
     }
+    if isinstance(error, QuotaActionSelectionConflictError):
+        # The requested Todo could not be reconciled with the projection. Report
+        # the real conflict and the next read to make, rather than the generic
+        # "quota collection failed" and a pointer at receipt writeback.
+        payload.update(
+            {
+                "reason": str(error),
+                "status": "quota_action_selection_conflict",
+                "recommended_action": error.recommended_action,
+                "action_selection_conflict": {
+                    "kind": error.kind.value,
+                    "requested_todo_id": error.requested_todo_id,
+                    "selected_todo_id": error.selected_todo_id,
+                    "qualification_state": error.qualification_state,
+                },
+            }
+        )
     if isinstance(error, QuotaIdentityPreconditionError):
         payload.update(
             {

loopx/control_plane/quota/error_codes.py

@@ -59,13 +59,67 @@ def __init__(
         super().__init__(reason)
 
 
+class QuotaActionSelectionConflictKind(StrEnum):
+    """Why a requested ``--todo-id`` could not be reconciled with the projection."""
+
+    UNQUALIFIED = "unqualified"
+    CONFLICT = "conflict"
+
+
+class QuotaActionSelectionConflictError(RuntimeError):
+    """Public-safe diagnostic for an unreconcilable requested action selection.
+
+    A guard bound to a ``--todo-id`` has to agree with the current projection.
+    When it cannot, this error names what was requested, what the projection
+    currently selects, and what the caller should do next, so the failure is not
+    reported as an opaque quota collection failure.
+    """
+
+    error_code = "quota_action_selection_conflict"
+
+    def __init__(
+        self,
+        kind: QuotaActionSelectionConflictKind,
+        *,
+        requested_todo_id: str | None,
+        selected_todo_id: str | None = None,
+        qualification_state: str | None = None,
+    ) -> None:
+        self.kind = kind
+        self.requested_todo_id = requested_todo_id
+        self.selected_todo_id = selected_todo_id
+        self.qualification_state = qualification_state
+        if kind is QuotaActionSelectionConflictKind.UNQUALIFIED:
+            reason = (
+                "the current projection carries no typed action-selection "
+                "qualification, so the requested Todo "
+                f"{requested_todo_id or '(none)'} cannot be reconciled with the "
+                "delivery frontier"
+            )
+        else:
+            reason = (
+                f"requested Todo {requested_todo_id or '(none)'} is neither the "
+                "projection's current selection "
+                f"({selected_todo_id or 'none'}) nor deferred or rejected by it "
+                f"(qualification state: {qualification_state or 'absent'})"
+            )
+        self.recommended_action = (
+            "rerun `loopx quota should-run` without --todo-id to read the current "
+            "selection, then bind that Todo, a deferred Todo, or the Todo the "
+            "recovery obligation must settle"
+        )
+        super().__init__(reason)
+
+
 def quota_error_code(exc: BaseException) -> str:
     if isinstance(exc, json.JSONDecodeError):
         return "quota_state_invalid_json"
     if isinstance(exc, QuotaCommandValidationError):
         return "quota_invalid_arguments"
     if isinstance(exc, QuotaIdentityPreconditionError):
         return exc.error_code
+    if isinstance(exc, QuotaActionSelectionConflictError):
+        return exc.error_code
     if isinstance(exc, HeartbeatReceiptIdentityConflictError):
         return "heartbeat_receipt_identity_conflict"
     if isinstance(exc, PermissionError):

tests/control_plane/test_quota_action_selection_conflict.py

@@ -0,0 +1,117 @@
+"""An unreconcilable `--todo-id` guard names its own cause and next read."""
+
+from __future__ import annotations
+
+import argparse
+from pathlib import Path
+
+import pytest
+
+from loopx.cli_commands.quota import _apply_requested_quota_action_selection_preflight
+from loopx.cli_commands.quota_failure_report import quota_failure_payload
+from loopx.control_plane.quota.error_codes import (
+    QuotaActionSelectionConflictError,
+    QuotaActionSelectionConflictKind,
+    quota_error_code,
+)
+
+
+REQUESTED_TODO_ID = "todo_requested_selection"
+SELECTED_TODO_ID = "todo_projected_selection"
+
+
+def _payload(**overrides: object) -> dict[str, object]:
+    payload: dict[str, object] = {
+        "ok": True,
+        "should_run": True,
+        "effective_action": "normal_run",
+        "selected_todo": None,
+        "execution_obligation": {"must_attempt_work": True},
+        "interaction_contract": {"agent_channel": {"must_attempt": True}},
+    }
+    payload.update(overrides)
+    return payload
+
+
+def _raise(payload: dict[str, object]) -> QuotaActionSelectionConflictError:
+    with pytest.raises(QuotaActionSelectionConflictError) as raised:
+        _apply_requested_quota_action_selection_preflight(
+            payload,
+            requested_todo_id=REQUESTED_TODO_ID,
+            receipt_bound_todo_id=None,
+            receipt_bound_replan_obligation_id=None,
+        )
+    return raised.value
+
+
+def _qualified_for(todo_id: str) -> dict[str, object]:
+    return {
+        "schema_version": "action_selection_qualification_v0",
+        "state": "qualified",
+        "requested_todo_id": REQUESTED_TODO_ID,
+        "selected_todo": {"todo_id": todo_id},
+    }
+
+
+def test_conflicting_qualification_names_requested_and_selected_todo() -> None:
+    error = _raise(
+        _payload(action_selection_qualification=_qualified_for(SELECTED_TODO_ID))
+    )
+
+    assert error.kind is QuotaActionSelectionConflictKind.CONFLICT
+    assert error.error_code == "quota_action_selection_conflict"
+    assert REQUESTED_TODO_ID in str(error)
+    assert SELECTED_TODO_ID in str(error)
+    assert "qualified" in str(error)
+    assert quota_error_code(error) == "quota_action_selection_conflict"
+
+
+def test_missing_qualification_is_typed_rather_than_unexplained() -> None:
+    error = _raise(_payload(selected_todo={"todo_id": SELECTED_TODO_ID}))
+
+    assert error.kind is QuotaActionSelectionConflictKind.UNQUALIFIED
+    assert REQUESTED_TODO_ID in str(error)
+    assert "no typed action-selection qualification" in str(error)
+
+
+def test_a_qualified_selection_for_the_requested_todo_is_not_a_conflict() -> None:
+    is_conflict = _apply_requested_quota_action_selection_preflight(
+        _payload(action_selection_qualification=_qualified_for(REQUESTED_TODO_ID)),
+        requested_todo_id=REQUESTED_TODO_ID,
+        receipt_bound_todo_id=None,
+        receipt_bound_replan_obligation_id=None,
+    )
+
+    assert is_conflict is False
+
+
+def test_failure_payload_reports_the_conflict_instead_of_collection_failure() -> None:
+    error = _raise(
+        _payload(action_selection_qualification=_qualified_for(SELECTED_TODO_ID))
+    )
+    args = argparse.Namespace(
+        quota_command="should-run",
+        goal_id="quota-conflict-fixture",
+        agent_id="agent-fixture",
+        runtime_root=None,
+        verbose=False,
+    )
+
+    payload = quota_failure_payload(
+        args,
+        registry_path=Path("/tmp/quota-conflict-registry.json"),
+        runtime_root_arg=None,
+        error=error,
+    )
+
+    assert payload["error_code"] == "quota_action_selection_conflict"
+    assert payload["status"] == "quota_action_selection_conflict"
+    assert payload["reason"] != "quota collection failed"
+    assert payload["reason"] == str(error)
+    assert "heartbeat receipt writeback" not in str(payload["recommended_action"])
+    assert payload["action_selection_conflict"] == {
+        "kind": "conflict",
+        "requested_todo_id": REQUESTED_TODO_ID,
+        "selected_todo_id": SELECTED_TODO_ID,
+        "qualification_state": "qualified",
+    }