harp — Streaming & back-patch dictation (slice A)

Brief

For agentic workers: REQUIRED SUB-SKILL — use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Each step is one 2–5 min action; checkboxes ([ ]) track progress.

Goal: Make harp stream transcription into the focused window in real time, committing a stable prefix and back-patching the volatile tail (delete + retype) as later passes revise it; remove the cloud LLM / command mode entirely.

Architecture: A pure, I/O-free StreamingTranscriber (LocalAgreement-2 over a re-decoded rolling window with buffer trim) feeds a TranscriptState(committed, tail) to an IncrementalTyper that emits the minimal backspace+type diff. The daemon's record path drives the loop.

Tech stack: Python 3.12, faster-whisper (already local), numpy, asyncio, evdev/uinput, pytest + pytest-asyncio, ruff, uv.

Vertical slice 1 = thinnest end-to-end path: pure streaming core + incremental typer wired into the daemon so a faked short utterance produces live, back-patched typed output (Tasks 1–5, end-to-end daemon test green at Task 5). Removal/CLI/tuning follow as Tasks 6–7.

File structure

Task 1 — Streaming core (LocalAgreement-2, no trim yet)

Files: Create src/harp/streaming.py; Test tests/test_streaming.py.

[ ] Step 1 — write failing tests. Create tests/test_streaming.py:

import numpy as np
from harp.streaming import (
    StreamingTranscriber, TranscriptState, longest_common_prefix,
)


def test_longest_common_prefix():
    assert longest_common_prefix("the cat sat", "the cat ran") == "the cat "
    assert longest_common_prefix("", "abc") == ""
    assert longest_common_prefix("abc", "abc") == "abc"


def _audio(seconds=1.0, sr=16000):
    return np.zeros(int(seconds * sr), dtype=np.float32)


def test_agreement_commits_stable_prefix():
    # Scripted hypotheses returned on successive step() calls.
    scripted = iter(["the quick", "the quick brown", "the quick brown fox"])

    def fake(audio, prompt, lang):
        return next(scripted)

    st = StreamingTranscriber(transcribe=fake)
    st.feed(_audio())
    s1 = st.step()                       # first hyp, nothing agreed yet
    assert s1.committed == ""
    assert s1.tail == "the quick"
    st.feed(_audio())
    s2 = st.step()                       # "the quick" agreed across 1&2
    assert s2.committed == "the quick "
    assert s2.tail == "brown"
    st.feed(_audio())
    s3 = st.step()
    assert s3.committed == "the quick brown "
    assert s3.tail == "fox"
    assert s3.full == "the quick brown fox"


def test_finalize_commits_tail():
    scripted = iter(["hello wor", "hello world"])

    def fake(audio, prompt, lang):
        return next(scripted)

    st = StreamingTranscriber(transcribe=fake)
    st.feed(_audio())
    st.step()
    st.feed(_audio())
    st.step()
    final = st.finalize()
    assert final.committed == "hello world"
    assert final.tail == ""


def test_step_on_empty_buffer_is_safe():
    st = StreamingTranscriber(transcribe=lambda a, p, l: "x")
    s = st.step()
    assert s == TranscriptState("", "")

[ ] Step 2 — run, expect fail. Run: cd repos/harp && uv run pytest tests/test_streaming.py -q. Expected: collection/import error (harp.streaming missing).

[ ] Step 3 — implement. Create src/harp/streaming.py:

"""Pure, I/O-free streaming transcription core (LocalAgreement-2)."""

from dataclasses import dataclass
from typing import Callable, Optional

import numpy as np

TranscribeFn = Callable[[np.ndarray, Optional[str], Optional[str]], str]


@dataclass(frozen=True)
class TranscriptState:
    """Immutable snapshot: committed text never changes; tail may be rewritten."""

    committed: str
    tail: str

    @property
    def full(self) -> str:
        return (self.committed + self.tail).strip()


def longest_common_prefix(a: str, b: str) -> str:
    """Character-level longest common prefix of two strings."""
    n = 0
    for ca, cb in zip(a, b):
        if ca != cb:
            break
        n += 1
    return a[:n]


