refactor(status): move lifecycle projections into bounded module (#2973)

Author: huangruiteng

loopx/control_plane/status/lifecycle_projection.py

@@ -0,0 +1,110 @@
+"""Goal lifecycle projections inside the `status` bounded context."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from ..runtime.run_compaction import (
+    compact_controller_readiness,
+    compact_human_reward,
+    compact_operator_gate,
+)
+from ..work_items.attention_fields import (
+    operator_gate_attention_fields as _operator_gate_attention_fields,
+    readiness_attention_fields as _readiness_attention_fields,
+)
+from ..work_items.lifecycle import (
+    goal_lifecycle_fields as _goal_lifecycle_fields,
+    ordered_lifecycle_flags as _ordered_lifecycle_flags,
+    primary_lifecycle_phase as _primary_lifecycle_phase,
+    run_lifecycle_flags as _run_lifecycle_flags,
+    run_lifecycle_phase as _run_lifecycle_phase,
+)
+from ...operator_gate import (
+    DEFAULT_OPERATOR_GATE,
+    normalize_operator_question,
+)
+
+
+CONNECTED_ADAPTER_STATUSES = {
+    "connected",
+    "connected-read-only",
+    "pre-tick-runnable",
+}
+LIFECYCLE_PRIORITY = (
+    "controller_ready",
+    "reward_judged",
+    "operator_approved",
+    "controller_gated",
+    "operator_gated",
+    "adapter_inspected",
+    "mapped",
+    "refreshed",
+    "connected",
+    "registered",
+    "planned",
+    "run_recorded",
+)
+
+
+def ordered_lifecycle_flags(flags: list[str]) -> list[str]:
+    return _ordered_lifecycle_flags(
+        flags,
+        lifecycle_priority=LIFECYCLE_PRIORITY,
+    )
+
+
+def primary_lifecycle_phase(flags: list[str], fallback: str = "registered") -> str:
+    return _primary_lifecycle_phase(
+        flags,
+        lifecycle_priority=LIFECYCLE_PRIORITY,
+        fallback=fallback,
+    )
+
+
+def run_lifecycle_flags(run: dict[str, Any] | None) -> list[str]:
+    return _run_lifecycle_flags(
+        run,
+        lifecycle_priority=LIFECYCLE_PRIORITY,
+        compact_human_reward=compact_human_reward,
+        compact_operator_gate=compact_operator_gate,
+        compact_controller_readiness=compact_controller_readiness,
+    )
+
+
+def run_lifecycle_phase(run: dict[str, Any] | None) -> str:
+    return _run_lifecycle_phase(
+        run,
+        lifecycle_priority=LIFECYCLE_PRIORITY,
+        compact_human_reward=compact_human_reward,
+        compact_operator_gate=compact_operator_gate,
+        compact_controller_readiness=compact_controller_readiness,
+    )
+
+
+def goal_lifecycle_fields(goal: dict[str, Any], current_run: dict[str, Any] | None) -> dict[str, Any]:
+    return _goal_lifecycle_fields(
+        goal,
+        current_run,
+        lifecycle_priority=LIFECYCLE_PRIORITY,
+        connected_adapter_statuses=CONNECTED_ADAPTER_STATUSES,
+        compact_human_reward=compact_human_reward,
+        compact_operator_gate=compact_operator_gate,
+        compact_controller_readiness=compact_controller_readiness,
+    )
+
+
+def readiness_attention_fields(run: dict[str, Any] | None) -> dict[str, Any]:
+    return _readiness_attention_fields(
+        run,
+        compact_controller_readiness=compact_controller_readiness,
+    )
+
+
+def operator_gate_attention_fields(run: dict[str, Any] | None) -> dict[str, Any]:
+    return _operator_gate_attention_fields(
+        run,
+        compact_operator_gate=compact_operator_gate,
+        normalize_operator_question=normalize_operator_question,
+        default_operator_gate=DEFAULT_OPERATOR_GATE,
+    )

loopx/status.py

@@ -91,10 +91,6 @@
 from .control_plane.work_items.attention_routing import (
     goal_attention as _goal_attention_read_model,
 )
-from .control_plane.work_items.attention_fields import (
-    operator_gate_attention_fields as _operator_gate_attention_fields_read_model,
-    readiness_attention_fields as _readiness_attention_fields_read_model,
-)
 from .control_plane.work_items.autonomous_replan_ack import (
     AUTONOMOUS_REPLAN_ACK_MATERIAL_RUN_WINDOW,
     compact_autonomous_replan_ack,
@@ -217,13 +213,6 @@
 from .control_plane.goals.goal_vision import (
     compact_goal_vision_packet as _compact_goal_vision_packet_read_model,
 )
-from .control_plane.work_items.lifecycle import (
-    goal_lifecycle_fields as _goal_lifecycle_fields_read_model,
-    ordered_lifecycle_flags as _ordered_lifecycle_flags_read_model,
-    primary_lifecycle_phase as _primary_lifecycle_phase_read_model,
-    run_lifecycle_flags as _run_lifecycle_flags_read_model,
-    run_lifecycle_phase as _run_lifecycle_phase_read_model,
-)
 from .control_plane.runtime.session_runtime import (
     compact_session_runtime_projection_from_run,
     legacy_runtime_goal_attention as _legacy_runtime_goal_attention_read_model,
@@ -2619,67 +2608,60 @@ def latest_run(goal: dict[str, Any]) -> dict[str, Any] | None:
 
 
 def ordered_lifecycle_flags(flags: list[str]) -> list[str]:
-    return _ordered_lifecycle_flags_read_model(
-        flags,
-        lifecycle_priority=LIFECYCLE_PRIORITY,
+    from .control_plane.status.lifecycle_projection import (
+        ordered_lifecycle_flags as _ordered_lifecycle_flags,
     )
 
+    return _ordered_lifecycle_flags(flags)
+
 
 def primary_lifecycle_phase(flags: list[str], fallback: str = "registered") -> str:
-    return _primary_lifecycle_phase_read_model(
-        flags,
-        lifecycle_priority=LIFECYCLE_PRIORITY,
-        fallback=fallback,
+    from .control_plane.status.lifecycle_projection import (
+        primary_lifecycle_phase as _primary_lifecycle_phase,
     )
 
+    return _primary_lifecycle_phase(flags, fallback=fallback)
+
 
 def run_lifecycle_flags(run: dict[str, Any] | None) -> list[str]:
-    return _run_lifecycle_flags_read_model(
-        run,
-        lifecycle_priority=LIFECYCLE_PRIORITY,
-        compact_human_reward=compact_human_reward,
-        compact_operator_gate=compact_operator_gate,
-        compact_controller_readiness=compact_controller_readiness,
+    from .control_plane.status.lifecycle_projection import (
+        run_lifecycle_flags as _run_lifecycle_flags,
     )
 
+    return _run_lifecycle_flags(run)
+
 
 def run_lifecycle_phase(run: dict[str, Any] | None) -> str:
-    return _run_lifecycle_phase_read_model(
-        run,
-        lifecycle_priority=LIFECYCLE_PRIORITY,
-        compact_human_reward=compact_human_reward,
-        compact_operator_gate=compact_operator_gate,
-        compact_controller_readiness=compact_controller_readiness,
+    from .control_plane.status.lifecycle_projection import (
+        run_lifecycle_phase as _run_lifecycle_phase,
     )
 
+    return _run_lifecycle_phase(run)
+
 
 def goal_lifecycle_fields(goal: dict[str, Any], current_run: dict[str, Any] | None) -> dict[str, Any]:
-    return _goal_lifecycle_fields_read_model(
-        goal,
-        current_run,
-        lifecycle_priority=LIFECYCLE_PRIORITY,
-        connected_adapter_statuses=CONNECTED_ADAPTER_STATUSES,
-        compact_human_reward=compact_human_reward,
-        compact_operator_gate=compact_operator_gate,
-        compact_controller_readiness=compact_controller_readiness,
+    from .control_plane.status.lifecycle_projection import (
+        goal_lifecycle_fields as _goal_lifecycle_fields,
     )
 
+    return _goal_lifecycle_fields(goal, current_run)
+
 
 def readiness_attention_fields(run: dict[str, Any] | None) -> dict[str, Any]:
-    return _readiness_attention_fields_read_model(
-        run,
-        compact_controller_readiness=compact_controller_readiness,
+    from .control_plane.status.lifecycle_projection import (
+        readiness_attention_fields as _readiness_attention_fields,
     )
 
+    return _readiness_attention_fields(run)
+
 
 def operator_gate_attention_fields(run: dict[str, Any] | None) -> dict[str, Any]:
-    return _operator_gate_attention_fields_read_model(
-        run,
-        compact_operator_gate=compact_operator_gate,
-        normalize_operator_question=normalize_operator_question,
-        default_operator_gate=DEFAULT_OPERATOR_GATE,
+    from .control_plane.status.lifecycle_projection import (
+        operator_gate_attention_fields as _operator_gate_attention_fields,
     )
 
+    return _operator_gate_attention_fields(run)
+
 
 def compact_server_planning_contract(value: Any) -> dict[str, Any]:
     return _compact_server_planning_contract_read_model(