Files
cleveragents-core/benchmarks/agent_skills_registry_bench.py
freemo 7a298ede6e
CI / benchmark-publish (pull_request) Has been skipped
CI / lint (pull_request) Successful in 16s
CI / quality (pull_request) Successful in 30s
CI / build (pull_request) Successful in 24s
CI / security (pull_request) Successful in 40s
CI / typecheck (pull_request) Successful in 1m0s
CI / integration_tests (pull_request) Successful in 4m14s
CI / unit_tests (pull_request) Successful in 16m25s
CI / docker (pull_request) Successful in 55s
CI / benchmark-regression (pull_request) Successful in 22m55s
CI / coverage (pull_request) Successful in 38m4s
CI / lint (push) Successful in 13s
CI / build (push) Successful in 15s
CI / quality (push) Successful in 28s
CI / typecheck (push) Successful in 31s
CI / benchmark-regression (push) Has been skipped
CI / security (push) Successful in 31s
CI / integration_tests (push) Successful in 3m19s
CI / benchmark-publish (push) Successful in 14m23s
CI / unit_tests (push) Successful in 14m44s
CI / docker (push) Successful in 38s
CI / coverage (push) Successful in 32m22s
feat(concurrency): add plan and project locks
Implemented plan-level and project-level locking with configurable timeouts.
Added locks table via Alembic migration storing owner_id, resource_type,
resource_id, acquired_at, and expires_at. Locks enforced in
PlanLifecycleService transitions. Support for re-entrant acquisition,
lock renewal, graceful shutdown release, and startup cleanup of expired
locks. Added diagnostics check for stale lock reporting.

ISSUES CLOSED: #327
2026-02-25 10:48:05 -05:00

114 lines
3.3 KiB
Python

"""ASV benchmarks for Agent Skills discovery scan overhead.
Measures the performance of:
- ``parse_agent_skills_paths`` path parsing
- ``scan_directory`` filesystem scanning
- ``build_tool_spec`` ToolSpec construction
- ``register_discovered_skills`` registration in ToolRegistry
"""
from __future__ import annotations
import shutil
import sys
import tempfile
from pathlib import Path
from typing import ClassVar
try:
from cleveragents.skills.discovery import (
DiscoveredAgentSkill,
build_tool_spec,
parse_agent_skills_paths,
register_discovered_skills,
scan_directory,
)
from cleveragents.tool.registry import ToolRegistry
except ModuleNotFoundError:
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from cleveragents.skills.discovery import (
DiscoveredAgentSkill,
build_tool_spec,
parse_agent_skills_paths,
register_discovered_skills,
scan_directory,
)
from cleveragents.tool.registry import ToolRegistry
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _create_skill_md(folder: Path, name: str) -> None:
"""Create a SKILL.md with minimal front-matter."""
folder.mkdir(parents=True, exist_ok=True)
content = (
f"---\nname: {name}\ndescription: Benchmark skill {name}\n---\n\n# {name}\n"
)
(folder / "SKILL.md").write_text(content, encoding="utf-8")
def _make_discovered(name: str) -> DiscoveredAgentSkill:
return DiscoveredAgentSkill(
name=name,
description=f"Benchmark skill {name}",
path=f"/tmp/bench/{name}",
)
# ---------------------------------------------------------------------------
# Benchmark suite
# ---------------------------------------------------------------------------
class ParsePathsSuite:
"""Benchmark path string parsing."""
def setup(self) -> None:
self.paths_str = ",".join(f"/opt/skills/dir-{i}" for i in range(20))
def time_parse_20_paths(self) -> None:
parse_agent_skills_paths(self.paths_str)
class ScanDirectorySuite:
"""Benchmark filesystem scanning."""
def setup(self) -> None:
self.tmpdir = Path(tempfile.mkdtemp())
for i in range(50):
_create_skill_md(self.tmpdir / f"skill-{i}", f"skill-{i}")
def teardown(self) -> None:
shutil.rmtree(str(self.tmpdir), ignore_errors=True)
def time_scan_50_skills(self) -> None:
scan_directory(self.tmpdir)
class BuildToolSpecSuite:
"""Benchmark ToolSpec construction from discovered skills."""
def setup(self) -> None:
self.skills = [_make_discovered(f"tool-{i}") for i in range(100)]
def time_build_100_toolspecs(self) -> None:
for skill in self.skills:
build_tool_spec(skill)
class RegisterSuite:
"""Benchmark tool registration with conflict handling."""
params: ClassVar[list[int]] = [10, 50, 100]
param_names: ClassVar[list[str]] = ["count"]
def setup(self, count: int) -> None:
self.skills = [_make_discovered(f"tool-{i}") for i in range(count)]
def time_register(self, count: int) -> None:
registry = ToolRegistry()
register_discovered_skills(self.skills, registry, on_conflict="skip")