feat(domain): align plan model with spec

Step A2.beta

benchmarks/plan_model_bench.py

@@ -1,166 +1,179 @@
-"""ASV benchmarks for Plan domain model validation and serialization.
+"""Airspeed Velocity benchmarks for v3 Plan domain model.
 
-Measures the performance of:
-- Plan model construction (Pydantic validation)
-- Plan.as_cli_dict() serialization
-- ProjectLink alias validation
-- PlanInvariant construction
-- Phase transition validation via can_transition()
+Measures validation, serialization, and as_cli_dict performance for the
+Plan model across different field configurations.
 """
 
 from __future__ import annotations
 
-import sys
-from pathlib import Path
-
-try:
-    from cleveragents.domain.models.core.plan import (
-        AutomationLevel,
-        InvariantScope,
-        NamespacedName,
-        Plan,
-        PlanIdentity,
-        PlanInvariant,
-        PlanPhase,
-        ProcessingState,
-        ProjectLink,
-        can_transition,
-    )
-except ModuleNotFoundError:
-    sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
-    from cleveragents.domain.models.core.plan import (
-        AutomationLevel,
-        InvariantScope,
-        NamespacedName,
-        Plan,
-        PlanIdentity,
-        PlanInvariant,
-        PlanPhase,
-        ProcessingState,
-        ProjectLink,
-        can_transition,
-    )
+from cleveragents.domain.models.core.plan import (
+    AutomationProfileProvenance,
+    AutomationProfileRef,
+    InvariantSource,
+    NamespacedName,
+    Plan,
+    PlanIdentity,
+    PlanInvariant,
+    PlanPhase,
+    ProcessingState,
+    ProjectLink,
+    SubplanConfig,
+    SubplanStatus,
+    can_transition,
+)
 
 
-def _make_plan() -> Plan:
-    """Create a fully-populated Plan for benchmarking."""
+# Valid ULID constants for benchmarks
+_ULID_A = "01HGZ6FE0AQDYTR4BXVQZ6EA00"
+_ULID_B = "01HGZ6FE0AQDYTR4BXVQZ6EB00"
+
+
+def _minimal_plan() -> Plan:
+    """Create a minimal Plan with only required fields."""
     return Plan(
-        identity=PlanIdentity(plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV"),
+        identity=PlanIdentity(plan_id=_ULID_A),
         namespaced_name=NamespacedName(
             server=None, namespace="local", name="bench-plan"
         ),
-        description="Benchmark plan for validation and serialization",
+        description="Benchmark plan",
         action_name="local/bench-action",
-        definition_of_done="All benchmarks pass",
-        phase=PlanPhase.STRATEGIZE,
-        state=ProcessingState.QUEUED,
-        automation_level=AutomationLevel.REVIEW_BEFORE_APPLY,
-        automation_profile_name="org/bench-profile",
-        effective_profile_snapshot={
-            "max_file_changes": 50,
-            "require_tests": True,
-        },
+    )
+
+
+def _rich_plan() -> Plan:
+    """Create a Plan with all optional fields populated."""
+    return Plan(
+        identity=PlanIdentity(
+            plan_id=_ULID_A,
+            parent_plan_id=_ULID_B,
+            root_plan_id=_ULID_B,
+        ),
+        namespaced_name=NamespacedName(
+            server="prod", namespace="myorg", name="rich-plan"
+        ),
+        description="Fully populated benchmark plan",
+        definition_of_done="All tests pass with >97% coverage",
+        action_name="myorg/code-coverage",
+        phase=PlanPhase.EXECUTE,
+        processing_state=ProcessingState.PROCESSING,
+        automation_profile=AutomationProfileRef(
+            profile_name="trusted",
+            provenance=AutomationProfileProvenance.ACTION,
+        ),
         project_links=[
-            ProjectLink(project_name="local/project-a", alias="main"),
-            ProjectLink(project_name="local/project-b", alias="lib", read_only=True),
+            ProjectLink(project_name="local/api-svc", alias="api", read_only=False),
+            ProjectLink(project_name="local/docs", read_only=True),
         ],
         invariants=[
-            PlanInvariant(
-                text="Global invariant",
-                scope=InvariantScope.GLOBAL,
-                source_name="system",
-            ),
-            PlanInvariant(
-                text="Project invariant",
-                scope=InvariantScope.PROJECT,
-                source_name="my-project",
-            ),
-            PlanInvariant(
-                text="Plan invariant",
-                scope=InvariantScope.PLAN,
-                source_name=None,
-            ),
+            PlanInvariant(text="No public API changes", source=InvariantSource.ACTION),
+            PlanInvariant(text="Keep backward compat", source=InvariantSource.PROJECT),
+        ],
+        arguments={"target": "src/main.py", "mode": "strict", "verbose": True},
+        arguments_order=["target", "mode", "verbose"],
+        strategy_actor="local/planner",
+        execution_actor="local/coder",
+        review_actor="local/reviewer",
+        error_message=None,
+        subplan_config=SubplanConfig(max_parallel=3),
+        subplan_statuses=[
+            SubplanStatus(subplan_id=_ULID_B, action_name="local/sub-action"),
         ],
-        strategy_actor="openai/gpt-4",
-        execution_actor="openai/gpt-4",
-        created_by="bench-user",
-        tags=["benchmark", "test"],
-        reusable=True,
-        read_only=False,
     )
 
 
 class PlanValidationSuite:
-    """Benchmark Plan model construction (Pydantic validation)."""
+    """Benchmark Plan model validation (construction)."""
 
-    def time_plan_construction(self) -> None:
-        """Benchmark creating a fully-populated Plan object."""
-        _make_plan()
+    def time_minimal_plan_creation(self) -> None:
+        """Time creating a minimal Plan with only required fields."""
+        _minimal_plan()
 
-    def time_plan_minimal_construction(self) -> None:
-        """Benchmark creating a minimal Plan object."""
-        Plan(
-            identity=PlanIdentity(plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV"),
-            namespaced_name=NamespacedName(
-                server=None, namespace="local", name="min-plan"
-            ),
-            description="Minimal plan",
-            action_name="local/min-action",
-            phase=PlanPhase.STRATEGIZE,
-            state=ProcessingState.QUEUED,
-        )
+    def time_rich_plan_creation(self) -> None:
+        """Time creating a Plan with all optional fields populated."""
+        _rich_plan()
 
-    def time_project_link_with_alias(self) -> None:
-        """Benchmark ProjectLink construction with alias validation."""
-        ProjectLink(
-            project_name="local/my-project",
-            alias="my-alias",
-            read_only=False,
-        )
-
-    def time_plan_invariant_construction(self) -> None:
-        """Benchmark PlanInvariant construction with text validation."""
-        PlanInvariant(
-            text="No secrets in code",
-            scope=InvariantScope.GLOBAL,
-            source_name="system",
+    def time_plan_model_validate_dict(self) -> None:
+        """Time Plan.model_validate from a raw dict."""
+        Plan.model_validate(
+            {
+                "identity": {"plan_id": _ULID_A},
+                "namespaced_name": {"namespace": "local", "name": "bench-plan"},
+                "description": "Benchmark plan",
+                "action_name": "local/bench-action",
+            }
         )
 
 
 class PlanSerializationSuite:
-    """Benchmark Plan serialization methods."""
+    """Benchmark Plan model serialization."""
 
     def setup(self) -> None:
-        """Create a plan for serialization benchmarks."""
-        self.plan = _make_plan()
+        self.minimal = _minimal_plan()
+        self.rich = _rich_plan()
 
-    def time_as_cli_dict(self) -> None:
-        """Benchmark Plan.as_cli_dict() ordered dict generation."""
-        self.plan.as_cli_dict()
+    def time_minimal_model_dump(self) -> None:
+        """Time model_dump for a minimal Plan."""
+        self.minimal.model_dump()
 
-    def time_model_dump(self) -> None:
-        """Benchmark Pydantic model_dump() serialization."""
-        self.plan.model_dump()
+    def time_rich_model_dump(self) -> None:
+        """Time model_dump for a fully populated Plan."""
+        self.rich.model_dump()
 
-    def time_model_dump_json(self) -> None:
-        """Benchmark Pydantic model_dump_json() JSON serialization."""
-        self.plan.model_dump_json()
+    def time_minimal_model_dump_json(self) -> None:
+        """Time model_dump_json for a minimal Plan."""
+        self.minimal.model_dump_json()
+
+    def time_rich_model_dump_json(self) -> None:
+        """Time model_dump_json for a fully populated Plan."""
+        self.rich.model_dump_json()
+
+
+class PlanCliDictSuite:
+    """Benchmark Plan.as_cli_dict output generation."""
+
+    def setup(self) -> None:
+        self.minimal = _minimal_plan()
+        self.rich = _rich_plan()
+
+    def time_minimal_as_cli_dict(self) -> None:
+        """Time as_cli_dict for a minimal Plan."""
+        self.minimal.as_cli_dict()
+
+    def time_rich_as_cli_dict(self) -> None:
+        """Time as_cli_dict for a Plan with all optional fields."""
+        self.rich.as_cli_dict()
 
 
 class PhaseTransitionSuite:
     """Benchmark phase transition validation."""
 
-    def time_valid_transition(self) -> None:
-        """Benchmark checking a valid phase transition."""
+    def time_can_transition_valid(self) -> None:
+        """Time a valid phase transition check."""
         can_transition(PlanPhase.STRATEGIZE, PlanPhase.EXECUTE)
 
-    def time_invalid_transition(self) -> None:
-        """Benchmark checking an invalid phase transition."""
-        can_transition(PlanPhase.STRATEGIZE, PlanPhase.APPLY)
+    def time_can_transition_invalid(self) -> None:
+        """Time an invalid phase transition check."""
+        can_transition(PlanPhase.STRATEGIZE, PlanPhase.APPLIED)
 
     def time_all_transitions(self) -> None:
-        """Benchmark checking all possible phase transitions."""
-        phases = list(PlanPhase)
-        for from_phase in phases:
-            for to_phase in phases:
+        """Time checking all possible phase transition pairs."""
+        for from_phase in PlanPhase:
+            for to_phase in PlanPhase:
                 can_transition(from_phase, to_phase)
+
+
+class NamespacedNameSuite:
+    """Benchmark NamespacedName parsing and serialization."""
+
+    def time_parse_simple(self) -> None:
+        """Time parsing a simple name."""
+        NamespacedName.parse("my-action")
+
+    def time_parse_full(self) -> None:
+        """Time parsing a full server:namespace/name string."""
+        NamespacedName.parse("prod:myorg/my-action")
+
+    def time_str_representation(self) -> None:
+        """Time string representation of a NamespacedName."""
+        name = NamespacedName(server="prod", namespace="myorg", name="my-action")
+        str(name)