metrics/timings.py

"""
Execution timing collector — tracks end-to-end job durations.
"""

from __future__ import annotations

import threading
from collections import defaultdict
from typing import Dict, List


class TimingCollector:
    """Tracks durations grouped by operation label."""

    MAX_SAMPLES = 500

    def __init__(self) -> None:
        self._lock = threading.RLock()
        self._samples: Dict[str, List[float]] = defaultdict(list)

    def record(self, label: str, duration_ms: float) -> None:
        with self._lock:
            self._samples[label].append(duration_ms)
            if len(self._samples[label]) > self.MAX_SAMPLES:
                self._samples[label] = self._samples[label][-self.MAX_SAMPLES:]

    def snapshot(self) -> dict:
        with self._lock:
            out = {}
            for label, samples in self._samples.items():
                if not samples:
                    continue
                s = sorted(samples)
                out[label] = {
                    "count": len(samples),
                    "avg_ms": round(sum(samples) / len(samples), 3),
                    "p50_ms": round(s[len(s) // 2], 3),
                    "p95_ms": round(s[int(len(s) * 0.95)], 3),
                    "min_ms": round(s[0], 3),
                    "max_ms": round(s[-1], 3),
                }
            return out

    def reset(self) -> None:
        with self._lock:
            self._samples.clear()