refactor(status): move dreaming projections into bounded module (#2974)

Author: huangruiteng

loopx/control_plane/status/dreaming_projection.py

@@ -0,0 +1,52 @@
+"""Dreaming and planning projections inside the `status` bounded context."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from ...operator_gate import normalize_operator_question
+from ..goals.dreaming import (
+    compact_dreaming_lane_badge as _compact_dreaming_lane_badge,
+    compact_dreaming_proposal as _compact_dreaming_proposal,
+    compact_server_planning_contract as _compact_server_planning_contract,
+    dreaming_attention_fields as _dreaming_attention_fields,
+)
+from ..runtime.public_safety import (
+    public_safe_compact_list,
+    public_safe_compact_text,
+)
+from ..runtime.status_classifications import DREAMING_ADVISORY_CLASSIFICATIONS
+
+
+def compact_server_planning_contract(value: Any) -> dict[str, Any]:
+    return _compact_server_planning_contract(
+        value,
+        public_safe_compact_text=public_safe_compact_text,
+        public_safe_compact_list=public_safe_compact_list,
+    )
+
+
+def compact_dreaming_proposal(run: dict[str, Any] | None) -> dict[str, Any] | None:
+    return _compact_dreaming_proposal(
+        run,
+        dreaming_advisory_classifications=DREAMING_ADVISORY_CLASSIFICATIONS,
+        public_safe_compact_text=public_safe_compact_text,
+        public_safe_compact_list=public_safe_compact_list,
+    )
+
+
+def compact_dreaming_lane_badge(proposal: dict[str, Any] | None) -> dict[str, Any] | None:
+    return _compact_dreaming_lane_badge(
+        proposal,
+        public_safe_compact_text=public_safe_compact_text,
+    )
+
+
+def dreaming_attention_fields(run: dict[str, Any] | None) -> dict[str, Any]:
+    return _dreaming_attention_fields(
+        run,
+        dreaming_advisory_classifications=DREAMING_ADVISORY_CLASSIFICATIONS,
+        public_safe_compact_text=public_safe_compact_text,
+        public_safe_compact_list=public_safe_compact_list,
+        normalize_operator_question=normalize_operator_question,
+    )

loopx/status.py

@@ -103,12 +103,6 @@
 from .control_plane.work_items.backlog_hygiene import (
     MAX_BACKLOG_HYGIENE_EVIDENCE_ITEMS as _MAX_BACKLOG_HYGIENE_EVIDENCE_ITEMS_READ_MODEL,
 )
-from .control_plane.goals.dreaming import (
-    compact_dreaming_lane_badge as _compact_dreaming_lane_badge_read_model,
-    compact_dreaming_proposal as _compact_dreaming_proposal_read_model,
-    compact_server_planning_contract as _compact_server_planning_contract_read_model,
-    dreaming_attention_fields as _dreaming_attention_fields_read_model,
-)
 from .control_plane.work_items.delivery_signals import (
     classification_contains_any as _classification_contains_any_read_model,
     delivery_batch_scale_for_run as _delivery_batch_scale_for_run_read_model,
@@ -136,7 +130,7 @@
 from .control_plane.runtime.status_classifications import (
     BLOCKING_CLASSIFICATIONS,
     CODEX_READY_CLASSIFICATIONS,
-    DREAMING_ADVISORY_CLASSIFICATIONS,
+    DREAMING_ADVISORY_CLASSIFICATIONS,  # noqa: F401
     HANDOFF_READY_CLASSIFICATIONS,
     USER_OR_CONTROLLER_CLASSIFICATIONS,
 )
@@ -284,6 +278,7 @@
 
 
 _PUBLIC_COMPAT_REEXPORTS = {
+    "DREAMING_ADVISORY_CLASSIFICATIONS": "loopx.control_plane.runtime.status_classifications",
     "TODO_PROJECTION_DETAIL_POINTER_SCHEMA_VERSION": "loopx.control_plane.work_items.project_asset",
     "TODO_PROJECTION_VIEW_SCHEMA_VERSION": "loopx.control_plane.work_items.project_asset",
     "project_asset_summary_is_public_safe": "loopx.control_plane.work_items.project_asset",
@@ -2664,38 +2659,36 @@ def operator_gate_attention_fields(run: dict[str, Any] | None) -> dict[str, Any]
 
 
 def compact_server_planning_contract(value: Any) -> dict[str, Any]:
-    return _compact_server_planning_contract_read_model(
-        value,
-        public_safe_compact_text=public_safe_compact_text,
-        public_safe_compact_list=public_safe_compact_list,
+    from .control_plane.status.dreaming_projection import (
+        compact_server_planning_contract as _compact_server_planning_contract,
     )
 
+    return _compact_server_planning_contract(value)
+
 
 def compact_dreaming_proposal(run: dict[str, Any] | None) -> dict[str, Any] | None:
-    return _compact_dreaming_proposal_read_model(
-        run,
-        dreaming_advisory_classifications=DREAMING_ADVISORY_CLASSIFICATIONS,
-        public_safe_compact_text=public_safe_compact_text,
-        public_safe_compact_list=public_safe_compact_list,
+    from .control_plane.status.dreaming_projection import (
+        compact_dreaming_proposal as _compact_dreaming_proposal,
     )
 
+    return _compact_dreaming_proposal(run)
+
 
 def compact_dreaming_lane_badge(proposal: dict[str, Any] | None) -> dict[str, Any] | None:
-    return _compact_dreaming_lane_badge_read_model(
-        proposal,
-        public_safe_compact_text=public_safe_compact_text,
+    from .control_plane.status.dreaming_projection import (
+        compact_dreaming_lane_badge as _compact_dreaming_lane_badge,
     )
 
+    return _compact_dreaming_lane_badge(proposal)
+
 
 def dreaming_attention_fields(run: dict[str, Any] | None) -> dict[str, Any]:
-    return _dreaming_attention_fields_read_model(
-        run,
-        dreaming_advisory_classifications=DREAMING_ADVISORY_CLASSIFICATIONS,
-        public_safe_compact_text=public_safe_compact_text,
-        public_safe_compact_list=public_safe_compact_list,
-        normalize_operator_question=normalize_operator_question,
+    from .control_plane.status.dreaming_projection import (
+        dreaming_attention_fields as _dreaming_attention_fields,
     )
 
+    return _dreaming_attention_fields(run)
+
 
 def legacy_runtime_goal_attention(
     goal: dict[str, Any],

tests/control_plane/test_status_classification_reexport.py

@@ -2,6 +2,7 @@
 
 import loopx.control_plane.runtime.status_classifications as classifications
 import loopx.status as status
+from loopx.status import DREAMING_ADVISORY_CLASSIFICATIONS
 
 
 REEXPORTED_NAMES = (
@@ -19,6 +20,7 @@ def test_status_reexports_classification_sets_unchanged() -> None:
         read_model_value = getattr(classifications, name)
         assert status_value == read_model_value
         assert type(status_value) is type(read_model_value)
+    assert DREAMING_ADVISORY_CLASSIFICATIONS == classifications.DREAMING_ADVISORY_CLASSIFICATIONS
 
 
 def test_status_classification_membership_is_stable() -> None: