feat(domain): align plan model with spec

This commit is contained in:
2026-02-12 22:58:15 -05:00
parent 3adc9c1f47
commit 36b4ec2b8d
45 changed files with 5971 additions and 864 deletions
+166
View File
@@ -0,0 +1,166 @@
"""ASV benchmarks for Plan domain model validation and serialization.
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()
"""
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,
)
def _make_plan() -> Plan:
"""Create a fully-populated Plan for benchmarking."""
return Plan(
identity=PlanIdentity(plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV"),
namespaced_name=NamespacedName(
server=None, namespace="local", name="bench-plan"
),
description="Benchmark plan for validation and serialization",
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,
},
project_links=[
ProjectLink(project_name="local/project-a", alias="main"),
ProjectLink(project_name="local/project-b", alias="lib", 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,
),
],
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)."""
def time_plan_construction(self) -> None:
"""Benchmark creating a fully-populated Plan object."""
_make_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_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",
)
class PlanSerializationSuite:
"""Benchmark Plan serialization methods."""
def setup(self) -> None:
"""Create a plan for serialization benchmarks."""
self.plan = _make_plan()
def time_as_cli_dict(self) -> None:
"""Benchmark Plan.as_cli_dict() ordered dict generation."""
self.plan.as_cli_dict()
def time_model_dump(self) -> None:
"""Benchmark Pydantic model_dump() serialization."""
self.plan.model_dump()
def time_model_dump_json(self) -> None:
"""Benchmark Pydantic model_dump_json() JSON serialization."""
self.plan.model_dump_json()
class PhaseTransitionSuite:
"""Benchmark phase transition validation."""
def time_valid_transition(self) -> None:
"""Benchmark checking a valid phase transition."""
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_all_transitions(self) -> None:
"""Benchmark checking all possible phase transitions."""
phases = list(PlanPhase)
for from_phase in phases:
for to_phase in phases:
can_transition(from_phase, to_phase)