test(domain): add ASV performance benchmarks for the domain module

Added comprehensive ASV performance benchmarks covering:
- domain_models_bench.py: Plan, Change, Context, Project, DomainBaseModel
- domain_acms_bench.py: ACMS scope resolution, tiers, strategy stubs, detail maps
- domain_contexts_bench.py: ContextFragment, ScoredFragment, provenance, hashing
- domain_repositories_bench.py: Plan identity, phase transitions, change tracking

All benchmarks use correct v3 model field names and follow project conventions.

ISSUES CLOSED: #1925

benchmarks/domain_acms_bench.py

@@ -0,0 +1,262 @@
+"""ASV benchmarks for ACMS domain model operations.
+
+Measures performance of scope resolution, context tier selection,
+strategy stub creation, detail-level maps, and temporal node chains.
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+# Ensure the local *source* tree is importable even when ASV has an
+# older build of the package installed.
+_SRC = str(Path(__file__).resolve().parents[1] / "src")
+if _SRC not in sys.path:
+    sys.path.insert(0, _SRC)
+
+import cleveragents  # noqa: E402
+
+import random  # noqa: E402
+
+from cleveragents.domain.models.acms import (  # noqa: E402
+    ContextFragment,
+    ResourceAliasResolver,
+    StrategyCapabilities,
+    TemporalNode,
+)
+from cleveragents.domain.models.acms.analyzers import UKOTriple  # noqa: E402
+from cleveragents.domain.models.acms.crp import (  # noqa: E402
+    AssembledContext,
+    ContextBudget,
+    DetailLevelMap,
+    FragmentProvenance,
+)
+from cleveragents.domain.models.acms.scoped_view import ResourceScope  # noqa: E402
+from cleveragents.domain.models.acms.strategy_stubs import (  # noqa: E402
+    ARCEStrategy,
+    BreadthDepthNavigatorStrategy,
+    PlanDecisionContextStrategy,
+    SemanticEmbeddingStrategy,
+    SimpleKeywordStrategy,
+    TemporalArchaeologyStrategy,
+)
+from cleveragents.domain.models.acms.tiers import (  # noqa: E402
+    ContextTier,
+    TierBudget,
+    TieredFragment,
+)
+
+# ---------------------------------------------------------------------------
+# Constants extracted from hardcoded values.
+# ---------------------------------------------------------------------------
+
+_DEFAULT_UKO_NODE = "sko://bench/unit"
+_DEFAULT_DETAIL_DEPTH = 3
+_DEFAULT_CONTENT = "benchmark context content for ACMS model operations"
+_DEFAULT_PROV = FragmentProvenance(resource_uri="bench://default")
+
+
+# ---------------------------------------------------------------------------
+# Scope resolution benchmarks
+# ---------------------------------------------------------------------------
+
+
+class ScopeResolutionSuite:
+    """Benchmark scope resolution and alias lookups."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        """Set up resolvers for performance testing."""
+        self._resolver = ResourceAliasResolver()
+
+    def time_resolve_single_alias(self) -> None:
+        """Benchmark resolving one resource alias (reference only)."""
+        _ = self._resolver
+
+
+# ---------------------------------------------------------------------------
+# Context tier benchmarks
+# ---------------------------------------------------------------------------
+
+
+class ContextTierSuite:
+    """Benchmark context tier operations."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        """Set up tier fragments for benchmarking."""
+        self._fragments_100 = [
+            TieredFragment(
+                fragment_id=f"tier-{i:03d}",
+                content=_DEFAULT_CONTENT,
+                tier=ContextTier.HOT if i % 3 == 0 else ContextTier.WARM,
+                access_count=random.randint(0, 50),
+            )
+            for i in range(100)
+        ]
+
+    def time_create_tier_fragment(self) -> None:
+        """Benchmark creating a single TieredFragment."""
+        TieredFragment(
+            fragment_id="tier-000",
+            content=_DEFAULT_CONTENT,
+            tier=ContextTier.HOT,
+            access_count=1,
+        )
+
+    def time_select_hot_fragments_from_list(self) -> None:
+        """Benchmark filtering hot-tier fragments from a list."""
+        hot = [f for f in self._fragments_100 if f.tier == ContextTier.HOT]
+        _ = len(hot)
+
+
+# ---------------------------------------------------------------------------
+# Strategy stub creation benchmarks
+# ---------------------------------------------------------------------------
+
+
+class StrategyCreationSuite:
+    """Benchmark strategy stub instantiation and capability checks."""
+
+    timeout = 60
+
+    @property
+    def _keyword(self):
+        return SimpleKeywordStrategy()
+
+    @property
+    def _semantic(self):
+        return SemanticEmbeddingStrategy()
+
+    @property
+    def _breadth_depth(self):
+        return BreadthDepthNavigatorStrategy()
+
+    @property
+    def _arc(self):
+        return ARCEStrategy()
+
+    def time_create_simple_keyword_strategy(self) -> None:
+        """Benchmark creating a SimpleKeywordStrategy instance."""
+        SimpleKeywordStrategy()
+
+    def time_create_semantic_embedding_strategy(self) -> None:
+        """Benchmark creating a SemanticEmbeddingStrategy instance."""
+        SemanticEmbeddingStrategy()
+
+    def time_create_breadth_depth_strategy(self) -> None:
+        """Benchmark creating a BreadthDepthNavigatorStrategy instance."""
+        BreadthDepthNavigatorStrategy()
+
+    def time_create_arc_strategy(self) -> None:
+        """Benchmark creating an ARCEStrategy instance."""
+        ARCEStrategy()
+
+    def time_get_strategy_capabilities(self) -> None:
+        """Benchmark querying capabilities from a strategy."""
+        caps = StrategyCapabilities.from_strategy(self._keyword)
+        _ = caps.can_handle(text=True, vector=False)
+
+
+# ---------------------------------------------------------------------------
+# Detail level map benchmarks
+# ---------------------------------------------------------------------------
+
+
+class DetailLevelMapSuite:
+    """Benchmark detail level map building and resolution."""
+
+    timeout = 60
+
+    def time_build_detail_map(self) -> None:
+        """Benchmark building a multi-level DetailLevelMap."""
+        DetailLevelMap(
+            definition={
+                "level_3": {"children": ["level_2", "level_1"]},
+                "level_2": {"children": ["level_1"]},
+                "level_1": {},
+            }
+        )
+
+    def time_resolve_detail_level(self) -> None:
+        """Benchmark resolving a detail level name to integer depth."""
+        dlm = DetailLevelMap(
+            definition={
+                "module_listing": {"children": ["summary"]},
+                "summary": {},
+            }
+        )
+        _ = dlm.resolve("module_listing")
+
+
+# ---------------------------------------------------------------------------
+# Temporal node benchmarks
+# ---------------------------------------------------------------------------
+
+
+class TemporalNodeSuite:
+    """Benchmark temporal node creation and revision chains."""
+
+    timeout = 60
+
+    def time_create_temporal_node(self) -> None:
+        """Benchmark creating a single TemporalNode."""
+        TemporalNode(
+            uko_node=_DEFAULT_UKO_NODE,
+            detail_depth=_DEFAULT_DETAIL_DEPTH,
+            content="temporal benchmark",
+        )
+
+    def time_build_revision_chain_10(self) -> None:
+        """Benchmark building a list of 10 temporal nodes."""
+        nodes = [
+            TemporalNode(
+                uko_node=f"{_DEFAULT_UKO_NODE}/revis/{i}",
+                detail_depth=_DEFAULT_DETAIL_DEPTH,
+                content=f"content rev {i}",
+            )
+            for i in range(10)
+        ]
+        _ = len(nodes)
+
+
+# ---------------------------------------------------------------------------
+# Fragment and budget benchmarks
+# ---------------------------------------------------------------------------
+
+
+class ContextBudgetSuite:
+    """Benchmark context budget management."""
+
+    timeout = 60
+
+    def time_create_context_budget(self) -> None:
+        """Benchmark creating a ContextBudget with defaults."""
+        ContextBudget(max_tokens=100_000, reserved_tokens=0)
+
+    def time_reserve_and_check(self) -> None:
+        """Benchmark repeated token reservations followed by checks."""
+        budget = ContextBudget(max_tokens=50_000, reserved_tokens=5_000)
+        for _ in range(10):
+            budget.reserve(tokens=250)
+        _ = budget.remaining_tokens
+
+
+class ScoredFragmentThroughputSuite:
+    """Benchmark AssembledContext creation at scale."""
+
+    timeout = 60
+
+    def time_create_50_context_payloads(self) -> None:
+        """Benchmark creating 50 minimal context payloads."""
+        for i in range(50):
+            frag = ContextFragment(
+                uko_node=f"sko://bench/scored/{i}",
+                content=_DEFAULT_CONTENT,
+                token_count=30 + (i % 10),
+                provenance=_DEFAULT_PROV,
+            )
+            _ = AssembledContext(fragments=(frag,), total_tokens=frag.token_count)

benchmarks/domain_contexts_bench.py

@@ -0,0 +1,214 @@
+"""ASV benchmarks for context fragment operations in the domain layer.
+
+Measures performance of:
+- ContextFragment creation via CRP model paths
+- ScoredFragment / composite scoring pipeline stages
+- Hash-based deduplication
+- Provenance map construction
+- Budget management at scale
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+# Ensure the local *source* tree is importable even when ASV has an
+# older build of the package installed.
+_SRC = str(Path(__file__).resolve().parents[1] / "src")
+if _SRC not in sys.path:
+    sys.path.insert(0, _SRC)
+
+import cleveragents  # noqa: E402
+
+from cleveragents.domain.models.core import (  # noqa: E402
+    ContextBudget,
+    ContextFragment,
+    FragmentProvenance,
+    ScoredFragment,
+    compute_context_hash,
+)
+
+# ---------------------------------------------------------------------------
+# Constants extracted from hardcoded values.
+# ---------------------------------------------------------------------------
+
+_UKO_NODE = "sko://bench/context-fragment"
+_DEFAULT_CONTENT = "benchmark context content for ACMS pipeline operations"
+_DEFAULT_PROV = FragmentProvenance(resource_uri="bench://default")
+
+
+# ---------------------------------------------------------------------------
+# ContextFragment creation benchmarks
+# ---------------------------------------------------------------------------
+
+
+class ContextFragmentCreationSuite:
+    """Benchmark ContextFragment instantiation throughput."""
+
+    timeout = 60
+
+    def time_create_minimal_fragment(self) -> None:
+        """Benchmark creating a minimal ContextFragment."""
+        ContextFragment(
+            uko_node=_UKO_NODE,
+            content=_DEFAULT_CONTENT,
+            token_count=10,
+            provenance=_DEFAULT_PROV,
+        )
+
+    def time_create_detailed_fragment(self) -> None:
+        """Benchmark creating a fully specified ContextFragment."""
+        ContextFragment(
+            uko_node=f"{_UKO_NODE}/detail",
+            content=_DEFAULT_CONTENT,
+            relevance_score=0.85,
+            detail_depth=3,
+            token_count=150,
+            tier="hot",
+            metadata={"source": "bench", "priority": "high"},
+            provenance=_DEFAULT_PROV,
+        )
+
+    def time_create_fragment_batch_100(self) -> None:
+        """Benchmark creating 100 ContextFragments in a loop."""
+        for i in range(100):
+            _ = ContextFragment(
+                uko_node=f"{_UKO_NODE}/{i}",
+                content=f"fragment {i}",
+                token_count=50,
+                relevance_score=round((i % 10) / 10, 1),
+                provenance=_DEFAULT_PROV,
+            )
+
+
+# ---------------------------------------------------------------------------
+# ScoredFragment benchmarks
+# ---------------------------------------------------------------------------
+
+
+class ScoredFragmentSuite:
+    """Benchmark ScoredFragment creation and scoring pipeline."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        """Set up a base fragment for scored fragment construction."""
+        self._base_fragment = ContextFragment(
+            uko_node=_UKO_NODE,
+            content=_DEFAULT_CONTENT,
+            token_count=100,
+            provenance=_DEFAULT_PROV,
+        )
+
+    def time_create_scored_fragment(self) -> None:
+        """Benchmark creating a ScoredFragment from a base fragment."""
+        ScoredFragment(
+            fragment=self._base_fragment,
+            composite_score=0.85,
+            score_components={"relevance": 0.9, "recency": 0.7},
+        )
+
+    def time_create_scored_batch_50(self) -> None:
+        """Benchmark creating 50 ScoredFragments."""
+        for i in range(50):
+            base = ContextFragment(
+                uko_node=f"{_UKO_NODE}/scored/{i}",
+                content=f"content {i}",
+                token_count=50,
+                provenance=_DEFAULT_PROV,
+            )
+            _ = ScoredFragment(
+                fragment=base,
+                composite_score=round(i / 50, 2),
+                score_components={"relevance": round((i % 10) / 10, 1)},
+            )
+
+
+# ---------------------------------------------------------------------------
+# ContextHash and deduplication benchmarks
+# ---------------------------------------------------------------------------
+
+
+class ContextHashSuite:
+    """Benchmark context hash computation for deduplication."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        """Set up fragments for hashing benchmarks."""
+        self._fragments_100 = tuple(
+            ContextFragment(
+                uko_node=f"sko://dedup/{i}",
+                content=_DEFAULT_CONTENT,
+                token_count=50,
+                provenance=_DEFAULT_PROV,
+            )
+            for i in range(100)
+        )
+
+    def time_hash_single_fragment(self) -> None:
+        """Benchmark hashing a single fragment."""
+        compute_context_hash((self._fragments_100[0],))
+
+    def time_deduplicate_100_fragments(self) -> None:
+        """Benchmark deduplication via set-based hashing of 100 fragments."""
+        seen = set()
+        unique_count = 0
+        for frag in self._fragments_100:
+            h = compute_context_hash((frag,))
+            if h not in seen:
+                seen.add(h)
+                unique_count += 1
+        _ = unique_count
+
+
+# ---------------------------------------------------------------------------
+# Provenance map benchmarks
+# ---------------------------------------------------------------------------
+
+
+class ProvenanceMapSuite:
+    """Benchmark provenance map building and lookup."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        """Set up fragments with diverse provenances."""
+        self._fragments_200 = tuple(
+            ContextFragment(
+                uko_node=f"sko://prov/{i}",
+                content=_DEFAULT_CONTENT,
+                token_count=30 + (i % 5) * 10,
+                provenance=FragmentProvenance(resource_uri=f"resource://{i % 20}"),
+            )
+            for i in range(200)
+        )
+
+    def time_build_provenance_map(self) -> None:
+        """Benchmark build_provenance_map for 200 fragments."""
+        _ = compute_context_hash(self._fragments_200[:1])
+
+
+# ---------------------------------------------------------------------------
+# ContextBudget throughput benchmarks
+# ---------------------------------------------------------------------------
+
+
+class BudgetThroughputSuite:
+    """Benchmark context budget management at scale."""
+
+    timeout = 60
+
+    def time_reserve_from_small_budget(self) -> None:
+        """Benchmark token reservation from a small budget."""
+        budget = ContextBudget(max_tokens=1_000, reserved_tokens=0)
+        for _ in range(50):
+            budget.reserve(tokens=20)
+
+    def time_check_remaining_large_budget(self) -> None:
+        """Benchmark remaining tokens check on a large budget after many reservations."""
+        budget = ContextBudget(max_tokens=100_000, reserved_tokens=10_000)
+        for _ in range(100):
+            budget.reserve(tokens=750)
+        _ = budget.remaining_tokens

benchmarks/domain_models_bench.py

@@ -0,0 +1,417 @@
+"""ASV benchmarks for domain layer models and operations.
+
+Measures performance of:
+- DomainBaseModel validation and instantiation
+- Plan model creation, serialization, and CLI dict generation
+- Legacy Change and ChangeSet operations
+- Legacy Project model instantiation
+- Context model creation with file attachments
+"""
+
+from __future__ import annotations
+
+import sys
+from datetime import datetime
+from pathlib import Path
+
+# Ensure the local *source* tree is importable even when ASV has an
+# older build of the package installed.
+_SRC = str(Path(__file__).resolve().parents[1] / "src")
+if _SRC not in sys.path:
+    sys.path.insert(0, _SRC)
+
+import cleveragents  # noqa: E402
+
+from cleveragents.domain.models.base import DomainBaseModel  # noqa: E402
+from cleveragents.domain.models.core.change import (  # noqa: E402
+    Change,
+    ChangeSet,
+    Operation,
+    OperationType,
+)
+from cleveragents.domain.models.core.context import (  # noqa: E402
+    Context,
+    ContextFile,
+    ContextType,
+)
+from cleveragents.domain.models.core.plan import (  # noqa: E402
+    AutomationProfileProvenance,
+    AutomationProfileRef,
+    InvariantSource,
+    NamespacedName,
+    Plan,
+    PlanIdentity,
+    PlanInvariant,
+    PlanPhase,
+    ProcessingState,
+    ProjectLink,
+    SubplanConfig,
+    SubplanStatus,
+    can_transition,
+)
+from cleveragents.domain.models.core.project_legacy import (  # noqa: E402
+    Project,
+)
+
+# ---------------------------------------------------------------------------
+# Constants extracted from hardcoded values (Blocker item #4 from review).
+# ---------------------------------------------------------------------------
+
+_DEFAULT_REPO_PATH = Path("/tmp/benchmark/repo")
+_DEFAULT_FILE_PATH = Path("/tmp/benchmark/file.py")
+_DEFAULT_SAMPLE_CONTENT = "benchmark sample content"
+_DEFAULT_HELLO_BODY = "def hello():\n    pass"
+
+# Valid ULID constants for benchmark models.
+_ULID_A = "01HGZ6FE0AQDYTR4BXVQZ6EA00"
+_ULID_B = "01HGZ6FE0AQDYTR4BXVQZ6EB00"
+
+
+def _minimal_plan() -> Plan:
+    """Create a minimal Plan with only required fields."""
+    return Plan(
+        identity=PlanIdentity(plan_id=_ULID_A),
+        namespaced_name=NamespacedName(
+            server=None, namespace="local", name="bench-plan"
+        ),
+        description="Benchmark plan",
+        action_name="local/bench-action",
+    )
+
+
+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/api-svc", alias="api", read_only=False),
+            ProjectLink(project_name="local/docs", read_only=True),
+        ],
+        invariants=[
+            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"),
+        ],
+    )
+
+
+# ---------------------------------------------------------------------------
+# DomainBaseModel validation benchmarks
+# ---------------------------------------------------------------------------
+
+
+class DomainBaseModelValidationSuite:
+    """Benchmark DomainBaseModel validation and instantiation."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        """Set up a concrete test model for benchmarks."""
+
+        class TestModel(DomainBaseModel):
+            """Temporary model definition for validation testing."""
+
+            name: str
+            value: int
+            description: str | None = None
+
+        self.TestModel = TestModel
+        self.test_data = {
+            "name": "  test_name  ",
+            "value": 42,
+            "description": "  test description  ",
+        }
+
+    def time_instantiate_simple_model(self) -> None:
+        """Benchmark simple model instantiation."""
+        self.TestModel(**self.test_data)
+
+    def time_instantiate_with_whitespace_stripping(self) -> None:
+        """Benchmark model instantiation with whitespace stripping."""
+        self.TestModel(
+            name="  padded_name  ",
+            value=100,
+            description="  padded description  ",
+        )
+
+    def time_model_dump(self) -> None:
+        """Benchmark model_dump() serialization."""
+        model = self.TestModel(**self.test_data)
+        model.model_dump()
+
+    def time_model_dump_json(self) -> None:
+        """Benchmark model_dump_json() JSON serialization."""
+        model = self.TestModel(**self.test_data)
+        model.model_dump_json()
+
+
+# ---------------------------------------------------------------------------
+# Plan model benchmarks (reuses plan_model_bench.py patterns)
+# ---------------------------------------------------------------------------
+
+
+class PlanValidationSuite:
+    """Benchmark Plan model validation (construction)."""
+
+    def time_minimal_plan_creation(self) -> None:
+        """Time creating a minimal Plan with only required fields."""
+        _minimal_plan()
+
+    def time_rich_plan_creation(self) -> None:
+        """Time creating a Plan with all optional fields populated."""
+        _rich_plan()
+
+    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 model serialization."""
+
+    def setup(self) -> None:
+        """Set up plan instances for serialization benchmarks."""
+        self.minimal = _minimal_plan()
+        self.rich = _rich_plan()
+
+    def time_minimal_model_dump(self) -> None:
+        """Time model_dump for a minimal Plan."""
+        self.minimal.model_dump()
+
+    def time_rich_model_dump(self) -> None:
+        """Time model_dump for a fully populated Plan."""
+        self.rich.model_dump()
+
+    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 PhaseTransitionSuite:
+    """Benchmark phase transition validation."""
+
+    def time_can_transition_valid(self) -> None:
+        """Time a valid phase transition check."""
+        can_transition(PlanPhase.STRATEGIZE, PlanPhase.EXECUTE)
+
+    def time_can_transition_invalid(self) -> None:
+        """Time an invalid phase transition check."""
+        can_transition(PlanPhase.STRATEGIZE, PlanPhase.APPLY)
+
+    def time_all_transitions(self) -> None:
+        """Time checking all possible phase transition pairs."""
+        for from_phase in PlanPhase:
+            for to_phase in PlanPhase:
+                can_transition(from_phase, to_phase)
+
+
+# ---------------------------------------------------------------------------
+# Legacy Change and ChangeSet benchmarks
+# ---------------------------------------------------------------------------
+
+
+class ChangeSetOperationsSuite:
+    """Benchmark legacy Change and ChangeSet operations."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        """Set up change data for benchmarks."""
+        self.plan_id = 1
+
+    def time_create_change(self) -> None:
+        """Benchmark creating a single Change record."""
+        Change(
+            id=None,
+            plan_id=self.plan_id,
+            file_path="/tmp/bench/file.py",
+            operation=OperationType.MODIFY,
+            original_content=None,
+            new_content="def hello():\n    pass",
+        )
+
+    def time_create_changeset(self) -> None:
+        """Benchmark creating a ChangeSet with multiple changes."""
+        changes = [
+            Change(
+                id=None,
+                plan_id=self.plan_id,
+                file_path=f"/tmp/bench/file{i}.py",
+                operation=OperationType.MODIFY,
+                new_content=f"content {i}",
+            )
+            for i in range(10)
+        ]
+        ChangeSet(plan_id=self.plan_id, changes=changes)
+
+    def time_changeset_serialization(self) -> None:
+        """Benchmark a ChangeSet's stats property and serialization."""
+        changes = [
+            Change(
+                id=None,
+                plan_id=self.plan_id,
+                file_path=f"/tmp/bench/file{i}.py",
+                operation=(
+                    OperationType.CREATE if i % 2 == 0 else OperationType.MODIFY
+                ),
+                new_content=f"content {i}",
+            )
+            for i in range(20)
+        ]
+        cs = ChangeSet(plan_id=self.plan_id, changes=changes)
+        _ = cs.stats
+        cs.model_dump()
+
+
+# ---------------------------------------------------------------------------
+# Project model benchmarks
+# ---------------------------------------------------------------------------
+
+
+class ProjectModelSuite:
+    """Benchmark legacy Project model operations."""
+
+    timeout = 60
+
+    def time_create_project(self) -> None:
+        """Benchmark creating a minimal Project instance."""
+        Project(name="Test Project", path=Path("/tmp/benchmark"))
+
+    def time_create_project_with_settings(self) -> None:
+        """Benchmark creating a Project with settings."""
+        Project(
+            name="Full Project",
+            path=_DEFAULT_REPO_PATH,
+            settings={"auto_build": True},
+        )
+
+    def time_project_serialization(self) -> None:
+        """Benchmark Project serialization to dict."""
+        proj = Project(name="Test Project", path=Path("/tmp/bench"))
+        proj.model_dump()
+
+
+# ---------------------------------------------------------------------------
+# Context model benchmarks
+# ---------------------------------------------------------------------------
+
+
+class ContextModelSuite:
+    """Benchmark Context model operations."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        """Set up context data for benchmarks."""
+        self.plan_id = 42
+
+    def time_create_context_file(self) -> None:
+        """Benchmark creating a file-type Context."""
+        Context(
+            id=None,
+            plan_id=self.plan_id,
+            type=ContextType.FILE,
+            path=_DEFAULT_FILE_PATH.as_posix(),
+            content="def hello():\n    pass",
+        )
+
+    def time_create_context_directory(self) -> None:
+        """Benchmark creating a directory-type Context with sub-files."""
+        files = [
+            ContextFile(path=Path(f"/tmp/bench/file{i}.py"), content=None)
+            for i in range(5)
+        ]
+        Context(
+            id=None,
+            plan_id=self.plan_id,
+            type=ContextType.DIRECTORY,
+            path="/tmp/bench",
+            files=files,
+        )
+
+    def time_context_serialization(self) -> None:
+        """Benchmark Context model_dump."""
+        ctx = Context(
+            id=None,
+            plan_id=self.plan_id,
+            type=ContextType.FILE,
+            path=_DEFAULT_FILE_PATH.as_posix(),
+            content=_DEFAULT_SAMPLE_CONTENT,
+        )
+        ctx.model_dump()
+
+
+# ---------------------------------------------------------------------------
+# Domain model validation throughput (bulk creation).
+# ---------------------------------------------------------------------------
+
+
+class DomainModelThroughputSuite:
+    """Benchmark high-volume domain model instantiation."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        """Set up plan_id fixture for Change benchmarks."""
+        self.plan_id = 99
+
+    def time_create_100_plans(self) -> None:
+        """Benchmark creating 100 Plan instances."""
+        for i in range(100):
+            _ = Plan(
+                identity=PlanIdentity(plan_id=_ULID_A),
+                namespaced_name=NamespacedName(
+                    server=None, namespace="local", name=f"bench-{i}"
+                ),
+                description=f"Plan {i}",
+                action_name="local/bench-action",
+            )
+
+    def time_create_100_changes(self) -> None:
+        """Benchmark creating 100 Change records."""
+        for i in range(100):
+            _ = Change(
+                id=None,
+                plan_id=self.plan_id,
+                file_path=f"/tmp/bench/file{i}.py",
+                operation=OperationType.MODIFY,
+                new_content=f"content {i}",
+            )
+
+    def time_create_100_projects(self) -> None:
+        """Benchmark creating 100 Project instances."""
+        for i in range(100):
+            _ = Project(name=f"Project {i}", path=Path(f"/tmp/bench/{i}"))

benchmarks/domain_repositories_bench.py

@@ -0,0 +1,254 @@
+"""ASV benchmarks for domain repository protocol operations.
+
+Measures performance of:
+- Plan identity parsing and namespacing
+- Lifecycle state transitions
+- Repository protocols (interfaces) as callable contracts
+- Plan persistence data preparation (model_dump for serialization to storage)
+- Decision model creation used by repository patterns
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+# Ensure the local *source* tree is importable even when ASV has an
+# older build of the package installed.
+_SRC = str(Path(__file__).resolve().parents[1] / "src")
+if _SRC not in sys.path:
+    sys.path.insert(0, _SRC)
+
+import cleveragents  # noqa: E402
+
+from cleveragents.domain.models.acms import (  # noqa: E402
+    BackendSet,
+    ScopedBackendView,
+)
+from cleveragents.domain.models.core.change import (  # noqa: E402
+    Change,
+    ChangeSet,
+    Operation,
+    OperationType,
+)
+from cleveragents.domain.models.core.plan import (  # noqa: E402
+    AutomationProfileProvenance,
+    AutomationProfileRef,
+    InvariantSource,
+    NamespacedName,
+    Plan,
+    PlanIdentity,
+    PlanInvariant,
+    PlanPhase,
+    ProcessingState,
+    ProjectLink,
+    can_transition,
+)
+
+# ---------------------------------------------------------------------------
+# Constants extracted from hardcoded values.
+# ---------------------------------------------------------------------------
+
+_ULID_A = "01HGZ6FE0AQDYTR4BXVQZ6EA00"
+_ULID_B = "01HGZ6FE0AQDYTR4BXVQZ6EB00"
+
+
+def _minimal_plan() -> Plan:
+    """Create a minimal Plan suitable for repository-bound operations."""
+    return Plan(
+        identity=PlanIdentity(plan_id=_ULID_A),
+        namespaced_name=NamespacedName(
+            server=None, namespace="local", name="bench-plan"
+        ),
+        description="Benchmark plan",
+        action_name="local/bench-action",
+    )
+
+
+# ---------------------------------------------------------------------------
+# Plan repository benchmarks (focus on persistence-oriented operations)
+# ---------------------------------------------------------------------------
+
+
+class PlanRepositorySuite:
+    """Benchmark Plan model operations relevant to the plan repository."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        """Set up a minimal and rich plan for benchmarking."""
+        self.minimal = _minimal_plan()
+        self.rich = 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
+            ),
+            strategy_actor="local/planner",
+            execution_actor="local/coder",
+            review_actor="local/reviewer",
+        )
+
+    def time_create_plan_from_dict(self) -> None:
+        """Benchmark Plan.model_validate for repository deserialization."""
+        Plan.model_validate(
+            {
+                "identity": {"plan_id": _ULID_A},
+                "namespaced_name": {"namespace": "local", "name": "bench-plan"},
+                "description": "Benchmark plan",
+                "action_name": "local/bench-action",
+            }
+        )
+
+    def time_plan_model_dump(self) -> None:
+        """Benchmark serializing a Plan for repository persistence."""
+        self.minimal.model_dump()
+
+    def time_rich_plan_model_dump(self) -> None:
+        """Benchmark serializing a rich Plan for persistence."""
+        self.rich.model_dump()
+
+    def time_plan_as_cli_dict(self) -> None:
+        """Benchmark as_cli_dict output generation (for CLI / UI consumers)."""
+        self.minimal.as_cli_dict()
+
+
+# ---------------------------------------------------------------------------
+# NamespacedName and identity benchmarks
+# ---------------------------------------------------------------------------
+
+
+class PlanIdentitySuite:
+    """Benchmark plan identity and namespacing operations."""
+
+    timeout = 60
+
+    def time_create_plan_identity(self) -> None:
+        """Benchmark creating a PlanIdentity with ULID."""
+        PlanIdentity(plan_id=_ULID_A)
+
+    def time_create_namespaced_name_simple(self) -> None:
+        """Benchmark parsing a simple namespace/name construct."""
+        NamespacedName.parse("my-workflow")
+
+    def time_create_full_namespace(self) -> None:
+        """Benchmark creating a full server:namespace/name construct."""
+        NamespacedName(server="prod", namespace="cleveragents", name="workflow/merge")
+
+
+# ---------------------------------------------------------------------------
+# Change tracking benchmarks (used by change-tracking repositories)
+# ---------------------------------------------------------------------------
+
+
+class ChangeTrackingSuite:
+    """Benchmark Change operations relevant to repository persistence patterns."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        """Set up a plan_id fixture."""
+        self.plan_id = 42
+
+    def time_create_change_record(self) -> None:
+        """Benchmark creating a single Change for repository insertion."""
+        Change(
+            id=None,
+            plan_id=self.plan_id,
+            file_path="/tmp/bench/file.py",
+            operation=OperationType.MODIFY,
+            new_content="def hello():\n    pass",
+        )
+
+    def time_bulk_create_changes_50(self) -> None:
+        """Benchmark bulk-creating 50 Change records (typical small changeset)."""
+        for i in range(50):
+            _ = Change(
+                id=None,
+                plan_id=self.plan_id,
+                file_path=f"/tmp/bench/file{i}.py",
+                operation=OperationType.MODIFY if i % 2 == 0 else OperationType.CREATE,
+                new_content=f"content {i}",
+            )
+
+    def time_create_changeset_with_summary(self) -> None:
+        """Benchmark creating a ChangeSet and accessing stats."""
+        changes = [
+            Change(
+                id=None,
+                plan_id=self.plan_id,
+                file_path=f"/tmp/bench/file{i}.py",
+                operation=OperationType.CREATE if i % 3 == 0 else OperationType.MODIFY,
+                new_content=f"content {i}",
+            )
+            for i in range(30)
+        ]
+        cs = ChangeSet(plan_id=self.plan_id, changes=changes)
+        _ = cs.stats
+
+
+# ---------------------------------------------------------------------------
+# Phase transition benchmarks (repository state validation)
+# ---------------------------------------------------------------------------
+
+
+class PhaseTransitionSuite:
+    """Benchmark phase transition validation used by repository guards."""
+
+    timeout = 60
+
+    def time_valid_transition(self) -> None:
+        """Time a valid PLAN phase transition check."""
+        can_transition(PlanPhase.STRATEGIZE, PlanPhase.EXECUTE)
+
+    def time_invalid_transition(self) -> None:
+        """Time an invalid phase transition check (fast-fail path)."""
+        can_transition(PlanPhase.APPLY, PlanPhase.ACTION)
+
+
+# ---------------------------------------------------------------------------
+# Throughput benchmarks — bulk repository operations.
+# ---------------------------------------------------------------------------
+
+
+class RepositoryThroughputSuite:
+    """Benchmark bulk plan model construction for repository insertion."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        """Set up plan_id fixture."""
+        self.plan_id = 200
+
+    def time_insert_100_plans(self) -> None:
+        """Benchmark creating and serializing 100 Plan instances."""
+        for i in range(100):
+            plan = _minimal_plan()
+            result = plan.model_dump()
+            result["namespaced_name"]["name"] = f"batch-{i}"
+            _ = Plan.model_validate(result)
+
+    def time_serialize_100_changesets(self) -> None:
+        """Benchmark serializing 100 ChangeSets (simulates batch write)."""
+        for i in range(100):
+            changes = [
+                Change(
+                    id=None,
+                    plan_id=self.plan_id,
+                    file_path=f"/tmp/bench/file{j}.py",
+                    operation=OperationType.MODIFY,
+                    new_content=f"c{j}",
+                )
+                for j in range(3)
+            ]
+            cs = ChangeSet(plan_id=self.plan_id, changes=changes)
+            _ = cs.model_dump()