Fix Phantom audit findings (2026-07-05)

scripts/bench_full_diagnostic.py

@@ -0,0 +1,123 @@
+"""Timing harness for full_diagnostic on a 60 s stereo file (synthetic audio).
+
+Generates an in-memory 60 s stereo fixture (no real audio committed), then
+reports two numbers:
+
+- COLD: best-of-3 wall-clock time for full_diagnostic with the shared
+  ``analysis_cache`` cleared before each run (and before warmup), so the timed
+  calls measure the actual analysis cost — apples-to-apples with the Task 6
+  baseline (2084.6 ms). This is the gate metric.
+- WARM: one re-run on the same bytes WITHOUT clearing the cache, measuring the
+  cache-hit path (a subsequent same-content tool call).
+
+Used as before/after evidence for the true-peak-once optimization (P-02), the
+content-hash memoization (P-01 follow-up), and re-used by later performance
+tasks on this branch.
+
+Run: uv run python scripts/bench_full_diagnostic.py
+"""
+
+import os
+import time
+import tempfile
+
+import numpy as np
+import soundfile as sf
+
+from phantom._cache import analysis_cache
+from phantom.audio import load_audio
+from phantom.server import full_diagnostic
+from phantom.spectral import _octave_band_energies, analyze_spectrum
+
+
+def _time_cold(p: str) -> float:
+    """Time one full_diagnostic with a freshly cleared cache (cold path)."""
+    analysis_cache.clear()
+    s = time.perf_counter()
+    full_diagnostic(p)
+    return time.perf_counter() - s
+
+
+def _time_warm(p: str) -> float:
+    """Time one full_diagnostic WITHOUT clearing (cache-hit path)."""
+    s = time.perf_counter()
+    full_diagnostic(p)
+    return time.perf_counter() - s
+
+
+def _timed(fn) -> float:
+    """Return the wall-clock seconds to run ``fn`` once."""
+    s = time.perf_counter()
+    fn()
+    return time.perf_counter() - s
+
+
+def _probe_spectral(p: str) -> None:
+    """P-05 gate probe: time analyze_spectrum vs its octave-band pass alone.
+
+    ``analyze_spectrum`` makes two framed FFT passes over the same mono signal:
+    the main 2048/1024 spectral pass and the octave-band 4096/2048 pass (now in
+    ``_octave_band_energies``, shared with masking). The audit (P-05) proposed
+    fusing these into a single pass, but only if the band pass is a material
+    share of spectral time. This probe measures that share so the gate can be
+    decided from data.
+
+    Loads the fixture once, then times both on the same in-memory mono signal
+    (best of 3 each, after a warmup) so import/JIT costs don't skew the split.
+    """
+    audio = load_audio(p)
+    mono = audio.mono
+    sr = audio.sample_rate
+
+    # Warmup both paths so Essentia/JIT costs don't land in the timed runs.
+    analyze_spectrum(audio)
+    _octave_band_energies(mono, sr)
+
+    spectral_s = min(_timed(lambda: analyze_spectrum(audio)) for _ in range(3))
+    band_s = min(_timed(lambda: _octave_band_energies(mono, sr)) for _ in range(3))
+
+    share = (band_s / spectral_s * 100.0) if spectral_s > 0 else 0.0
+    print(f"P-05 probe -- analyze_spectrum 60s: {spectral_s * 1000:.1f} ms (best of 3)")
+    print(f"P-05 probe -- octave-band pass only: {band_s * 1000:.1f} ms (best of 3)")
+    print(f"P-05 probe -- band pass share of spectral: {share:.1f}%")
+
+
+def main() -> None:
+    sr = 44100
+    n = sr * 60
+    t = np.arange(n) / sr
+    rng = np.random.default_rng(0)
+    left = (0.4 * np.sin(2 * np.pi * 220 * t) + 0.1 * rng.standard_normal(n)).astype(
+        "float32"
+    )
+    right = (0.4 * np.sin(2 * np.pi * 223 * t) + 0.1 * rng.standard_normal(n)).astype(
+        "float32"
+    )
+    with tempfile.TemporaryDirectory() as d:
+        os.environ["PHANTOM_OUTPUT_DIR"] = d
+        p = os.path.join(d, "bench.wav")
+        sf.write(p, np.column_stack([left, right]), sr)
+
+        # Warmup on the cold path (cache cleared) so JIT/import costs don't skew.
+        _time_cold(p)
+
+        # COLD: cache cleared before each run — measures true analysis cost.
+        cold = min(_time_cold(p) for _ in range(3))
+
+        # WARM: prime the cache once, then time a same-bytes re-run (cache hits).
+        full_diagnostic(p)
+        warm = min(_time_warm(p) for _ in range(3))
+
+        print(
+            f"full_diagnostic 60s stereo COLD (cache cleared): {cold * 1000:.1f} ms (best of 3)"
+        )
+        print(
+            f"full_diagnostic 60s stereo WARM (cache-hit path): {warm * 1000:.1f} ms (best of 3)"
+        )
+
+        # P-05 gate probe: split analyze_spectrum time into its two FFT passes.
+        _probe_spectral(p)
+
+
+if __name__ == "__main__":
+    main()

