VoiceAgent/vocabulary.py

"""Tell the recognizer which words to expect, and fix the ones it still gets wrong.

Two mechanisms, deliberately separate, because they fail differently:

- **Biasing.** `SFSpeechRecognitionRequest.contextualStrings` nudges the decoder
  towards terms it would otherwise never produce. Measured at 23.5% -> 16.5%
  word error rate on technical speech. It is a hint, so it fails softly.
- **Repair.** Deterministic substitutions applied after recognition, for
  mistakes that recur identically ("the coral voice" is always Kokoro). Exact
  and testable, but blind to context, so the rules must be specific.

Relevance beats coverage, and by more than expected. Eighteen apt terms measured
16.5% WER; padding them out to 100 with names harvested from the project put it
back to 23.5%, i.e. no better than no biasing at all. Padding with a thousand
*random dictionary* words cost only 1.1 points, so the damage is not volume —
it is that filenames and identifiers ("bot", "plan", "hack") are ordinary words
the recognizer would happily produce anyway, and biasing towards them drags
real speech onto them.

So auto-discovered terms are filtered to ones that are *not* ordinary English,
capped tightly, and always ranked behind the hand-written list. The curated file
is what carries the benefit; discovery is a small bonus with a real downside.
"""

import re
import subprocess
from collections import OrderedDict
from pathlib import Path

from loguru import logger

# Apple documents a ceiling of 100, but measurement says stay well under it:
# quality decays fast once ordinary words get in.
MAX_TERMS = 40

# Discovered terms are the risky kind, so they get a small share of the budget.
MAX_DISCOVERED = 12

_SYSTEM_DICTIONARY = Path("/usr/share/dict/words")

# Words that are never worth biasing towards and only crowd out real terms.
_STOPWORDS = {
    "and", "are", "but", "for", "from", "has", "have", "into", "not", "our",
    "out", "the", "that", "this", "was", "were", "will", "with", "you",
    "your", "main", "init", "self", "test", "tests", "src", "lib", "util",
    "utils", "readme", "true", "false", "none", "null", "class", "def",
}

_IDENTIFIER = re.compile(r"^[A-Za-z][A-Za-z0-9_.-]*$")
_CAMEL_BOUNDARY = re.compile(r"(?<=[a-z0-9])(?=[A-Z])")
# Things worth learning from Claude's replies: CamelCase, dotted paths,
# snake_case, and words carrying digits. Ordinary prose is skipped.
_TECHNICAL = re.compile(r"\b(?:[A-Za-z]+[._][A-Za-z0-9._-]+|[a-z]+[A-Z][A-Za-z]*|[A-Za-z]*\d[A-Za-z0-9]*)\b")

_SKIP_DIRS = {".git", ".venv", "__pycache__", "node_modules", ".cache", "dist", "build"}


def _speakable(term: str) -> str:
    """Turn an identifier into something a person could say.

    Apple's guidance is one or two words per phrase, spoken without a pause, so
    `sounddevice_transport` is useless as-is but `sounddevice transport` is not.
    """
    term = term.replace("_", " ").replace("-", " ").replace(".", " ")
    term = _CAMEL_BOUNDARY.sub(" ", term)
    return " ".join(term.split())


