From 801246ecfdb4d01f87edd30892df5c2220279e66 Mon Sep 17 00:00:00 2001
From: Alireza Eliaderani <172368+cubny@users.noreply.github.com>
Date: Thu, 9 Jul 2026 08:47:23 +0200
Subject: [PATCH 1/3] Add real-time CLI spectrum visualizer

Add a classic bar-style audio equalizer that animates in the terminal
while music plays. Bars rise from the bottom using Unicode block glyphs
(black & white); on by default in an interactive terminal, disable with
--no-spectrum.

- SpectrumAnalyzer (analysis/realtime.py): windowed rfft, log-spaced
  frequency bands (one per column), dB normalization, fast-attack/
  slow-decay smoothing.
- SpectrumRenderer + SpectrumDisplay (ui/spectrum.py): vectorized
  bottom-up bar rendering and an in-place multi-row ANSI block reusing
  StatusLine._format for the status row.
- AudioOutput (audio/output.py): a lock-free ring buffer tap in the
  playback callback so the visualizer analyzes the audio actually being
  heard (the generator runs seconds ahead behind the buffer), pulled at
  25fps every frame rather than once per chunk.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 src/focus/analysis/realtime.py | 115 +++++++++++++++++++
 src/focus/audio/output.py      |  49 ++++++++
 src/focus/cli.py               |  71 +++++++++++-
 src/focus/ui/spectrum.py       | 125 ++++++++++++++++++++
 tests/test_spectrum.py         | 204 +++++++++++++++++++++++++++++++++
 5 files changed, 558 insertions(+), 6 deletions(-)
 create mode 100644 src/focus/analysis/realtime.py
 create mode 100644 src/focus/ui/spectrum.py
 create mode 100644 tests/test_spectrum.py

diff --git a/src/focus/analysis/realtime.py b/src/focus/analysis/realtime.py
new file mode 100644
index 0000000..5319ff2
--- /dev/null
+++ b/src/focus/analysis/realtime.py
@@ -0,0 +1,115 @@
+"""Real-time spectrum analysis for the live terminal visualizer.
+
+Reuses the same FFT primitives as :mod:`focus.analysis.fft` (``scipy.fft``), but
+maintains a rolling window and per-band smoothing state so it can be driven at
+video frame rates from the playback loop.
+"""
+
+from __future__ import annotations
+
+import numpy as np
+from scipy.fft import rfft, rfftfreq
+
+
+class SpectrumAnalyzer:
+    """Turns a stream of audio chunks into normalized log-spaced band levels.
+
+    All work happens on the caller's thread (the asyncio event loop); the
+    PortAudio callback is never involved. ``push`` is cheap (a downmix + buffer
+    roll); ``compute`` performs one ``rfft`` per call.
+    """
+
+    def __init__(
+        self,
+        sample_rate: int = 48000,
+        fft_size: int = 2048,
+        f_min: float = 30.0,
+        f_max: float = 16000.0,
+        floor_db: float = -70.0,
+        ceil_db: float = -12.0,
+        attack: float = 0.6,
+        decay: float = 0.18,
+    ) -> None:
+        self.sample_rate = sample_rate
+        self.fft_size = fft_size
+        self.f_min = f_min
+        self.f_max = min(f_max, sample_rate / 2.0)
+        self.floor_db = floor_db
+        self.ceil_db = ceil_db
+        self.attack = attack
+        self.decay = decay
+
+        self._buf = np.zeros(fft_size, dtype=np.float32)
+        self._window = np.hanning(fft_size).astype(np.float32)
+        self._freqs = rfftfreq(fft_size, 1.0 / sample_rate)
+        # Coherent gain of the window: normalizing by it makes a full-scale
+        # sine read ~0 dBFS, so the floor/ceil dB thresholds are meaningful and
+        # independent of fft_size.
+        self._norm = self._window.sum() / 2.0
+
+        # Smoothing state and cached band edges, both keyed by num_bands.
+        self._levels: np.ndarray | None = None
+        self._band_bins: list[tuple[int, int]] | None = None
+        self._num_bands: int | None = None
+
+    def push(self, chunk: np.ndarray) -> None:
+        """Feed a new audio chunk (mono or stereo float array)."""
+        if chunk is None or len(chunk) == 0:
+            return
+        mono = chunk.mean(axis=1) if chunk.ndim == 2 else chunk
+        mono = np.asarray(mono, dtype=np.float32)
+        n = len(mono)
+        if n >= self.fft_size:
+            self._buf[:] = mono[-self.fft_size :]
+        else:
+            self._buf = np.roll(self._buf, -n)
+            self._buf[-n:] = mono
+
+    def push_silence(self) -> None:
+        """Zero the rolling window so bars fall to the floor (e.g. while paused)."""
+        self._buf[:] = 0.0
+
+    def _ensure_bands(self, num_bands: int) -> None:
+        if self._num_bands == num_bands and self._band_bins is not None:
+            return
+        edges = np.logspace(
+            np.log10(self.f_min), np.log10(self.f_max), num_bands + 1
+        )
+        idx = np.searchsorted(self._freqs, edges)
+        bins: list[tuple[int, int]] = []
+        max_bin = len(self._freqs)
+        for i in range(num_bands):
+            lo = int(idx[i])
+            hi = int(idx[i + 1])
+            # Guarantee at least one bin per band so high bands aren't empty.
+            if hi <= lo:
+                hi = min(lo + 1, max_bin)
+            bins.append((lo, hi))
+        self._band_bins = bins
+        self._num_bands = num_bands
+        self._levels = np.zeros(num_bands, dtype=np.float32)
+
+    def compute(self, num_bands: int) -> np.ndarray:
+        """Return smoothed band levels in ``[0, 1]`` (length ``num_bands``)."""
+        self._ensure_bands(num_bands)
+        assert self._band_bins is not None and self._levels is not None
+
+        mag = np.abs(rfft(self._buf * self._window)) / self._norm
+        db = 20.0 * np.log10(mag + 1e-9)
+
+        target = np.empty(num_bands, dtype=np.float32)
+        span = self.ceil_db - self.floor_db
+        for i, (lo, hi) in enumerate(self._band_bins):
+            band_db = db[lo:hi].max() if hi > lo else self.floor_db
+            target[i] = np.clip((band_db - self.floor_db) / span, 0.0, 1.0)
+
+        # Fast attack, slow decay; never dip below the instantaneous target.
+        prev = self._levels
+        rising = target >= prev
+        smoothed = np.where(
+            rising,
+            self.attack * target + (1.0 - self.attack) * prev,
+            np.maximum(prev - self.decay, target),
+        )
+        self._levels = np.clip(smoothed, 0.0, 1.0).astype(np.float32)
+        return self._levels
diff --git a/src/focus/audio/output.py b/src/focus/audio/output.py
index af059c4..ea7cd6c 100644
--- a/src/focus/audio/output.py
+++ b/src/focus/audio/output.py
@@ -59,6 +59,14 @@ class AudioOutput:
     _blocks_written: int = field(default=0, init=False)
     _blocks_played: int = field(default=0, init=False)
 