src/phantom/_cache.py

@@ -14,7 +14,7 @@
 import logging
 import threading
 from collections import OrderedDict
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Callable
 
 if TYPE_CHECKING:
     from phantom.audio import AudioData
@@ -45,6 +45,31 @@ def __init__(self, max_entries: int = 8) -> None:
     # Internal helpers
     # ------------------------------------------------------------------
 
+    def _content_digest(self, audio: AudioData) -> "hashlib._Hash":
+        """Return a SHA-256 object pre-fed with the per-instance content bytes.
+
+        The digest state (sample bytes + sample rate + channel count) is
+        per-instance-constant — samples are never mutated after construction
+        (established in Task 5's review) — so it is computed once and memoized
+        on the ``AudioData`` instance under ``_content_hash``. This is the same
+        ``audio.__dict__`` mechanism used for ``AudioData.mono`` (Task 5) and
+        ``_true_peaks`` (Task 6), and avoids re-hashing the (potentially large)
+        sample buffer on every cache ``get``/``put``.
+
+        Callers must ``.copy()`` the returned object before feeding the
+        per-function suffix so the memoized base state is never mutated.
+        """
+        cached = audio.__dict__.get("_content_hash")
+        if cached is not None:
+            return cached
+
+        h = hashlib.sha256()
+        h.update(audio.samples.tobytes())
+        h.update(str(audio.sample_rate).encode())
+        h.update(str(audio.num_channels).encode())
+        audio.__dict__["_content_hash"] = h
+        return h
+
     def _hash_audio(self, audio: AudioData, func_name: str) -> str:
         """Compute a SHA-256 cache key from audio content and function name.
 
@@ -53,11 +78,14 @@ def _hash_audio(self, audio: AudioData, func_name: str) -> str:
         - Sample rate (same bytes at different rates = different audio)
         - Channel count (mono vs stereo reshape = different analysis)
         - Function name (prevents cross-function result collisions)
+
+        The content-digest portion (sample bytes + rate + channels) is memoized
+        per instance via :meth:`_content_digest`; only the cheap ``func_name``
+        suffix is fed per call. The resulting key is byte-identical to feeding
+        all four parts into a single fresh SHA-256, so the key format is
+        unchanged.
         """
-        h = hashlib.sha256()
-        h.update(audio.samples.tobytes())
-        h.update(str(audio.sample_rate).encode())
-        h.update(str(audio.num_channels).encode())
+        h = self._content_digest(audio).copy()
         h.update(func_name.encode())
         return h.hexdigest()
 
@@ -99,6 +127,34 @@ def put(self, audio: AudioData, func_name: str, result: Any) -> None:
                         "Cache evicted oldest entry (key=%s...)", evicted_key[:12]
                     )
 
