refactor(control-plane): move goal latest run into quota bounded context (#2933)

Author: huangruiteng

loopx/control_plane/quota/recent_runs.py

@@ -10,6 +10,19 @@
 NON_WORK_RUN_CLASSIFICATIONS = {"state_refreshed"}
 
 
+def goal_latest_run(goal: dict[str, Any]) -> dict[str, Any]:
+    """Return the latest run object for one goal, or an empty mapping."""
+
+    latest_runs = (
+        goal.get("latest_runs")
+        if isinstance(goal.get("latest_runs"), list)
+        else []
+    )
+    if latest_runs and isinstance(latest_runs[0], dict):
+        return latest_runs[0]
+    return {}
+
+
 def _run_is_unchanged_monitor_observation(run: dict[str, Any]) -> bool:
     if str(run.get("classification") or "") != QUOTA_MONITOR_POLL_CLASSIFICATION:
         return False

loopx/quota.py

@@ -78,6 +78,7 @@
 )
 from .control_plane.quota.recent_runs import (
     build_monitor_debt_arbitration as _build_monitor_debt_arbitration,
+    goal_latest_run as _goal_latest_run,
     goal_latest_runs as _goal_latest_runs,
     recent_external_monitor_observation_unchanged as _recent_external_monitor_observation_unchanged,
 )
@@ -507,13 +508,6 @@ def quota_status(
     return payload
 
 
-def _latest_run(goal: dict[str, Any]) -> dict[str, Any]:
-    latest_runs = goal.get("latest_runs") if isinstance(goal.get("latest_runs"), list) else []
-    if latest_runs and isinstance(latest_runs[0], dict):
-        return latest_runs[0]
-    return {}
-
-
 def _quota_sort_key(item: dict[str, Any]) -> tuple[int, float, int, str]:
     quota = item.get("quota") if isinstance(item.get("quota"), dict) else {}
     state = str(quota.get("state") or "waiting")
@@ -910,7 +904,7 @@ def build_quota_plan(status_payload: dict[str, Any], *, mode: str = "status") ->
             if isinstance(attention.get("project_asset"), dict)
             else {}
         )
-        latest = _latest_run(goal)
+        latest = _goal_latest_run(goal)
         waiting_on = attention.get("waiting_on") or "none"
         lifecycle_phase = attention.get("lifecycle_phase") or goal.get("lifecycle_phase")
         lifecycle_flags = attention.get("lifecycle_flags") or goal.get("lifecycle_flags")

tests/control_plane/test_quota_recent_runs.py

@@ -0,0 +1,16 @@
+from __future__ import annotations
+
+from loopx.control_plane.quota.recent_runs import goal_latest_run
+
+
+def test_goal_latest_run_returns_first_compact_run() -> None:
+    assert goal_latest_run({}) == {}
+    assert goal_latest_run({"latest_runs": []}) == {}
+    assert goal_latest_run(
+        {
+            "latest_runs": [
+                {"generated_at": "2026-08-08T00:00:00+00:00"},
+                {"generated_at": "2026-08-07T00:00:00+00:00"},
+            ]
+        }
+    ) == {"generated_at": "2026-08-08T00:00:00+00:00"}