class StreamingTranscriber:
    """
    Re-decodes a rolling audio window every step() and commits the longest
    word-aligned prefix that agrees across two successive hypotheses
    (LocalAgreement-2). Buffer trimming is added in Task 2.
    """

    def __init__(
        self,
        transcribe: TranscribeFn,
        samplerate: int = 16000,
        window: float = 30.0,
        overlap: float = 5.0,
        language: Optional[str] = None,
    ) -> None:
        self._transcribe = transcribe
        self._sr = samplerate
        self._window = window
        self._overlap = overlap
        self._language = language
        self._buf = np.zeros(0, dtype=np.float32)
        self._committed = ""
        self._prev_hyp = ""  # hypothesis text *after* the committed prefix

    def feed(self, pcm: np.ndarray) -> None:
        self._buf = np.concatenate(
            [self._buf, np.asarray(pcm, dtype=np.float32).flatten()]
        )

    def _window_audio(self) -> np.ndarray:
        max_samples = int(self._window * self._sr)
        if self._buf.shape[0] <= max_samples:
            return self._buf
        return self._buf[-max_samples:]

    def _decode(self) -> str:
        audio = self._window_audio()
        if audio.size == 0:
            return ""
        prompt = self._committed[-200:] or None
        return self._transcribe(audio, prompt, self._language).strip()

    @staticmethod
    def _word_prefix(text: str) -> str:
        """Trim a string back to its last whole-word (space) boundary."""
        i = text.rfind(" ")
        return text[: i + 1] if i != -1 else ""

    def step(self) -> TranscriptState:
        if self._buf.size == 0:
            return TranscriptState(self._committed, "")
        hyp = self._decode()
        agreed = self._word_prefix(longest_common_prefix(self._prev_hyp, hyp))
        if agreed:
            self._committed += agreed
            hyp = hyp[len(agreed):]
        self._prev_hyp = hyp
        return TranscriptState(self._committed, hyp)

    def finalize(self) -> TranscriptState:
        """End of session: force-commit whatever the last decode yields."""
        if self._buf.size > 0:
            self._committed += self._decode()
        self._prev_hyp = ""
        self._buf = np.zeros(0, dtype=np.float32)
        return TranscriptState(self._committed, "")

Note on test_finalize_commits_tail: after the two step() calls, "hello " is committed and finalize() re-decodes (scripted iterator exhausted → re-raises StopIteration only if called again; it is not — finalize calls _decode once, consuming the 2nd scripted value is already done, so provide a 3rd). Fix the test fixture: use scripted = iter(["hello wor", "hello world", "hello world"]) so finalize's decode returns the final string and committed == "hello world" (the "hello " prefix plus "world" tail re-decoded as the full string — assert final.committed.replace(" ", " ").strip() == "hello world"). Apply this corrected fixture and assertion in Step 1.

[ ] Step 4 — run, expect pass. Run: uv run pytest tests/test_streaming.py -q. Expected: 4 passed.

[ ] Step 5 — commit.

git add src/harp/streaming.py tests/test_streaming.py
git commit -m "feat(streaming): LocalAgreement-2 core (no trim yet)"

Task 2 — Buffer trim / long-session cap

Files: Modify src/harp/streaming.py; Test tests/test_streaming.py.

Without trimming, a 2-hour session re-decodes an ever-growing buffer. After a commit, drop audio older than (committed_audio − overlap). We track committed audio by counting samples consumed: when the buffer exceeds window seconds and a commit just happened, hard-trim to the last overlap seconds and rely on the initial_prompt (already self._committed[-200:]) for cross-seam continuity.

[ ] Step 1 — add failing test to tests/test_streaming.py:

def test_buffer_trims_after_commit_when_over_window():
    scripted = iter(["alpha beta", "alpha beta gamma", "alpha beta gamma delta"])

    def fake(audio, prompt, lang):
        return next(scripted)

    st = StreamingTranscriber(transcribe=fake, window=1.0, overlap=0.25)
    st.feed(_audio(seconds=2.0))      # 2s > window=1s
    st.step()
    st.feed(_audio(seconds=2.0))
    st.step()                         # commit happens -> trim fires
    assert st._buf.shape[0] == int(0.25 * 16000)  # trimmed to overlap

