"""ASV benchmarks for ChangeSet models and invocation tracking. Measures the performance of: - ChangeEntry creation and validation - SpecChangeSet sorted_entries and grouped_by_resource - InMemoryInvocationTracker operations - Path normalization """ from __future__ import annotations import sys from datetime import UTC, datetime from pathlib import Path try: from cleveragents.domain.models.core.change import ( ChangeEntry, ChangeOperation, InMemoryInvocationTracker, SpecChangeSet, ToolInvocation, normalize_change_path, ) except ModuleNotFoundError: sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) from cleveragents.domain.models.core.change import ( ChangeEntry, ChangeOperation, InMemoryInvocationTracker, SpecChangeSet, ToolInvocation, normalize_change_path, ) def _make_entries(n: int) -> list[ChangeEntry]: """Create *n* ChangeEntry objects with varied resources.""" entries: list[ChangeEntry] = [] for i in range(n): minutes, seconds = divmod(i, 60) entries.append( ChangeEntry( plan_id="plan-bench", resource_id=f"res-{i % 5}", tool_name="builtin/file-write", operation=ChangeOperation.CREATE, path=f"src/file_{i}.py", after_hash=f"hash-{i}", timestamp=datetime(2026, 1, 1, 0, minutes, seconds, tzinfo=UTC), ) ) return entries class ChangeEntrySuite: """Benchmark ChangeEntry creation and validation.""" def time_create_entry(self) -> None: ChangeEntry( plan_id="plan-1", resource_id="res-1", tool_name="builtin/file-write", operation=ChangeOperation.CREATE, path="src/new.py", after_hash="abc123", ) def time_create_entry_with_integrity_check(self) -> None: e = ChangeEntry( plan_id="plan-1", resource_id="res-1", tool_name="builtin/file-edit", operation=ChangeOperation.MODIFY, path="src/edit.py", before_hash="before", after_hash="after", ) _ = e.has_integrity_hashes class SpecChangeSetSuite: """Benchmark SpecChangeSet operations.""" def setup(self) -> None: self.entries = _make_entries(100) self.changeset = SpecChangeSet(plan_id="plan-bench", entries=self.entries) def time_sorted_entries(self) -> None: self.changeset.sorted_entries() def time_grouped_by_resource(self) -> None: self.changeset.grouped_by_resource() def time_summary(self) -> None: self.changeset.summary() def time_add_change(self) -> None: cs = SpecChangeSet(plan_id="plan-bench") for entry in self.entries[:10]: cs.add_change(entry) class InvocationTrackerSuite: """Benchmark InMemoryInvocationTracker.""" def setup(self) -> None: self.tracker = InMemoryInvocationTracker() for i in range(50): self.tracker.track( ToolInvocation( plan_id=f"plan-{i % 3}", tool_name=f"tool-{i}", skill_name=f"skill-{i % 5}", sequence_number=i, change_ids=[f"chg-{i}"], ) ) def time_get_invocations(self) -> None: self.tracker.get_invocations("plan-0") def time_get_invocations_for_skill(self) -> None: self.tracker.get_invocations_for_skill("plan-0", "skill-0") def time_get_changes(self) -> None: self.tracker.get_changes("plan-0") class PathNormalizationSuite: """Benchmark path normalization.""" def time_normalize_absolute(self) -> None: normalize_change_path("/home/user/project/src/main.py", "/home/user/project") def time_normalize_relative(self) -> None: normalize_change_path("src/main.py", "")