+    def clear(self) -> None:
+        """Remove all cached entries.
+
+        Used to measure the cold (cache-miss) analysis path in the timing
+        harness, so successive runs don't measure cache hits.
+        """
+        with self._lock:
+            self._store.clear()
+
 
 # Module-level singleton for use by analysis/comparison modules.
 analysis_cache = AnalysisCache()
+
+
+def _cached_analysis(
+    audio: AudioData, func_name: str, func: Callable[[AudioData], Any]
+) -> Any:
+    """Run an analysis function with cache lookup/store.
+
+    Checks the shared ``analysis_cache`` first. On miss, runs ``func(audio)``
+    and stores the result under ``func_name`` for subsequent calls with the
+    same audio content. Composite tools (``full_diagnostic``,
+    ``batch_diagnostic``), the ``analyze`` CLI, and the ``compare_*`` tools all
+    route through this helper so they share per-analyzer results (P-01).
+    """
+    result = analysis_cache.get(audio, func_name)
+    if result is not _MISSING:
+        return result
+    result = func(audio)
+    analysis_cache.put(audio, func_name, result)
+    return result

src/phantom/_resample.py

@@ -82,3 +82,28 @@ def resample_to_match(audio: AudioData, target_sr: int) -> AudioData:
         num_samples=num_samples,
         file_path=audio.file_path,
     )
+
+
+def align_sample_rates(*audios: AudioData) -> tuple[AudioData, ...]:
+    """Upsample all inputs to the highest sample rate; identity when equal.
+
+    Returns the inputs in their original order.  When every input already
+    shares the same sample rate, the same objects are returned unchanged
+    (identity fast-path).  Otherwise each lower-rate input is upsampled to
+    ``max`` via :func:`resample_to_match`; inputs already at the target rate
+    are returned unchanged.
+
+    Args:
+        *audios: One or more :class:`AudioData` objects to align.
+
+    Returns:
+        A tuple of :class:`AudioData` in input order, all sharing the highest
+        sample rate among the inputs.
+    """
+    rates = {a.sample_rate for a in audios}
+    if len(rates) <= 1:
+        return tuple(audios)
+    target = max(rates)
+    return tuple(
+        resample_to_match(a, target) if a.sample_rate != target else a for a in audios
+    )

src/phantom/_truepeak.py

@@ -0,0 +1,57 @@
+"""Shared per-channel true-peak computation (P-02, P-06).
+
+``es.TruePeakDetector`` (x4-oversampled FIR per ITU-R BS.1770-4) is the single
+most expensive operation in the analysis pipeline. Both loudness measurement
+and inter-sample-peak problem detection need it, and in ``full_diagnostic`` both
+run on the same file — previously computing it twice.
+
+``channel_true_peaks`` computes per-channel ``(sample_peak, true_peak)`` once and
+memoizes the result on the ``AudioData`` instance so loudness and problem
+detection share a single computation (P-02). A single ``TruePeakDetector`` is
+constructed and reused across channels (P-06); the detector is stateless across
+the two channels here (validated by the numeric-parity test in the suite).
+
+The memo lives in ``audio.__dict__["_true_peaks"]`` — the same mechanism
+``functools.cached_property`` uses for ``AudioData.mono`` / ``mono_rms``, and
+proven safe with this Pydantic v2 model.
+"""
+
+from __future__ import annotations
+
+import numpy as np
+import essentia.standard as es
+
+from phantom.audio import AudioData
+
+
+def channel_true_peaks(audio: AudioData) -> list[tuple[float, float]]:
+    """Per-channel ``(sample_peak, true_peak)`` via ITU-R BS.1770-4 x4 oversampling.
+
+    Constructs ONE ``TruePeakDetector`` and reuses it across channels (P-06).
+    The result is memoized on the ``AudioData`` instance so loudness and problem
+    detection share a single computation (P-02).
+
+    Args:
+        audio: AudioData object to analyze.
+
+    Returns:
+        A list of ``(sample_peak, true_peak)`` tuples, one per channel, where
+        both values are linear (not dB) absolute maxima.
+    """
+    cached = audio.__dict__.get("_true_peaks")
+    if cached is not None:
+        return cached
+
+    tp_algo = es.TruePeakDetector(
+        version=4, sampleRate=audio.sample_rate, oversamplingFactor=4
+    )
+    out: list[tuple[float, float]] = []
+    for ch in range(audio.num_channels):
+        sig = audio.samples[:, ch]
+        sample_peak = float(np.max(np.abs(sig))) if len(sig) else 0.0
+        _, tp_output = tp_algo(sig)
+        true_peak = float(np.max(np.abs(tp_output))) if len(tp_output) else 0.0
+        out.append((sample_peak, true_peak))
+
+    audio.__dict__["_true_peaks"] = out
+    return out