[ ] Step 2 — run, expect fail. Run: uv run pytest tests/test_streaming.py::test_buffer_trims_after_commit_when_over_window -q. Expected: FAIL (_buf not trimmed).

[ ] Step 3 — implement. Add the trim method and call it from step() right after a commit. Replace the step method body's commit branch:

    def _maybe_trim(self) -> None:
        max_samples = int(self._window * self._sr)
        if self._buf.shape[0] <= max_samples:
            return
        keep = int(self._overlap * self._sr)
        self._buf = self._buf[-keep:] if keep > 0 else np.zeros(
            0, dtype=np.float32
        )
        # Hypothesis no longer aligns to the trimmed buffer; reset agreement.
        self._prev_hyp = ""

    def step(self) -> TranscriptState:
        if self._buf.size == 0:
            return TranscriptState(self._committed, "")
        hyp = self._decode()
        agreed = self._word_prefix(longest_common_prefix(self._prev_hyp, hyp))
        if agreed:
            self._committed += agreed
            hyp = hyp[len(agreed):]
            self._prev_hyp = hyp
            self._maybe_trim()
        else:
            self._prev_hyp = hyp
        return TranscriptState(self._committed, hyp)

(Delete the old step definition; this replaces it. The _prev_hyp reset on trim accepts one missed agreement cycle at the seam in exchange for a bounded buffer — correct for long sessions.)

[ ] Step 4 — run full core suite, expect pass. Run: uv run pytest tests/test_streaming.py -q. Expected: all passed (5).

[ ] Step 5 — commit.

git add src/harp/streaming.py tests/test_streaming.py
git commit -m "feat(streaming): bounded buffer trim for long sessions"

Task 3 — IncrementalTyper (back-patch diff)

Files: Modify src/harp/input.py (append class); Test tests/test_incremental_typer.py.

[ ] Step 1 — write failing tests. Create tests/test_incremental_typer.py:

from harp.input import IncrementalTyper


class FakeTyper:
    """Records ops; emulates a screen buffer for assertions."""

    def __init__(self):
        self.screen = ""

    def backspace(self, count: int) -> None:
        if count > 0:
            self.screen = self.screen[:-count]

    def type_text(self, text: str) -> None:
        self.screen += text

    def filter_text(self, text: str) -> str:
        return text


def test_append_only_growth():
    f = FakeTyper()
    it = IncrementalTyper(f)
    it.update("the quick")
    it.update("the quick brown")
    assert f.screen == "the quick brown"


def test_backpatch_rewrites_divergent_tail():
    f = FakeTyper()
    it = IncrementalTyper(f)
    it.update("the kwik brown")
    it.update("the quick brown")            # 'kwik' -> 'quick'
    assert f.screen == "the quick brown"


def test_minimal_diff_preserves_common_prefix():
    f = FakeTyper()
    it = IncrementalTyper(f)
    it.update("hello world")
    f.type_text("")  # no-op marker
    ops = []
    f.backspace = lambda c, ops=ops: ops.append(("bs", c)) or None
    f.type_text = lambda t, ops=ops: ops.append(("tt", t)) or None
    it.update("hello there")
    # Only "world" (5) backspaced, "there" typed -- prefix "hello " kept.
    assert ops == [("bs", 5), ("tt", "there")]


def test_backspace_cap_falls_back_to_append_only():
    f = FakeTyper()
    it = IncrementalTyper(f, max_backspaces=3)
    it.update("aaaa bbbb cccc")
    it.update("xxxx")                       # would need huge rewrite
    # Cap exceeded -> append-only: original text kept, new appended.
    assert f.screen == "aaaa bbbb ccccxxxx"
    assert it.append_only is True


def test_pause_defers_until_safe():
    f = FakeTyper()
    paused = {"v": True}
    it = IncrementalTyper(f, is_paused=lambda: paused["v"])
    it.update("typed while user busy")
    assert f.screen == ""                   # deferred
    paused["v"] = False
    it.update("typed while user busy now")
    assert f.screen == "typed while user busy now"

