feat(effects): add performance monitoring to effect pipeline

- Add PerformanceMonitor to collect per-effect timings
- Track effect duration (ms), buffer chars in/out per frame
- Store last 60 frames in ring buffer
- Add /effects stats NTFY command to view performance data
- Add tests for performance monitoring system

engine/effects/__init__.py

@@ -8,6 +8,7 @@ from engine.effects.legacy import (
     noise,
     vis_trunc,
 )
+from engine.effects.performance import PerformanceMonitor, get_monitor, set_monitor
 from engine.effects.registry import EffectRegistry, get_registry, set_registry
 from engine.effects.types import EffectConfig, EffectContext, PipelineConfig
 
@@ -27,6 +28,9 @@ __all__ = [
     "get_registry",
     "set_registry",
     "get_effect_chain",
+    "get_monitor",
+    "set_monitor",
+    "PerformanceMonitor",
     "handle_effects_command",
     "show_effects_menu",
     "fade_line",

engine/effects/chain.py

@@ -1,11 +1,22 @@
+import time
+
+from engine.effects.performance import PerformanceMonitor, get_monitor
 from engine.effects.registry import EffectRegistry
 from engine.effects.types import EffectContext
 
 
 class EffectChain:
-    def __init__(self, registry: EffectRegistry):
+    def __init__(
+        self, registry: EffectRegistry, monitor: PerformanceMonitor | None = None
+    ):
         self._registry = registry
         self._order: list[str] = []
+        self._monitor = monitor
+
+    def _get_monitor(self) -> PerformanceMonitor:
+        if self._monitor is not None:
+            return self._monitor
+        return get_monitor()
 
     def set_order(self, names: list[str]) -> None:
         self._order = list(names)
@@ -36,12 +47,25 @@ class EffectChain:
         return True
 
     def process(self, buf: list[str], ctx: EffectContext) -> list[str]:
+        monitor = self._get_monitor()
+        frame_number = ctx.frame_number
+        monitor.start_frame(frame_number)
+
+        frame_start = time.perf_counter()
         result = list(buf)
         for name in self._order:
             plugin = self._registry.get(name)
             if plugin and plugin.config.enabled:
+                chars_in = sum(len(line) for line in result)
+                effect_start = time.perf_counter()
                 try:
                     result = plugin.process(result, ctx)
                 except Exception:
                     plugin.config.enabled = False
+                elapsed = time.perf_counter() - effect_start
+                chars_out = sum(len(line) for line in result)
+                monitor.record_effect(name, elapsed * 1000, chars_in, chars_out)
+
+        total_elapsed = time.perf_counter() - frame_start
+        monitor.end_frame(frame_number, total_elapsed * 1000)
         return result

engine/effects/controller.py

@@ -1,3 +1,4 @@
+from engine.effects.performance import get_monitor
 from engine.effects.registry import get_registry
 
 
@@ -16,6 +17,7 @@ def handle_effects_command(cmd: str) -> str:
         /effects <name> off - disable an effect
         /effects <name> intensity <0.0-1.0> - set intensity
         /effects reorder <name1>,<name2>,... - reorder pipeline
+        /effects stats - show performance statistics
     """
     parts = cmd.strip().split()
     if not parts or parts[0] != "/effects":
@@ -34,6 +36,9 @@ def handle_effects_command(cmd: str) -> str:
             result.append(f"Order: {chain.get_order()}")
         return "\n".join(result)
 
+    if parts[1] == "stats":
+        return _format_stats()
+
     if len(parts) < 3:
         return "Usage: /effects <name> on|off|intensity <value>"
 
@@ -72,6 +77,30 @@ def handle_effects_command(cmd: str) -> str:
     return f"Unknown action: {action}"
 
 
+def _format_stats() -> str:
+    monitor = get_monitor()
+    stats = monitor.get_stats()
+
+    if "error" in stats:
+        return stats["error"]
+
+    lines = ["Performance Stats:"]
+
+    pipeline = stats["pipeline"]
+    lines.append(
+        f"  Pipeline: avg={pipeline['avg_ms']:.2f}ms min={pipeline['min_ms']:.2f}ms max={pipeline['max_ms']:.2f}ms (over {stats['frame_count']} frames)"
+    )
+
+    if stats["effects"]:
+        lines.append("  Per-effect (avg ms):")
+        for name, effect_stats in stats["effects"].items():
+            lines.append(
+                f"    {name}: avg={effect_stats['avg_ms']:.2f}ms min={effect_stats['min_ms']:.2f}ms max={effect_stats['max_ms']:.2f}ms"
+            )
+
+    return "\n".join(lines)
+
+
 def show_effects_menu() -> str:
     """Generate effects menu text for display."""
     registry = get_registry()

engine/effects/performance.py

@@ -0,0 +1,103 @@
+from collections import deque
+from dataclasses import dataclass
+
+
+@dataclass
+class EffectTiming:
+    name: str
+    duration_ms: float
+    buffer_chars_in: int
+    buffer_chars_out: int
+
+
+@dataclass
+class FrameTiming:
+    frame_number: int
+    total_ms: float
+    effects: list[EffectTiming]
+
+
+class PerformanceMonitor:
+    """Collects and stores performance metrics for effect pipeline."""
+
+    def __init__(self, max_frames: int = 60):
+        self._max_frames = max_frames
+        self._frames: deque[FrameTiming] = deque(maxlen=max_frames)
+        self._current_frame: list[EffectTiming] = []
+
+    def start_frame(self, frame_number: int) -> None:
+        self._current_frame = []
+
+    def record_effect(
+        self, name: str, duration_ms: float, chars_in: int, chars_out: int
+    ) -> None:
+        self._current_frame.append(
+            EffectTiming(
+                name=name,
+                duration_ms=duration_ms,
+                buffer_chars_in=chars_in,
+                buffer_chars_out=chars_out,
+            )
+        )
+
+    def end_frame(self, frame_number: int, total_ms: float) -> None:
+        self._frames.append(
+            FrameTiming(
+                frame_number=frame_number,
+                total_ms=total_ms,
+                effects=self._current_frame,
+            )
+        )
+
+    def get_stats(self) -> dict:
+        if not self._frames:
+            return {"error": "No timing data available"}
+
+        total_times = [f.total_ms for f in self._frames]
+        avg_total = sum(total_times) / len(total_times)
+        min_total = min(total_times)
+        max_total = max(total_times)
+
+        effect_stats: dict[str, dict] = {}
+        for frame in self._frames:
+            for effect in frame.effects:
+                if effect.name not in effect_stats:
+                    effect_stats[effect.name] = {"times": [], "total_chars": 0}
+                effect_stats[effect.name]["times"].append(effect.duration_ms)
+                effect_stats[effect.name]["total_chars"] += effect.buffer_chars_out
+
+        for name, stats in effect_stats.items():
+            times = stats["times"]
+            stats["avg_ms"] = sum(times) / len(times)
+            stats["min_ms"] = min(times)
+            stats["max_ms"] = max(times)
+            del stats["times"]
+
+        return {
+            "frame_count": len(self._frames),
+            "pipeline": {
+                "avg_ms": avg_total,
+                "min_ms": min_total,
+                "max_ms": max_total,
+            },
+            "effects": effect_stats,
+        }
+
+    def reset(self) -> None:
+        self._frames.clear()
+        self._current_frame = []
+
+
+_monitor: PerformanceMonitor | None = None
+
+
+def get_monitor() -> PerformanceMonitor:
+    global _monitor
+    if _monitor is None:
+        _monitor = PerformanceMonitor()
+    return _monitor
+
+
+def set_monitor(monitor: PerformanceMonitor) -> None:
+    global _monitor
+    _monitor = monitor