src/phantom/audio.py

@@ -13,6 +13,7 @@
 from __future__ import annotations
 
 import os
+from functools import cached_property
 from typing import Optional
 
 import numpy as np
@@ -92,17 +93,25 @@ def right(self) -> np.ndarray:
             )
         return self.samples[:, 1]
 
-    @property
+    @cached_property
     def mono(self) -> np.ndarray:
-        """Return a mono mixdown of the audio.
+        """Return a (memoized) mono mixdown of the audio.
 
-        For mono input, returns samples[:, 0] directly.
-        For stereo input, returns the mean of left and right channels.
+        For mono input, returns samples[:, 0]. For stereo, the mean of the
+        first two channels. Computed once per instance (P-03).
         """
         if self.num_channels == 1:
             return self.samples[:, 0]
         return np.mean(self.samples[:, :2], axis=1)
 
+    @cached_property
+    def mono_rms(self) -> float:
+        """Full-signal RMS of the mono mixdown, computed once (P-04)."""
+        m = self.mono
+        if not np.issubdtype(m.dtype, np.floating):
+            m = m.astype(np.float64)
+        return float(np.sqrt(np.mean(m**2)))
+
 
 def load_audio(
     path: str,

src/phantom/cli/_formatting.py

@@ -7,6 +7,7 @@
 from __future__ import annotations
 
 import json
+import re
 import sys
 
 import plotext as plt
@@ -16,6 +17,12 @@
 
 from phantom.exceptions import RECOMMENDED_PYTHON, DependencyMissingError, PhantomError
 
+# Local copy of server.py's _PATH_REGEX — strips Unix/Windows absolute paths from
+# error messages so CLI output never leaks internal filesystem layout. Kept as a
+# local copy (rather than importing from phantom.server) to avoid a CLI->server
+# import edge.
+_PATH_REGEX = re.compile(r"([A-Za-z]:\\[^\s:,)]+\\|/[^\s:,)]+/)+")
+
 # ---------------------------------------------------------------------------
 # Severity styling (D-06)
 # ---------------------------------------------------------------------------
@@ -214,8 +221,9 @@ def render_error(exc: Exception, console: Console) -> None:
     """Render an exception as a styled Rich Panel.
 
     - ``DependencyMissingError``: yellow panel with install instructions
-    - ``PhantomError``: red panel with error message
-    - Other: red panel titled "Unexpected Error"
+    - ``PhantomError``: red panel with error message (absolute paths redacted)
+    - Other: red panel titled "Unexpected Error" with a generic message so
+      internal paths / details never leak (mirrors ``server.py``).
     """
     if isinstance(exc, DependencyMissingError):
         console.print(
@@ -229,15 +237,15 @@ def render_error(exc: Exception, console: Console) -> None:
     elif isinstance(exc, PhantomError):
         console.print(
             Panel(
-                str(exc),
+                _PATH_REGEX.sub("", str(exc)),
                 title="Error",
                 border_style="red",
             )
         )
     else:
         console.print(
             Panel(
-                str(exc),
+                "Internal error — check the logs for details.",
                 title="Unexpected Error",
                 border_style="red",
             )