[ ] Step 2 — run, expect fail. Run: uv run pytest tests/test_incremental_typer.py -q. Expected: ImportError (IncrementalTyper missing).

[ ] Step 3 — implement. Append to src/harp/input.py (after WaylandTyper):

from typing import Callable, Optional

from harp.streaming import longest_common_prefix


class IncrementalTyper:
    """
    Emits the minimal backspace+type diff to make the focused window match a
    target string, preserving the common prefix already on screen. Defers
    while the user is typing; falls back to append-only past a backspace cap.
    """

    def __init__(
        self,
        typer,
        max_backspaces: int = 80,
        is_paused: Optional[Callable[[], bool]] = None,
    ) -> None:
        self._typer = typer
        self._max_backspaces = max_backspaces
        self._is_paused = is_paused or (lambda: False)
        self._rendered = ""
        self._pending: Optional[str] = None
        self.append_only = False

    def update(self, target: str) -> None:
        target = self._typer.filter_text(target)
        if self._is_paused():
            self._pending = target
            return
        if self._pending is not None:
            self._pending = None
        self._reconcile(target)

    def _reconcile(self, target: str) -> None:
        if target == self._rendered:
            return
        if self.append_only:
            if target.startswith(self._rendered):
                self._typer.type_text(target[len(self._rendered):])
                self._rendered = target
            else:
                extra = target[len(longest_common_prefix(self._rendered, target)):]
                self._typer.type_text(extra)
                self._rendered += extra
            return
        common = longest_common_prefix(self._rendered, target)
        to_delete = len(self._rendered) - len(common)
        if to_delete > self._max_backspaces:
            self.append_only = True
            self._typer.type_text(target[len(self._rendered):]
                                  if target.startswith(self._rendered) else target)
            self._rendered = self._rendered + (
                target[len(self._rendered):] if target.startswith(self._rendered)
                else target)
            return
        if to_delete > 0:
            self._typer.backspace(to_delete)
        self._typer.type_text(target[len(common):])
        self._rendered = target

    def flush(self) -> None:
        """Apply any deferred update (called when pause clears)."""
        if self._pending is not None and not self._is_paused():
            t, self._pending = self._pending, None
            self._reconcile(t)

The deferred-update path: update() while paused stores _pending; the next non-paused update() reconciles directly to the newest target (newer supersedes the stale pending — correct). flush() exists for the daemon to drain a pending update when the pause clears without a new transcription.

[ ] Step 4 — run, expect pass. Run: uv run pytest tests/test_incremental_typer.py -q. Expected: 5 passed.

[ ] Step 5 — commit.

git add src/harp/input.py tests/test_incremental_typer.py
git commit -m "feat(input): IncrementalTyper back-patch diff with cap + pause defer"

Task 4 — Config: add stream knobs, remove cloud/command fields

Files: Modify src/harp/config.py; the existing tests/ have no dedicated config test — verification is via test_main.py (Task 6) and the daemon tests (Task 5).

[ ] Step 1 — edit config.py. In HarpConfig: delete the entire LLM (Post-processing) Settings block (llm_api_key, llm_base_url, llm_model), the send_clipboard field, and the command_prompt field. Replace the STT Behavior block:

# STT Behavior (real-time streaming)
stream_window: float = Field(
    default=30.0, description="Rolling re-decode window in seconds"
)
stream_overlap: float = Field(
    default=5.0, description="Audio retained across a buffer trim, seconds"
)
stream_slide_interval: float = Field(
    default=1.0,
    description="Seconds between streaming re-decode passes (tuned per model)",
)

(Remove continuous, stt_min_chunk_size, stt_slide_interval, stt_overlap — superseded. Keep all local_*, type_result, copy_result, toggle, full_mode, device.)

[ ] Step 2 — verify it imports & defaults resolve. Run: uv run python -c "from harp.config import HarpConfig; c=HarpConfig(); print(c.stream_slide_interval, c.copy_result)". Expected: 1.0 False, no error.

[ ] Step 3 — commit.

git add src/harp/config.py
git commit -m "feat(config): stream_* knobs; drop llm/command/continuous config"

