"""ASV benchmarks for decision persistence (serialization round-trips). Measures performance of: - model_dump / model_validate round-trips - JSON (model_dump_json / model_validate) round-trips - ContextSnapshot persistence overhead - Decision tree serialization and reconstruction """ from __future__ import annotations import json import sys from pathlib import Path from typing import Any try: from cleveragents.domain.models.core.decision import ( ArtifactRef, ContextSnapshot, Decision, DecisionType, ResourceRef, ) except ModuleNotFoundError: sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) from cleveragents.domain.models.core.decision import ( ArtifactRef, ContextSnapshot, Decision, DecisionType, ResourceRef, ) from ulid import ULID # --------------------------------------------------------------------------- # Module-level fixtures # --------------------------------------------------------------------------- _PLAN_ID = str(ULID()) _PARENT_ID = str(ULID()) _RESOURCE_ID = str(ULID()) _ALL_DECISION_TYPES: list[DecisionType] = list(DecisionType) def _make_decision(**overrides: Any) -> Decision: """Build a Decision with sensible defaults, merged with *overrides*.""" defaults: dict[str, Any] = { "plan_id": _PLAN_ID, "sequence_number": 0, "decision_type": DecisionType.PROMPT_DEFINITION, "question": "What approach should we take?", "chosen_option": "Build a REST API", } dt = overrides.get("decision_type", defaults["decision_type"]) if dt != DecisionType.PROMPT_DEFINITION and "parent_decision_id" not in overrides: defaults["parent_decision_id"] = _PARENT_ID defaults.update(overrides) return Decision(**defaults) # --------------------------------------------------------------------------- # Suite 1: model_dump / model_validate round-trips # --------------------------------------------------------------------------- class DecisionDumpRoundTripSuite: """Benchmark model_dump -> model_validate round-trips.""" def setup(self) -> None: self.root = _make_decision() self.child = _make_decision( parent_decision_id=_PARENT_ID, decision_type=DecisionType.STRATEGY_CHOICE, alternatives_considered=["Flask", "Django", "Starlette"], confidence_score=0.88, rationale="Async support and auto docs", ) self.full = _make_decision( parent_decision_id=_PARENT_ID, decision_type=DecisionType.IMPLEMENTATION_CHOICE, confidence_score=0.92, context_snapshot=ContextSnapshot( hot_context_hash="sha256:bench", hot_context_ref="store://bench", relevant_resources=[ ResourceRef(resource_id=_RESOURCE_ID, path="src/main.py"), ResourceRef(resource_id=str(ULID()), path="src/utils.py"), ], actor_state_ref="checkpoint://bench", ), artifacts_produced=[ ArtifactRef(artifact_path="src/api.py", artifact_type="file"), ArtifactRef(artifact_path="patches/fix.patch", artifact_type="patch"), ], ) self.root_dump = self.root.model_dump() self.child_dump = self.child.model_dump() self.full_dump = self.full.model_dump() def time_root_dump(self) -> None: """Benchmark root decision model_dump.""" self.root.model_dump() def time_root_validate(self) -> None: """Benchmark root decision model_validate.""" Decision.model_validate(self.root_dump) def time_root_roundtrip(self) -> None: """Benchmark root decision full round-trip.""" Decision.model_validate(self.root.model_dump()) def time_child_roundtrip(self) -> None: """Benchmark child decision full round-trip.""" Decision.model_validate(self.child.model_dump()) def time_full_roundtrip(self) -> None: """Benchmark fully-populated decision round-trip.""" Decision.model_validate(self.full.model_dump()) # --------------------------------------------------------------------------- # Suite 2: JSON round-trips # --------------------------------------------------------------------------- class DecisionJsonRoundTripSuite: """Benchmark JSON serialization round-trips.""" def setup(self) -> None: self.root = _make_decision() self.full = _make_decision( parent_decision_id=_PARENT_ID, decision_type=DecisionType.TOOL_INVOCATION, confidence_score=0.75, context_snapshot=ContextSnapshot( hot_context_hash="sha256:json_bench", relevant_resources=[ ResourceRef(resource_id=str(ULID()), path=f"src/f{i}.py") for i in range(5) ], ), artifacts_produced=[ ArtifactRef(artifact_path=f"out/file_{i}.py") for i in range(5) ], ) self.root_json = self.root.model_dump_json() self.full_json = self.full.model_dump_json() def time_root_dump_json(self) -> None: """Benchmark root decision model_dump_json.""" self.root.model_dump_json() def time_root_json_roundtrip(self) -> None: """Benchmark root decision JSON round-trip.""" Decision.model_validate(json.loads(self.root.model_dump_json())) def time_full_json_roundtrip(self) -> None: """Benchmark fully-populated decision JSON round-trip.""" Decision.model_validate(json.loads(self.full.model_dump_json())) def time_root_json_validate(self) -> None: """Benchmark root decision model_validate from JSON.""" Decision.model_validate(json.loads(self.root_json)) def time_full_json_validate(self) -> None: """Benchmark full decision model_validate from JSON.""" Decision.model_validate(json.loads(self.full_json)) # --------------------------------------------------------------------------- # Suite 3: ContextSnapshot persistence # --------------------------------------------------------------------------- class DecisionSnapshotPersistenceSuite: """Benchmark ContextSnapshot serialization overhead.""" def setup(self) -> None: self.empty_snap = ContextSnapshot() self.small_snap = ContextSnapshot( hot_context_hash="sha256:small", hot_context_ref="store://small", relevant_resources=[ ResourceRef(resource_id=str(ULID()), path="src/a.py"), ], actor_state_ref="checkpoint://small", ) self.large_snap = ContextSnapshot( hot_context_hash="sha256:large", hot_context_ref="store://large", relevant_resources=[ ResourceRef(resource_id=str(ULID()), path=f"src/f{i}.py") for i in range(50) ], actor_state_ref="checkpoint://large", ) def time_empty_snapshot_roundtrip(self) -> None: """Benchmark empty snapshot round-trip.""" data = self.empty_snap.model_dump() ContextSnapshot.model_validate(data) def time_small_snapshot_roundtrip(self) -> None: """Benchmark small snapshot (1 resource) round-trip.""" data = self.small_snap.model_dump() ContextSnapshot.model_validate(data) def time_large_snapshot_roundtrip(self) -> None: """Benchmark large snapshot (50 resources) round-trip.""" data = self.large_snap.model_dump() ContextSnapshot.model_validate(data) # --------------------------------------------------------------------------- # Suite 4: Decision tree serialization # --------------------------------------------------------------------------- class DecisionTreeSerializationSuite: """Benchmark serializing and deserializing decision trees.""" def setup(self) -> None: self.small_tree = self._build_tree(depth=2) # 3 nodes self.medium_tree = self._build_tree(depth=3) # 7 nodes self.large_tree = self._build_tree(depth=4) # 15 nodes @staticmethod def _build_tree(depth: int) -> list[Decision]: """Build a binary decision tree of the given depth.""" plan_id = str(ULID()) nodes: list[Decision] = [] seq = 0 root = _make_decision( plan_id=plan_id, sequence_number=seq, question="Root?", chosen_option="Start", ) nodes.append(root) seq += 1 level = [root] for d in range(1, depth): next_level: list[Decision] = [] dt = ( DecisionType.STRATEGY_CHOICE if d == 1 else DecisionType.IMPLEMENTATION_CHOICE ) for parent in level: for ci in range(2): child = _make_decision( plan_id=plan_id, parent_decision_id=parent.decision_id, sequence_number=seq, decision_type=dt, question=f"Node d{d} c{ci}?", chosen_option=f"Opt d{d} c{ci}", ) nodes.append(child) next_level.append(child) seq += 1 level = next_level return nodes def time_small_tree_roundtrip(self) -> None: """Benchmark 3-node tree serialization round-trip.""" for n in self.small_tree: Decision.model_validate(n.model_dump()) def time_medium_tree_roundtrip(self) -> None: """Benchmark 7-node tree serialization round-trip.""" for n in self.medium_tree: Decision.model_validate(n.model_dump()) def time_large_tree_roundtrip(self) -> None: """Benchmark 15-node tree serialization round-trip.""" for n in self.large_tree: Decision.model_validate(n.model_dump()) def time_large_tree_json_roundtrip(self) -> None: """Benchmark 15-node tree JSON round-trip.""" for n in self.large_tree: Decision.model_validate(json.loads(n.model_dump_json()))