refactor(control-plane): extract quota registry and pause read models

loopx/control_plane/quota/goal_boundary.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import json
 import shlex
 from collections.abc import Callable, Mapping
 from pathlib import Path
@@ -63,6 +64,29 @@ def quota_execution_profile_summary(value: Any) -> dict[str, Any] | None:
     return compact or None
 
 
+def registry_goal_by_id(
+    status_payload: Mapping[str, Any],
+) -> dict[str, dict[str, Any]]:
+    """Return registry goals indexed by goal id from a status payload."""
+
+    registry_value = status_payload.get("registry")
+    if not registry_value:
+        return {}
+    registry_path = Path(str(registry_value)).expanduser()
+    try:
+        payload = json.loads(registry_path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError):
+        return {}
+    goals = payload.get("goals") if isinstance(payload, dict) else None
+    if not isinstance(goals, list):
+        return {}
+    return {
+        str(goal.get("id") or ""): goal
+        for goal in goals
+        if isinstance(goal, dict) and goal.get("id")
+    }
+
+
 def quota_execution_profile_boundary_summary(value: Any) -> dict[str, Any] | None:
     summary = quota_execution_profile_summary(value)
     if not summary:

loopx/control_plane/quota/states.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+from typing import Any
+
 
 QUOTA_STATE_ORDER = (
     "blocked_health",
@@ -10,3 +12,18 @@
     "throttled",
     "paused",
 )
+
+
+def quota_item_is_paused(item: dict[str, Any]) -> bool:
+    """Return True when a plan item carries a Goal-level hard pause."""
+
+    raw_quota = item.get("quota")
+    quota = raw_quota if isinstance(raw_quota, dict) else {}
+    if str(quota.get("state") or "") == "paused":
+        return True
+    compute = quota.get("compute")
+    return (
+        isinstance(compute, (int, float))
+        and not isinstance(compute, bool)
+        and compute <= 0
+    )

loopx/quota.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import json
 from collections.abc import Callable, Mapping
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
@@ -71,7 +70,7 @@
     refine_quota_recommended_action,
     resolve_quota_run_decision,
 )
-from .control_plane.quota.goal_boundary import effective_available_capabilities as _effective_available_capabilities, goal_boundary as _goal_boundary, quota_execution_profile_summary as _quota_execution_profile_summary
+from .control_plane.quota.goal_boundary import effective_available_capabilities as _effective_available_capabilities, goal_boundary as _goal_boundary, quota_execution_profile_summary as _quota_execution_profile_summary, registry_goal_by_id as _registry_goal_by_id
 from .control_plane.quota.monitor_poll import (
     QUOTA_MONITOR_POLL_CLASSIFICATION as QUOTA_MONITOR_POLL_CLASSIFICATION,
     build_quota_monitor_poll_event as build_quota_monitor_poll_event,
@@ -116,7 +115,7 @@
 from .control_plane.quota.spend_sources import (
     DEFAULT_SLOT_SPEND_SOURCE,
 )
-from .control_plane.quota.states import QUOTA_STATE_ORDER
+from .control_plane.quota.states import QUOTA_STATE_ORDER, quota_item_is_paused as _quota_item_is_paused
 from .control_plane.quota.policy_constants import (
     AUTONOMOUS_CANDIDATE_CONTEXT_FIELDS,
     DEFAULT_COMPUTE_QUOTA,
@@ -1103,25 +1102,6 @@ def _reward_lesson_projection_warning(
     }
 
 
-def _registry_goal_by_id(status_payload: dict[str, Any]) -> dict[str, dict[str, Any]]:
-    registry_value = status_payload.get("registry")
-    if not registry_value:
-        return {}
-    registry_path = Path(str(registry_value)).expanduser()
-    try:
-        payload = json.loads(registry_path.read_text(encoding="utf-8"))
-    except (OSError, json.JSONDecodeError):
-        return {}
-    goals = payload.get("goals") if isinstance(payload, dict) else None
-    if not isinstance(goals, list):
-        return {}
-    return {
-        str(goal.get("id") or ""): goal
-        for goal in goals
-        if isinstance(goal, dict) and goal.get("id")
-    }
-
-
 def _recovery_delivery_allowed(quota: dict[str, Any], *, plan_ok: bool) -> bool:
     return (
         bool(plan_ok)
@@ -2434,22 +2414,6 @@ def _build_quota_should_run_payload(
 QUOTA_PAUSED_MODE = "quota_paused"
 
 
-def _quota_item_is_paused(item: dict[str, Any]) -> bool:
-    """Return True when a plan item carries a Goal-level hard pause.
-
-    A paused Goal (`quota.compute<=0`) is a typed terminal decision: it is
-    evaluated before the selector builds any capability, workspace, replan,
-    monitor, or inbox candidate, so no lane can emit a contradicting execution
-    signal underneath the pause.
-    """
-
-    quota = item.get("quota") if isinstance(item.get("quota"), dict) else {}
-    if str(quota.get("state") or "") == "paused":
-        return True
-    compute = quota.get("compute")
-    return isinstance(compute, (int, float)) and not isinstance(compute, bool) and compute <= 0
-
-
 def _build_quota_paused_should_run_payload(
     status_payload: dict[str, Any],
     *,