feat(perf): Add performance guardrails registry & decorator, instrument validation aggregator

Adds PerformanceRegistry, perf_guard decorator and overhead comparison utility. Instruments run_structural_validation when perf_registry provided. Includes tests ensuring <8% overhead for moderate workload. Maintains physics invariants (read-only timing).

Author: fer

src/tnfr/performance/guardrails.py

@@ -0,0 +1,136 @@
+"""Performance guardrails instrumentation for TNFR Phase 3.
+
+Lightweight timing utilities ensuring added structural validation / telemetry
+instrumentation remains below configured overhead thresholds.
+
+Design Goals
+------------
+1. Zero external dependencies (stdlib only).
+2. Minimal footprint: single perf_counter measurement plus registry append.
+3. Opt-in: instrumentation only active when explicitly passed a registry.
+4. Composable: decorator or manual timing blocks.
+
+Physics Alignment
+-----------------
+Performance measurement is purely operational and never alters TNFR physics;
+it wraps functions that perform read-only structural computations. The
+guardrails act as a containment layer ensuring added monitoring does not
+fragment coherence through excessive latency.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from time import perf_counter
+from typing import Any, Callable, Dict, List
+
+__all__ = [
+    "PerformanceRegistry",
+    "perf_guard",
+    "compare_overhead",
+]
+
+
+@dataclass(slots=True)
+class PerformanceRecord:
+    label: str
+    elapsed: float
+    meta: Dict[str, Any] | None = None
+
+
+@dataclass(slots=True)
+class PerformanceRegistry:
+    """Collects performance timing records.
+
+    Methods
+    -------
+    record(label, elapsed, meta=None): Add a timing entry.
+    summary(): Aggregate statistics (count, mean, max, min, labels).
+    filter(label): Return list of records matching label.
+    """
+
+    records: List[PerformanceRecord] = field(default_factory=list)
+
+    def record(
+        self, label: str, elapsed: float, meta: Dict[str, Any] | None = None
+    ) -> None:
+        self.records.append(PerformanceRecord(label, float(elapsed), meta))
+
+    def filter(self, label: str) -> List[PerformanceRecord]:
+        return [r for r in self.records if r.label == label]
+
+    def summary(self) -> Dict[str, Any]:
+        if not self.records:
+            return {"count": 0}
+        total = sum(r.elapsed for r in self.records)
+        return {
+            "count": len(self.records),
+            "total": total,
+            "mean": total / len(self.records),
+            "max": max(r.elapsed for r in self.records),
+            "min": min(r.elapsed for r in self.records),
+            "labels": sorted({r.label for r in self.records}),
+        }
+
+
+def perf_guard(label: str, registry: PerformanceRegistry | None) -> Callable:
+    """Decorator adding a single perf_counter measurement if registry provided.
+
+    Parameters
+    ----------
+    label : str
+        Logical name for the operation (e.g. "validation" or "telemetry").
+    registry : PerformanceRegistry | None
+        Active registry; if None instrumentation is skipped.
+    """
+
+    def decorator(fn: Callable) -> Callable:
+        def wrapped(*args, **kwargs):  # type: ignore[override]
+            if registry is None:
+                return fn(*args, **kwargs)
+            start = perf_counter()
+            result = fn(*args, **kwargs)
+            registry.record(label, perf_counter() - start, meta={
+                "fn": fn.__name__,
+                "arg_count": len(args),
+                "kw_count": len(kwargs),
+            })
+            return result
+
+        wrapped.__name__ = fn.__name__  # preserve for introspection
+        wrapped.__doc__ = fn.__doc__
+        return wrapped
+
+    return decorator
+
+
+def compare_overhead(
+    baseline_fn: Callable[[], Any],
+    instrumented_fn: Callable[[], Any],
+    *,
+    runs: int = 5000,
+) -> Dict[str, float]:
+    """Compare overhead ratio between baseline and instrumented call sets.
+
+    Returns timing dict with baseline, instrumented and ratio
+    (instrumented - baseline) / baseline.
+    """
+    # Warmup
+    for _ in range(10):
+        baseline_fn()
+        instrumented_fn()
+    b_start = perf_counter()
+    for _ in range(runs):
+        baseline_fn()
+    b_elapsed = perf_counter() - b_start
+    i_start = perf_counter()
+    for _ in range(runs):
+        instrumented_fn()
+    i_elapsed = perf_counter() - i_start
+    ratio = (i_elapsed - b_elapsed) / b_elapsed if b_elapsed > 0 else 0.0
+    return {
+        "baseline": b_elapsed,
+        "instrumented": i_elapsed,
+        "ratio": ratio,
+        "runs": float(runs),
+    }

src/tnfr/validation/aggregator.py

@@ -66,6 +66,7 @@
     compute_phase_curvature,
     estimate_coherence_length,
 )
+from ..performance.guardrails import PerformanceRegistry
 
 __all__ = [
     "ValidationReport",
@@ -132,6 +133,8 @@ def run_structural_validation(
     xi_c_watch_multiplier: float = 3.0,
     # Optional baselines for drift calculations
     baseline_structural_potential: Dict[Any, float] | None = None,
+    # Performance instrumentation (opt-in)
+    perf_registry: PerformanceRegistry | None = None,
 ) -> ValidationReport:
     """Run enhanced structural validation aggregating grammar + field safety.
 