class Vocabulary:
    """Ranked terms for the recognizer, plus repair rules for its known mistakes.

    Terms are ordered by how likely they are to matter: words the user wrote
    down, then the project around them, then whatever Claude has been talking
    about lately. Only the first ``limit`` survive.
    """

    def __init__(
        self,
        *,
        project_dir: Path | None = None,
        vocabulary_file: Path | None = None,
        corrections_file: Path | None = None,
        limit: int = MAX_TERMS,
    ):
        self._limit = limit
        self._user: list[str] = []
        self._project: list[str] = []
        self._observed: OrderedDict[str, None] = OrderedDict()
        self._corrections: list[tuple[re.Pattern, str]] = []

        if vocabulary_file and vocabulary_file.exists():
            self._user = _read_terms(vocabulary_file)
            logger.debug(f"Vocabulary: {len(self._user)} terms from {vocabulary_file.name}")
        if corrections_file and corrections_file.exists():
            self._corrections = _read_corrections(corrections_file)
            logger.debug(f"Vocabulary: {len(self._corrections)} repair rules")
        if project_dir:
            self._project = _terms_from_project(project_dir)

    def add_terms(self, terms: list[str]):
        """Add terms that rank alongside the hand-written ones.

        For names the user certainly says — project names out of the brain —
        which are as authoritative as anything typed into vocabulary.txt.
        """
        for term in terms:
            speakable = _speakable(term)
            if _worth_keeping(speakable) and speakable not in self._user:
                self._user.append(speakable)

    def observe(self, text: str):
        """Learn technical words from something Claude just said.

        What the assistant is discussing is a good predictor of what the user is
        about to say back, so its replies feed the next recognition.
        """
        for match in _TECHNICAL.findall(text or ""):
            term = _speakable(match)
            if _worth_keeping(term) and _is_distinctive(term):
                self._observed.pop(term, None)  # move to most-recent
                self._observed[term] = None
        # Keep the tail bounded; only the newest can reach the term list anyway.
        while len(self._observed) > self._limit * 2:
            self._observed.popitem(last=False)

    def terms(self) -> list[str]:
        """The ranked, de-duplicated, capped term list.

        What Claude said a moment ago predicts the next utterance better than an
        arbitrary filename does, so recent observations get a reserved share of
        the budget rather than queueing behind the whole project.
        """
        budget = max(0, self._limit - len(self._user))
        discovered = min(budget, MAX_DISCOVERED)
        recent = list(reversed(self._observed))[: (discovered + 1) // 2]
        project = self._project[: discovered - len(recent)]
        ranked = [*self._user, *recent, *project]

        seen: dict[str, str] = {}
        for term in ranked:
            key = term.lower()
            if key not in seen:
                seen[key] = term
            if len(seen) >= self._limit:
                break
        return list(seen.values())

    def prompt_block(self, limit: int = 60) -> str:
        """The stable terms, for Claude's system prompt.

        Only the written-down and project-derived terms go here: the prompt is
        fixed for the session, so terms learned mid-conversation could not
        appear anyway.
        """
        stable = list(dict.fromkeys([*self._user, *self._project]))[:limit]
        return ", ".join(stable)

    def repair(self, text: str) -> str:
        """Apply the substitution rules to a transcript."""
        for pattern, replacement in self._corrections:
            text = pattern.sub(replacement, text)
        return text


def _worth_keeping(term: str) -> bool:
    return (
        len(term) >= 3
        and term.lower() not in _STOPWORDS
        and not term.isdigit()
        and len(term) <= 40
    )


def _load_dictionary() -> frozenset[str]:
    try:
        return frozenset(w.strip().lower() for w in _SYSTEM_DICTIONARY.read_text().splitlines())
    except OSError:
        return frozenset()


_ORDINARY_WORDS = _load_dictionary()


def _is_distinctive(term: str) -> bool:
    """Whether biasing towards this term could actually change an outcome.

    Only words the recognizer would not already produce are worth boosting.
    Biasing towards ordinary English measurably drags correct speech onto the
    wrong word, which is how a harvested term list erased the entire benefit.
    """
    words = term.lower().split()
    if len(words) > 1:
        return True  # multi-word phrases are specific enough to be safe
    return words[0] not in _ORDINARY_WORDS


def _read_terms(path: Path) -> list[str]:
    terms = []
    for line in path.read_text().splitlines():
        line = line.split("#", 1)[0].strip()
        if line and _worth_keeping(line):
            terms.append(line)
    return terms


def _read_corrections(path: Path) -> list[tuple[re.Pattern, str]]:
    rules = []
    for lineno, raw in enumerate(path.read_text().splitlines(), 1):
        line = raw.split("#", 1)[0].strip()
        if not line:
            continue
        if "=>" not in line:
            logger.warning(f"{path.name}:{lineno}: expected 'heard => replacement', got {raw!r}")
            continue
        heard, replacement = (part.strip() for part in line.split("=>", 1))
        if not heard:
            continue
        # Word-bounded and case-insensitive so "coral voice" matches mid-sentence
        # but "chorale" never does.
        rules.append((re.compile(rf"\b{re.escape(heard)}\b", re.IGNORECASE), replacement))
    return rules


def _terms_from_project(project_dir: Path) -> list[str]:
    """Names from the code the user is most likely to talk about."""
    terms: list[str] = []

    for path in sorted(project_dir.rglob("*")):
        if any(part in _SKIP_DIRS for part in path.parts):
            continue
        if path.is_file():
            term = _speakable(path.stem)
            if _worth_keeping(term) and _is_distinctive(term):
                terms.append(term)

    terms.extend(_git_terms(project_dir))

    for path in sorted(project_dir.glob("*.py")):
        try:
            source = path.read_text(errors="ignore")
        except OSError:
            continue
        for name in re.findall(r"^\s*(?:class|def)\s+([A-Za-z_][A-Za-z0-9_]*)", source, re.M):
            term = _speakable(name)
            if _worth_keeping(term) and _is_distinctive(term):
                terms.append(term)

    return list(dict.fromkeys(terms))


def _git_terms(project_dir: Path) -> list[str]:
    """Branch and author names, which are spoken far more often than they're typed."""
    terms: list[str] = []
    commands = (
        ["git", "branch", "--format=%(refname:short)"],
        ["git", "log", "-40", "--format=%an"],
    )
    for command in commands:
        try:
            result = subprocess.run(
                command, cwd=project_dir, capture_output=True, text=True, timeout=5
            )
        except (OSError, subprocess.SubprocessError):
            continue
        if result.returncode != 0:
            continue
        for line in result.stdout.splitlines():
            for piece in line.split():
                term = _speakable(piece)
                if _worth_keeping(term) and _is_distinctive(term):
                    terms.append(term)
    return terms