src/core/perf.py

"""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()