+    # Visualizer tap: a mono ring buffer of the audio actually sent to the
+    # speakers, written from the PortAudio callback and read by the spectrum
+    # renderer on the event-loop thread. Lock-free (visual-only; a torn read is
+    # at most a one-frame glitch) so the callback never blocks.
+    _viz_size: int = field(default=8192, init=False)
+    _viz_ring: np.ndarray | None = field(default=None, init=False, repr=False)
+    _viz_write: int = field(default=0, init=False)
+
     def __post_init__(self):
         if not SOUNDDEVICE_AVAILABLE:
             raise ImportError(
@@ -67,6 +75,7 @@ def __post_init__(self):
         self._queue = queue.Queue(maxsize=self.buffersize)
         # Recovery fade-in over ~50ms for smooth transitions
         self._recovery_samples = int(0.05 * self.sample_rate)
+        self._viz_ring = np.zeros(self._viz_size, dtype=np.float32)
 
     def start(self) -> None:
         """Prepare for audio output (stream starts when buffer is filled)."""
@@ -153,6 +162,42 @@ def _audio_callback(self, outdata: np.ndarray, frames: int, time_info, status) -
         if self.volume != 1.0:
             outdata *= self.volume
 
+        # Capture what's actually being played for the spectrum visualizer.
+        self._capture_visualizer(outdata)
+
+    def _capture_visualizer(self, outdata: np.ndarray) -> None:
+        """Write a mono downmix of the played block into the ring buffer."""
+        ring = self._viz_ring
+        if ring is None:
+            return
+        mono = outdata.mean(axis=1) if outdata.ndim == 2 else outdata
+        n = len(mono)
+        size = self._viz_size
+        w = self._viz_write
+        end = w + n
+        if end <= size:
+            ring[w:end] = mono
+        else:
+            first = size - w
+            ring[w:] = mono[:first]
+            ring[: end - size] = mono[first:]
+        self._viz_write = end % size
+
+    def latest_samples(self, n: int) -> np.ndarray:
+        """Return the most recent ``n`` played mono samples, in order (a copy).
+
+        Safe to call from another thread; reads may momentarily straddle a
+        callback write, which only ever causes a harmless one-frame glitch.
+        """
+        ring = self._viz_ring
+        if ring is None:
+            return np.zeros(n, dtype=np.float32)
+        n = min(n, self._viz_size)
+        w = self._viz_write
+        if w >= n:
+            return ring[w - n : w].copy()
+        return np.concatenate((ring[self._viz_size - (n - w) :], ring[:w]))
+
     def write(self, audio: np.ndarray) -> None:
         """Write audio data to the output buffer.
 
@@ -277,6 +322,10 @@ def pause(self) -> None:
         self._in_underrun = False
         self._underrun_fade_pos = 0
         self._recovery_fade_pos = 0
+        # Silence the visualizer tap so the bars fall to the floor while paused
+        # (the callback is torn down here, so it can no longer clear the ring).
+        if self._viz_ring is not None:
+            self._viz_ring[:] = 0.0
 
     def resume(self) -> None:
         """Resume playback. The stream is recreated lazily once the buffer re-fills."""
diff --git a/src/focus/cli.py b/src/focus/cli.py
index 42b792d..641c2c6 100644
--- a/src/focus/cli.py
+++ b/src/focus/cli.py
@@ -12,6 +12,9 @@
 from focus.profiles import FocusProfile, get_profile, list_profiles
 from focus.ui.transport import KeyboardController, PlaybackState, StatusLine
 
+# Spectrum visualizer is optional; imported lazily in _run_session so the CLI
+# still works if numpy/scipy are missing (guarded there).
+
 # Check for optional dependencies
 try:
     import numpy as np
@@ -150,6 +153,12 @@ def show_profiles():
     default=9,
     help="Duration of each track before rotation (1-9 minutes, default: 9)",
 )
+@click.option(
+    "--spectrum/--no-spectrum",
+    is_flag=True,
+    default=True,
+    help="Live audio spectrum visualizer (interactive terminal only, default: on)",
+)
 def start_session(
     profile: str,
     frequency: float | None,
@@ -163,6 +172,7 @@ def start_session(
     limiter: bool,
     verbose: bool,
     track_duration: int,
+    spectrum: bool,
 ):
     """Start a focus music session.
 
@@ -187,6 +197,7 @@ def start_session(
         limiter=limiter,
         verbose=verbose,
         track_duration=track_duration,
+        spectrum=spectrum,
     )
 
 
@@ -203,6 +214,7 @@ def launch_session(
     limiter: bool = True,
     verbose: bool = False,
     track_duration: int = 9,
+    spectrum: bool = True,
 ):
     """Resolve a profile, apply overrides, and run a session.
 
@@ -271,6 +283,7 @@ def launch_session(
                 limiter=limiter,
                 verbose=verbose,
                 track_duration=track_duration,
+                spectrum=spectrum,
             )
         )
     except KeyboardInterrupt:
@@ -291,6 +304,7 @@ async def _run_session(
     limiter: bool = True,
     verbose: bool = False,
     track_duration: int = 9,
+    spectrum: bool = True,
 ):
     """Run the audio generation session."""
     try:
@@ -355,6 +369,8 @@ def make_client(phase: str):
     state = None
     keyboard = None
     status_line = None
+    display = None
+    spectrum_task = None
     if interactive:
         state = PlaybackState(
             profile_name=profile.name,
@@ -363,16 +379,32 @@ def make_client(phase: str):
             status="connecting",
         )
         if not verbose:
-            # The live status line and -v logging both want the bottom line;
-            # when verbose, the logs already convey state, so skip the line.
-            status_line = StatusLine()
-            status_line.start()
-            status_line.render(state)
+            # The live status line and -v logging both want the bottom region;
+            # when verbose, the logs already convey state, so skip the display.
+            if spectrum:
+                # The spectrum display owns the bottom rows and draws the same
+                # status text as its last row (via StatusLine._format).
+                try:
+                    from focus.analysis.realtime import SpectrumAnalyzer
+                    from focus.ui.spectrum import SpectrumDisplay
+
+                    analyzer = SpectrumAnalyzer(sample_rate=sample_rate)
+                    display = SpectrumDisplay(state, analyzer)
+                    display.start()
+                    display.render()
+                except ImportError:
+                    display = None
+            if display is None:
+                status_line = StatusLine()
+                status_line.start()
+                status_line.render(state)
 
         # Redraw immediately on each keypress so pause / volume / help toggles
         # are reflected instantly instead of on the next audio chunk.
         def _on_key():
-            if status_line is not None:
+            if display is not None:
+                display.render()
+            elif status_line is not None:
                 status_line.render(state)
 
         keyboard = KeyboardController(state, on_change=_on_key)
@@ -380,6 +412,8 @@ def _on_key():
             keyboard.start()
         except Exception:
             keyboard.stop()
+            if display is not None:
+                display.finish()
             if status_line is not None:
                 status_line.finish()
             raise
@@ -432,6 +466,23 @@ def _on_key():
     session_complete = False
 
     try:
+        # Drive the spectrum redraw at a fixed frame rate, decoupled from the
+        # irregular arrival of audio chunks. Created inside the try so the
+        # finally block always cancels it and restores the terminal.
+        if display is not None:
+            # Feed the visualizer from the playback tap (what's actually heard),
+            # now that the output stream exists.
+            display.source = output.latest_samples
+
+            async def _spectrum_loop():
+                try:
+                    while state is None or not state.quit_requested:
+                        display.render()
+                        await asyncio.sleep(1.0 / display.fps)
+                except asyncio.CancelledError:
+                    pass
+
+            spectrum_task = asyncio.create_task(_spectrum_loop())
         # Outer loop: each iteration consumes one generator until it ends or an
         # interactive control (pause / next take) asks us to reconnect.
         while not session_complete:
@@ -644,8 +695,16 @@ def _on_key():
             traceback.print_exc()
     finally:
         # Restore the terminal before any further output
+        if spectrum_task is not None:
+            spectrum_task.cancel()
+            try:
+                await spectrum_task
+            except asyncio.CancelledError:
+                pass
         if keyboard is not None:
             keyboard.stop()
+        if display is not None:
+            display.finish()
         if status_line is not None:
             status_line.finish()
         # Flush any remaining buffered audio to file (for Ctrl+C case with duration set)
diff --git a/src/focus/ui/spectrum.py b/src/focus/ui/spectrum.py
new file mode 100644
index 0000000..8656d1c
--- /dev/null
+++ b/src/focus/ui/spectrum.py
@@ -0,0 +1,125 @@
+"""Terminal spectrum visualizer: a classic bar analyzer rising from the bottom.
+
+Bars are drawn with Unicode block glyphs (black & white). Rendering uses the
+same in-place multi-row ANSI pattern as :mod:`focus.ui.launcher` (cursor-up +
+erase-to-end each frame), and reuses :meth:`StatusLine._format` for the bottom
+status row so the status text stays single-sourced.
+"""
+
+from __future__ import annotations
+
+import shutil
+import sys
+
+import numpy as np
+
+from focus.analysis.realtime import SpectrumAnalyzer
+from focus.ui.transport import PlaybackState, StatusLine
+
+# Glyph ramp indexed by eighths of a cell: 0 = empty, 8 = full block.
+_GLYPHS = np.array(list(" ▁▂▃▄▅▆▇█"))
+
+
+class SpectrumRenderer:
+    """Pure, testable conversion of band levels into terminal rows.
+
+    Bars rise from the bottom: the last row is densest. Each returned string is a
+    full row of ``len(values)`` glyphs; rows are ordered top -> bottom.
+    """
+
+    @staticmethod
+    def render_rows(values: np.ndarray, height: int) -> list[str]:
+        if height < 1:
+            return []
+        # Filled eighths per column, then per (row, column) how much of that cell
+        # is filled. Vectorized so a wide terminal stays cheap at 25+ fps.
+        eighths = np.clip(np.asarray(values, dtype=float), 0.0, 1.0) * (height * 8)
+        # Row 0 is the top; a row's distance from the bottom scales its threshold.
+        from_bottom = np.arange(height - 1, -1, -1)[:, None]
+        cell = eighths[None, :] - from_bottom * 8  # (height, num_bands)
+        idx = np.clip(np.floor(cell), 0, 8).astype(int)
+        grid = _GLYPHS[idx]
+        return ["".join(row) for row in grid]
+
+
+class SpectrumDisplay:
+    """Owns the in-place terminal block: spectrum bars plus a status row."""
+
+    def __init__(
+        self,
+        state: PlaybackState,
+        analyzer: SpectrumAnalyzer,
+        stream=None,
+        height: int = 12,
+        fps: int = 25,
+        source=None,
+    ) -> None:
+        self.state = state
+        self.analyzer = analyzer
+        self.stream = stream or sys.stdout
+        self.height = height
+        self.fps = fps
+        # Callable returning the latest N played mono samples. Pulled every frame
+        # so the bars track what's actually heard (not the producer, which runs
+        # seconds ahead behind the playback buffer). May be set after construction
+        # once the audio output exists.
+        self.source = source
+        self._active = False
+        self._prev_lines = 0
+
+    def start(self) -> None:
+        self._active = True
+        self._prev_lines = 0
+        # Hide the cursor for a flicker-free repaint.
+        self.stream.write("\x1b[?25l")
+        self.stream.flush()
+
+    def push(self, chunk: np.ndarray) -> None:
+        self.analyzer.push(chunk)
+
+    def render(self) -> None:
+        if not self._active:
+            return
+        # Refresh the analysis window from the live playback tap each frame.
+        if self.source is not None:
+            self.analyzer.push(self.source(self.analyzer.fft_size))
+        size = shutil.get_terminal_size(fallback=(80, 24))
+        cols = max(1, size.columns)
+        lines = max(1, size.lines)
+
+        status = StatusLine._format(self.state)
+        if len(status) > cols - 1:
+            status = status[: cols - 1]
+
+        # Very short terminals: fall back to a status-only line.
+        if lines < 4:
+            rows: list[str] = []
+        else:
+            h = max(1, min(self.height, lines - 2))
+            num_bands = cols
+            values = self.analyzer.compute(num_bands)
+            rows = SpectrumRenderer.render_rows(values, h)
+
+        block = "\n".join(rows + [status])
+        out = ["\x1b[?7l"]  # disable autowrap for the repaint
+        if self._prev_lines:
+            out.append(f"\x1b[{self._prev_lines}A")
+        out.append("\r\x1b[J")  # move to column 0, erase from cursor to end
+        out.append(block)
+        out.append("\x1b[?7h")  # restore autowrap
+        self.stream.write("".join(out))
+        self.stream.flush()
+        self._prev_lines = len(rows)  # cursor ends on the status row
+
+    def finish(self) -> None:
+        if not self._active:
+            return
+        out = ["\x1b[?7l"]
+        if self._prev_lines:
+            out.append(f"\x1b[{self._prev_lines}A")
+        out.append("\r\x1b[J")  # erase the whole block
+        out.append("\x1b[?7h\x1b[?25h")  # restore autowrap + show cursor
+        self.stream.write("".join(out))
+        self.stream.flush()
+        self._active = False
+        self._prev_lines = 0
diff --git a/tests/test_spectrum.py b/tests/test_spectrum.py
new file mode 100644
index 0000000..e07cc67
--- /dev/null
+++ b/tests/test_spectrum.py
@@ -0,0 +1,204 @@
+"""Tests for the real-time spectrum analyzer and terminal visualizer."""
+
+import io
+
+import numpy as np
+import pytest
+from click.testing import CliRunner
+
+from focus.analysis.realtime import SpectrumAnalyzer
+from focus.audio.output import SOUNDDEVICE_AVAILABLE, AudioOutput
+from focus.cli import main
+from focus.ui.spectrum import SpectrumDisplay, SpectrumRenderer
+from focus.ui.transport import PlaybackState
+
+requires_sounddevice = pytest.mark.skipif(
+    not SOUNDDEVICE_AVAILABLE, reason="sounddevice/PortAudio unavailable"
+)
+
+
+def _sine(freq: float, sr: int = 48000, n: int = 24000, amp: float = 0.8) -> np.ndarray:
+    t = np.arange(n) / sr
+    mono = (amp * np.sin(2 * np.pi * freq * t)).astype(np.float32)
+    return np.column_stack([mono, mono])
+
+
+class TestSpectrumAnalyzer:
+    def test_output_shape_and_range(self):
+        a = SpectrumAnalyzer()
+        a.push(_sine(1000))
+        v = a.compute(60)
+        assert len(v) == 60
+        assert float(v.min()) >= 0.0
+        assert float(v.max()) <= 1.0
+
+    def test_tone_peaks_in_matching_band(self):
+        a = SpectrumAnalyzer()
+        a.push(_sine(1000))
+        for _ in range(6):  # let the fast attack settle
+            v = a.compute(60)
+        peak = int(v.argmax())
+        assert v[peak] > 0.7
+        # The peak band's FFT-bin range must actually cover 1 kHz.
+        lo, hi = a._band_bins[peak]
+        band_freqs = a._freqs[lo:hi]
+        assert band_freqs.min() <= 1000.0 <= band_freqs.max() + a._freqs[1]
+        # A far-away low band should be well below the peak.
+        assert v[0] < v[peak] - 0.4
+
+    def test_silence_is_near_zero(self):
+        a = SpectrumAnalyzer()
+        a.push_silence()
+        for _ in range(20):
+            v = a.compute(40)
+        assert float(v.max()) < 1e-3
+
+    def test_slow_decay_after_drop(self):
+        a = SpectrumAnalyzer(decay=0.1)
+        a.push(_sine(1000))
+        for _ in range(6):
+            high = a.compute(60).copy()
+        peak = int(high.argmax())
+        a.push_silence()
+        after_one = a.compute(60)
+        # One frame of silence must not collapse the bar; it decays gradually.
+        assert after_one[peak] >= high[peak] - 0.1 - 1e-6
+        assert after_one[peak] > 0.0
+
+    def test_rise_is_fast(self):
+        a = SpectrumAnalyzer(attack=0.6)
+        a.push_silence()
+        a.compute(60)
+        a.push(_sine(1000))
+        one = a.compute(60)
+        peak = int(one.argmax())
+        # A single frame of attack should already be a large fraction of target.
+        assert one[peak] > 0.4
+
+
+class TestSpectrumRenderer:
+    def test_full_column_is_all_blocks(self):
+        rows = SpectrumRenderer.render_rows(np.array([1.0]), 5)
+        assert rows == ["█"] * 5
+
+    def test_empty_column_is_all_spaces(self):
+        rows = SpectrumRenderer.render_rows(np.array([0.0]), 5)
+        assert rows == [" "] * 5
+
+    def test_bars_rise_from_the_bottom(self):
+        rows = SpectrumRenderer.render_rows(np.array([0.5]), 4)
+        # Top rows empty, bottom rows full (rises from the bottom).
+        assert rows[0] == " "
+        assert rows[-1] == "█"
+
+    def test_row_width_matches_band_count(self):
+        rows = SpectrumRenderer.render_rows(np.array([0.2, 0.5, 0.9]), 3)
+        assert all(len(r) == 3 for r in rows)
+
+
+class TestSpectrumDisplay:
+    def test_render_brackets_and_status_and_cursor_up(self, monkeypatch):
+        monkeypatch.setenv("COLUMNS", "40")
+        monkeypatch.setenv("LINES", "24")
+        state = PlaybackState(profile_name="deep-work", status="playing")
+        analyzer = SpectrumAnalyzer()
+        analyzer.push(_sine(1000))
+        buf = io.StringIO()
+        display = SpectrumDisplay(state, analyzer, stream=buf, height=8)
+        display.start()
+        display.render()
+        out = buf.getvalue()
+        assert "\x1b[?25l" in out  # cursor hidden on start
+        assert "\x1b[?7l" in out and "\x1b[?7h" in out  # autowrap toggled
+        assert "deep-work" in out  # status row reuses StatusLine._format
+        # Second render moves the cursor up over the previously drawn block.
+        buf.truncate(0)
+        buf.seek(0)
+        display.render()
+        # 8 spectrum rows joined to the status row => 8 newlines => move up 8.
+        assert "\x1b[8A" in buf.getvalue()
+
+    def test_finish_restores_cursor_and_autowrap(self, monkeypatch):
+        monkeypatch.setenv("COLUMNS", "40")
+        monkeypatch.setenv("LINES", "24")
+        state = PlaybackState(profile_name="deep-work")
+        display = SpectrumDisplay(state, SpectrumAnalyzer(), stream=io.StringIO())
+        display.start()
+        display.render()
+        display.stream.truncate(0)
+        display.stream.seek(0)
+        display.finish()
+        out = display.stream.getvalue()
+        assert "\x1b[?25h" in out  # cursor restored
+        assert "\x1b[?7h" in out  # autowrap restored
+
+    def test_short_terminal_falls_back_to_status_only(self, monkeypatch):
+        monkeypatch.setenv("COLUMNS", "40")
+        monkeypatch.setenv("LINES", "3")
+        state = PlaybackState(profile_name="deep-work", status="playing")
+        buf = io.StringIO()
+        display = SpectrumDisplay(state, SpectrumAnalyzer(), stream=buf)
+        display.start()
+        display.render()
+        # Only the status row is drawn (no preceding spectrum rows).
+        assert display._prev_lines == 0
+        assert "deep-work" in buf.getvalue()
+
+
+class TestSourceDrivenRender:
+    def test_render_pulls_a_fresh_window_from_the_source_each_frame(self, monkeypatch):
+        monkeypatch.setenv("COLUMNS", "40")
+        monkeypatch.setenv("LINES", "24")
+        calls = {"n": 0}
+
+        def source(n):
+            calls["n"] += 1
+            return _sine(1000)[:, 0]  # mono window
+
+        display = SpectrumDisplay(
+            PlaybackState(profile_name="x"),
+            SpectrumAnalyzer(),
+            stream=io.StringIO(),
+            source=source,
+        )
+        display.start()
+        display.render()
+        display.render()
+        assert calls["n"] == 2  # one pull per frame, so bars track live audio
+
+
+@requires_sounddevice
+class TestAudioOutputVisualizerTap:
+    def test_latest_samples_returns_recent_played_audio(self):
+        o = AudioOutput()
+        for i in range(6):
+            o._capture_visualizer(np.full((2048, 2), float(i), dtype=np.float32))
+        tail = o.latest_samples(2048)
+        assert len(tail) == 2048
+        assert np.all(tail == 5.0)  # most recent block
+
+    def test_ring_wraps_around(self):
+        o = AudioOutput()
+        ramp = np.arange(3000, dtype=np.float32)
+        for _ in range(3):  # 9000 samples into an 8192 ring forces a wrap
+            o._capture_visualizer(np.column_stack([ramp, ramp]))
+        last = o.latest_samples(500)
+        assert np.all(last == np.arange(2500, 3000))
+
+    def test_pause_clears_the_tap(self):
+        o = AudioOutput()
+        o._capture_visualizer(np.ones((2048, 2), dtype=np.float32))
+        o.pause()
+        assert np.all(o.latest_samples(2048) == 0.0)  # bars fall while paused
+
+
+class TestCliGate:
+    def test_spectrum_flag_is_noop_in_non_tty(self):
+        # In a non-tty CliRunner context the visualizer path is skipped entirely;
+        # --spectrum must not crash and must not emit the ANSI block.
+        runner = CliRunner()
+        result = runner.invoke(
+            main, ["start", "--mock", "--spectrum", "--duration", "60"]
+        )
+        # Duration < minimum is rejected cleanly; the point is no import/setup crash.
+        assert "\x1b[?25l" not in result.output

From e7add4e785f407dfdfb12c508c97c97e01d752d8 Mon Sep 17 00:00:00 2001
From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com>
Date: Thu, 9 Jul 2026 19:15:34 +0000
Subject: [PATCH 2/3] Fix CI failure by applying Ruff formatting to spectrum
 files

---
 src/focus/analysis/realtime.py | 4 +---
 tests/test_spectrum.py         | 4 +---
 2 files changed, 2 insertions(+), 6 deletions(-)

diff --git a/src/focus/analysis/realtime.py b/src/focus/analysis/realtime.py
index 5319ff2..12dc723 100644
--- a/src/focus/analysis/realtime.py
+++ b/src/focus/analysis/realtime.py
@@ -72,9 +72,7 @@ def push_silence(self) -> None:
     def _ensure_bands(self, num_bands: int) -> None:
         if self._num_bands == num_bands and self._band_bins is not None:
             return
-        edges = np.logspace(
-            np.log10(self.f_min), np.log10(self.f_max), num_bands + 1
-        )
+        edges = np.logspace(np.log10(self.f_min), np.log10(self.f_max), num_bands + 1)
         idx = np.searchsorted(self._freqs, edges)
         bins: list[tuple[int, int]] = []
         max_bin = len(self._freqs)
diff --git a/tests/test_spectrum.py b/tests/test_spectrum.py
index e07cc67..ccece1b 100644
--- a/tests/test_spectrum.py
+++ b/tests/test_spectrum.py
@@ -197,8 +197,6 @@ def test_spectrum_flag_is_noop_in_non_tty(self):
         # In a non-tty CliRunner context the visualizer path is skipped entirely;
         # --spectrum must not crash and must not emit the ANSI block.
         runner = CliRunner()
-        result = runner.invoke(
-            main, ["start", "--mock", "--spectrum", "--duration", "60"]
-        )
+        result = runner.invoke(main, ["start", "--mock", "--spectrum", "--duration", "60"])
         # Duration < minimum is rejected cleanly; the point is no import/setup crash.
         assert "\x1b[?25l" not in result.output

From 62a5e660a75a09c5523b6079c45243a3061209b8 Mon Sep 17 00:00:00 2001
From: Alireza Eliaderani <172368+cubny@users.noreply.github.com>
Date: Thu, 9 Jul 2026 21:44:03 +0200
Subject: [PATCH 3/3] Address PR review feedback

- Shift the analyzer's rolling window in place instead of np.roll, so
  pushes smaller than fft_size don't allocate. (The 25fps render path
  already took the in-place branch, since it pulls exactly fft_size
  samples; this covers the remaining callers.) Pinned by tests, as the
  overlapping slice copy is a subtle numpy guarantee.
- Promote StatusLine._format to a public module-level format_status_line,
  so SpectrumDisplay composes the status row without reaching across
  modules into a private helper.
- Stop the CLI gate test from opening a real sounddevice stream and
  playing audio during the suite: stub _run_session, and assert the
  --spectrum/--no-spectrum flag actually threads through. Cover the
  duration-below-minimum fast-fail separately.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 src/focus/analysis/realtime.py |  4 ++-
 src/focus/cli.py               |  2 +-
 src/focus/ui/spectrum.py       |  6 ++--
 src/focus/ui/transport.py      | 39 ++++++++++++----------
 tests/test_spectrum.py         | 59 ++++++++++++++++++++++++++++++----
 tests/test_ui.py               | 12 +++++--
 6 files changed, 90 insertions(+), 32 deletions(-)

diff --git a/src/focus/analysis/realtime.py b/src/focus/analysis/realtime.py
index 12dc723..f689aec 100644
--- a/src/focus/analysis/realtime.py
+++ b/src/focus/analysis/realtime.py
@@ -62,7 +62,9 @@ def push(self, chunk: np.ndarray) -> None:
         if n >= self.fft_size:
             self._buf[:] = mono[-self.fft_size :]
         else:
-            self._buf = np.roll(self._buf, -n)
+            # Shift left in place. np.roll would allocate a new array on every
+            # push; numpy handles the overlapping slice copy correctly.
+            self._buf[:-n] = self._buf[n:]
             self._buf[-n:] = mono
 
     def push_silence(self) -> None:
diff --git a/src/focus/cli.py b/src/focus/cli.py
index 641c2c6..91ba030 100644
--- a/src/focus/cli.py
+++ b/src/focus/cli.py
@@ -383,7 +383,7 @@ def make_client(phase: str):
             # when verbose, the logs already convey state, so skip the display.
             if spectrum:
                 # The spectrum display owns the bottom rows and draws the same
-                # status text as its last row (via StatusLine._format).
+                # status text as its last row (via format_status_line).
                 try:
                     from focus.analysis.realtime import SpectrumAnalyzer
                     from focus.ui.spectrum import SpectrumDisplay
diff --git a/src/focus/ui/spectrum.py b/src/focus/ui/spectrum.py
index 8656d1c..e735a38 100644
--- a/src/focus/ui/spectrum.py
+++ b/src/focus/ui/spectrum.py
@@ -2,7 +2,7 @@
 
 Bars are drawn with Unicode block glyphs (black & white). Rendering uses the
 same in-place multi-row ANSI pattern as :mod:`focus.ui.launcher` (cursor-up +
-erase-to-end each frame), and reuses :meth:`StatusLine._format` for the bottom
+erase-to-end each frame), and reuses :func:`format_status_line` for the bottom
 status row so the status text stays single-sourced.
 """
 
@@ -14,7 +14,7 @@
 import numpy as np
 
 from focus.analysis.realtime import SpectrumAnalyzer
-from focus.ui.transport import PlaybackState, StatusLine
+from focus.ui.transport import PlaybackState, format_status_line
 
 # Glyph ramp indexed by eighths of a cell: 0 = empty, 8 = full block.
 _GLYPHS = np.array(list(" ▁▂▃▄▅▆▇█"))
@@ -87,7 +87,7 @@ def render(self) -> None:
         cols = max(1, size.columns)
         lines = max(1, size.lines)
 
-        status = StatusLine._format(self.state)
+        status = format_status_line(self.state)
         if len(status) > cols - 1:
             status = status[: cols - 1]
 
diff --git a/src/focus/ui/transport.py b/src/focus/ui/transport.py
index ff150a0..f885f82 100644
--- a/src/focus/ui/transport.py
+++ b/src/focus/ui/transport.py
@@ -166,7 +166,7 @@ def start(self) -> None:
     def render(self, state: PlaybackState) -> None:
         if not self._active:
             return
-        line = self._format(state)
+        line = format_status_line(state)
         # Truncate to one less than the terminal width (cosmetic right-edge
         # cleanup). This counts code points, not display columns, so glyphs that
         # render 2 wide (♫, ⏸, ↑↓ on some terminals) can still overflow — hence
@@ -191,20 +191,25 @@ def finish(self) -> None:
             self.stream.flush()
             self._active = False
 
-    @staticmethod
-    def _format(s: PlaybackState) -> str:
-        token = _STATUS_TOKENS.get(s.status, s.status)
-        vol = f"{int(round(s.volume * 100))}%"
-        info = (
-            f"♫ {s.profile_name} · {s.modulation_freq:.0f}Hz @ {s.modulation_depth:.0%}"
-            f"   {token}   {_format_time(s.elapsed_seconds)}"
-            f"   vol {vol}   buf {s.buffer_seconds:.1f}s"
+
+def format_status_line(s: PlaybackState) -> str:
+    """Render the one-line status text shared by the status line and spectrum.
+
+    Public so other displays (e.g. :class:`focus.ui.spectrum.SpectrumDisplay`)
+    can compose it as a row without reaching into private helpers.
+    """
+    token = _STATUS_TOKENS.get(s.status, s.status)
+    vol = f"{int(round(s.volume * 100))}%"
+    info = (
+        f"♫ {s.profile_name} · {s.modulation_freq:.0f}Hz @ {s.modulation_depth:.0%}"
+        f"   {token}   {_format_time(s.elapsed_seconds)}"
+        f"   vol {vol}   buf {s.buffer_seconds:.1f}s"
+    )
+    if s.show_help:
+        hints = (
+            "[space/p] pause  [n] next take  [↑↓ or +/-] volume  "
+            "[?] hide help  [q] quit  [Ctrl+C] stop"
         )
-        if s.show_help:
-            hints = (
-                "[space/p] pause  [n] next take  [↑↓ or +/-] volume  "
-                "[?] hide help  [q] quit  [Ctrl+C] stop"
-            )
-        else:
-            hints = "[space] pause  [n] next  [↑↓] volume  [?] help  [q] quit"
-        return f"{info}   ·   {hints}"
+    else:
+        hints = "[space] pause  [n] next  [↑↓] volume  [?] help  [q] quit"
+    return f"{info}   ·   {hints}"
diff --git a/tests/test_spectrum.py b/tests/test_spectrum.py
index ccece1b..dd31db7 100644
--- a/tests/test_spectrum.py
+++ b/tests/test_spectrum.py
@@ -46,6 +46,20 @@ def test_tone_peaks_in_matching_band(self):
         # A far-away low band should be well below the peak.
         assert v[0] < v[peak] - 0.4
 
+    def test_push_shifts_window_in_place_for_small_chunks(self):
+        # Chunks smaller than fft_size shift the rolling window left in place
+        # (an overlapping slice copy); the newest samples must land at the end.
+        a = SpectrumAnalyzer(fft_size=8)
+        a.push(np.arange(1, 6, dtype=np.float32))  # 5 samples into an 8-window
+        assert np.array_equal(a._buf, [0, 0, 0, 1, 2, 3, 4, 5])
+        a.push(np.arange(6, 9, dtype=np.float32))  # 3 more, oldest fall off
+        assert np.array_equal(a._buf, [1, 2, 3, 4, 5, 6, 7, 8])
+
+    def test_push_keeps_tail_of_oversized_chunk(self):
+        a = SpectrumAnalyzer(fft_size=4)
+        a.push(np.arange(10, dtype=np.float32))
+        assert np.array_equal(a._buf, [6, 7, 8, 9])
+
     def test_silence_is_near_zero(self):
         a = SpectrumAnalyzer()
         a.push_silence()
@@ -110,7 +124,7 @@ def test_render_brackets_and_status_and_cursor_up(self, monkeypatch):
         out = buf.getvalue()
         assert "\x1b[?25l" in out  # cursor hidden on start
         assert "\x1b[?7l" in out and "\x1b[?7h" in out  # autowrap toggled
-        assert "deep-work" in out  # status row reuses StatusLine._format
+        assert "deep-work" in out  # status row reuses format_status_line
         # Second render moves the cursor up over the previously drawn block.
         buf.truncate(0)
         buf.seek(0)
@@ -193,10 +207,41 @@ def test_pause_clears_the_tap(self):
 
 
 class TestCliGate:
-    def test_spectrum_flag_is_noop_in_non_tty(self):
-        # In a non-tty CliRunner context the visualizer path is skipped entirely;
-        # --spectrum must not crash and must not emit the ANSI block.
-        runner = CliRunner()
-        result = runner.invoke(main, ["start", "--mock", "--spectrum", "--duration", "60"])
-        # Duration < minimum is rejected cleanly; the point is no import/setup crash.
+    @staticmethod
+    def _invoke(monkeypatch, args):
+        """Run the CLI with the session stubbed out (no audio device, no stream)."""
+        captured = {}
+
+        async def fake_run_session(profile, use_mock, duration, output_path=None, **kwargs):
+            captured.update(kwargs)
+
+        monkeypatch.setattr("focus.cli._run_session", fake_run_session)
+        result = CliRunner().invoke(main, args)
+        return result, captured
+
+    def test_spectrum_flag_threads_through(self, monkeypatch):
+        result, captured = self._invoke(monkeypatch, ["start", "--mock", "--spectrum"])
+        assert result.exit_code == 0
+        assert captured["spectrum"] is True
+
+    def test_no_spectrum_flag_threads_through(self, monkeypatch):
+        result, captured = self._invoke(monkeypatch, ["start", "--mock", "--no-spectrum"])
+        assert result.exit_code == 0
+        assert captured["spectrum"] is False
+
+    def test_spectrum_defaults_on(self, monkeypatch):
+        _, captured = self._invoke(monkeypatch, ["start", "--mock"])
+        assert captured["spectrum"] is True
+
+    def test_no_ansi_block_in_non_tty(self, monkeypatch):
+        # CliRunner is not a tty, so the visualizer must never hide the cursor
+        # or emit its in-place block into piped output.
+        result, _ = self._invoke(monkeypatch, ["start", "--mock", "--spectrum"])
         assert "\x1b[?25l" not in result.output
+        assert "\x1b[J" not in result.output
+
+    def test_duration_below_minimum_fast_fails(self):
+        # Guards the real fast-fail path; no session (and no audio) is started.
+        result = CliRunner().invoke(main, ["start", "--mock", "--duration", "59"])
+        assert result.exit_code != 0
+        assert "at least 60 seconds" in result.output
diff --git a/tests/test_ui.py b/tests/test_ui.py
index 9632a97..ea2171e 100644
--- a/tests/test_ui.py
+++ b/tests/test_ui.py
@@ -18,7 +18,13 @@
 from focus.cli import main
 from focus.profiles import FocusProfile
 from focus.ui import launcher
-from focus.ui.transport import KeyboardController, PlaybackState, StatusLine, _format_time
+from focus.ui.transport import (
+    KeyboardController,
+    PlaybackState,
+    StatusLine,
+    _format_time,
+    format_status_line,
+)
 
 # The pty-backed tests exercise the raw terminal readers; pty is Unix-only.
 requires_pty = pytest.mark.skipif(
@@ -87,14 +93,14 @@ def test_format_time(self):
 
     def test_format_includes_profile_and_hints(self):
         s = PlaybackState(profile_name="deep-work", modulation_freq=18.0, status="playing")
-        line = StatusLine._format(s)
+        line = format_status_line(s)
         assert "deep-work" in line
         assert "18Hz" in line
         assert "[q] quit" in line
 
     def test_help_toggle_changes_hints(self):
         s = PlaybackState(show_help=True)
-        assert "next take" in StatusLine._format(s)
+        assert "next take" in format_status_line(s)
 
     def test_render_truncates_to_terminal_width(self, monkeypatch):
         # The rendered payload must stay under the terminal width so writing it