Task 5 — Daemon: streaming loop + remove command mode (vertical slice complete)

Files: Modify src/harp/daemon.py; delete src/harp/api.py; delete tests/test_api.py; modify tests/test_daemon_async.py, tests/test_daemon.py.

[ ] Step 1 — rewrite the failing/old async tests first. In tests/test_daemon_async.py: (a) remove patch("harp.daemon.LLMClient") from the async_daemon fixture; (b) delete test_stop_recording_command_mode and test_background_transcription_loop_continuous entirely; (c) replace test_stop_recording_success with the streaming-loop test below; (d) add the error-resilience test:

@pytest.mark.asyncio
async def test_streaming_session_types_committed_text(async_daemon):
    async_daemon.config.type_result = True
    scripted = iter(["the cat", "the cat sat", "the cat sat down"])
    async_daemon.whisper_engine.transcribe.side_effect = (
        lambda *a, **k: next(scripted)
    )
    async_daemon.typer.filter_text.side_effect = lambda x: x
    chunk = np.zeros(16000, dtype=np.float32)
    async_daemon.audio_streamer.get_current_buffer.return_value = chunk
    async_daemon.audio_streamer.stop_recording.return_value = chunk

    async_daemon._start_recording()
    for _ in range(3):
        await async_daemon._stream_tick()
    await async_daemon._stop_recording()

    assert async_daemon.state == DaemonState.IDLE
    typed = "".join(
        c.args[0] for c in async_daemon.typer.type_text.call_args_list
    )
    assert "the cat sat down" in typed.replace("  ", " ")


@pytest.mark.asyncio
async def test_streaming_tick_survives_transcribe_error(async_daemon):
    async_daemon.config.type_result = True
    async_daemon.whisper_engine.transcribe.side_effect = RuntimeError("boom")
    async_daemon.typer.filter_text.side_effect = lambda x: x
    async_daemon.audio_streamer.get_current_buffer.return_value = np.zeros(
        16000, dtype=np.float32
    )
    async_daemon._start_recording()
    await async_daemon._stream_tick()        # must not raise
    assert async_daemon.state == DaemonState.RECORDING

[ ] Step 2 — run, expect fail. Run: uv run pytest tests/test_daemon_async.py -q. Expected: FAIL (_stream_tick undefined; harp.daemon.LLMClient patch removed but still imported).

[ ] Step 3 — rewrite daemon.py. Apply these exact changes:

