Drop the voice pipeline

The assistant becomes a text daemon behind the quickshell canvas. STT,
TTS, the mic button loop, the presentation gate, the audio levels and the
overlay config table go; the socket, the tool runner, the never-swap model
rule and the config merge stay. The reply loop goes with the voice and
returns with the text protocol. The voice version is tagged voice-v1, so
later voice work starts from the tag instead of from dormant code.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

1 files changed, 33 insertions, 463 deletions

diff --git a/assistant.py b/assistant.py
index a6d83f4..b69f199 100644
--- a/assistant.py
+++ b/assistant.py
@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-# desktop-assistant: push-to-talk local voice assistant.
+# desktop-assistant: local text assistant daemon.
 # Copyright (C) 2026 Danilo M. <danix@danix.xyz>
 #
 # This program is free software; you can redistribute it and/or modify it
@@ -14,62 +14,46 @@
 # You should have received a copy of the GNU General Public License along
 # with this program; if not, write to the Free Software Foundation, Inc.,
 # 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
-"""Mic button on: record. Mic button off: transcribe (whisper-server),
-ask the loaded llama-server model, speak the reply (Kokoro).
+"""Questions arrive as JSON lines on a Unix socket, the loaded llama-server
+model answers with tools, replies stream back on the same socket, and every
+finished turn is kept in a SQLite history with full-text search.
 
 Run: python3 assistant.py        (--test runs the self-check)
 """
 
-import array
-import ctypes
 import glob
 import json
-import math
 import os
-import queue
-import random
 import re
 import signal
 import socket
+import sqlite3
 import subprocess
 import sys
 import threading
 import time
 import tomllib
-import unicodedata
+import traceback
 import urllib.parse
 import urllib.request
-import wave
 
 CONFIG_DIR = os.path.join(os.environ.get("XDG_CONFIG_HOME") or os.path.expanduser("~/.config"),
                           "desktop-assistant")
-RAW = os.path.join(os.environ.get("XDG_RUNTIME_DIR", "/tmp"), "desktop-assistant.raw")
-WAV = RAW[:-3] + "wav"
+DB_PATH = os.path.join(os.environ.get("XDG_DATA_HOME") or os.path.expanduser("~/.local/share"),
+                       "desktop-assistant", "history.db")
 DEFAULTS = {
-    "card": "Microphone",                 # ALSA card id of the USB mic
     "llm": "http://localhost:8181/v1",
-    "system": ("You are a voice assistant. Your replies are spoken aloud, so answer "
-               "briefly in plain sentences: no markdown, no lists, no emoji. "
-               "Reply in the language the user speaks, Italian or English. "
-               "Write numbers, times and dates in digits, never spelled out. "
-               "The user's words come from speech recognition and may contain "
-               "misheard words: read them by how they sound. "
-               "Facts about the user's computer, devices, accounts, mail, calendar "
-               "or money come only from a tool: call one, and if none can tell, "
-               "say you do not know. Never guess them."),
+    "system": ("You are a desktop assistant. Answer in the language the user writes, "
+               "Italian or English. Markdown is rendered: use lists, tables and fenced "
+               "code blocks when they help, and keep answers as short as the question "
+               "allows. Facts about the user's computer, devices, accounts, mail, "
+               "calendar or money come only from a tool: call one, and if none can "
+               "tell, say you do not know. Never guess them."),
     "tool_rounds": 3,
     "temperature": None,                  # None: the llama-server preset's
-    "whisper": {"model": "/data/voice-models/whisper/ggml-large-v3-turbo-q5_0.bin",
-                "device": 0, "port": 8182},
-    "kokoro": {"model": "/data/voice-models/kokoro/kokoro-v1.0.onnx",
-               "voices": "/data/voice-models/kokoro/voices-v1.0.bin"},
-    "voice": {"it": {"lang": "it", "blend": {"if_sara": 0.8, "af_bella": 0.2}},
-              "en": {"lang": "en-us", "blend": {"af_heart": 1.0}}},
     "env": {},
-    "overlay": {"name": "Assistant", "position": "center", "hide_delay": 3.0,
-                "user_color": "accent", "assistant_color": "green"},
 }
-OPEN = {"env", "blend"}                   # free-form tables: replaced whole, not merged
+OPEN = {"env"}                            # free-form tables: replaced whole, not merged
 
 
 def merge(base, over, where=""):
@@ -115,12 +99,9 @@ def schema(tools):
 
 
 CFG = load_config(os.path.join(CONFIG_DIR, "config.toml"))
-WHISPER = f"http://127.0.0.1:{CFG['whisper']['port']}"
-KLANG = {l: v["lang"] for l, v in CFG["voice"].items()}
 TOOLS_DIR = os.path.join(CONFIG_DIR, "tools")
 TOOLS = load_tools(TOOLS_DIR)
 TOOL_SCHEMA = schema(TOOLS) or None
-FRAME = 1 / 30                            # one audio level per frame, seconds
 SOCK = os.path.join(os.environ.get("XDG_RUNTIME_DIR", "/tmp"), "desktop-assistant.sock")
 
 
@@ -128,48 +109,8 @@ def log(*a):
     print(time.strftime("%H:%M:%S"), *a, flush=True)
 
 
-def mic_on():
-    out = subprocess.run(["amixer", "-c", CFG["card"], "cget", "numid=2"],
-                         capture_output=True, text=True).stdout
-    return out.rstrip().endswith("values=on")
-
-
-def presenting(d=None):
-    # During calls the user turns on presentation mode; the button then
-    # belongs to the call, not to us. Presentation is the OR of the manual
-    # flag and the game detector's (quickshell status registry: the files are
-    # its interface); `statusctl presentation get` reads only the manual half.
-    d = d or os.environ.get("XDG_RUNTIME_DIR", "/tmp")
-
-    def flag(name):
-        try:
-            with open(os.path.join(d, f"status.{name}")) as f:
-                return f.read().strip() == "1"
-        except OSError:                   # a missing file means off
-            return False
-    # Why the press is ignored, for the notification; "" when it is not.
-    return "gaming" if flag("gaming") else "presentation" if flag("presentation") else ""
-
-
-notify_id = ""
-
-
-def notify_ignored(reason):
-    # Normal urgency: DND (on during presentation mode) holds it back from
-    # popping over a call or a game, and it waits in the notification list.
-    # Each press replaces the last (-r), so one game makes one entry.
-    global notify_id
-    cmd = ["notify-send", "-a", "desktop-assistant", "-u", "normal", "-p"]
-    if notify_id:
-        cmd += ["-r", notify_id]
-    cmd += ["Voice assistant paused", f"The button press was ignored: {reason} mode is on."]
-    r = subprocess.run(cmd, capture_output=True, text=True)
-    notify_id = r.stdout.strip() or notify_id
-
-
-# The overlay's feed: JSON lines on a Unix socket, one event per line. With
-# no client connected every event is dropped, so the voice loop never waits
-# on a UI.
+# The canvas's feed: JSON lines on a Unix socket. With no client connected
+# every event is dropped.
 clients, clients_lock = [], threading.Lock()
 now_state = {"state": "idle"}             # repeated in hello for a late client
 ctx = {"model": None, "max": 0}           # context size of the model in use
@@ -190,7 +131,7 @@ def emit(event):
 
 
 def hello():
-    return {"type": "hello", **CFG["overlay"], "ctx_max": ctx["max"], **now_state}
+    return {"type": "hello", "ctx_max": ctx["max"], **now_state}
 
 
 def serve(path=SOCK):
@@ -233,78 +174,6 @@ def context_size(model):
     return ctx["max"]
 
 
-def latin(text):
-    return all(unicodedata.name(c, "").startswith("LATIN") for c in text if c.isalpha())
-
-
-def scale(rms):
-    # Loudness as the overlay draws it: -60 dBFS and below is 0, full scale
-    # is 1. Logarithmic, so quiet speech still moves the ring.
-    if rms <= 0:
-        return 0.0
-    return max(0.0, min(1.0, (20 * math.log10(rms / 32768) + 60) / 60))
-
-
-def pcm_level(data):
-    # 16-bit little-endian mono, as arecord writes it; a trailing odd byte is
-    # half a sample and ignored.
-    samples = array.array("h", data[: len(data) // 2 * 2])
-    if not samples:
-        return 0.0
-    return scale(math.sqrt(sum(s * s for s in samples) / len(samples)))
-
-
-def envelope(audio, sr):
-    # One level per FRAME of float audio in -1..1 (Kokoro's output).
-    n = int(sr * FRAME)
-    frames = audio[: len(audio) // n * n].reshape(-1, n)
-    return [scale(float(r) * 32768) for r in (frames.astype("float64") ** 2).mean(axis=1) ** 0.5]
-
-
-def transcribe(lang="auto"):
-    # Always send language: whisper-server's own default is en, not auto.
-    # Beam search: 9/10 test clips right vs 8/10 greedy, +0.06 s.
-    out = subprocess.run(["curl", "-s", "-F", f"file=@{WAV}", "-F", f"language={lang}",
-                          "-F", "beam_size=5",
-                          "-F", "response_format=verbose_json", WHISPER + "/inference"],
-                         capture_output=True, text=True).stdout
-    d = json.loads(out)
-    return d.get("language_probabilities", {}), " ".join(d["text"].split())
-
-
-def stt():
-    # large-v3-turbo's detector is reliable for this speaker (small's was
-    # not). Only it and en are spoken: if detection lands elsewhere, or the
-    # text is non-Latin (a hallucination, both are Latin script), redo the
-    # pass forced to the likelier of the two.
-    p, text = transcribe()
-    lang = max(KLANG, key=lambda l: p.get(l, 0))
-    if max(p, key=p.get, default=lang) != lang or not latin(text):
-        text = transcribe(lang)[1]
-    return lang, text
-
-
-IT_WORDS = set("il lo la gli le di che è non per una sono ho hai con mi ti ci del della "
-               "come questo questa anche ma più se nel alla puoi posso sei".split())
-EN_WORDS = set("the is are you i to of and it that for with not this what be have "
-               "can your my do there".split())
-
-
-def lang_of(sentence, fallback):
-    # The voice must follow the reply's language, not the user's: the model
-    # may answer in the other one. ponytail: stopword count, a real language
-    # identifier if this misfires.
-    words = re.findall(r"[a-zàèéìòù]+", sentence.lower())
-    it = sum(w in IT_WORDS for w in words)
-    en = sum(w in EN_WORDS for w in words)
-    return "it" if it > en else "en" if en > it else fallback
-
-
-def split_sentences(buf):
-    parts = re.split(r"(?<=[.!?…])\s+|\n+", buf)
-    return [p.strip() for p in parts[:-1] if p.strip()], parts[-1]
-
-
 def run_tool(tools, name, arguments, d=TOOLS_DIR):
     # Whatever goes wrong becomes the tool's answer, so the model can say it
     # could not find out; a tool never takes the assistant down.
@@ -353,15 +222,6 @@ def collect_calls(calls, fragments):
         c["arguments"] += fn.get("arguments") or ""
 
 
-def announcement(tools, calls, lang):
-    # Something to hear while a slow tool runs: there is no screen.
-    for c in calls.values():
-        phrases = tools.get(c["name"], {}).get("announce", {}).get(lang)
-        if phrases:
-            return random.choice(phrases)
-    return None
-
-
 def llm_model():
     # The router keeps one model resident; asking for another would swap it
     # out under other clients, so use whatever is not unloaded.
@@ -404,299 +264,38 @@ def ask(messages, calls, tools=None):
                 yield delta["content"]
 
 
-turn = 0          # bumped by every button press: older replies and audio are stale
-player = None     # the aplay speaking now, killed by a new press
-
-
-def speaker(q, lq):
-    global player
-    from kokoro_onnx import Kokoro
-    k = Kokoro(CFG["kokoro"]["model"], CFG["kokoro"]["voices"])
-    # Each voice is a weighted blend of Kokoro style vectors.
-    voices = {l: sum(w * k.get_voice_style(n) for n, w in v["blend"].items())
-              for l, v in CFG["voice"].items()}
-    play, playing = None, None            # aplay and the turn it speaks for
-    start, written = 0.0, 0               # playback clock of the open stream
-    while True:
-        t, item = q.get()
-        if t != turn:                     # from a reply the user cut off
-            continue
-        if play and playing != t:         # never reuse a cut-off stream
-            play.kill()
-            play = None
-        try:
-            if item is None:              # end of reply
-                if play:
-                    play.stdin.close()
-                    play.wait()
-                    play = None
-                if t == turn:
-                    emit({"type": "state", "state": "idle"})
-                continue
-            if isinstance(item, float):   # seconds of silence
-                if play:
-                    play.stdin.write(bytes(int(sr * item) * 2))
-                    written += int(sr * item)
-                continue
-            text, lang = item
-            audio, sr = k.create(text, voice=voices[lang], lang=KLANG[lang])
-            if t != turn:                 # cut off while synthesising
-                continue
-            if play is None:
-                play = subprocess.Popen(["aplay", "-q", "-r", str(sr), "-f", "S16_LE", "-c", "1"],
-                                        stdin=subprocess.PIPE)
-                playing, player = t, play
-                # A new stream wakes the suspended sink and the first ~0.2 s is
-                # lost: lead with silence so it eats that, not the first word.
-                play.stdin.write(bytes(int(sr * 0.3) * 2))
-                start, written = time.monotonic() + 0.3, 0
-            # This sentence plays after everything written so far. If synthesis
-            # fell behind (an underrun), playback restarts from now.
-            # ponytail: no sound-server latency correction; add an offset here
-            # if the waves visibly lead the voice.
-            now = time.monotonic()
-            if start + written / sr < now:
-                start = now - written / sr
-            due = start + written / sr
-            for i, value in enumerate(envelope(audio, sr)):
-                lq.put((t, due + i * FRAME, value))
-            written += len(audio)
-            # The pipe buffer lets synthesis of the next sentence overlap playback.
-            play.stdin.write((audio.clip(-1, 1) * 32767).astype("<i2").tobytes())
-        except BrokenPipeError:           # killed by a new press mid-write
-            play = None
-
-
-def user_levels(rec, mine):
-    # The user's waves: the loudness of what arecord wrote since last time.
-    pos = 0
-    while rec.poll() is None and turn == mine:
-        time.sleep(FRAME)
-        try:
-            with open(RAW, "rb") as f:
-                f.seek(pos)
-                data = f.read()
-        except OSError:
-            continue
-        if len(data) >= 2:
-            pos += len(data) // 2 * 2
-            emit({"type": "level", "who": "user", "value": pcm_level(data)})
-
-
-def level_emitter(lq):
-    # The reply's waves, each frame sent when its audio plays rather than when
-    # it was synthesised (synthesis runs ahead of playback).
-    while True:
-        t, due, value = lq.get()
-        while t == turn and (wait := due - time.monotonic()) > 0:
-            time.sleep(min(wait, 0.05))
-        if t == turn:
-            emit({"type": "level", "who": "assistant", "value": value})
-
-
-def die_with_parent():
-    # Runs in the child before exec: the kernel sends SIGTERM when we die,
-    # however we die, so no whisper-server or amixer is ever orphaned.
-    ctypes.CDLL("libc.so.6").prctl(1, signal.SIGTERM)   # PR_SET_PDEATHSIG
-
-
-def start_whisper():
-    def up():
-        try:
-            urllib.request.urlopen(WHISPER, timeout=1)
-            return True
-        except OSError:
-            return False
-    if up():
-        return
-    p = subprocess.Popen(["whisper-server", "-m", CFG["whisper"]["model"],
-                          "-dev", str(CFG["whisper"]["device"]),
-                          "--host", "127.0.0.1", "--port", str(CFG["whisper"]["port"])],
-                         stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
-                         preexec_fn=die_with_parent)
-    for _ in range(60):
-        if up():
-            return
-        time.sleep(0.5)
-    p.kill()
-    sys.exit("whisper-server did not come up")
-
-
-def reply(history, q, mine):
-    # Runs on its own thread; returns as soon as a newer press makes it stale.
-    # ponytail: a press during this STT can overwrite WAV mid-upload, one
-    # garbled question at worst; per-turn file names if that ever shows.
-    pcm = open(RAW, "rb").read()
-    if len(pcm) < 16000 * 2 * 0.3:               # under 0.3 s: a stray press
-        emit({"type": "state", "state": "idle"})
-        return
-    # Trailing silence: whisper drops the last word when audio ends mid-word,
-    # which it does when the button is pressed while still talking.
-    with wave.open(WAV, "wb") as w:
-        w.setnchannels(1)
-        w.setsampwidth(2)
-        w.setframerate(16000)
-        w.writeframes(pcm + bytes(16000 * 2 // 2))
-    emit({"type": "state", "state": "transcribing"})
-    t = time.perf_counter()
-    lang, text = stt()
-    if turn != mine:
-        return
-    if not text or text.startswith("["):         # [BLANK_AUDIO] and friends
-        emit({"type": "state", "state": "idle"})
-        return
-    log(f"you ({lang}, {time.perf_counter() - t:.2f}s): {text}")
-    emit({"type": "state", "state": "thinking"})
-    question = {"role": "user", "content": text}
-    # The model has no clock: without the date it makes one up for tools.
-    # Also stamped on the question itself, which history does not keep: with
-    # the time only in the system prompt, "what time is it?" asked again an
-    # hour later copied the old answer from history. Stamping every message
-    # in history fixed that but lost tool arguments (account filter 0/6).
-    now = time.strftime("%A %d %B %Y, %H:%M")
-    # ponytail: last 10 turns, summarise if long sessions need memory
-    msgs = ([{"role": "system", "content": f"{CFG['system']} It is {now}."}]
-            + [m for past in history[-10:] for m in past]
-            + [{"role": "user", "content": f"[{now}] {text}"}])
-    first = len(msgs)                            # this turn's tool messages start here
-    t, said, voice, announced = time.perf_counter(), [], lang, False
-
-    def say(s):
-        nonlocal voice
-        if not said:
-            log(f"first sentence after {time.perf_counter() - t:.2f}s")
-        if now_state["state"] != "speaking":
-            emit({"type": "state", "state": "speaking"})
-        said.append(s)
-        voice = lang_of(s, voice)
-        q.put((mine, (re.sub(r"[*_#`]", "", s), voice)))
-
-    for rnd in range(CFG["tool_rounds"] + 1):
-        calls, buf = {}, ""
-        # The last round offers no tools, which forces a spoken answer.
-        for delta in ask(msgs, calls, TOOL_SCHEMA if rnd < CFG["tool_rounds"] else None):
-            if turn != mine:
-                log("interrupted")
-                return
-            buf += delta
-            done, buf = split_sentences(buf)
-            for s in done:
-                say(s)
-        if buf.strip():
-            say(buf.strip())
-        if not calls or turn != mine:
-            break
-        msgs.append({"role": "assistant", "content": "", "tool_calls": [
-            {"id": c["id"], "type": "function",
-             "function": {"name": c["name"], "arguments": c["arguments"]}}
-            for c in calls.values()]})
-        phrase = None if announced else announcement(TOOLS, calls, lang)
-        if phrase:
-            # The question's language: the reply has no text yet to follow.
-            q.put((mine, (phrase, lang)))
-            # A beat after it: straight into the answer sounds rushed.
-            q.put((mine, 0.6))
-            announced = True
-        # ponytail: a press does not kill a running tool, its result is dropped
-        for c in calls.values():
-            emit({"type": "state", "state": "tool", "tool": c["name"]})
-            result = run_tool(TOOLS, c["name"], c["arguments"])
-            log(f"tool {c['name']}({c['arguments']}): {result[:120]!r}")
-            msgs.append({"role": "tool", "tool_call_id": c["id"], "content": result})
-        if turn == mine:
-            emit({"type": "state", "state": "thinking"})
-    if turn != mine:
-        log("interrupted")
-        return
-    q.put((mine, None))
-    # History keeps the tool calls: with only what was said, the model saw
-    # facts answered without tools and invented the next ones (no tool call
-    # 9/10 on a replayed session, 0/10 with the calls kept). Results are
-    # replaced by a placeholder, for context: same 10/10, and a follow-up on
-    # an old result calls the tool again (10/10). A cut off exchange is
-    # dropped whole, so user and assistant still alternate.
-    past = [{**m, "content": "(shown to the user at the time)"} if m["role"] == "tool" else m
-            for m in msgs[first:]]
-    history.append([question] + past + [{"role": "assistant", "content": " ".join(said)}])
-    log("assistant:", " ".join(said))
-
 
 def main():
-    global turn
-    # SIGTERM must run the cleanup below too, or the recording is left behind.
+    # SIGTERM runs the finally below, so the socket is not left behind.
     signal.signal(signal.SIGTERM, lambda *_: sys.exit(0))
-    start_whisper()
     try:
-        context_size(llm_model())         # so the overlay's first hello has it
+        context_size(llm_model())         # so the first hello has it
     except OSError:
         pass                              # llama-server down: asked again per request
     serve()
-    q, lq = queue.Queue(), queue.Queue()
-    threading.Thread(target=speaker, args=(q, lq), daemon=True).start()
-    threading.Thread(target=level_emitter, args=(lq,), daemon=True).start()
-    history, rec = [], None
-    events = subprocess.Popen(["stdbuf", "-oL", "amixer", "-c", CFG["card"], "events"],
-                              stdout=subprocess.PIPE, text=True, preexec_fn=die_with_parent)
-    prev = mic_on()
-    log(f"ready, mic is {'on' if prev else 'off'}")
+    log(f"ready on {SOCK}")
     try:
-        for line in events.stdout:
-            if "Mic Capture Switch" not in line:
-                continue
-            on = mic_on()
-            if on == prev:
-                continue
-            prev = on
-            reason = presenting() if on else ""
-            if reason:
-                log(f"{reason} mode: ignored")
-                notify_ignored(reason)
-            elif on:
-                # Barge-in: a press silences the reply now and makes it stale.
-                turn += 1
-                if player:
-                    player.kill()
-                emit({"type": "state", "state": "listening"})
-                # A 33 ms period: arecord writes the file in periods, and the
-                # default (~125 ms) left the user's waves at 8 levels a second.
-                rec = subprocess.Popen(["arecord", "-q", "-t", "raw", "-f", "S16_LE", "-r", "16000",
-                                        "-c", "1", "--period-time=33333", RAW],
-                                       stderr=subprocess.DEVNULL, preexec_fn=die_with_parent)
-                threading.Thread(target=user_levels, args=(rec, turn), daemon=True).start()
-            elif rec:
-                rec.send_signal(signal.SIGINT)
-                rec.wait()
-                rec = None
-                threading.Thread(target=reply, args=(history, q, turn), daemon=True).start()
+        while True:
+            signal.pause()
     except KeyboardInterrupt:
         pass
     finally:
-        for f in (RAW, WAV, SOCK):              # the user's voice: never keep it
-            if os.path.exists(f):
-                os.remove(f)
+        if os.path.exists(SOCK):
+            os.remove(SOCK)
 
 
 def test():
-    assert split_sentences("Ciao! Come stai? Io") == (["Ciao!", "Come stai?"], "Io")
-    assert split_sentences("Pi is 3.14 today") == ([], "Pi is 3.14 today")
-    assert split_sentences("one\n\ntwo") == (["one"], "two")
-    assert latin("Perché no? Hi!") and not latin("早, 早!")
-    assert lang_of("Non posso fornire il meteo in tempo reale.", "en") == "it"
-    assert lang_of("That is a very exciting place to be.", "it") == "en"
-    assert lang_of("Tokyo!", "it") == "it"
-    c = merge(DEFAULTS, {"card": "X", "whisper": {"port": 9},
-                         "voice": {"it": {"blend": {"a": 1.0}}}, "env": {"K": "v"}})
-    assert c["card"] == "X" and c["whisper"] == {**DEFAULTS["whisper"], "port": 9}
-    assert c["voice"]["it"] == {"lang": "it", "blend": {"a": 1.0}}      # blend replaced whole
-    assert c["voice"]["en"] == DEFAULTS["voice"]["en"] and c["env"] == {"K": "v"}
-    assert DEFAULTS["card"] == "Microphone"                             # defaults untouched
-    for bad in ({"whisper": {"prot": 1}}, {"cards": "X"}, {"whisper": 3}):
+    import tempfile
+
+    c = merge(DEFAULTS, {"tool_rounds": 1, "env": {"K": "v"}})
+    assert c["tool_rounds"] == 1 and c["env"] == {"K": "v"}
+    assert DEFAULTS["tool_rounds"] == 3                                 # defaults untouched
+    for bad in ({"card": "X"}, {"whisper": {"port": 1}}):       # voice-era keys are gone
         try:
             merge(DEFAULTS, bad)
             assert False, bad
         except SystemExit as e:
             assert "config:" in str(e)
-    import tempfile
     with tempfile.TemporaryDirectory() as d:
         with open(os.path.join(d, "echo.sh"), "w") as f:
             f.write('#!/bin/sh\nprintf "%s|" "$@"\n')
@@ -724,13 +323,6 @@ def test():
         assert run_tool(tools, "nope", "{}", d) == "error: no tool named nope"
         assert run_tool(tools, "echo", "{bad", d) == "error: arguments are not valid JSON"
         assert run_tool(tools, "echo", "[1]", d) == "error: arguments are not valid JSON"
-    with tempfile.TemporaryDirectory() as d:
-        assert presenting(d) == ""
-        for name, value, want in (("gaming", "1", "gaming"), ("gaming", "0", ""),
-                                  ("presentation", "1\n", "presentation")):
-            with open(os.path.join(d, f"status.{name}"), "w") as f:
-                f.write(value)
-            assert presenting(d) == want, (name, value)
     calls = {}
     collect_calls(calls, [{"index": 0, "id": "c1", "type": "function",
                            "function": {"name": "weather", "arguments": "{"}}])
@@ -739,22 +331,7 @@ def test():
     collect_calls(calls, [{"index": 1, "id": "c2", "function": {"name": "now", "arguments": "{}"}}])
     assert calls == {0: {"id": "c1", "name": "weather", "arguments": '{"place":"Rome"}'},
                      1: {"id": "c2", "name": "now", "arguments": "{}"}}
-    tools = {"now": {}, "search": {"announce": {"it": ["Cerco."], "en": ["Searching."]}}}
-    assert announcement(tools, {0: {"name": "now"}}, "it") is None
-    assert announcement(tools, {0: {"name": "now"}, 1: {"name": "search"}}, "en") == "Searching."
-    assert announcement(tools, {0: {"name": "unknown"}}, "it") is None
-    import numpy as np
-    assert scale(0) == 0.0 and scale(1) == 0.0            # -90 dBFS clamps to 0
-    assert scale(32768) == 1.0
-    assert abs(scale(32768 * 10 ** (-30 / 20)) - 0.5) < 1e-9
-    assert pcm_level(b"") == 0.0
-    assert pcm_level(array.array("h", [1000] * 100).tobytes()) == scale(1000)
-    assert pcm_level(b"\x00") == 0.0                      # half a sample is ignored
-    env = envelope(np.zeros(24000, dtype=np.float32), 24000)
-    assert len(env) == 30 and max(env) == 0.0             # 1 s at 30 frames/s
-    env = envelope(np.full(24000, 0.5, dtype=np.float32), 24000)
-    assert all(abs(v - scale(16384)) < 1e-6 for v in env)
-    import socket
+
     a, b = socket.socketpair()
     a.setblocking(False)
     clients.append(a)
@@ -775,14 +352,7 @@ def test():
         c.connect(path)
         f = c.makefile()
         first = json.loads(f.readline())
-        assert first["type"] == "hello" and first["name"] == CFG["overlay"]["name"]
-        assert first["state"] == "idle" and "ctx_max" in first
-        for _ in range(100):                             # accept() appends just after hello
-            if clients:
-                break
-            time.sleep(0.01)
-        emit({"type": "level", "who": "user", "value": 0.5})
-        assert json.loads(f.readline()) == {"type": "level", "who": "user", "value": 0.5}
+        assert first == {"type": "hello", "ctx_max": ctx["max"], "state": "idle"}
         c.close()
     print("ok")