File size: 1,512 Bytes
3f2f9aa c3b49d6 3f2f9aa c3b49d6 3f2f9aa | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 | """Lightweight per-step performance instrumentation.
Cheap timing hooks so the latency win from the persistent HTTP MCP server
(no per-tool-call subprocess spawn) is measurable rather than anecdotal.
Two things are timed:
- agent generate vs execute phases (src/agent.py)
- each MCP tool call (src/managers/tools/mcp_manager.py)
Each event is printed with a ``[perf]`` prefix (visible in HF Space logs) and
appended to a bounded module-level ring buffer so a test or a trace dump can
read the timings back without scraping stdout.
"""
from __future__ import annotations
import time
from collections import deque
from contextlib import contextmanager
# Bounded so a long-running container can't grow this unboundedly.
_EVENTS: deque[dict] = deque(maxlen=2000)
def record(kind: str, name: str, seconds: float) -> None:
"""Record a single timed event and echo it to stdout."""
seconds = round(seconds, 3)
_EVENTS.append({"kind": kind, "name": name, "seconds": seconds, "t": time.time()})
print(f"[perf] {kind} {name}: {seconds}s", flush=True)
@contextmanager
def timed(kind: str, name: str):
"""Context manager that records monotonic wall time of the block."""
t0 = time.monotonic()
try:
yield
finally:
record(kind, name, time.monotonic() - t0)
def get_events() -> list:
"""Return a snapshot of recorded events (oldest first)."""
return list(_EVENTS)
def reset() -> None:
"""Clear recorded events (used by tests)."""
_EVENTS.clear()
|