• Remove imports: from harp.api import LLMClient, import pyperclip (only used by clipboard-context), from rich.panel import Panel if unused after edits (keep if still used for the result panel — keep it), import re (used only by _get_clipboard_context — remove).
• Add imports: from harp.streaming import StreamingTranscriber, from harp.input import IncrementalTyper.
• In __init__: delete the self.llm_client = LLMClient(...) block, self._is_command_mode, self._transcription_task, self._latest_transcription. Add: self._transcriber: StreamingTranscriber | None = None, self._inc_typer = IncrementalTyper(self.typer, is_paused=lambda: self.pause_typing), self._stream_task: asyncio.Task | None = None.
• Delete methods _get_clipboard_context and _background_transcription_loop entirely.
• Replace _start_recording, _stop_recording, and add _stream_tick / _stream_loop with:

    def _start_recording(self) -> None:
        if self.state != DaemonState.IDLE:
            return
        self.state = DaemonState.RECORDING
        self._play_chime(start=True)
        self.audio_streamer.start_recording()
        self._transcriber = StreamingTranscriber(
            transcribe=self.whisper_engine.transcribe,
            window=self.config.stream_window,
            overlap=self.config.stream_overlap,
            language=self.config.local_language,
        )
        self._inc_typer = IncrementalTyper(
            self.typer, is_paused=lambda: self.pause_typing
        )
        self._last_consumed = 0
        self._stream_task = asyncio.create_task(self._stream_loop())
        self.console.print("[bold green]capturing...[/]")
        self._notify("Status", "capturing")

    async def _stream_tick(self) -> None:
        if self._transcriber is None:
            return
        buf = self.audio_streamer.get_current_buffer().flatten()
        new = buf[self._last_consumed:]
        self._last_consumed = buf.shape[0]
        if new.size:
            self._transcriber.feed(new)
        try:
            loop = asyncio.get_running_loop()
            state = await loop.run_in_executor(None, self._transcriber.step)
        except Exception:
            return  # skip this tick, keep buffer (resilient)
        if self.config.type_result:
            self._inc_typer.update(state.full)

    async def _stream_loop(self) -> None:
        try:
            while self.state == DaemonState.RECORDING:
                await asyncio.sleep(self.config.stream_slide_interval)
                await self._stream_tick()
        except asyncio.CancelledError:
            pass

    async def _stop_recording(self) -> None:
        if self.state != DaemonState.RECORDING:
            return
        self.state = DaemonState.PROCESSING
        self._play_chime(start=False)
        if self._stream_task:
            self._stream_task.cancel()
        audio = self.audio_streamer.stop_recording().flatten()
        if self._transcriber is not None:
            extra = audio[self._last_consumed:]
            if extra.size:
                self._transcriber.feed(extra)
            try:
                loop = asyncio.get_running_loop()
                final = await loop.run_in_executor(
                    None, self._transcriber.finalize
                )
            except Exception as e:
                self.console.print(f"[bold red]Finalize failed: {e}[/]")
                final = None
            if final is not None:
                text = final.committed.strip()
                self.console.print(
                    Panel(f"[italic green]{text}[/]",
                          title="[bold cyan]Result[/]",
                          border_style="cyan")
                )
                self._notify("Transcription Ready", text)
                if self.config.copy_result and text:
                    try:
                        import pyperclip
                        pyperclip.copy(text)
                    except Exception as e:
                        self.console.print(f"[yellow]Copy failed: {e}[/]")
                if self.config.type_result:
                    self._release_modifiers()
                    self._inc_typer.update(text)
        self.state = DaemonState.IDLE
        self._notify("Status", "idle")

• In _handle_events: delete the if self.state == DaemonState.IDLE: self._is_command_mode = is_shift line and the four shift-suppression add lines tied to command mode only if shift is no longer meaningful — keep shift suppression for the hotkey itself; just remove the _is_command_mode assignment. No other event-loop change.
• Delete src/harp/api.py and tests/test_api.py: git rm src/harp/api.py tests/test_api.py.
• In tests/test_daemon.py line ~22: remove the patch("harp.daemon.LLMClient"), line from its fixture.

[ ] Step 4 — run daemon suites, expect pass. Run: uv run pytest tests/test_daemon_async.py tests/test_daemon.py -q. Expected: all passed (no LLMClient/api import errors; streaming + error-resilience tests green). Fix any residual _is_command_mode / pyperclip / re reference the compiler flags.

[ ] Step 5 — commit (vertical slice 1 complete).

git add -A src/harp/daemon.py tests/test_daemon_async.py tests/test_daemon.py
git rm src/harp/api.py tests/test_api.py
git commit -m "feat(daemon): real-time streaming loop; remove command mode + LLM"

Task 6 — CLI surface cleanup

Files: Modify src/harp/__main__.py, tests/test_main.py.

[ ] Step 1 — fix the failing CLI test first. In tests/test_main.py test_cli_start_custom: remove the "--send-clipboard", "1000", and "--continuous", argv entries and the assert overrides["send_clipboard"] == 1000 / assert overrides["continuous"] is True lines. In test_cli_config_command: change mock_config.model_dump.return_value = {"llm_api_key": "test"} / assertion to a surviving field, e.g. {"stream_slide_interval": 1.0} and assert "stream_slide_interval: 1.0" in result.output.

[ ] Step 2 — run, expect fail. Run: uv run pytest tests/test_main.py -q. Expected: FAIL (CLI still defines/forwards removed options; assertions mismatch).

[ ] Step 3 — edit __main__.py. In run_daemon and the start command: remove the send_clipboard, command_prompt, continuous parameters, their typer.Option definitions, and their entries in the overrides dict. Add stream options:

