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