test(control-plane): add quota should-run parity fixtures

Author: huangruiteng

loopx/control_plane/testing/quota_should_run_parity.py

@@ -0,0 +1,57 @@
+"""Parity fixture helpers for `quota should-run` extraction."""
+
+from __future__ import annotations
+
+from collections.abc import Mapping
+from typing import Any
+
+from ...quota import build_quota_should_run
+
+
+def compact_quota_should_run_parity(
+    packet: Mapping[str, Any],
+) -> dict[str, Any]:
+    """Return a compact stable surface for old/new quota builder comparison."""
+
+    interaction = (
+        packet.get("interaction_contract")
+        if isinstance(packet.get("interaction_contract"), Mapping)
+        else {}
+    )
+    lane = (
+        packet.get("work_lane_contract")
+        if isinstance(packet.get("work_lane_contract"), Mapping)
+        else {}
+    )
+    scheduler = (
+        packet.get("scheduler_hint")
+        if isinstance(packet.get("scheduler_hint"), Mapping)
+        else {}
+    )
+    gate = packet.get("capability_gate")
+    gate = gate if isinstance(gate, Mapping) else {}
+    return {
+        "decision": packet.get("decision"),
+        "should_run": packet.get("should_run"),
+        "effective_action": packet.get("effective_action"),
+        "recommended_action": packet.get("recommended_action"),
+        "interaction_mode": interaction.get("mode"),
+        "interaction_action_required": interaction.get("action_required"),
+        "lane": lane.get("lane"),
+        "obligation": lane.get("obligation"),
+        "must_attempt_work": lane.get("must_attempt_work"),
+        "scheduler_action": scheduler.get("action"),
+        "scheduler_cadence_class": scheduler.get("cadence_class"),
+        "capability_gate_action": gate.get("action"),
+    }
+
+
+def build_quota_should_run_parity(
+    status_payload: Mapping[str, Any],
+    **kwargs: Any,
+) -> dict[str, Any]:
+    """Build `quota should-run` and reduce it to the compact parity surface."""
+
+    return compact_quota_should_run_parity(
+        build_quota_should_run(dict(status_payload), **kwargs)
+    )

tests/control_plane/test_quota_should_run_parity.py

@@ -0,0 +1,72 @@
+from __future__ import annotations
+
+from loopx.control_plane.testing.quota_fixtures import quota_status_payload
+from loopx.control_plane.testing.quota_should_run_parity import (
+    build_quota_should_run_parity,
+)
+
+GOAL_ID = "quota-parity-fixture"
+
+
+def test_advancement_run_parity_surface() -> None:
+    todo_text = "[P1] Advance the bounded slice."
+    payload = quota_status_payload(
+        goal_id=GOAL_ID,
+        status="active",
+        agent_todo_items=[
+            {
+                "index": 1,
+                "text": todo_text,
+                "role": "agent",
+                "status": "open",
+                "priority": "P1",
+                "task_class": "advancement_task",
+            }
+        ],
+        recommended_action=todo_text,
+        next_action=todo_text,
+    )
+
+    parity = build_quota_should_run_parity(payload, goal_id=GOAL_ID)
+
+    assert parity["decision"] == "run"
+    assert parity["should_run"] is True
+    assert parity["effective_action"] == "normal_run"
+    assert parity["lane"] == "advancement_task"
+    assert parity["obligation"] == "advance_one_bounded_segment"
+    assert parity["must_attempt_work"] is True
+    assert parity["interaction_mode"] == "bounded_delivery"
+    assert parity["capability_gate_action"] is None
+
+
+def test_capability_gate_repair_parity_surface() -> None:
+    todo_text = "[P1] Network-only slice."
+    payload = quota_status_payload(
+        goal_id=GOAL_ID,
+        status="active",
+        agent_todo_items=[
+            {
+                "index": 1,
+                "text": todo_text,
+                "role": "agent",
+                "status": "open",
+                "priority": "P1",
+                "task_class": "advancement_task",
+                "required_capabilities": ["network"],
+            }
+        ],
+        recommended_action=todo_text,
+        next_action=todo_text,
+    )
+
+    parity = build_quota_should_run_parity(
+        payload,
+        goal_id=GOAL_ID,
+        available_capabilities=["shell"],
+    )
+
+    assert parity["decision"] == "repair_bridge"
+    assert parity["should_run"] is True
+    assert parity["effective_action"] == "capability_bridge_repair"
+    assert parity["capability_gate_action"] == "repair_bridge"
+    assert parity["lane"] == "advancement_task"