"""ASV benchmarks for tool lifecycle runtime.""" from __future__ import annotations from typing import Any from cleveragents.tool.context import ( BoundResource, CancellationToken, Change, ChangeOperation, ToolExecutionContext, ) from cleveragents.tool.lifecycle import ( ToolDescriptor, ToolLifecycleCache, ToolResult, ) from cleveragents.tool.schema_validator import validate_tool_input class _DummyInstance: """Minimal mock for benchmarks.""" def discover(self) -> ToolDescriptor: return ToolDescriptor(name="bench/tool", description="bench", source="builtin") def activate(self, ctx: Any) -> None: pass def execute(self, params: Any, ctx: Any) -> ToolResult: return ToolResult() def deactivate(self, ctx: Any) -> None: pass class TimeToolExecutionContext: """Benchmark ToolExecutionContext creation and operations.""" def time_create_context(self) -> None: ToolExecutionContext(plan_id="plan-bench-001") def time_create_context_with_resources(self) -> None: resources = { f"slot_{i}": BoundResource( slot_name=f"slot_{i}", resource_id=f"res-{i:03d}", resource_type="git-checkout", access="read_write", ) for i in range(10) } ToolExecutionContext( plan_id="plan-bench-002", resources=resources, ) def time_record_change(self) -> None: ctx = ToolExecutionContext(plan_id="plan-bench-003") for i in range(100): ctx.record_change( Change( operation=ChangeOperation.MODIFY, resource_id=f"res-{i:03d}", ) ) def time_context_summary(self) -> None: ctx = ToolExecutionContext(plan_id="plan-bench-004") for _ in range(50): ctx.as_summary() class TimeCancellationToken: """Benchmark CancellationToken operations.""" def time_create_token(self) -> None: CancellationToken() def time_check_not_cancelled(self) -> None: token = CancellationToken() for _ in range(1000): _ = token.is_cancelled def time_cancel_and_check(self) -> None: token = CancellationToken() token.cancel() for _ in range(1000): _ = token.is_cancelled class TimeToolLifecycleCache: """Benchmark ToolLifecycleCache operations.""" def time_put_and_get(self) -> None: cache = ToolLifecycleCache() dummy = _DummyInstance() for i in range(100): cache.put(f"plan-{i}", f"tool/t-{i}", dummy) for i in range(100): cache.get(f"plan-{i}", f"tool/t-{i}") def time_remove_plan(self) -> None: cache = ToolLifecycleCache() dummy = _DummyInstance() for i in range(50): cache.put("plan-001", f"tool/t-{i}", dummy) cache.remove_plan("plan-001") def time_get_stats(self) -> None: cache = ToolLifecycleCache() dummy = _DummyInstance() for i in range(20): cache.put("plan-001", f"tool/t-{i}", dummy) for _ in range(50): cache.get_stats("plan-001") class TimeSchemaValidation: """Benchmark JSON Schema validation.""" def setup(self) -> None: self.schema: dict[str, Any] = { "type": "object", "properties": { "path": {"type": "string"}, "content": {"type": "string"}, "mode": {"type": "integer"}, }, "required": ["path"], } def time_validate_simple(self) -> None: validate_tool_input({"path": "test.txt"}, self.schema) def time_validate_full(self) -> None: validate_tool_input( {"path": "test.txt", "content": "hello", "mode": 644}, self.schema, ) class TimeToolDescriptor: """Benchmark ToolDescriptor creation.""" def time_create_descriptor(self) -> None: for i in range(100): ToolDescriptor( name=f"ns/tool-{i}", description=f"Tool {i}", ) def time_create_descriptor_with_schema(self) -> None: schema: dict[str, Any] = { "type": "object", "properties": {"path": {"type": "string"}}, } for i in range(100): ToolDescriptor( name=f"ns/tool-{i}", description=f"Tool {i}", input_schema=schema, output_schema=schema, )