stream_slide_interval: Optional[float] = typer.Option(
    None, "--slide", help="Seconds between streaming re-decode passes"
),

and thread it through run_daemon into overrides["stream_slide_interval"]. Leave --type, --copy, --device, --toggle, --full, --local-*, --language untouched.

[ ] Step 4 — run, expect pass. Run: uv run pytest tests/test_main.py -q. Expected: all passed.

[ ] Step 5 — commit.

git add src/harp/__main__.py tests/test_main.py
git commit -m "feat(cli): drop command/clipboard/continuous flags; add --slide"

Task 7 — Dependency cleanup, full verification, empirical tuning, docs

Files: pyproject.toml, docs/cli.md, README.md, CHANGELOG.md, TASKS.md.

[ ] Step 1 — remove the dependency. In pyproject.toml delete the "openai>=1.65.0", line from dependencies. Run uv lock to refresh uv.lock.

[ ] Step 2 — full suite + lint, expect green. Run: uv run pytest -q then uv run ruff check src tests. Expected: all tests pass; ruff clean. Also assert removal: grep -rn "openai\|LLMClient\|_is_command_mode" src returns nothing.

[ ] Step 3 — empirical cadence tuning. Run the daemon live for short and ~2-min dictations at each available model size; measure single-window decode wall-time (small int8 CPU vs medium, and CUDA if present). Set stream_slide_interval default so it exceeds measured decode time with headroom (rule: slide ≥ 1.3 × decode). Record the chosen number in config.py and the rationale in a one-paragraph note appended to this plan's Tuning results details block. If CPU small decode > ~2s, set the shipped default for the bundled model accordingly and document the CUDA-recommended value.

[ ] Step 4 — integration test (real WAV). Add to tests/test_integration_end_to_end.py a @pytest.mark.integration test that feeds tests/assets/ground_truth.wav through a real StreamingTranscriber (real LocalWhisperEngine.transcribe, smallest model) in ~1s slices and asserts the finalized text fuzzy-matches tests/assets/ground_truth.txt (reuse the existing fuzzy-match helper / fuzzywuzzy ratio > 80). Run: uv run pytest -q -m integration. Expected: pass.

[ ] Step 5 — docs & task ledger. Update docs/cli.md (remove command-mode / --send-clipboard / --continuous; document streaming behavior + --slide), README.md (reframe as local-first real-time dictation engine; drop LLM/command sections), CHANGELOG.md (new ## [Unreleased] entry: streaming + back-patch, removal of cloud LLM/command mode — BREAKING), and TASKS.md (move "Add post-processing hooks (#9)" reasoning if affected; add done entry referencing this plan; note slices B/D remain). Commit:

git add pyproject.toml uv.lock docs/ README.md CHANGELOG.md TASKS.md tests/
git commit -m "chore: drop openai dep; docs + changelog for streaming dictation"

Self-review

Spec coverage: stable-prefix/volatile-tail → T1+T2; LocalAgreement-2 → T1; buffer trim for 2h sessions → T2; IncrementalTyper diff/cap/pause → T3; drop command mode+LLM+openai+config → T4/T5/T6/T7; single streaming mode/Ctrl+Space → T5; I/O-free reusable core → T1 (no daemon imports in streaming.py); empirical cadence → T7 S3; error handling (decode error skip, coalesce via single in-flight tick, pause defer, backspace cap) → T3+T5; testing inline → tasks own their tests; non-goals (B/D) untouched. No gaps.

Placeholder scan: every code step has complete code; the only deferred values are the empirically-tuned cadence numbers, which the spec explicitly designates as measure-don't-guess (T7 S3 gives the exact rule and where to record them) — not a placeholder.

Type/name consistency: TranscriptState(committed, tail), .full, StreamingTranscriber.feed/step/finalize, IncrementalTyper(typer, max_backspaces, is_paused)/.update/.flush/.append_only, daemon _stream_tick/_stream_loop/_last_consumed, config stream_window/stream_overlap/stream_slide_interval — used identically across T1–T7.

Note carried into execution: T1 Step 3 flags a fixture correction for test_finalize_commits_tail (3-element scripted iterator + normalized assertion); apply as written.