@@ -169,6 +172,15 @@ def run_structural_validation(
 
     notes: List[str] = []
 
+    # Performance start (if instrumentation active)
+    start_time = None
+    if perf_registry is not None:
+        try:
+            import time as _t
+            start_time = _t.perf_counter()
+        except Exception:  # pragma: no cover
+            start_time = None
+
     # Grammar errors (read-only enrichment)
     grammar_errors: List[ExtendedGrammarError] = []
     if sequence is not None:
@@ -314,7 +326,7 @@ def run_structural_validation(
         "mean_node_distance": mean_node_distance,
     }
 
-    return ValidationReport(
+    report = ValidationReport(
         status=status,
         risk_level=risk_level,
         grammar_errors=grammar_errors,
@@ -323,3 +335,31 @@ def run_structural_validation(
         sequence=tuple(sequence or []),
         notes=notes,
     )
+
+    if perf_registry is not None and start_time is not None:
+        try:
+            import time as _t
+            perf_registry.record(
+                "validation",
+                _t.perf_counter() - start_time,
+                meta={
+                    "nodes": (
+                        G.number_of_nodes()
+                        if hasattr(G, "number_of_nodes")
+                        else None
+                    ),
+                    "edges": (
+                        G.number_of_edges()
+                        if hasattr(G, "number_of_edges")
+                        else None
+                    ),
+                    "sequence_len": (
+                        len(sequence) if sequence is not None else 0
+                    ),
+                    "status": status,
+                },
+            )
+        except Exception:  # pragma: no cover
+            pass
+
+    return report

tests/unit/performance/test_guardrails.py

@@ -0,0 +1,52 @@
+"""Tests for performance guardrails instrumentation.
+
+Ensures perf_guard adds minimal overhead (<10% ratio baseline) for a trivial
+function. Threshold is conservative to reduce flakiness across CI schedulers.
+"""
+from __future__ import annotations
+
+from time import perf_counter
+
+from tnfr.performance.guardrails import (
+    PerformanceRegistry,
+    perf_guard,
+    compare_overhead,
+)
+
+
+def _baseline_op() -> int:
+    # Moderate workload to reduce relative overhead impact of instrumentation
+    x = 0
+    for _ in range(2000):
+        x += 1
+    return x
+
+
+def test_perf_guard_overhead_ratio():
+    registry = PerformanceRegistry()
+
+    @perf_guard("test", registry)
+    def _instrumented() -> int:
+        return _baseline_op()
+
+    stats = compare_overhead(_baseline_op, _instrumented, runs=500)
+    # Overhead ratio should remain below 8% for moderate workload
+    assert stats["ratio"] < 0.08, stats
+    # Registry should have at least one record (warmup + runs)
+    assert registry.summary()["count"] >= 1
+
+
+def test_perf_registry_summary_fields():
+    registry = PerformanceRegistry()
+    # Manually record
+    start = perf_counter()
+    _baseline_op()
+    registry.record(
+        "manual",
+        perf_counter() - start,
+        meta={"kind": "baseline"},
+    )
+    summary = registry.summary()
+    assert summary["count"] == 1
+    assert "mean" in summary and summary["mean"] > 0
+    assert "labels" in summary and summary["labels"] == ["manual"]