diff --git a/.gitignore b/.gitignore index e7e3ecd..8b750b2 100644 --- a/.gitignore +++ b/.gitignore @@ -10,3 +10,4 @@ uv.lock backtalk.egg-info/ # your personal config, created at setup: never tracked, never touched by updates backtalk.json +.backtalk_session diff --git a/README.md b/README.md index 5f0609f..564019f 100644 --- a/README.md +++ b/README.md @@ -16,7 +16,7 @@ The hearing and the voice run local: free, offline models on your machine, no vo - **Type instead whenever you want.** Typing in the terminal is the same conversation, and the reply is still spoken. - **It asks before it acts, in plain words.** When your agent wants to do something real, it asks out loud the way a person would ("I want to change a note in your vault called Recipes") and waits. An exact spoken yes approves; "details" reads you the exact command; anything else denies, with your words passed back as the reason, so "no, put that in drafts instead" actually steers it. Most read-only work passes without interrupting. Prefer auto-approve? Say "stop asking for permission" (or "turn off the permission prompts") and confirm; it changes its own config, and the first ask of every session reminds you the phrase exists. - **The voice console.** Session control by voice, so you never go back to the keyboard: "clear the session", "compact the session", "switch to the deep model" / "back to the fast model", "set effort to low" (or medium, high, max; this one saves itself as your default), "usage report", "go hands free" / "push to talk mode" for the microphone, "stop asking for permission" / "start asking again" for approvals. Exact phrases, spoken alone. (Credit where due: this grew out of a community member's own build shared in the Discord.) -- **It can pick up where it left off.** Set `"resume_last_session": true` in the config and every launch reattaches to your previous conversation instead of starting cold, so closing the window stops costing you the thread. Off by default. And the built-in voice has a pace dial: `"speed"` in the config, 1.0 native, 1.15 brisker. (Credit where due: both grew out of a community proposal by aram-cloudstak.) +- **It can pick up where it left off.** Set `"resume_last_session": true` in the config and every launch reattaches to your previous conversation instead of starting cold, so closing the window (or a crash) stops costing you the thread. On a reattached launch the agent opens by saying where you left off and asking whether to continue or start fresh; say "start fresh" (or "clear the session") to start over, or just keep talking. Off by default. And the built-in voice has a pace dial: `"speed"` in the config, 1.0 native, 1.15 brisker. (Credit where due: both grew out of a community proposal by aram-cloudstak.) - **Music ducks while it speaks** (Spotify, macOS) and comes back up after. - **It thinks out loud.** While the agent works, you hear the processing sound from my videos, so a pause never reads as a dead line. Silence it with `"thinking_sound": ""` in the config. @@ -56,7 +56,7 @@ Two engines, and the setup wizard offers you both instead of quietly defaulting. ## Give it a face (optional) -backtalk writes tiny state files while it listens, thinks, and speaks, so anything can watch them and react in real time. +backtalk writes tiny state files while it listens, thinks, and speaks, so anything can watch them and react in real time: `.voice_state`, `.voice_waveform`, `.voice_loading_pid`, plus `.voice_heartbeat` (unix time, rewritten every ~2 s while backtalk is alive — stale means dead or hung) and `.voice_activity` (JSON `{ts, turn_started, line}` naming what the agent is doing mid-turn, e.g. `Read: foo.py`; deleted when the turn ends). All in `signals_dir`. - **[ai-visualizer](https://github.com/jaredrhod/ai-visualizer)** is the matching face: four full-screen visualizers, including the living circuit board from my videos. Point its `bus_dir` at this folder (or set `signals_dir` here to its folder) and it performs your actual conversation, idling, listening, thinking, and speaking along with the voice. - **[barehands](https://github.com/jaredrhod/barehands)**: point `barehands_state_dir` at its `state/` folder and the on-screen ring becomes your agent's face, breathing while idle, spinning while thinking, and pulsing with the voice while it talks. diff --git a/TROUBLESHOOTING.md b/TROUBLESHOOTING.md index 86de4f8..ccc0713 100644 --- a/TROUBLESHOOTING.md +++ b/TROUBLESHOOTING.md @@ -10,7 +10,7 @@ Written for humans AND for AI assistants. If you're an AI helping someone debug - **Mic permission never appeared / recording is silent**: launch from a normal terminal window, not a background service or launcher daemon: the process inherits the *terminal's* microphone permission. Check the input device: `python -m sounddevice` lists them. - **It hears me but answers slowly**: check `model` in `backtalk.json`. Full-size deep-work models make every reply noticeably slower; the fast tier is the point of a voice loop. Also confirm the model id is the FULL id, never a bare alias; aliases can silently resolve to an older model through the SDK's bundled CLI. - **First reply after launch is slow**: that's the one-time prompt-cache toll, mostly hidden behind the greeting. Warm turns are the real speed. -- **It starts cold and forgets the last conversation after a restart**: that is the default (a fresh session every launch is predictable). Want it to pick up where it left off? Tell your agent to set `"resume_last_session": true` in backtalk.json. From then on every launch reattaches to the previous conversation, and a stale saved session falls back to fresh with a log line instead of breaking the launch. The saved conversation lives on your machine as a Claude Code transcript (kept for 30 days from last use by default), and a long conversation never dies from length: it compacts itself automatically, older turns becoming a summary while recent ones stay verbatim. Start over any time by saying "clear the session." +- **It starts cold and forgets the last conversation after a restart**: that is the default (a fresh session every launch is predictable). Want it to pick up where it left off? Tell your agent to set `"resume_last_session": true` in backtalk.json. From then on every launch reattaches to the previous conversation and the agent opens by saying where you left off and asking continue-or-start-fresh ("start fresh" clears it; otherwise just keep talking); a stale saved session falls back to fresh with a log line instead of breaking the launch. The saved conversation lives on your machine as a Claude Code transcript (kept for 30 days from last use by default), and a long conversation never dies from length: it compacts itself automatically, older turns becoming a summary while recent ones stay verbatim. Start over any time by saying "clear the session." - **The voice talks too fast or too slow**: the built-in voice has a pace dial, `"speed"` in backtalk.json. 1.0 is native, 1.15 is brisker, 0.9 is slower. ElevenLabs pace lives in the `master` chain's atempo value instead. - **Updating, or an update that complains about local changes**: run `./update.sh` (`update.bat` on Windows) in this folder, or double-click the `Update` icon if setup left one. It shows what changed before applying it and can never touch your `backtalk.json`. If an older updater said "couldn't fast-forward" or mentioned local changes, run `./update.sh` once and it clears: it moves your config out of git's sight and everything flows after. - **The voice sounds robotic**: you're hearing Kokoro's base register, or the wrong voice for the language. Try `bm_george`, `bm_daniel`, `am_michael`, `af_heart`. Remember the first letter must match the language pipeline (`b…` British, `a…` American). diff --git a/backtalk/brain.py b/backtalk/brain.py index a0433cf..d6a9ea4 100644 --- a/backtalk/brain.py +++ b/backtalk/brain.py @@ -1,317 +1,447 @@ -# backtalk: talk to your Claude Code agent out loud. -# Copyright (C) 2026 Jared Rhodenizer -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU Affero General Public License as published -# by the Free Software Foundation, either version 3 of the License, or -# (at your option) any later version. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU Affero General Public License for more details. -# -# You should have received a copy of the GNU Affero General Public License -# along with this program. If not, see . -# -# SPDX-License-Identifier: AGPL-3.0-or-later -"""The warm brain — a persistent Claude session via the Agent SDK, -streaming. - -One ClaudeSDKClient lives for the whole voice session: no per-turn -process spawn, no per-turn context reload. Partial-message streaming -means sentences are yielded the moment they're complete, so the mouth -starts speaking while the rest of the thought is still forming. - -The session's cwd is YOUR agent's folder (agent_dir in backtalk.json) — -whatever CLAUDE.md lives there defines who is speaking. backtalk adds -only the spoken-delivery discipline (config.DISCIPLINE): the medium, -never the character. -""" -import asyncio -import os -import re -import warnings - -from claude_agent_sdk import ClaudeAgentOptions, ClaudeSDKClient - -try: - from claude_agent_sdk import CanUseToolShadowedWarning -except ImportError: # older SDKs: nothing to silence - CanUseToolShadowedWarning = None - -from backtalk.config import CFG, DISCIPLINE -from backtalk.vlog import log - -_SENTENCE_END = re.compile(r"(?<=[.!?])\s") - - -SESSION_FILE = os.path.join(CFG["signals_dir"], ".backtalk_session") - - -class WarmBrain: - def __init__(self, model: str | None = None, can_use_tool=None, - resume_id: str | None = None): - # Full model id ON PURPOSE — never a bare alias. The SDK - # resolves aliases through its own bundled CLI and can silently - # land on an older model. - self.model = model or CFG["model"] - # The spoken permission gate (main.py builds it). Wired at - # connect in EVERY mode, so a live mode flip needs no reconnect; - # bypass simply never consults it. - self._can_use_tool = can_use_tool - # Session usage, spoken on request ("usage report"). - self.session = {"turns": 0, "out_tokens": 0, "in_tokens": 0, - "cost": 0.0} - self._client: ClaudeSDKClient | None = None - # The session to reattach to at the FIRST start only (config key - # resume_last_session). Consumed on use: a desync rebuild in - # reset_turn() must always start FRESH: a rebuild means a turn - # went sideways mid-stream, the wrong moment to gamble on - # reattaching. (Community proposal, issue #1.) - self._resume_id = resume_id - # True while a query's response hasn't been consumed through its - # ResultMessage — i.e. the shared message pipe may hold leftovers. - self._dirty = False - - async def start(self): - mode = CFG["permission_mode"] - if mode == "default": - mode = "ask" # legacy alias, see config.py - # backtalk's "ask" = the SDK's "default" mode with gated calls - # routed to the spoken can_use_tool gate. - sdk_mode = "default" if mode == "ask" else mode - if sdk_mode == "bypassPermissions" and self._can_use_tool \ - and CanUseToolShadowedWarning: - # Deliberate auto-approve: the SDK warns that the callback is - # shadowed. That IS the chosen behavior, so boot quietly. - warnings.filterwarnings("ignore", - category=CanUseToolShadowedWarning) - resume, self._resume_id = self._resume_id, None # consume once - - def _opts(rid): - return ClaudeAgentOptions( - cwd=CFG["agent_dir"], - model=self.model, - system_prompt={"type": "preset", "preset": "claude_code", - "append": DISCIPLINE}, - include_partial_messages=True, - permission_mode=sdk_mode, - can_use_tool=self._can_use_tool, - add_dirs=CFG["extra_dirs"], - resume=rid, - ) - if resume: - try: - self._client = ClaudeSDKClient(options=_opts(resume)) - await self._client.connect() - log(f"[brain] resumed session {resume[:8]}") - return - except Exception as e: - # a stale or invalid saved session must never brick the - # launch. Fall back to a fresh conversation and say so. - log(f"[brain] resume failed ({str(e)[:80]}), " - f"starting fresh") - try: - await self._client.disconnect() - except Exception: - pass - self._client = ClaudeSDKClient(options=_opts(None)) - await self._client.connect() - - async def set_permission_mode(self, backtalk_mode: str): - """Live flip, no reconnect, conversation intact ("ask" maps to - the SDK's "default", whose gated calls hit the spoken gate).""" - if self._client: - sdk_mode = "default" if backtalk_mode == "ask" \ - else backtalk_mode - await self._client.set_permission_mode(sdk_mode) - - async def context_usage(self): - """The CLI's own context-window breakdown, or None.""" - try: - return await self._client.get_context_usage() - except Exception: - return None - - def _remember_session(self, rm): - """Persist the session id after a completed turn, so the next - launch can reattach (config: resume_last_session). Must never - break a turn; silence on any failure.""" - if not CFG.get("resume_last_session"): - return - sid = getattr(rm, "session_id", None) - if not sid: - return - try: - with open(SESSION_FILE, "w") as f: - f.write(sid) - except OSError: - pass - - def _tally(self, rm, count_turn=True): - """Session usage bookkeeping. Must never break a turn.""" - try: - u = getattr(rm, "usage", None) or {} - s = self.session - if count_turn: - s["turns"] += 1 - s["out_tokens"] += int(u.get("output_tokens") or 0) - s["in_tokens"] += (int(u.get("input_tokens") or 0) - + int(u.get("cache_read_input_tokens") - or 0)) - c = getattr(rm, "total_cost_usd", None) - if c: - s["cost"] += float(c) - except Exception: - pass - - async def command(self, cmd: str) -> str: - """Run a console slash command (/clear, /compact, /model, - /effort) through the normal stream and return whatever text the - CLI answered with (confirmations, errors). Slash-command replies - arrive as COMPLETE AssistantMessages, not stream deltas, so - ask_stream cannot see them. Bounded like reset_turn is: this - stream is not trusted to always deliver, and an unbounded await - here would deafen the whole voice loop. On timeout the pipe is - left marked dirty so the next reset_turn drains or rebuilds.""" - self._dirty = True - await self._client.query(cmd) - texts = [] - - async def _collect(): - async for msg in self._client.receive_response(): - t = type(msg).__name__ - if t == "AssistantMessage": - for b in getattr(msg, "content", []) or []: - txt = getattr(b, "text", None) - if txt: - texts.append(txt) - elif t == "ResultMessage": - self._dirty = False - self._tally(msg, count_turn=False) - self._remember_session(msg) - break - - try: - await asyncio.wait_for(_collect(), 90) - except asyncio.TimeoutError: - log(f"[brain] console command timed out: {cmd!r}") - return "error: the command timed out" - return " ".join(texts).strip() - - async def interrupt(self): - if self._client: - await self._client.interrupt() - - async def reset_turn(self, timeout: float = 8.0): - """Re-align the message pipe after an interrupted/failed turn. - - THE OFF-BY-ONE BUG, and why this method exists: the SDK client - has ONE shared message stream and receive_response() stops at - the FIRST ResultMessage it sees — there is no pairing between a - query and its response. A cancelled turn stops consuming - mid-stream, leaving the dead turn's remaining messages - (including its ResultMessage) buffered. The next query then - pairs with those leftovers: the first ask lands on the stale - ResultMessage and yields nothing, and every ask after that - answers the PREVIOUS question — for the rest of the session. - So: interrupt the dead turn, then drain the pipe through its - stale ResultMessage before the next query goes out. No-op when - the last turn was consumed clean.""" - if not self._client or not self._dirty: - return - try: - await asyncio.wait_for(self._client.interrupt(), 5) - except Exception: - pass # turn may already be over — the drain below is the point - - async def _drain() -> int: - n = 0 - async for msg in self._client.receive_response(): - n += 1 - if type(msg).__name__ == "ResultMessage": - break - return n - - try: - drained = await asyncio.wait_for(_drain(), timeout) - log(f"[brain] interrupted turn drained ({drained} stale messages)") - self._dirty = False - except Exception: - # Can't re-align — rebuild the session rather than run - # desynced. Loses this voice session's conversation memory; - # better than answering every question one turn late for the - # rest of the day. - log("[brain] stream desynced beyond repair — rebuilding the " - "session (conversation memory for this session resets)") - try: - await self._client.disconnect() - except Exception: - pass - self._client = None - await self.start() - self._dirty = False - - async def stop(self): - if self._client: - await self._client.disconnect() - self._client = None - - async def ask_stream(self, utterance: str): - """Yield complete sentences as they stream out of the model.""" - self._dirty = True # in flight until its ResultMessage - await self._client.query(utterance) - buf = "" - async for msg in self._client.receive_response(): - t = type(msg).__name__ - if t == "StreamEvent": - ev = getattr(msg, "event", {}) or {} - if ev.get("type") == "content_block_delta": - delta = ev.get("delta", {}) or {} - if delta.get("type") == "text_delta": - buf += delta.get("text", "") - # emit any complete sentences - while True: - m = _SENTENCE_END.search(buf) - if not m: - break - sentence, buf = (buf[:m.end()].strip(), - buf[m.end():]) - if sentence: - yield sentence - elif ev.get("type") == "content_block_stop": - # End of a speech block (e.g. right before a tool - # call): flush NOW. Without this, pre-tool filler - # ("On it — let me grab that.") sits silent in the - # buffer through the whole tool run, then plays - # glued to the answer: long dead air, then two - # thoughts at once. - tail = buf.strip() - buf = "" - if tail: - yield tail - elif t == "ResultMessage": - self._dirty = False # turn fully consumed — pipe aligned - self._tally(msg) - self._remember_session(msg) - break - tail = buf.strip() - if tail: - yield tail - - -if __name__ == "__main__": - import time - - async def demo(): - b = WarmBrain() - await b.start() - for prompt in ("Voice check: greet me in one sentence.", - "And what's two plus two, spoken like yourself?"): - t0 = time.time() - async for s in b.ask_stream(prompt): - print(f" ({time.time()-t0:4.1f}s) {s}", flush=True) - await b.stop() - - asyncio.run(demo()) +# backtalk: talk to your Claude Code agent out loud. +# Copyright (C) 2026 Jared Rhodenizer +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as published +# by the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . +# +# SPDX-License-Identifier: AGPL-3.0-or-later +"""The warm brain — a persistent Claude session via the Agent SDK, +streaming. + +One ClaudeSDKClient lives for the whole voice session: no per-turn +process spawn, no per-turn context reload. Partial-message streaming +means sentences are yielded the moment they're complete, so the mouth +starts speaking while the rest of the thought is still forming. + +The session's cwd is YOUR agent's folder (agent_dir in backtalk.json) — +whatever CLAUDE.md lives there defines who is speaking. backtalk adds +only the spoken-delivery discipline (config.DISCIPLINE): the medium, +never the character. +""" +import asyncio +import os +import re +import warnings + +import anyio + +from claude_agent_sdk import ClaudeAgentOptions, ClaudeSDKClient + +try: + from claude_agent_sdk import CanUseToolShadowedWarning +except ImportError: # older SDKs: nothing to silence + CanUseToolShadowedWarning = None + +from backtalk import signals +from backtalk.config import CFG, DISCIPLINE +from backtalk.vlog import log + +_SENTENCE_END = re.compile(r"(?<=[.!?])\s") + + +SESSION_FILE = os.path.join(CFG["signals_dir"], ".backtalk_session") + + +class WarmBrain: + def __init__(self, model: str | None = None, can_use_tool=None, + resume_id: str | None = None): + # Full model id ON PURPOSE — never a bare alias. The SDK + # resolves aliases through its own bundled CLI and can silently + # land on an older model. + self.model = model or CFG["model"] + # The spoken permission gate (main.py builds it). Wired at + # connect in EVERY mode, so a live mode flip needs no reconnect; + # bypass simply never consults it. + self._can_use_tool = can_use_tool + # Session usage, spoken on request ("usage report"). + self.session = {"turns": 0, "out_tokens": 0, "in_tokens": 0, + "cost": 0.0} + self._client: ClaudeSDKClient | None = None + # The session to reattach to at the FIRST start only (config key + # resume_last_session). Consumed on use: a desync rebuild in + # reset_turn() must always start FRESH: a rebuild means a turn + # went sideways mid-stream, the wrong moment to gamble on + # reattaching. (Community proposal, issue #1.) + self._resume_id = resume_id + # True once start() actually reattached to a saved conversation + # (main.py speaks a where-were-we recap instead of the silent + # warmup ping in that case). + self.resumed = False + # True while a query's response hasn't been consumed through its + # ResultMessage — i.e. the shared message pipe may hold leftovers. + self._dirty = False + + async def start(self): + mode = CFG["permission_mode"] + if mode == "default": + mode = "ask" # legacy alias, see config.py + # backtalk's "ask" = the SDK's "default" mode with gated calls + # routed to the spoken can_use_tool gate. + sdk_mode = "default" if mode == "ask" else mode + if sdk_mode == "bypassPermissions" and self._can_use_tool \ + and CanUseToolShadowedWarning: + # Deliberate auto-approve: the SDK warns that the callback is + # shadowed. That IS the chosen behavior, so boot quietly. + warnings.filterwarnings("ignore", + category=CanUseToolShadowedWarning) + resume, self._resume_id = self._resume_id, None # consume once + + def _opts(rid): + return ClaudeAgentOptions( + cwd=CFG["agent_dir"], + model=self.model, + system_prompt={"type": "preset", "preset": "claude_code", + "append": DISCIPLINE}, + include_partial_messages=True, + permission_mode=sdk_mode, + can_use_tool=self._can_use_tool, + add_dirs=CFG["extra_dirs"], + # SDK default is 1 MB per stream-json message; a 1080p + # screenshot read is ~5 MB base64 and killed the session. + max_buffer_size=16 * 1024 * 1024, + resume=rid, + ) + if resume: + try: + self._client = ClaudeSDKClient(options=_opts(resume)) + await self._client.connect() + log(f"[brain] resumed session {resume[:8]}") + self.resumed = True + return + except Exception as e: + # a stale or invalid saved session must never brick the + # launch. Fall back to a fresh conversation and say so. + log(f"[brain] resume failed ({str(e)[:80]}), " + f"starting fresh") + try: + await self._client.disconnect() + except Exception: + pass + self._client = ClaudeSDKClient(options=_opts(None)) + await self._client.connect() + + async def set_permission_mode(self, backtalk_mode: str): + """Live flip, no reconnect, conversation intact ("ask" maps to + the SDK's "default", whose gated calls hit the spoken gate).""" + if self._client: + sdk_mode = "default" if backtalk_mode == "ask" \ + else backtalk_mode + await self._client.set_permission_mode(sdk_mode) + + async def context_usage(self): + """The CLI's own context-window breakdown, or None.""" + try: + return await self._client.get_context_usage() + except Exception: + return None + + def _remember_session(self, rm): + """Persist the session id after a completed turn, so the next + launch can reattach (config: resume_last_session). Must never + break a turn; silence on any failure.""" + if not CFG.get("resume_last_session"): + return + sid = getattr(rm, "session_id", None) + if not sid: + return + try: + with open(SESSION_FILE, "w") as f: + f.write(sid) + except OSError: + pass + + def _tally(self, rm, count_turn=True): + """Session usage bookkeeping. Must never break a turn.""" + try: + u = getattr(rm, "usage", None) or {} + s = self.session + if count_turn: + s["turns"] += 1 + s["out_tokens"] += int(u.get("output_tokens") or 0) + s["in_tokens"] += (int(u.get("input_tokens") or 0) + + int(u.get("cache_read_input_tokens") + or 0)) + c = getattr(rm, "total_cost_usd", None) + if c: + s["cost"] += float(c) + except Exception: + pass + + async def command(self, cmd: str) -> str: + """Run a console slash command (/clear, /compact, /model, + /effort) through the normal stream and return whatever text the + CLI answered with (confirmations, errors). Slash-command replies + arrive as COMPLETE AssistantMessages, not stream deltas, so + ask_stream cannot see them. Bounded like reset_turn is: this + stream is not trusted to always deliver, and an unbounded await + here would deafen the whole voice loop. On timeout the pipe is + left marked dirty so the next reset_turn drains or rebuilds.""" + self._dirty = True + await self._client.query(cmd) + texts = [] + + async def _collect(): + async for msg in self._client.receive_response(): + t = type(msg).__name__ + if t == "AssistantMessage": + for b in getattr(msg, "content", []) or []: + txt = getattr(b, "text", None) + if txt: + texts.append(txt) + elif t == "ResultMessage": + self._dirty = False + self._tally(msg, count_turn=False) + self._remember_session(msg) + break + + try: + await asyncio.wait_for(_collect(), 90) + except asyncio.TimeoutError: + log(f"[brain] console command timed out: {cmd!r}") + return "error: the command timed out" + return " ".join(texts).strip() + + async def interrupt(self): + if self._client: + await self._client.interrupt() + + async def reset_turn(self, timeout: float = 8.0): + """Re-align the message pipe after an interrupted/failed turn. + + THE OFF-BY-ONE BUG, and why this method exists: the SDK client + has ONE shared message stream and receive_response() stops at + the FIRST ResultMessage it sees — there is no pairing between a + query and its response. A cancelled turn stops consuming + mid-stream, leaving the dead turn's remaining messages + (including its ResultMessage) buffered. The next query then + pairs with those leftovers: the first ask lands on the stale + ResultMessage and yields nothing, and every ask after that + answers the PREVIOUS question — for the rest of the session. + So: interrupt the dead turn, then drain the pipe through its + stale ResultMessage before the next query goes out. No-op when + the last turn was consumed clean.""" + if not self._client or not self._dirty: + return + try: + await asyncio.wait_for(self._client.interrupt(), 5) + except Exception: + pass # turn may already be over — the drain below is the point + + async def _drain() -> int: + n = 0 + async for msg in self._client.receive_response(): + n += 1 + if type(msg).__name__ == "ResultMessage": + break + return n + + try: + drained = await asyncio.wait_for(_drain(), timeout) + log(f"[brain] interrupted turn drained ({drained} stale messages)") + self._dirty = False + except Exception: + # Can't re-align — rebuild the session rather than run + # desynced. Loses this voice session's conversation memory; + # better than answering every question one turn late for the + # rest of the day. + log("[brain] stream desynced beyond repair — rebuilding the " + "session (conversation memory for this session resets)") + try: + await self._client.disconnect() + except Exception: + pass + self._client = None + await self.start() + self._dirty = False + + async def stop(self): + if self._client: + await self._client.disconnect() + self._client = None + + def _drain_idle(self) -> bool: + """Throw away whatever a BACKGROUND turn left in the pipe. + + THE OTHER OFF-BY-ONE: the agent can take turns nobody asked for + — background-task notifications (a finished Bash job, a Monitor + event, a timeout) wake the model while the mic is quiet, and it + answers. Nothing here is reading the stream at that moment, so + that answer (text + ResultMessage) sits buffered. The next real + question then pairs with it: the person hears the reply to the + notification, and every answer after that is one question late. + reset_turn can't see it (the turn wasn't ours, _dirty is False). + So before every query: pull everything already buffered, non- + blocking, and log what got dropped. Returns True when the last + drained message shows a background turn still in flight (text + without its ResultMessage) — the caller then waits for that + turn to finish before sending, or the same pairing breaks.""" + q = getattr(self._client, "_query", None) + rx = getattr(q, "_message_receive", None) + if rx is None: + return False + n, open_turn, texts = 0, False, [] + while True: + try: + m = rx.receive_nowait() + except anyio.WouldBlock: + break + except Exception: + break + t = m.get("type") if isinstance(m, dict) else None + if t in ("end", "error"): + # Lifecycle markers — put them back for receive_messages + # to handle; nothing after them matters. + try: + q._message_send.send_nowait(m) + except Exception: + pass + break + n += 1 + if t == "result": + open_turn = False + elif t == "assistant": + open_turn = True + for b in (m.get("message", {}) or {}).get("content", []) or []: + if isinstance(b, dict) and b.get("type") == "text": + texts.append((b.get("text") or "").strip()) + elif t in ("user", "stream_event"): + open_turn = True + if n: + log(f"[brain] dropped {n} buffered messages from a background " + f"turn (not spoken; see below)") + for x in texts: + if x: + log(f"[brain] (unspoken) {x[:300]}") + return open_turn + + async def ask_stream(self, utterance: str): + """Yield complete sentences as they stream out of the model.""" + if self._drain_idle(): + # A background turn is mid-flight: let it finish (bounded) + # so our question can't pair with its ResultMessage. + async def _finish(): + async for msg in self._client.receive_response(): + if type(msg).__name__ == "ResultMessage": + break + try: + await asyncio.wait_for(_finish(), 30) + log("[brain] waited out an in-flight background turn") + except Exception: + log("[brain] background turn didn't finish in 30s — " + "sending anyway") + self._dirty = True # in flight until its ResultMessage + await self._client.query(utterance) + buf = "" + # The activity file on the bus is cleared in the finally: it + # covers the clean finish (ResultMessage), a cancelled turn (the + # interrupt lands at the await below and unwinds through here), + # and a failed one, so no stale "Read: foo.py" ever outlives its + # turn on the face. + try: + async for msg in self._client.receive_response(): + t = type(msg).__name__ + if t == "StreamEvent": + ev = getattr(msg, "event", {}) or {} + if ev.get("type") == "content_block_delta": + delta = ev.get("delta", {}) or {} + if delta.get("type") == "text_delta": + buf += delta.get("text", "") + # emit any complete sentences + while True: + m = _SENTENCE_END.search(buf) + if not m: + break + sentence, buf = (buf[:m.end()].strip(), + buf[m.end():]) + if sentence: + yield sentence + elif ev.get("type") == "content_block_stop": + # End of a speech block (e.g. right before a + # tool call): flush NOW. Without this, pre-tool + # filler ("On it — let me grab that.") sits + # silent in the buffer through the whole tool + # run, then plays glued to the answer: long + # dead air, then two thoughts at once. + tail = buf.strip() + buf = "" + if tail: + yield tail + elif t == "AssistantMessage": + # The complete message lands right as its tool + # calls run: print what the agent is DOING while + # the voice is quiet, so a long silence never reads + # as a dead line — and put the same line on the bus + # so the face can show it. + for b in getattr(msg, "content", []) or []: + if type(b).__name__ == "ToolUseBlock": + line = _tool_line(b.name, b.input, + prefix=False) + log(f"[tool] {line}") + signals.activity(line) + elif t == "ResultMessage": + self._dirty = False # turn fully consumed — aligned + self._tally(msg) + self._remember_session(msg) + break + finally: + signals.turn_end() + tail = buf.strip() + if tail: + yield tail + + +def _tool_line(name: str, inp, prefix: bool = True) -> str: + """One line per tool call — the file, command, pattern or query that + says what the agent is up to, not the raw JSON. prefix=True adds the + terminal's "[tool] " tag; the bare form goes on the signal bus.""" + inp = inp if isinstance(inp, dict) else {} + if name in ("Read", "Write", "Edit", "NotebookEdit"): + what = inp.get("file_path") or inp.get("notebook_path") or "" + elif name == "Bash": + what = inp.get("description") or inp.get("command") or "" + elif name in ("Grep", "Glob"): + what = inp.get("pattern") or "" + if inp.get("path"): + what = f"{what} in {inp['path']}" + elif name == "WebFetch": + what = inp.get("url") or "" + elif name == "WebSearch": + what = inp.get("query") or "" + elif name in ("Agent", "Task"): + what = inp.get("description") or "" + elif name == "Skill": + what = inp.get("skill") or "" + else: + what = ", ".join(f"{k}={str(v)[:40]}" + for k, v in list(inp.items())[:3]) + what = " ".join(str(what).split()) + if len(what) > 100: + what = what[:97] + "..." + line = f"{name}: {what}" if what else name + return f"[tool] {line}" if prefix else line + + +if __name__ == "__main__": + import time + + async def demo(): + b = WarmBrain() + await b.start() + for prompt in ("Voice check: greet me in one sentence.", + "And what's two plus two, spoken like yourself?"): + t0 = time.time() + async for s in b.ask_stream(prompt): + print(f" ({time.time()-t0:4.1f}s) {s}", flush=True) + await b.stop() + + asyncio.run(demo()) diff --git a/backtalk/config.py b/backtalk/config.py index 7aa4a1f..a915ddf 100644 --- a/backtalk/config.py +++ b/backtalk/config.py @@ -178,7 +178,8 @@ "by exact spoken phrases, never by you. Permissions: 'stop " "asking for permission' (then 'confirm'), or 'start asking " "again'. Microphone: 'go hands free', or 'push to talk mode'. " - "Also: 'clear the session', 'compact the session', 'switch to " + "Also: 'clear the session' (or 'start fresh'), 'compact the " + "session', 'switch to " "the deep model', 'back to the fast model', 'set effort to low' " "(or medium, high, max), and 'usage report'. You cannot flip " "these live yourself, so when asked, give the person the exact " diff --git a/backtalk/main.py b/backtalk/main.py index bca7497..5a9966f 100644 --- a/backtalk/main.py +++ b/backtalk/main.py @@ -1,1029 +1,1083 @@ -# backtalk: talk to your Claude Code agent out loud. -# Copyright (C) 2026 Jared Rhodenizer -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU Affero General Public License as published -# by the Free Software Foundation, either version 3 of the License, or -# (at your option) any later version. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU Affero General Public License for more details. -# -# You should have received a copy of the GNU Affero General Public License -# along with this program. If not, see . -# -# SPDX-License-Identifier: AGPL-3.0-or-later -"""backtalk — talk to your Claude Code agent out loud. - -Flow: hold the key and speak -> local transcription -> your agent's warm -Claude session streams the reply -> sentences go to the mouth the moment -they complete (~1-2s to first audio on warm turns). The greeting plays -over a hidden warmup query so the first real turn is already hot. - -Typing in this terminal is a first-class turn too: same conversation, -spoken reply, and typing while it talks interrupts it. - -THE VOICE CONSOLE: exact phrases, spoken (or typed) alone, control the -session itself so you never go back to the keyboard: "clear the -session" / "compact the session" / "switch to the deep model" / "back -to the fast model" / "set effort to low" (or medium, high, max) / -"usage report" / "go hands free" and "push to talk mode" (the MIC) / -"stop asking for permission" and "start asking again" (permissions, -called auto-approve, a different axis than the microphone on purpose). -And with permission_mode "ask" (the default), gated tool calls ASK OUT -LOUD and your spoken yes or no decides them; any other answer is -passed back to the agent as the reason. - -Flags: - --open-mic start in hands-free listening for this session (the - config key mic_mode makes it the standing default, and - the voice can switch live either way: "go hands free" / - "push to talk mode"). Know the tradeoff: room audio (a - video, music, another voice assistant) can trigger - replies to speech never meant for the agent. The talk - key keeps working: it interrupts, and holding it always - gets you heard. - --barge-in with --open-mic: keep listening WHILE speaking. - HEADPHONES REQUIRED — with open speakers the mic hears - the reply and the agent interrupts itself. - --model X override the model for this session (full id). - -Say "goodbye " / "end voice mode" to hang up. Ctrl-C works. -""" -import asyncio -import json -import queue -import sys -import threading -import time - -from backtalk import signals -from backtalk.brain import WarmBrain -from backtalk.config import CFG -from backtalk.ears import Ears, record_held, warm as warm_ears -from backtalk.mouth import Mouth -from backtalk.ptt import PTTListener -from backtalk.vlog import log - -NAME = CFG["name"] -QUIT_PHRASES = CFG["quit_phrases"] - -# ---- THE SPOKEN PERMISSION GATE (permission_mode "ask", the default). -# When the agent wants a gated tool, the SDK routes the decision here: -# the ask is spoken, the turn pauses (the SDK waits indefinitely; the -# timeout below is ours), and the NEXT utterance or typed line is the -# answer. "yes" approves; anything else denies, with the user's own -# words passed back as the reason. Silence means no. -PERM_TIMEOUT_S = 75 -_PERM = {"fut": None, "asked_at": 0.0, # pending ask + when it was posed - "hinted": False} # escape-hatch hint said yet? -_CONFIRM = {"verb": None, "at": 0.0} # pending "say confirm" + when -_INTERRUPT_ANSWER = "\x00interrupt" # sentinel: turn is being killed -# Live AUTO-APPROVE is OUR flag, not an SDK mode flip: the CLI refuses -# a live switch INTO bypassPermissions unless it was launched with the -# danger flag, so instead the gate below auto-approves silently while -# this is on. Same behavior, no reconnect, conversation intact. A -# session that BOOTS in bypassPermissions never consults the gate at -# all; saying "start asking again" flips the SDK side live (that -# direction is allowed) and turns this off. ONLY the explicit -# bypassPermissions value arms this: any other mode (acceptEdits, plan) -# passes through to the SDK and keeps the spoken gate for whatever the -# SDK routes here. (Auto-approve is about PERMISSIONS; hands-free -# LISTENING is about the microphone: see _MIC below. Two different -# axes, deliberately never sharing a name.) -_AUTOAPPROVE = {"on": False} -# The microphone mode, switchable live by voice. "ptt" = mic closed -# except while the key is held. "open" = hands-free listening (VAD). -# The key keeps working in open mode: it interrupts, and holding it -# always gets you heard. gen bumps on every switch so an in-flight -# open-mic capture from before the switch gets discarded, never -# processed. -_MIC = {"mode": "ptt", "gen": 0, "btn": False} - -# Approvals are EXACT matches after normalization, never prefixes: -# "yesterday", "yes or no", and "yes, but do not overwrite" must all -# fail. Anything that is not an exact yes DENIES, with the words passed -# back to the agent as the reason. Deny is always the default. -_YES = {"yes", "yeah", "yep", "yup", "sure", "approve", "approved", - "go ahead", "do it", "yes please", "yes sir", "yes boss", - "yes go ahead", "go for it", "green light", "okay", "ok", "y"} -_CHAIN_MARKS = ("&&", "||", ";", "|", "$(", "`", "\n") - - -def _norm_speech(text): - """Lowercase, every non-letter to space, collapse. Whisper loves - interior commas ("yes, confirm"); end-stripping alone misses them.""" - out = [] - for ch in text.lower(): - out.append(ch if "a" <= ch <= "z" else " ") - return " ".join("".join(out).split()) - - -def _deny_pending(reason=_INTERRUPT_ANSWER): - """Resolve a pending spoken ask as a deny. Called whenever the turn - that posed it is being interrupted, so the ask can never outlive its - turn and hijack a later utterance (or stall the pipe drain).""" - f = _PERM["fut"] - if f is not None and not f.done(): - f.set_result(reason) - - -def _human_what(tool, tool_input, ctx): - """The SHORT spoken form, built for a person who has never seen a - terminal: plain words, no paths, no syntax. Built by code, never by - the model, so it cannot understate; and every ask offers "details", - which reads the full literal form below. (Field case: the gate read - whole file paths and command syntax at a brand-new user.)""" - d = tool_input or {} - if tool in ("Write", "Edit", "MultiEdit", "NotebookEdit"): - path = str(d.get("file_path") or d.get("notebook_path") - or "a file").replace("\\", "/") - name = path.rsplit("/", 1)[-1] - import os as _os - homes = [CFG.get("agent_dir", "")] + list(CFG.get("extra_dirs") - or []) - in_vault = any(h and path.startswith(str(h).rstrip("/") + "/") - for h in (CFG.get("extra_dirs") or [])) - verb = "edit" if "Edit" in tool else "create or change" - if in_vault and name.endswith(".md"): - return f"{verb} a note in your vault called {name[:-3]}" - return f"{verb} a file called {name}" - if tool == "Bash": - cmd = " ".join(str(d.get("command", "")).split()) - first = (cmd.split() or ["a"])[0].rsplit("/", 1)[-1] - chained = any(m in cmd for m in _CHAIN_MARKS) - return (f"run a {first} command in the terminal" - + (", with several chained parts" if chained else "")) - if tool == "WebFetch": - url = str(d.get("url", "")) - host = url.split("//", 1)[-1].split("/", 1)[0] or "a site" - return f"read a web page at {host}" - name = getattr(ctx, "display_name", None) or tool - return f"use the {name} tool" - - -_DETAILS = {"details", "the details", "give me details", - "give me the details", "what command", "what is it", - "say more", "more", "what exactly", "the exact command"} - - -def _full_detail(tool, tool_input, ctx): - """The full literal form, spoken only when the person asks for - "details". Never lets a long command hide its tail: truncation is - DISCLOSED and shell chaining is called out (the agent composes - tool_input itself, so this line must not be steerable into - understatement).""" - d = tool_input or {} - if tool == "Bash": - cmd = " ".join(str(d.get("command", "")).split()) - chained = any(m in cmd for m in _CHAIN_MARKS) - line = ("a chained command: " if chained else - "run a command: ") + cmd[:90] - if len(cmd) > 90: - line += (f", and {len(cmd) - 90} more characters. " - "Check the log before approving") - return line - if tool in ("Write", "Edit", "MultiEdit", "NotebookEdit"): - path = str(d.get("file_path") or d.get("notebook_path") - or "a file").replace("\\", "/") - bits = path.rsplit("/", 2) - name = "/".join(bits[-2:]) if len(bits) >= 2 else path - return f"{'edit' if 'Edit' in tool else 'write'} the file {name}" - if tool == "WebFetch": - return f"fetch a web page: {str(d.get('url', ''))[:70]}" - desc = (getattr(ctx, "description", None) or "").strip() - name = getattr(ctx, "display_name", None) or tool - return f"use {name}" + (f", {desc[:70]}" if desc else "") - - -def make_permission_gate(mouth): - from claude_agent_sdk import (PermissionResultAllow, - PermissionResultDeny) - - async def gate(tool, tool_input, ctx): - if _AUTOAPPROVE["on"]: - return PermissionResultAllow(behavior="allow") - what = _human_what(tool, tool_input, ctx) - detail = _full_detail(tool, tool_input, ctx) - loop = asyncio.get_running_loop() - signals.static_stop() - log(f"[perm] asking: {what}") - log(f"[perm] detail: {detail}") - if tool == "Bash": # the FULL command always reaches the log - log(f"[perm] full command: {str((tool_input or {}).get('command', ''))[:2000]}") - ask = f"Permission check. I want to {what}. Yes, no, or details?" - if not _PERM["hinted"]: - # the escape hatch announces itself exactly once, at the - # moment it becomes relevant (a field case: a new user - # couldn't find the phrase to turn the checks off) - _PERM["hinted"] = True - ask += (" And any time you're done with these checks, say " - "stop asking for permission.") - mouth.say(ask) - answer = None - try: - deadline = loop.time() + PERM_TIMEOUT_S - while answer is None: - fut = loop.create_future() - _PERM["fut"] = fut - _PERM["asked_at"] = time.monotonic() - while True: - try: - got = await asyncio.wait_for( - asyncio.shield(fut), 1.0) - break - except asyncio.TimeoutError: - if loop.time() >= deadline: - fut.cancel() - mouth.say("No answer, so I didn't do it.") - log("[perm] timed out, denied") - return PermissionResultDeny( - behavior="deny", - message="No spoken answer within the " - "timeout; the action was not " - "approved.", - interrupt=False) - # keep the ring honest while we wait - if not mouth.speaking: - signals.set_state("listening") - if (got != _INTERRUPT_ANSWER - and _norm_speech(got) in _DETAILS): - # read the full literal form, then ask again with a - # fresh clock: asking for details is engagement, - # not silence - log("[perm] details requested") - mouth.say(f"The details: I want to {detail}. " - "Yes or no?") - deadline = loop.time() + PERM_TIMEOUT_S - continue - answer = got - finally: - _PERM["fut"] = None - if answer == _INTERRUPT_ANSWER: - log("[perm] turn interrupted, denied silently") - return PermissionResultDeny( - behavior="deny", - message="Interrupted by the user; the turn is being " - "cancelled.", - interrupt=False) - approved = _norm_speech(answer) in _YES - # the model keeps working either way: restore the working state - signals.set_state("thinking") - signals.static_start() - if approved: - log("[perm] approved by voice") - return PermissionResultAllow(behavior="allow") - log(f"[perm] denied: {answer!r}") - return PermissionResultDeny( - behavior="deny", - message=f'Denied by voice. The user said: "{answer[:500]}"', - interrupt=False) - return gate - - -# ---- THE VOICE CONSOLE: session verbs, spoken. Exact phrases only, -# spoken alone, so ordinary sentences can never trigger them. (Grown -# from a community member's own build shared in the Discord.) -CONSOLE_VERBS = { - "clear": ("clear the session", "clear the context", - "clear context", "fresh slate", "slash clear"), - "compact": ("compact the session", "compact the context", - "compact context", "slash compact"), - "deep": ("switch to the deep model", "use the deep model", - "slash model deep"), - "fast": ("switch to the fast model", "use the fast model", - "back to the fast model", "slash model fast"), - "usage": ("usage report", "slash usage"), - "micopen": ("go hands free", "hands free mode", - "hands free listening", "open mic", "open the mic"), - "micptt": ("push to talk", "push to talk mode", - "back to push to talk", "back to the button"), - "noask": ("stop asking for permission", - "stop asking permission", - "stop asking me for permission", - "turn off the permission prompt", - "turn off the permission prompts", - "turn off the permissions prompt", - "turn off the permissions prompts", - "turn off permissions", "turn off permission checks", - "disable the permission checks", - "disable permission checks", "auto approve", - "auto approve mode"), - "ask": ("start asking again", "ask before acting", - "ask for permission again"), -} -_EFFORTS = ("low", "medium", "high", "xhigh", "max") - - -def console_match(text): - norm = " ".join(text.lower().replace("-", " ").split()).strip(" .,!?") - for verb, phrases in CONSOLE_VERBS.items(): - if norm in phrases: - return verb - for lvl in _EFFORTS: - if norm in (f"set effort to {lvl}", f"effort {lvl}", - f"slash effort {lvl}"): - return f"effort:{lvl}" - return None - - -def _write_config_key(key, value): - """The agent rewrites the config; the person never hand-edits it. - Returns True on a persisted write. A file that fails to PARSE is - left untouched (rewriting from {} would wipe every other setting); - the in-memory CFG updates either way so the session behaves.""" - from backtalk.config import CONFIG_PATH - CFG[key] = value - try: - data = json.loads(CONFIG_PATH.read_text()) - except FileNotFoundError: - data = {} - except (OSError, ValueError) as e: - log(f"[console] config not writable/parsable, session-only: {e}") - return False - data[key] = value - try: - CONFIG_PATH.write_text(json.dumps(data, indent=2) + "\n") - except OSError as e: - log(f"[console] config write failed, session-only: {e}") - return False - return True - - -def _fmt_tokens(n): - if n >= 1_000_000: - return f"about {round(n / 1_000_000, 1):g} million tokens" - if n >= 1000: - return f"about {round(n / 1000)} thousand tokens" - return f"{n} tokens" - - -def _spoken_usage(sess, ctx_usage): - """A short CFO brief of the session, written for the ear: plain - numerals only (the TTS reads "40" fine; symbols come out garbled).""" - turns = sess["turns"] - parts = [f"{turns} turn{'s' if turns != 1 else ''} this session", - _fmt_tokens(sess["out_tokens"]) + " spoken out"] - cents = round(sess["cost"] * 100) - if cents >= 1: - parts.append(f"roughly {cents} cents" if cents < 100 - else f"roughly {round(cents / 100)} dollars") - try: - cats = (getattr(ctx_usage, "categories", None) - or (ctx_usage or {}).get("categories") or []) - # the breakdown includes "Free space" and the autocompact - # buffer; only OCCUPIED categories belong in the spoken number - total = sum(int(c.get("tokens") or 0) for c in cats - if isinstance(c, dict) - and "free" not in str(c.get("name", "")).lower() - and "buffer" not in str(c.get("name", "")).lower()) - if total: - parts.append(_fmt_tokens(total) - + " sitting in the context window") - except Exception: - pass - return ". ".join(parts) + "." - -_PASTE_ON = "\x1b[200~" # bracketed-paste markers (we enable the mode below) -_PASTE_OFF = "\x1b[201~" - - -def _clean_typed(line: str) -> str: - """Scrub terminal-copy artifacts: blockquote gutter glyphs and stray - whitespace (copying from a CLI chat render drags bars along).""" - line = line.strip() - while line[:1] in ("▎", "│", ">"): - line = line[1:].lstrip() - return line - - -def _join_paste(body: str) -> str: - """Pasted blob -> one clean message (gutters scrubbed, lines joined).""" - parts = [_clean_typed(l) for l in body.split("\n")] - return " ".join(" ".join(p for p in parts if p).split()) - - -def _typed_reader_pipe(q: "queue.Queue[str]", fd: int): - """Non-tty stdin (pipes/tests): line assembly with paste markers.""" - import os - pend = "" - while True: - try: - b = os.read(fd, 65536) - except OSError: - return - if not b: - return - pend += b.decode("utf-8", "replace") - while True: - if _PASTE_ON in pend: - if _PASTE_OFF not in pend: - break - head, rest = pend.split(_PASTE_ON, 1) - body, pend = rest.split(_PASTE_OFF, 1) - *hlines, hpart = head.split("\n") - for l in hlines: - l = _clean_typed(l) - if l: - q.put(l) - text = _join_paste(hpart + body) - if text: - q.put(text) - continue - if "\n" in pend: - line, pend = pend.split("\n", 1) - line = _clean_typed(line) - if line: - q.put(line) - continue - break - - -def _typed_reader_simple(q: "queue.Queue[str]"): - """Windows (no termios): plain line input on a thread. Pastes work; - they just echo normally instead of collapsing to a count.""" - while True: - try: - line = _clean_typed(input()) - except (EOFError, OSError): - return - if line: - q.put(line) - - -def _typed_reader(q: "queue.Queue[str]"): - """Terminal stdin -> typed messages (daemon thread). Typed lines are - first-class turns: same pipeline as a spoken utterance, spoken reply. - - On a POSIX tty we OWN the input line (cbreak: no kernel echo, no - canonical buffering — the little line editor below echoes keys, - handles backspace, and assembles bracketed pastes invisibly). The - kernel's canonical mode is unfixable for pastes: it echoes the - markers as visible junk and holds unfinished marker lines hostage. - Pastes show as `[pasted N chars]`; Enter sends everything as ONE - message. Ctrl-C still works (ISIG stays on); termios restored at - exit.""" - import atexit - import os - fd = sys.stdin.fileno() - if not os.isatty(fd): - _typed_reader_pipe(q, fd) - return - try: - import termios - import tty as _tty - except ImportError: # Windows: no termios — simple reader - _typed_reader_simple(q) - return - old = termios.tcgetattr(fd) - _tty.setcbreak(fd) # ECHO+ICANON off, ISIG kept - sys.stdout.write("\x1b[?2004h") # bracket pastes, please - sys.stdout.flush() - - def _restore(): - try: - termios.tcsetattr(fd, termios.TCSADRAIN, old) - except Exception: - pass - sys.stdout.write("\x1b[?2004l") - sys.stdout.flush() - atexit.register(_restore) - - MARKS = (_PASTE_ON, _PASTE_OFF) - - def _partial_tail(s: str) -> int: - """Length of a trailing partial paste-marker (hold it for the - next read).""" - for m in MARKS: - for k in range(min(len(s), len(m) - 1), 0, -1): - if m.startswith(s[-k:]): - return k - return 0 - - buf = "" # the input line being composed - paste = None # accumulating paste body, or None - pend = "" - while True: - try: - b = os.read(fd, 4096) - except OSError: - _restore() - return - if not b: - _restore() - return - pend += b.decode("utf-8", "replace") - keep = _partial_tail(pend) - proc = pend[:len(pend) - keep] if keep else pend - pend = pend[len(pend) - keep:] if keep else "" - i = 0 - while i < len(proc): - if paste is not None: - j = proc.find(_PASTE_OFF, i) - if j < 0: - paste += proc[i:] - break - paste += proc[i:j] - i = j + len(_PASTE_OFF) - text = _join_paste(paste) - paste = None - if text: - if buf and not buf.endswith(" "): - buf += " " - buf += text - sys.stdout.write(text if len(text) <= 60 - else f"[pasted {len(text)} chars]") - sys.stdout.flush() - continue - if proc.startswith(_PASTE_ON, i): - paste = "" - i += len(_PASTE_ON) - continue - ch = proc[i] - i += 1 - if ch in ("\r", "\n"): - sys.stdout.write("\n") - sys.stdout.flush() - line = buf.strip() - buf = "" - if line: - q.put(line) - elif ch in ("\x7f", "\x08"): # backspace - if buf: - buf = buf[:-1] - sys.stdout.write("\b \b") - sys.stdout.flush() - elif ch >= " " or ch == "\t": # printable: echo + collect - buf += ch - sys.stdout.write(ch) - sys.stdout.flush() - - -async def speak_reply(brain: WarmBrain, mouth: Mouth, text: str): - """First sentence ships alone (fast start); the rest go in - 2-sentence breaths — fuller chunks get livelier prosody (single - short sentences come out flat).""" - t0 = time.time() - first = True - batch: list[str] = [] - - def emit(raw: str): - nonlocal first, batch - # TTS hygiene: backticks, markdown fences and angle-bracket tag - # syntax are never speakable — the mouth gets clean prose only. - s = raw.replace("`", "").replace("<<", "").replace(">>", "").strip() - if not s: - return - if first: - log(f"[{NAME}] ({time.time()-t0:.1f}s to first) {s}") - mouth.say_chunk(s) - first = False - else: - log(f"[{NAME}] {s}") - batch.append(s) - if len(batch) >= 2: - mouth.say_chunk(" ".join(batch)) - batch = [] - - try: - async for sentence in brain.ask_stream(text): - emit(sentence) - if batch: - mouth.say_chunk(" ".join(batch)) - if first: - # Zero sentences yielded (brain error / empty turn): nothing - # will ever dequeue, so nothing resets the bus — park it here. - signals.static_stop() - signals.set_state("idle") - except asyncio.CancelledError: - try: - await brain.interrupt() - except Exception: - pass - raise - - -async def amain(): - open_mic = "--open-mic" in sys.argv - barge_in = "--barge-in" in sys.argv - model = None - if "--model" in sys.argv: - try: - model = sys.argv[sys.argv.index("--model") + 1] - except IndexError: - pass - - CFG_BOOT_MODE = CFG["permission_mode"] - _AUTOAPPROVE["on"] = CFG_BOOT_MODE == "bypassPermissions" - _MIC["mode"] = "open" if (open_mic - or CFG.get("mic_mode") == "open") else "ptt" - # resume_last_session: reattach to the saved conversation, if any - resume_id = None - if CFG.get("resume_last_session"): - try: - from backtalk.brain import SESSION_FILE - with open(SESSION_FILE) as f: - resume_id = f.read().strip() or None - except OSError: - resume_id = None - - mouth = Mouth() - ears = Ears() - brain = WarmBrain(model=model, - can_use_tool=make_permission_gate(mouth), - resume_id=resume_id) - - mode = ("hands-free listening (the talk key still works)" - if _MIC["mode"] == "open" - else f"push-to-talk ({CFG['ptt_key']})") - log(f"[backtalk] up — agent={NAME} dir={CFG['agent_dir']} " - f"model={brain.model} mic={mode} " - f"(say 'goodbye {NAME.lower()}' to hang up)") - mouth.say(CFG["greeting"]) - - loop = asyncio.get_event_loop() - # Warm the engines while the greeting plays: the STT model load and - # the brain's prompt-cache toll both hide behind the spoken line. - loop.run_in_executor(None, warm_ears) - # THE BRAIN CONNECT, guarded. This is the one startup step that - # needs a signed-in Claude Code, internet, and available usage. - # When it fails or hangs, the mouth still works, so SAY SO instead - # of dying silently with the face stuck on idle (a real field - # case: the greeting played, then nothing, and on Windows the - # window closed before anyone could read the error). - log("[backtalk] connecting the brain...") - try: - await asyncio.wait_for(brain.start(), 120) - - async def _warmup(): - async for _ in brain.ask_stream( - "Warmup ping - reply with the single word: ready"): - pass - await asyncio.wait_for(_warmup(), 180) - except (Exception, asyncio.TimeoutError) as e: - kind = ("timed out" if isinstance(e, asyncio.TimeoutError) - else f"failed: {e!r}"[:220]) - log(f"[backtalk] BRAIN CONNECT {kind}") - mouth.say("Bad news. The voice and the face are fine, but I " - "couldn't reach my brain, the Claude Code session. " - "Check this window for the error. The usual causes: " - "Claude Code isn't signed in, the internet is down, " - "or the plan is out of usage.") - mouth.wait_done(timeout=30) - raise SystemExit(1) - log("[backtalk] brain warm") - # the hidden warmup ping is plumbing, not conversation - brain.session.update(turns=0, out_tokens=0, in_tokens=0, cost=0.0) - # a configured effort level applies at launch (saved by the spoken - # "set effort to X", or written by the person's agent on request) - boot_effort = str(CFG.get("effort") or "").strip().lower() - if boot_effort in _EFFORTS: - await brain.command(f"/effort {boot_effort}") - log(f"[backtalk] effort set to {boot_effort} (from config)") - elif boot_effort: - log(f"[backtalk] ignoring unknown effort {boot_effort!r} in config") - - speak_task: asyncio.Task | None = None - typed_q: "queue.Queue[str]" = queue.Queue() - threading.Thread(target=_typed_reader, args=(typed_q,), daemon=True).start() - typed_fut: asyncio.Future | None = None - - async def run_console(verb): - """One voice-console verb. The current reply was already - cancelled and awaited by handle(); the pipe gets drained here - before the command goes out. A verb that blows up must never - take the whole voice session down with it.""" - try: - await _run_console_inner(verb) - except Exception as e: - log(f"[console] {verb} failed: {e}") - mouth.say("That command hit an error. Check the log.") - signals.set_state("idle") - - async def _run_console_inner(verb): - _deny_pending() - await brain.reset_turn() - say_after = None - if verb == "clear": - resp = await brain.command("/clear") - say_after = "Cleared. Fresh slate." - elif verb == "compact": - mouth.say("Compacting. One moment.") - resp = await brain.command("/compact") - say_after = "Compacted. Same conversation, smaller footprint." - elif verb == "deep": - mouth.say("Switching to the deep model. Heads up, replies " - "get slower. Say back to the fast model when " - "you're done.") - resp = await brain.command(f"/model {CFG['deep_model']}") - say_after = "Deep model online, for this session only." - elif verb == "fast": - resp = await brain.command(f"/model {CFG['model']}") - say_after = "Back on the fast model." - elif verb.startswith("effort:"): - lvl = verb.split(":", 1)[1] - resp = await brain.command(f"/effort {lvl}") - saved = _write_config_key("effort", lvl) - say_after = (f"Effort set to {lvl}, and saved as your " - "default." if saved else - f"Effort set to {lvl} for this session. The " - "config file couldn't be written, so it won't " - "stick past a restart.") - elif verb == "usage": - resp = "" - mouth.say(_spoken_usage(brain.session, - await brain.context_usage())) - elif verb == "micopen": - resp = "" - if _MIC["mode"] == "open": - mouth.say("Already in hands-free listening.") - else: - _MIC["mode"] = "open" - _MIC["gen"] += 1 - _write_config_key("mic_mode", "open") - log("[console] mic_mode -> open (hands-free listening)") - mouth.say("Hands-free listening on. I'm always " - "listening now, so anything said in the room " - "can reach me. The talk key still works, and " - "holding it always gets you heard. Say push " - "to talk mode to bring the button back.") - elif verb == "micptt": - resp = "" - if _MIC["mode"] == "ptt": - mouth.say("Already on push to talk.") - else: - _MIC["mode"] = "ptt" - _MIC["gen"] += 1 - _write_config_key("mic_mode", "ptt") - log("[console] mic_mode -> ptt") - key = str(CFG.get("ptt_key", "home")).replace("_", " ") - mouth.say(f"Push to talk. Hold the {key} key and " - "talk; the mic stays closed otherwise.") - elif verb == "noask": - resp = "" - _CONFIRM["verb"] = "noask" - _CONFIRM["at"] = time.monotonic() - mouth.say("Auto-approve means I act without asking " - "permission, and it becomes your saved default. " - "Say confirm to switch.") - elif verb == "noask:confirmed": - resp = "" - saved = _write_config_key("permission_mode", - "bypassPermissions") - _AUTOAPPROVE["on"] = True - log("[console] permission_mode -> bypassPermissions" - + (" (saved)" if saved else " (session only)")) - mouth.say(("Auto-approve on, and saved as your default. " - if saved else - "Auto-approve on for this session. The config " - "file couldn't be written, so it won't stick " - "past a restart. ") - + "Say start asking again any time to flip it " - "back.") - elif verb == "ask": - resp = "" - saved = _write_config_key("permission_mode", "ask") - _AUTOAPPROVE["on"] = False - flipped = True - if CFG_BOOT_MODE == "bypassPermissions": - # a bypass-booted session never consults the gate, so - # the SDK itself must flip (the safe direction is - # allowed live). If that fails, saying "done" would be - # a lie: the agent would keep acting silently. - try: - await brain.set_permission_mode("ask") - except Exception as e: - flipped = False - log(f"[console] live flip to ask FAILED: {e}") - log("[console] permission_mode -> ask" - + (" (saved)" if saved else " (session only)")) - if flipped: - mouth.say("Done. I'll ask out loud before real " - "actions" - + (", and that's saved as your default." - if saved else - ". The config file couldn't be written, " - "so tell me again after a restart.")) - else: - mouth.say("I saved asking as your default, but this " - "session couldn't switch over. Restart the " - "voice line to get asking back.") - else: - resp = "" - if say_after: - # the CLI answers slash commands with its own text - # (confirmations, API errors); an error outranks our line - low = (resp or "").lower() - if resp and ("error" in low or "invalid" in low): - mouth.say(resp[:160]) - log(f"[console] {verb} answered: {resp[:120]}") - else: - mouth.say(say_after) - signals.set_state("idle") - - async def handle(text: str, spoke_from: float | None = None) -> bool: - """Process one utterance; returns False on quit. spoke_from is - when the utterance STARTED (the PTT press), so an answer can be - told apart from speech that began before the ask even existed.""" - nonlocal speak_task - log(f"[you] {text}") - # A pending spoken permission ask owns the next utterance IF - # that utterance started after the ask was posed. Speech that - # began earlier is the user interrupting the turn, not - # answering a question they never heard: the ask resolves as a - # silent deny and the utterance falls through as a normal - # interrupt. Quit wins either way, but only as an EXACT phrase - # here ("No! Don't hang up, skip it" must stay a deny reason, - # not kill the session). - if _PERM["fut"] is not None and not _PERM["fut"].done(): - started_after = (spoke_from is None - or spoke_from >= _PERM["asked_at"]) - if _norm_speech(text) in {_norm_speech(q) - for q in QUIT_PHRASES}: - _PERM["fut"].set_result("no") - # falls through to the quit body below - elif started_after: - _PERM["fut"].set_result(text) - return True - else: - _deny_pending() - # A pending auto-approve confirm owns it too, for two minutes; - # after that it expires and speech flows normally again. - verb = None - if _CONFIRM["verb"]: - pend, _CONFIRM["verb"] = _CONFIRM["verb"], None - expired = time.monotonic() - _CONFIRM["at"] > 120 - if not expired and _norm_speech(text) in ( - "confirm", "confirmed", "yes confirm", - "yes confirmed"): - verb = pend + ":confirmed" - elif not expired and not any(q in text.lower() - for q in QUIT_PHRASES): - mouth.say("Staying as we are.") - return True - if any(q in text.lower() for q in QUIT_PHRASES): - if speak_task and not speak_task.done(): - speak_task.cancel() - mouth.shut_up() - mouth.say(CFG["signoff"]) - mouth.wait_done(timeout=15) - return False - if speak_task and not speak_task.done(): - log("[turn] interrupted mid-reply by new input") - _deny_pending() # an ask never outlives its turn - speak_task.cancel() - mouth.shut_up() - if speak_task: - # Let the cancellation fully land (its brain.interrupt() - # included) BEFORE anything else touches the brain — - # otherwise the dead turn's stop signal can race in after - # the new query and kill the new answer (half of the - # off-by-one bug; see brain.reset_turn for the other half). - try: - await speak_task - except asyncio.CancelledError: - pass - except Exception: - pass - speak_task = None - verb = verb or console_match(text) - if verb: - await run_console(verb) - return True - signals.set_state("thinking") - signals.static_start() - # Clean the pipe: drain the interrupted turn's leftovers so the - # new question can't pair with a stale ResultMessage. A gate - # that fired in the meantime resolves first, or the drain would - # wait on a ResultMessage the CLI is withholding for an answer. - _deny_pending() - await brain.reset_turn() - speak_task = asyncio.create_task(speak_reply(brain, mouth, text)) - return True - - try: - # ONE loop, two mic modes, switchable live (_MIC). The talk key - # is constructed and honored in BOTH modes: in hands-free - # listening it is the interrupt and the guaranteed way to be - # heard over room noise. The open mic joins the wait-set only - # in "open" mode; a mode switch bumps _MIC["gen"], the abort - # callable closes the in-flight open mic promptly, and any - # capture born under an old gen is discarded unprocessed. - ptt = PTTListener(CFG["ptt_key"]) - press_fut: asyncio.Future | None = None - mic_fut: asyncio.Future | None = None - mic_gen_seen = _MIC["gen"] - # The open mic yields while the BUTTON records (or the double - # capture would turn one held utterance into two turns), and, - # without barge-in, while the mouth speaks. - mic_gate = (lambda: _MIC["btn"] - or (not barge_in and mouth.speaking)) - mic_fails = 0 - while True: - if _MIC["gen"] != mic_gen_seen: - mic_gen_seen = _MIC["gen"] - # consume futures that completed under the old mode so - # a stale press or capture can't fire after a switch - if press_fut is not None and press_fut.done(): - press_fut.result(); press_fut = None - if mic_fut is not None and mic_fut.done(): - mic_fut.result(); mic_fut = None - if typed_fut is None: - typed_fut = loop.run_in_executor(None, typed_q.get) - if press_fut is None: - press_fut = loop.run_in_executor(None, ptt.wait_press) - waiters = {press_fut, typed_fut} - if _MIC["mode"] == "open": - if mic_fut is None: - g = _MIC["gen"] - mic_fut = loop.run_in_executor( - None, lambda g=g: (g, ears.listen_once( - gate=mic_gate, - abort=lambda: _MIC["gen"] != g))) - waiters.add(mic_fut) - done, _ = await asyncio.wait( - waiters, return_when=asyncio.FIRST_COMPLETED) - if typed_fut in done: - text = typed_fut.result(); typed_fut = None - if text and not await handle(text): - return - continue - if mic_fut is not None and mic_fut in done: - try: - g, text = mic_fut.result() - except Exception as e: - mic_fut = None - mic_fails += 1 - log(f"[ears] open mic failed ({mic_fails}): {e!r}") - if mic_fails >= 3: - _MIC["mode"] = "ptt" - _MIC["gen"] += 1 - mic_fails = 0 - mouth.say("The open microphone keeps failing, " - "so I'm switching to push to talk. " - "Hold the key to reach me, and " - "check this window for the error.") - continue - mic_fut = None - if g != _MIC["gen"]: - continue # captured before a switch - if text and not await handle(text): - return - continue - if press_fut in done: - press_fut.result(); press_fut = None - press_t = time.monotonic() - perm_wait = (_PERM["fut"] is not None - and not _PERM["fut"].done()) - if speak_task and not speak_task.done() and not perm_wait: - log("[turn] interrupted mid-reply — key pressed") - speak_task.cancel() # the button = interrupt - # During a permission ask the TURN stays alive; the - # press only silences playback and records the answer. - mouth.shut_up() - signals.static_stop() # button kills the static too - signals.set_state("listening") - mouth.ducker.speech_start() # duck NOW, while you talk - print("[ptt] recording (release to send)...", flush=True) - _MIC["btn"] = True # open mic yields to the button - try: - text = await loop.run_in_executor( - None, lambda: record_held(ptt.is_held)) - except Exception as e: - log(f"[ears] record/transcribe failed: {e!r}") - mouth.say("My ears hit an error. Check this " - "window for the details.") - text = None - finally: - _MIC["btn"] = False - mouth.ducker.speech_end(0.2) # snap back fast on release - if not text: - log("[ptt] (tap or empty — ignored)") - signals.set_state("idle") - continue - if not await handle(text, spoke_from=press_t): - return - except KeyboardInterrupt: - pass - finally: - _MIC["gen"] += 1 # abort any live open-mic capture promptly - if speak_task and not speak_task.done(): - speak_task.cancel() - mouth.shutdown() # restores the music on Ctrl-C / crash paths too - signals.static_stop() - signals.set_state("idle") - await brain.stop() - log("[backtalk] hung up") - - -def main(): - try: - asyncio.run(amain()) - except KeyboardInterrupt: - print("\n[backtalk] interrupted — hanging up", flush=True) - - -if __name__ == "__main__": - main() +# backtalk: talk to your Claude Code agent out loud. +# Copyright (C) 2026 Jared Rhodenizer +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as published +# by the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . +# +# SPDX-License-Identifier: AGPL-3.0-or-later +"""backtalk — talk to your Claude Code agent out loud. + +Flow: hold the key and speak -> local transcription -> your agent's warm +Claude session streams the reply -> sentences go to the mouth the moment +they complete (~1-2s to first audio on warm turns). The greeting plays +over a hidden warmup query so the first real turn is already hot. + +Typing in this terminal is a first-class turn too: same conversation, +spoken reply, and typing while it talks interrupts it. + +THE VOICE CONSOLE: exact phrases, spoken (or typed) alone, control the +session itself so you never go back to the keyboard: "clear the +session" / "compact the session" / "switch to the deep model" / "back +to the fast model" / "set effort to low" (or medium, high, max) / +"usage report" / "go hands free" and "push to talk mode" (the MIC) / +"stop asking for permission" and "start asking again" (permissions, +called auto-approve, a different axis than the microphone on purpose). +And with permission_mode "ask" (the default), gated tool calls ASK OUT +LOUD and your spoken yes or no decides them; any other answer is +passed back to the agent as the reason. + +Flags: + --open-mic start in hands-free listening for this session (the + config key mic_mode makes it the standing default, and + the voice can switch live either way: "go hands free" / + "push to talk mode"). Know the tradeoff: room audio (a + video, music, another voice assistant) can trigger + replies to speech never meant for the agent. The talk + key keeps working: it interrupts, and holding it always + gets you heard. + --barge-in with --open-mic: keep listening WHILE speaking. + HEADPHONES REQUIRED — with open speakers the mic hears + the reply and the agent interrupts itself. + --model X override the model for this session (full id). + +Say "goodbye " / "end voice mode" to hang up. Ctrl-C works. +""" +import asyncio +import json +import os +import queue +import sys +import threading +import time + +from backtalk import signals +from backtalk.brain import WarmBrain +from backtalk.config import CFG +from backtalk.ears import Ears, record_held, warm as warm_ears +from backtalk.mouth import Mouth +from backtalk.ptt import PTTListener +from backtalk.vlog import log + +NAME = CFG["name"] +QUIT_PHRASES = CFG["quit_phrases"] + +# ---- THE SPOKEN PERMISSION GATE (permission_mode "ask", the default). +# When the agent wants a gated tool, the SDK routes the decision here: +# the ask is spoken, the turn pauses (the SDK waits indefinitely; the +# timeout below is ours), and the NEXT utterance or typed line is the +# answer. "yes" approves; anything else denies, with the user's own +# words passed back as the reason. Silence means no. +PERM_TIMEOUT_S = 75 +_PERM = {"fut": None, "asked_at": 0.0, # pending ask + when it was posed + "hinted": False} # escape-hatch hint said yet? +_CONFIRM = {"verb": None, "at": 0.0} # pending "say confirm" + when +_INTERRUPT_ANSWER = "\x00interrupt" # sentinel: turn is being killed +# Live AUTO-APPROVE is OUR flag, not an SDK mode flip: the CLI refuses +# a live switch INTO bypassPermissions unless it was launched with the +# danger flag, so instead the gate below auto-approves silently while +# this is on. Same behavior, no reconnect, conversation intact. A +# session that BOOTS in bypassPermissions never consults the gate at +# all; saying "start asking again" flips the SDK side live (that +# direction is allowed) and turns this off. ONLY the explicit +# bypassPermissions value arms this: any other mode (acceptEdits, plan) +# passes through to the SDK and keeps the spoken gate for whatever the +# SDK routes here. (Auto-approve is about PERMISSIONS; hands-free +# LISTENING is about the microphone: see _MIC below. Two different +# axes, deliberately never sharing a name.) +_AUTOAPPROVE = {"on": False} +# The microphone mode, switchable live by voice. "ptt" = mic closed +# except while the key is held. "open" = hands-free listening (VAD). +# The key keeps working in open mode: it interrupts, and holding it +# always gets you heard. gen bumps on every switch so an in-flight +# open-mic capture from before the switch gets discarded, never +# processed. +_MIC = {"mode": "ptt", "gen": 0, "btn": False} + +# Approvals are EXACT matches after normalization, never prefixes: +# "yesterday", "yes or no", and "yes, but do not overwrite" must all +# fail. Anything that is not an exact yes DENIES, with the words passed +# back to the agent as the reason. Deny is always the default. +_YES = {"yes", "yeah", "yep", "yup", "sure", "approve", "approved", + "go ahead", "do it", "yes please", "yes sir", "yes boss", + "yes go ahead", "go for it", "green light", "okay", "ok", "y"} +_CHAIN_MARKS = ("&&", "||", ";", "|", "$(", "`", "\n") + + +def _norm_speech(text): + """Lowercase, every non-letter to space, collapse. Whisper loves + interior commas ("yes, confirm"); end-stripping alone misses them.""" + out = [] + for ch in text.lower(): + out.append(ch if "a" <= ch <= "z" else " ") + return " ".join("".join(out).split()) + + +def _deny_pending(reason=_INTERRUPT_ANSWER): + """Resolve a pending spoken ask as a deny. Called whenever the turn + that posed it is being interrupted, so the ask can never outlive its + turn and hijack a later utterance (or stall the pipe drain).""" + f = _PERM["fut"] + if f is not None and not f.done(): + f.set_result(reason) + + +def _human_what(tool, tool_input, ctx): + """The SHORT spoken form, built for a person who has never seen a + terminal: plain words, no paths, no syntax. Built by code, never by + the model, so it cannot understate; and every ask offers "details", + which reads the full literal form below. (Field case: the gate read + whole file paths and command syntax at a brand-new user.)""" + d = tool_input or {} + if tool in ("Write", "Edit", "MultiEdit", "NotebookEdit"): + path = str(d.get("file_path") or d.get("notebook_path") + or "a file").replace("\\", "/") + name = path.rsplit("/", 1)[-1] + import os as _os + homes = [CFG.get("agent_dir", "")] + list(CFG.get("extra_dirs") + or []) + in_vault = any(h and path.startswith(str(h).rstrip("/") + "/") + for h in (CFG.get("extra_dirs") or [])) + verb = "edit" if "Edit" in tool else "create or change" + if in_vault and name.endswith(".md"): + return f"{verb} a note in your vault called {name[:-3]}" + return f"{verb} a file called {name}" + if tool == "Bash": + cmd = " ".join(str(d.get("command", "")).split()) + first = (cmd.split() or ["a"])[0].rsplit("/", 1)[-1] + chained = any(m in cmd for m in _CHAIN_MARKS) + return (f"run a {first} command in the terminal" + + (", with several chained parts" if chained else "")) + if tool == "WebFetch": + url = str(d.get("url", "")) + host = url.split("//", 1)[-1].split("/", 1)[0] or "a site" + return f"read a web page at {host}" + name = getattr(ctx, "display_name", None) or tool + return f"use the {name} tool" + + +_DETAILS = {"details", "the details", "give me details", + "give me the details", "what command", "what is it", + "say more", "more", "what exactly", "the exact command"} + + +def _full_detail(tool, tool_input, ctx): + """The full literal form, spoken only when the person asks for + "details". Never lets a long command hide its tail: truncation is + DISCLOSED and shell chaining is called out (the agent composes + tool_input itself, so this line must not be steerable into + understatement).""" + d = tool_input or {} + if tool == "Bash": + cmd = " ".join(str(d.get("command", "")).split()) + chained = any(m in cmd for m in _CHAIN_MARKS) + line = ("a chained command: " if chained else + "run a command: ") + cmd[:90] + if len(cmd) > 90: + line += (f", and {len(cmd) - 90} more characters. " + "Check the log before approving") + return line + if tool in ("Write", "Edit", "MultiEdit", "NotebookEdit"): + path = str(d.get("file_path") or d.get("notebook_path") + or "a file").replace("\\", "/") + bits = path.rsplit("/", 2) + name = "/".join(bits[-2:]) if len(bits) >= 2 else path + return f"{'edit' if 'Edit' in tool else 'write'} the file {name}" + if tool == "WebFetch": + return f"fetch a web page: {str(d.get('url', ''))[:70]}" + desc = (getattr(ctx, "description", None) or "").strip() + name = getattr(ctx, "display_name", None) or tool + return f"use {name}" + (f", {desc[:70]}" if desc else "") + + +def make_permission_gate(mouth): + from claude_agent_sdk import (PermissionResultAllow, + PermissionResultDeny) + + async def gate(tool, tool_input, ctx): + if _AUTOAPPROVE["on"]: + return PermissionResultAllow(behavior="allow") + what = _human_what(tool, tool_input, ctx) + detail = _full_detail(tool, tool_input, ctx) + loop = asyncio.get_running_loop() + signals.static_stop() + log(f"[perm] asking: {what}") + log(f"[perm] detail: {detail}") + if tool == "Bash": # the FULL command always reaches the log + log(f"[perm] full command: {str((tool_input or {}).get('command', ''))[:2000]}") + ask = f"Permission check. I want to {what}. Yes, no, or details?" + if not _PERM["hinted"]: + # the escape hatch announces itself exactly once, at the + # moment it becomes relevant (a field case: a new user + # couldn't find the phrase to turn the checks off) + _PERM["hinted"] = True + ask += (" And any time you're done with these checks, say " + "stop asking for permission.") + mouth.say(ask) + answer = None + try: + deadline = loop.time() + PERM_TIMEOUT_S + while answer is None: + fut = loop.create_future() + _PERM["fut"] = fut + _PERM["asked_at"] = time.monotonic() + while True: + try: + got = await asyncio.wait_for( + asyncio.shield(fut), 1.0) + break + except asyncio.TimeoutError: + if loop.time() >= deadline: + fut.cancel() + mouth.say("No answer, so I didn't do it.") + log("[perm] timed out, denied") + return PermissionResultDeny( + behavior="deny", + message="No spoken answer within the " + "timeout; the action was not " + "approved.", + interrupt=False) + # keep the ring honest while we wait + if not mouth.speaking: + signals.set_state("listening") + if (got != _INTERRUPT_ANSWER + and _norm_speech(got) in _DETAILS): + # read the full literal form, then ask again with a + # fresh clock: asking for details is engagement, + # not silence + log("[perm] details requested") + mouth.say(f"The details: I want to {detail}. " + "Yes or no?") + deadline = loop.time() + PERM_TIMEOUT_S + continue + answer = got + finally: + _PERM["fut"] = None + if answer == _INTERRUPT_ANSWER: + log("[perm] turn interrupted, denied silently") + return PermissionResultDeny( + behavior="deny", + message="Interrupted by the user; the turn is being " + "cancelled.", + interrupt=False) + approved = _norm_speech(answer) in _YES + # the model keeps working either way: restore the working state + signals.set_state("thinking") + signals.static_start() + if approved: + log("[perm] approved by voice") + return PermissionResultAllow(behavior="allow") + log(f"[perm] denied: {answer!r}") + return PermissionResultDeny( + behavior="deny", + message=f'Denied by voice. The user said: "{answer[:500]}"', + interrupt=False) + return gate + + +# ---- THE VOICE CONSOLE: session verbs, spoken. Exact phrases only, +# spoken alone, so ordinary sentences can never trigger them. (Grown +# from a community member's own build shared in the Discord.) +CONSOLE_VERBS = { + "clear": ("clear the session", "clear the context", + "clear context", "fresh slate", "slash clear", + # natural answers to the resumed-launch "continue or + # start fresh?" question + "start fresh", "new session", "start a new session"), + "compact": ("compact the session", "compact the context", + "compact context", "slash compact"), + "deep": ("switch to the deep model", "use the deep model", + "slash model deep"), + "fast": ("switch to the fast model", "use the fast model", + "back to the fast model", "slash model fast"), + "usage": ("usage report", "slash usage"), + "micopen": ("go hands free", "hands free mode", + "hands free listening", "open mic", "open the mic"), + "micptt": ("push to talk", "push to talk mode", + "back to push to talk", "back to the button"), + "noask": ("stop asking for permission", + "stop asking permission", + "stop asking me for permission", + "turn off the permission prompt", + "turn off the permission prompts", + "turn off the permissions prompt", + "turn off the permissions prompts", + "turn off permissions", "turn off permission checks", + "disable the permission checks", + "disable permission checks", "auto approve", + "auto approve mode"), + "ask": ("start asking again", "ask before acting", + "ask for permission again"), +} +_EFFORTS = ("low", "medium", "high", "xhigh", "max") + +# Spoken in place of the silent warmup ping when a launch reattached to +# the previous conversation (config: resume_last_session). +RESUME_RECAP = ( + "We just relaunched and reattached to this conversation. In two " + "short spoken sentences: say what we were in the middle of, then " + "ask whether I want to continue or start fresh. Don't do any work " + "yet. If I want fresh, I'll say 'start fresh'; otherwise I'll just " + "keep talking.") + + +def console_match(text): + norm = " ".join(text.lower().replace("-", " ").split()).strip(" .,!?") + for verb, phrases in CONSOLE_VERBS.items(): + if norm in phrases: + return verb + for lvl in _EFFORTS: + if norm in (f"set effort to {lvl}", f"effort {lvl}", + f"slash effort {lvl}"): + return f"effort:{lvl}" + return None + + +def _write_config_key(key, value): + """The agent rewrites the config; the person never hand-edits it. + Returns True on a persisted write. A file that fails to PARSE is + left untouched (rewriting from {} would wipe every other setting); + the in-memory CFG updates either way so the session behaves.""" + from backtalk.config import CONFIG_PATH + CFG[key] = value + try: + data = json.loads(CONFIG_PATH.read_text()) + except FileNotFoundError: + data = {} + except (OSError, ValueError) as e: + log(f"[console] config not writable/parsable, session-only: {e}") + return False + data[key] = value + try: + CONFIG_PATH.write_text(json.dumps(data, indent=2) + "\n") + except OSError as e: + log(f"[console] config write failed, session-only: {e}") + return False + return True + + +def _fmt_tokens(n): + if n >= 1_000_000: + return f"about {round(n / 1_000_000, 1):g} million tokens" + if n >= 1000: + return f"about {round(n / 1000)} thousand tokens" + return f"{n} tokens" + + +def _spoken_usage(sess, ctx_usage): + """A short CFO brief of the session, written for the ear: plain + numerals only (the TTS reads "40" fine; symbols come out garbled).""" + turns = sess["turns"] + parts = [f"{turns} turn{'s' if turns != 1 else ''} this session", + _fmt_tokens(sess["out_tokens"]) + " spoken out"] + cents = round(sess["cost"] * 100) + if cents >= 1: + parts.append(f"roughly {cents} cents" if cents < 100 + else f"roughly {round(cents / 100)} dollars") + try: + cats = (getattr(ctx_usage, "categories", None) + or (ctx_usage or {}).get("categories") or []) + # the breakdown includes "Free space" and the autocompact + # buffer; only OCCUPIED categories belong in the spoken number + total = sum(int(c.get("tokens") or 0) for c in cats + if isinstance(c, dict) + and "free" not in str(c.get("name", "")).lower() + and "buffer" not in str(c.get("name", "")).lower()) + if total: + parts.append(_fmt_tokens(total) + + " sitting in the context window") + except Exception: + pass + return ". ".join(parts) + "." + +_PASTE_ON = "\x1b[200~" # bracketed-paste markers (we enable the mode below) +_PASTE_OFF = "\x1b[201~" + + +def _clean_typed(line: str) -> str: + """Scrub terminal-copy artifacts: blockquote gutter glyphs and stray + whitespace (copying from a CLI chat render drags bars along).""" + line = line.strip() + while line[:1] in ("▎", "│", ">"): + line = line[1:].lstrip() + return line + + +def _join_paste(body: str) -> str: + """Pasted blob -> one clean message (gutters scrubbed, lines joined).""" + parts = [_clean_typed(l) for l in body.split("\n")] + return " ".join(" ".join(p for p in parts if p).split()) + + +def _typed_reader_pipe(q: "queue.Queue[str]", fd: int): + """Non-tty stdin (pipes/tests): line assembly with paste markers.""" + import os + pend = "" + while True: + try: + b = os.read(fd, 65536) + except OSError: + return + if not b: + return + pend += b.decode("utf-8", "replace") + while True: + if _PASTE_ON in pend: + if _PASTE_OFF not in pend: + break + head, rest = pend.split(_PASTE_ON, 1) + body, pend = rest.split(_PASTE_OFF, 1) + *hlines, hpart = head.split("\n") + for l in hlines: + l = _clean_typed(l) + if l: + q.put(l) + text = _join_paste(hpart + body) + if text: + q.put(text) + continue + if "\n" in pend: + line, pend = pend.split("\n", 1) + line = _clean_typed(line) + if line: + q.put(line) + continue + break + + +def _typed_reader_simple(q: "queue.Queue[str]"): + """Windows (no termios): plain line input on a thread. Pastes work; + they just echo normally instead of collapsing to a count.""" + while True: + try: + line = _clean_typed(input()) + except (EOFError, OSError): + return + if line: + q.put(line) + + +def _typed_reader(q: "queue.Queue[str]"): + """Terminal stdin -> typed messages (daemon thread). Typed lines are + first-class turns: same pipeline as a spoken utterance, spoken reply. + + On a POSIX tty we OWN the input line (cbreak: no kernel echo, no + canonical buffering — the little line editor below echoes keys, + handles backspace, and assembles bracketed pastes invisibly). The + kernel's canonical mode is unfixable for pastes: it echoes the + markers as visible junk and holds unfinished marker lines hostage. + Pastes show as `[pasted N chars]`; Enter sends everything as ONE + message. Ctrl-C still works (ISIG stays on); termios restored at + exit.""" + import atexit + import os + fd = sys.stdin.fileno() + if not os.isatty(fd): + _typed_reader_pipe(q, fd) + return + try: + import termios + import tty as _tty + except ImportError: # Windows: no termios — simple reader + _typed_reader_simple(q) + return + old = termios.tcgetattr(fd) + _tty.setcbreak(fd) # ECHO+ICANON off, ISIG kept + sys.stdout.write("\x1b[?2004h") # bracket pastes, please + sys.stdout.flush() + + def _restore(): + try: + termios.tcsetattr(fd, termios.TCSADRAIN, old) + except Exception: + pass + sys.stdout.write("\x1b[?2004l") + sys.stdout.flush() + atexit.register(_restore) + + MARKS = (_PASTE_ON, _PASTE_OFF) + + def _partial_tail(s: str) -> int: + """Length of a trailing partial paste-marker (hold it for the + next read).""" + for m in MARKS: + for k in range(min(len(s), len(m) - 1), 0, -1): + if m.startswith(s[-k:]): + return k + return 0 + + buf = "" # the input line being composed + paste = None # accumulating paste body, or None + pend = "" + while True: + try: + b = os.read(fd, 4096) + except OSError: + _restore() + return + if not b: + _restore() + return + pend += b.decode("utf-8", "replace") + keep = _partial_tail(pend) + proc = pend[:len(pend) - keep] if keep else pend + pend = pend[len(pend) - keep:] if keep else "" + i = 0 + while i < len(proc): + if paste is not None: + j = proc.find(_PASTE_OFF, i) + if j < 0: + paste += proc[i:] + break + paste += proc[i:j] + i = j + len(_PASTE_OFF) + text = _join_paste(paste) + paste = None + if text: + if buf and not buf.endswith(" "): + buf += " " + buf += text + sys.stdout.write(text if len(text) <= 60 + else f"[pasted {len(text)} chars]") + sys.stdout.flush() + continue + if proc.startswith(_PASTE_ON, i): + paste = "" + i += len(_PASTE_ON) + continue + ch = proc[i] + i += 1 + if ch in ("\r", "\n"): + sys.stdout.write("\n") + sys.stdout.flush() + line = buf.strip() + buf = "" + if line: + q.put(line) + elif ch in ("\x7f", "\x08"): # backspace + if buf: + buf = buf[:-1] + sys.stdout.write("\b \b") + sys.stdout.flush() + elif ch >= " " or ch == "\t": # printable: echo + collect + buf += ch + sys.stdout.write(ch) + sys.stdout.flush() + + +def _defence(raw: str, in_fence: bool) -> tuple[str, bool]: + """Split out fenced-code content (Joe, 2026-08-24): everything between + ``` markers is for the DASHBOARD, not the mouth — pasted paths, markup, + whole listing bodies. Returns (speakable text, new fence state). State + persists across chunks because a fence usually opens in one sentence + chunk and closes many chunks later.""" + parts = raw.split("```") + spoken, state = [], in_fence + for i, part in enumerate(parts): + if not state: + spoken.append(part) + if i < len(parts) - 1: + state = not state + return " ".join(spoken), state + + +async def speak_reply(brain: WarmBrain, mouth: Mouth, text: str): + """First sentence ships alone (fast start); the rest go in + 2-sentence breaths — fuller chunks get livelier prosody (single + short sentences come out flat).""" + t0 = time.time() + first = True + in_fence = False + batch: list[str] = [] + + def emit(raw: str): + nonlocal first, batch, in_fence + # Fenced blocks are dashboard-only — never spoken (they still + # reach the transcript untouched; this only mutes the mouth). + raw, in_fence = _defence(raw, in_fence) + # TTS hygiene: backticks, markdown fences and angle-bracket tag + # syntax are never speakable — the mouth gets clean prose only. + s = raw.replace("`", "").replace("<<", "").replace(">>", "").strip() + if not s: + return + if first: + log(f"[{NAME}] ({time.time()-t0:.1f}s to first) {s}") + mouth.say_chunk(s) + first = False + else: + log(f"[{NAME}] {s}") + batch.append(s) + if len(batch) >= 2: + mouth.say_chunk(" ".join(batch)) + batch = [] + + try: + async for sentence in brain.ask_stream(text): + emit(sentence) + if batch: + mouth.say_chunk(" ".join(batch)) + if first: + # Zero sentences yielded (brain error / empty turn): nothing + # will ever dequeue, so nothing resets the bus — park it here. + signals.static_stop() + signals.set_state("idle") + except asyncio.CancelledError: + try: + await brain.interrupt() + except Exception: + pass + raise + + +async def amain(): + open_mic = "--open-mic" in sys.argv + barge_in = "--barge-in" in sys.argv + model = None + if "--model" in sys.argv: + try: + model = sys.argv[sys.argv.index("--model") + 1] + except IndexError: + pass + + CFG_BOOT_MODE = CFG["permission_mode"] + _AUTOAPPROVE["on"] = CFG_BOOT_MODE == "bypassPermissions" + _MIC["mode"] = "open" if (open_mic + or CFG.get("mic_mode") == "open") else "ptt" + # resume_last_session: reattach to the saved conversation, if any + resume_id = None + if CFG.get("resume_last_session"): + try: + from backtalk.brain import SESSION_FILE + with open(SESSION_FILE) as f: + resume_id = f.read().strip() or None + except OSError: + resume_id = None + + mouth = Mouth() + ears = Ears() + brain = WarmBrain(model=model, + can_use_tool=make_permission_gate(mouth), + resume_id=resume_id) + # The bus heartbeat (.voice_heartbeat, every ~2 s) starts NOW, on + # this loop, so the face reads LINK LIVE from the first second of + # boot and LOST the moment this process dies or the loop wedges. + signals.heartbeat_start() + + mode = ("hands-free listening (the talk key still works)" + if _MIC["mode"] == "open" + else f"push-to-talk ({CFG['ptt_key']})") + log(f"[backtalk] up — agent={NAME} dir={CFG['agent_dir']} " + f"model={brain.model} mic={mode} " + f"(say 'goodbye {NAME.lower()}' to hang up)") + mouth.say(CFG["greeting"]) + + loop = asyncio.get_event_loop() + # Warm the engines while the greeting plays: the STT model load and + # the brain's prompt-cache toll both hide behind the spoken line. + loop.run_in_executor(None, warm_ears) + # THE BRAIN CONNECT, guarded. This is the one startup step that + # needs a signed-in Claude Code, internet, and available usage. + # When it fails or hangs, the mouth still works, so SAY SO instead + # of dying silently with the face stuck on idle (a real field + # case: the greeting played, then nothing, and on Windows the + # window closed before anyone could read the error). + log("[backtalk] connecting the brain...") + try: + await asyncio.wait_for(brain.start(), 120) + + async def _warmup(): + if brain.resumed: + # Reattached to the last conversation: the warmup turn + # is spoken, so the person hears where they left off + # and can choose to continue or start over. + await speak_reply(brain, mouth, RESUME_RECAP) + return + async for _ in brain.ask_stream( + "Warmup ping - reply with the single word: ready"): + pass + await asyncio.wait_for(_warmup(), 180) + except (Exception, asyncio.TimeoutError) as e: + kind = ("timed out" if isinstance(e, asyncio.TimeoutError) + else f"failed: {e!r}"[:220]) + log(f"[backtalk] BRAIN CONNECT {kind}") + mouth.say("Bad news. The voice and the face are fine, but I " + "couldn't reach my brain, the Claude Code session. " + "Check this window for the error. The usual causes: " + "Claude Code isn't signed in, the internet is down, " + "or the plan is out of usage.") + mouth.wait_done(timeout=30) + raise SystemExit(1) + log("[backtalk] brain warm") + # the hidden warmup ping is plumbing, not conversation + brain.session.update(turns=0, out_tokens=0, in_tokens=0, cost=0.0) + # a configured effort level applies at launch (saved by the spoken + # "set effort to X", or written by the person's agent on request) + boot_effort = str(CFG.get("effort") or "").strip().lower() + if boot_effort in _EFFORTS: + await brain.command(f"/effort {boot_effort}") + log(f"[backtalk] effort set to {boot_effort} (from config)") + elif boot_effort: + log(f"[backtalk] ignoring unknown effort {boot_effort!r} in config") + + speak_task: asyncio.Task | None = None + typed_q: "queue.Queue[str]" = queue.Queue() + threading.Thread(target=_typed_reader, args=(typed_q,), daemon=True).start() + typed_fut: asyncio.Future | None = None + + async def run_console(verb): + """One voice-console verb. The current reply was already + cancelled and awaited by handle(); the pipe gets drained here + before the command goes out. A verb that blows up must never + take the whole voice session down with it.""" + try: + await _run_console_inner(verb) + except Exception as e: + log(f"[console] {verb} failed: {e}") + mouth.say("That command hit an error. Check the log.") + signals.set_state("idle") + + async def _run_console_inner(verb): + _deny_pending() + await brain.reset_turn() + say_after = None + if verb == "clear": + resp = await brain.command("/clear") + # A fresh start is also a fresh launch: drop the saved + # resume id so a relaunch right after comes up cold instead + # of reattaching to an empty conversation. The file is + # rewritten on the next completed turn. + try: + from backtalk.brain import SESSION_FILE + os.remove(SESSION_FILE) + except OSError: + pass + say_after = "Cleared. Fresh slate." + elif verb == "compact": + mouth.say("Compacting. One moment.") + resp = await brain.command("/compact") + say_after = "Compacted. Same conversation, smaller footprint." + elif verb == "deep": + mouth.say("Switching to the deep model. Heads up, replies " + "get slower. Say back to the fast model when " + "you're done.") + resp = await brain.command(f"/model {CFG['deep_model']}") + say_after = "Deep model online, for this session only." + elif verb == "fast": + resp = await brain.command(f"/model {CFG['model']}") + say_after = "Back on the fast model." + elif verb.startswith("effort:"): + lvl = verb.split(":", 1)[1] + resp = await brain.command(f"/effort {lvl}") + saved = _write_config_key("effort", lvl) + say_after = (f"Effort set to {lvl}, and saved as your " + "default." if saved else + f"Effort set to {lvl} for this session. The " + "config file couldn't be written, so it won't " + "stick past a restart.") + elif verb == "usage": + resp = "" + mouth.say(_spoken_usage(brain.session, + await brain.context_usage())) + elif verb == "micopen": + resp = "" + if _MIC["mode"] == "open": + mouth.say("Already in hands-free listening.") + else: + _MIC["mode"] = "open" + _MIC["gen"] += 1 + _write_config_key("mic_mode", "open") + log("[console] mic_mode -> open (hands-free listening)") + mouth.say("Hands-free listening on. I'm always " + "listening now, so anything said in the room " + "can reach me. The talk key still works, and " + "holding it always gets you heard. Say push " + "to talk mode to bring the button back.") + elif verb == "micptt": + resp = "" + if _MIC["mode"] == "ptt": + mouth.say("Already on push to talk.") + else: + _MIC["mode"] = "ptt" + _MIC["gen"] += 1 + _write_config_key("mic_mode", "ptt") + log("[console] mic_mode -> ptt") + key = str(CFG.get("ptt_key", "home")).replace("_", " ") + mouth.say(f"Push to talk. Hold the {key} key and " + "talk; the mic stays closed otherwise.") + elif verb == "noask": + resp = "" + _CONFIRM["verb"] = "noask" + _CONFIRM["at"] = time.monotonic() + mouth.say("Auto-approve means I act without asking " + "permission, and it becomes your saved default. " + "Say confirm to switch.") + elif verb == "noask:confirmed": + resp = "" + saved = _write_config_key("permission_mode", + "bypassPermissions") + _AUTOAPPROVE["on"] = True + log("[console] permission_mode -> bypassPermissions" + + (" (saved)" if saved else " (session only)")) + mouth.say(("Auto-approve on, and saved as your default. " + if saved else + "Auto-approve on for this session. The config " + "file couldn't be written, so it won't stick " + "past a restart. ") + + "Say start asking again any time to flip it " + "back.") + elif verb == "ask": + resp = "" + saved = _write_config_key("permission_mode", "ask") + _AUTOAPPROVE["on"] = False + flipped = True + if CFG_BOOT_MODE == "bypassPermissions": + # a bypass-booted session never consults the gate, so + # the SDK itself must flip (the safe direction is + # allowed live). If that fails, saying "done" would be + # a lie: the agent would keep acting silently. + try: + await brain.set_permission_mode("ask") + except Exception as e: + flipped = False + log(f"[console] live flip to ask FAILED: {e}") + log("[console] permission_mode -> ask" + + (" (saved)" if saved else " (session only)")) + if flipped: + mouth.say("Done. I'll ask out loud before real " + "actions" + + (", and that's saved as your default." + if saved else + ". The config file couldn't be written, " + "so tell me again after a restart.")) + else: + mouth.say("I saved asking as your default, but this " + "session couldn't switch over. Restart the " + "voice line to get asking back.") + else: + resp = "" + if say_after: + # the CLI answers slash commands with its own text + # (confirmations, API errors); an error outranks our line + low = (resp or "").lower() + if resp and ("error" in low or "invalid" in low): + mouth.say(resp[:160]) + log(f"[console] {verb} answered: {resp[:120]}") + else: + mouth.say(say_after) + signals.set_state("idle") + + async def handle(text: str, spoke_from: float | None = None) -> bool: + """Process one utterance; returns False on quit. spoke_from is + when the utterance STARTED (the PTT press), so an answer can be + told apart from speech that began before the ask even existed.""" + nonlocal speak_task + log(f"[you] {text}") + # A pending spoken permission ask owns the next utterance IF + # that utterance started after the ask was posed. Speech that + # began earlier is the user interrupting the turn, not + # answering a question they never heard: the ask resolves as a + # silent deny and the utterance falls through as a normal + # interrupt. Quit wins either way, but only as an EXACT phrase + # here ("No! Don't hang up, skip it" must stay a deny reason, + # not kill the session). + if _PERM["fut"] is not None and not _PERM["fut"].done(): + started_after = (spoke_from is None + or spoke_from >= _PERM["asked_at"]) + if _norm_speech(text) in {_norm_speech(q) + for q in QUIT_PHRASES}: + _PERM["fut"].set_result("no") + # falls through to the quit body below + elif started_after: + _PERM["fut"].set_result(text) + return True + else: + _deny_pending() + # A pending auto-approve confirm owns it too, for two minutes; + # after that it expires and speech flows normally again. + verb = None + if _CONFIRM["verb"]: + pend, _CONFIRM["verb"] = _CONFIRM["verb"], None + expired = time.monotonic() - _CONFIRM["at"] > 120 + if not expired and _norm_speech(text) in ( + "confirm", "confirmed", "yes confirm", + "yes confirmed"): + verb = pend + ":confirmed" + elif not expired and not any(q in text.lower() + for q in QUIT_PHRASES): + mouth.say("Staying as we are.") + return True + if any(q in text.lower() for q in QUIT_PHRASES): + if speak_task and not speak_task.done(): + speak_task.cancel() + mouth.shut_up() + mouth.say(CFG["signoff"]) + mouth.wait_done(timeout=15) + return False + if speak_task and not speak_task.done(): + log("[turn] interrupted mid-reply by new input") + _deny_pending() # an ask never outlives its turn + speak_task.cancel() + mouth.shut_up() + if speak_task: + # Let the cancellation fully land (its brain.interrupt() + # included) BEFORE anything else touches the brain — + # otherwise the dead turn's stop signal can race in after + # the new query and kill the new answer (half of the + # off-by-one bug; see brain.reset_turn for the other half). + try: + await speak_task + except asyncio.CancelledError: + pass + except Exception: + pass + speak_task = None + verb = verb or console_match(text) + if verb: + await run_console(verb) + return True + signals.set_state("thinking") + signals.static_start() + signals.turn_begin() # .voice_activity: the turn clock starts + # Clean the pipe: drain the interrupted turn's leftovers so the + # new question can't pair with a stale ResultMessage. A gate + # that fired in the meantime resolves first, or the drain would + # wait on a ResultMessage the CLI is withholding for an answer. + _deny_pending() + await brain.reset_turn() + speak_task = asyncio.create_task(speak_reply(brain, mouth, text)) + return True + + try: + # ONE loop, two mic modes, switchable live (_MIC). The talk key + # is constructed and honored in BOTH modes: in hands-free + # listening it is the interrupt and the guaranteed way to be + # heard over room noise. The open mic joins the wait-set only + # in "open" mode; a mode switch bumps _MIC["gen"], the abort + # callable closes the in-flight open mic promptly, and any + # capture born under an old gen is discarded unprocessed. + ptt = PTTListener(CFG["ptt_key"]) + press_fut: asyncio.Future | None = None + mic_fut: asyncio.Future | None = None + mic_gen_seen = _MIC["gen"] + # The open mic yields while the BUTTON records (or the double + # capture would turn one held utterance into two turns), and, + # without barge-in, while the mouth speaks. + mic_gate = (lambda: _MIC["btn"] + or (not barge_in and mouth.speaking)) + mic_fails = 0 + while True: + if _MIC["gen"] != mic_gen_seen: + mic_gen_seen = _MIC["gen"] + # consume futures that completed under the old mode so + # a stale press or capture can't fire after a switch + if press_fut is not None and press_fut.done(): + press_fut.result(); press_fut = None + if mic_fut is not None and mic_fut.done(): + mic_fut.result(); mic_fut = None + if typed_fut is None: + typed_fut = loop.run_in_executor(None, typed_q.get) + if press_fut is None: + press_fut = loop.run_in_executor(None, ptt.wait_press) + waiters = {press_fut, typed_fut} + if _MIC["mode"] == "open": + if mic_fut is None: + g = _MIC["gen"] + mic_fut = loop.run_in_executor( + None, lambda g=g: (g, ears.listen_once( + gate=mic_gate, + abort=lambda: _MIC["gen"] != g))) + waiters.add(mic_fut) + done, _ = await asyncio.wait( + waiters, return_when=asyncio.FIRST_COMPLETED) + if typed_fut in done: + text = typed_fut.result(); typed_fut = None + if text and not await handle(text): + return + continue + if mic_fut is not None and mic_fut in done: + try: + g, text = mic_fut.result() + except Exception as e: + mic_fut = None + mic_fails += 1 + log(f"[ears] open mic failed ({mic_fails}): {e!r}") + if mic_fails >= 3: + _MIC["mode"] = "ptt" + _MIC["gen"] += 1 + mic_fails = 0 + mouth.say("The open microphone keeps failing, " + "so I'm switching to push to talk. " + "Hold the key to reach me, and " + "check this window for the error.") + continue + mic_fut = None + if g != _MIC["gen"]: + continue # captured before a switch + if text and not await handle(text): + return + continue + if press_fut in done: + press_fut.result(); press_fut = None + press_t = time.monotonic() + perm_wait = (_PERM["fut"] is not None + and not _PERM["fut"].done()) + if speak_task and not speak_task.done() and not perm_wait: + log("[turn] interrupted mid-reply — key pressed") + speak_task.cancel() # the button = interrupt + # During a permission ask the TURN stays alive; the + # press only silences playback and records the answer. + mouth.shut_up() + signals.static_stop() # button kills the static too + signals.set_state("listening") + mouth.ducker.speech_start() # duck NOW, while you talk + print("[ptt] recording (release to send)...", flush=True) + _MIC["btn"] = True # open mic yields to the button + try: + text = await loop.run_in_executor( + None, lambda: record_held(ptt.is_held)) + except Exception as e: + log(f"[ears] record/transcribe failed: {e!r}") + mouth.say("My ears hit an error. Check this " + "window for the details.") + text = None + finally: + _MIC["btn"] = False + mouth.ducker.speech_end(0.2) # snap back fast on release + if not text: + log("[ptt] (tap or empty — ignored)") + signals.set_state("idle") + continue + if not await handle(text, spoke_from=press_t): + return + except KeyboardInterrupt: + pass + finally: + _MIC["gen"] += 1 # abort any live open-mic capture promptly + if speak_task and not speak_task.done(): + speak_task.cancel() + mouth.shutdown() # restores the music on Ctrl-C / crash paths too + signals.static_stop() + signals.set_state("idle") + signals.heartbeat_stop() # LINK reads LOST at once, not in 6 s + await brain.stop() + log("[backtalk] hung up") + + +def main(): + try: + asyncio.run(amain()) + except KeyboardInterrupt: + print("\n[backtalk] interrupted — hanging up", flush=True) + + +if __name__ == "__main__": + main() diff --git a/backtalk/mouth.py b/backtalk/mouth.py index d59390a..d531a87 100644 --- a/backtalk/mouth.py +++ b/backtalk/mouth.py @@ -29,6 +29,15 @@ is cancellable mid-word: set the stop event and the speaker goes silent within one audio block plus the device buffer (~0.15s). +Two threads, not one: a synth thread renders chunks AHEAD of the one +playing (LOOKAHEAD deep), so when a chunk ends the next one's audio is +already sitting there finished and plays with no gap. One thread doing +render-then-play-then-render put a synth-latency-plus-prebuffer pause +(~1s) at every chunk boundary, while the text had long since reached the +screen. A generation counter ties the two threads together for barge-in: +shut_up() bumps it, and every chunk rendered or queued under the old +generation is dropped instead of played. + HARD-WON AUDIO LAW #1 — ONE long-lived OutputStream, reused for every sentence for the life of the process. A fresh stream per sentence gives an audible onset blip or a beat of dead air on plenty of audio setups @@ -248,17 +257,47 @@ def synth_stream(text: str, timeout: float = 30.0): yield KOKORO_RATE, pcm +LOOKAHEAD = 2 # chunks rendered ahead of the one playing +_DONE = object() # end-of-render marker on a _Render's pcm queue + + +class _Render: + """One chunk's audio in flight from the synth thread to the player. + PCM blocks stream through `pcm` as they render (rate is set before + the first one lands), _DONE closes it. `gen` is the barge-in + generation the chunk was ordered under; stale means shut_up() has + happened since and the chunk must never play.""" + __slots__ = ("text", "gen", "pcm", "rate") + + def __init__(self, text: str, gen: int): + self.text = text + self.gen = gen + self.pcm: queue.Queue = queue.Queue() + self.rate: int | None = None + + class Mouth: def __init__(self): from backtalk.ducking import Ducker - self._q: queue.Queue = queue.Queue() + self._q: queue.Queue = queue.Queue() # (gen, text) in + self._ready: queue.Queue = queue.Queue(maxsize=LOOKAHEAD) self._stop = threading.Event() self._speaking = threading.Event() + # Barge-in generation: shut_up() bumps it, and anything ordered + # under an older one is dropped wherever it's found. _pending + # counts current-generation chunks not yet played — "queue + # empty" stopped meaning "nothing left" once rendering ran ahead + # of playback, so the speaking flag and wait_done key off this. + self._gen = 0 + self._pending = 0 + self._lock = threading.Lock() # The one persistent output stream (audio law #1). - # Worker-thread-only — never touch from other threads. + # Player-thread-only — never touch from other threads. self._out: sd.OutputStream | None = None self._out_rate: int | None = None self.ducker = Ducker() # public: PTT ducks for the USER's voice too + self._synth = threading.Thread(target=self._render_loop, daemon=True) + self._synth.start() self._worker = threading.Thread(target=self._run, daemon=True) self._worker.start() @@ -269,7 +308,7 @@ def speaking(self) -> bool: def say(self, text: str): """Queue text (split to sentences) for speech.""" for s in split_sentences(text): - self._q.put(s) + self._enqueue(s) def say_chunk(self, text: str): """Queue text as ONE TTS request, no sentence splitting — fuller @@ -277,16 +316,28 @@ def say_chunk(self, text: str): dull).""" text = text.strip() if text: - self._q.put(text) + self._enqueue(text) + + def _enqueue(self, text: str): + with self._lock: + self._pending += 1 + self._q.put((self._gen, text)) def shut_up(self): - """Barge-in: stop current playback and flush everything queued.""" + """Barge-in: stop current playback and flush everything queued + or rendering. Bump the generation FIRST, then raise stop: the + player clears stop and then checks the generation, and that + order pair is what leaves no window for a dead chunk to play.""" + with self._lock: + self._gen += 1 + self._pending = 0 self._stop.set() - try: - while True: - self._q.get_nowait() - except queue.Empty: - pass + for q_ in (self._q, self._ready): + try: + while True: + q_.get_nowait() + except queue.Empty: + pass def shutdown(self): """Exit path: stop playback and restore the music SYNCHRONOUSLY @@ -295,31 +346,57 @@ def shutdown(self): self.ducker.restore_now() def wait_done(self, timeout: float | None = None): - """Block until the queue is drained and playback finished.""" + """Block until everything queued has played.""" import time deadline = None if timeout is None else time.time() + timeout - while (not self._q.empty()) or self._speaking.is_set(): + while self._pending > 0 or self._speaking.is_set(): time.sleep(0.05) if deadline and time.time() > deadline: return + def _render_loop(self): + """Synth thread: render chunks in order, LOOKAHEAD ahead of the + player. A render is handed over BEFORE it's filled, so the first + chunk still starts after the prebuffer rather than after the + whole render; _ready's bound is what throttles the lookahead.""" + while True: + gen, text = self._q.get() + if gen != self._gen: + continue + r = _Render(text, gen) + self._ready.put(r) + try: + for rate, pcm in synth_stream(text): + if gen != self._gen: + break # barged in: don't finish a dead chunk + r.rate = rate + r.pcm.put(pcm) + except Exception as e: + log(f"[mouth] synth error: {e}") + finally: + r.pcm.put(_DONE) + def _run(self): from backtalk import signals while True: - sentence = self._q.get() - if not sentence: - continue + r = self._ready.get() self._stop.clear() + if r.gen != self._gen: + continue self._speaking.set() self.ducker.speech_start() signals.static_stop() # thinking sound dies when speech starts signals.set_state("speaking") try: - self._play_stream(sentence) + self._play_render(r) except Exception as e: log(f"[mouth] synth/play error: {e}") finally: - if self._q.empty(): + with self._lock: + if r.gen == self._gen: + self._pending -= 1 + done = self._pending <= 0 + if done: self._speaking.clear() self.ducker.speech_end() signals.set_state("idle") @@ -363,26 +440,29 @@ def _drop_out(self): self._out = None self._out_rate = None - def _play_stream(self, sentence: str, block: int = 2205, + def _play_render(self, r: _Render, block: int = 2205, prebuffer_s: float = 0.75): - """Stream-synthesize and play with the head-start buffer (audio - law #2). stop() reacts ~50ms. The sample rate comes from - whichever engine actually answered.""" + """Play one render as it fills, with the head-start buffer (audio + law #2) — already satisfied instantly for any chunk that finished + rendering while the previous one played. stop() reacts ~50ms. + The sample rate comes from whichever engine actually answered.""" from backtalk import signals - gen = synth_stream(sentence) head: list = [] banked = 0 - rate = None - for rate_, pcm in gen: - rate = rate_ + finished = False + while True: + pcm = r.pcm.get() + if pcm is _DONE: + finished = True + break head.append(pcm) banked += len(pcm) - if banked >= int(rate * prebuffer_s): + if banked >= int(r.rate * prebuffer_s): break - if rate is None: + if not head: return try: - out = self._get_out(rate) + out = self._get_out(r.rate) def _write(pcm): for i in range(0, len(pcm), block): @@ -400,7 +480,10 @@ def _write(pcm): if not _write(pcm): self._cut() return - for _, pcm in gen: + while not finished: + pcm = r.pcm.get() + if pcm is _DONE: + break if not _write(pcm): self._cut() return diff --git a/backtalk/signals.py b/backtalk/signals.py index 84c459b..a3ce140 100644 --- a/backtalk/signals.py +++ b/backtalk/signals.py @@ -1,148 +1,281 @@ -# backtalk: talk to your Claude Code agent out loud. -# Copyright (C) 2026 Jared Rhodenizer -# -# This program is free software: you can redistribute it and/or modify -# it under the terms of the GNU Affero General Public License as published -# by the Free Software Foundation, either version 3 of the License, or -# (at your option) any later version. -# -# This program is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# GNU Affero General Public License for more details. -# -# You should have received a copy of the GNU Affero General Public License -# along with this program. If not, see . -# -# SPDX-License-Identifier: AGPL-3.0-or-later -"""The signal bus — tiny files any other program can watch. - -The voice line leaves notes; faces read the notes. That one dumb trick -is the whole integration surface: - - .voice_state idle | listening | thinking | speaking - .voice_waveform JSON {ts, samples: [64 floats]} while audio plays - .voice_loading_pid exists while the thinking sound is playing - -Written to signals_dir (default: the repo root). Visualizers built on -this contract just work. - -THE BAREHANDS SEAM: set barehands_state_dir in backtalk.json to a -barehands checkout's state/ folder and the same signals are mirrored in -its format (state/state as a bare word, state/wave.json normalized -0..1) — the on-screen ring becomes your agent's face with zero glue. - -Every write is wrapped: the bus must never crash the voice line. -""" -import json -import os -import subprocess -import sys -import time - -import numpy as np - -from backtalk.config import CFG - -_DIR = CFG["signals_dir"] -_STATE_FILE = os.path.join(_DIR, ".voice_state") -_WAVEFORM_FILE = os.path.join(_DIR, ".voice_waveform") -_LOADING_PID_FILE = os.path.join(_DIR, ".voice_loading_pid") - -_BH = CFG.get("barehands_state_dir") or "" -_BH_STATE = os.path.join(_BH, "state") if _BH else "" -_BH_WAVE = os.path.join(_BH, "wave.json") if _BH else "" - -_THINKING_SOUND = CFG.get("thinking_sound") or "" - -_WAVEFORM_MIN_INTERVAL = 1.0 / 15 # ~15 writes/sec is plenty for 60fps reads -_last_waveform_write = 0.0 -_static_proc: subprocess.Popen | None = None - - -def set_state(name: str): - """Write the state. Never raises — the show must go on.""" +# backtalk: talk to your Claude Code agent out loud. +# Copyright (C) 2026 Jared Rhodenizer +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as published +# by the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. +# +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . +# +# SPDX-License-Identifier: AGPL-3.0-or-later +"""The signal bus — tiny files any other program can watch. + +The voice line leaves notes; faces read the notes. That one dumb trick +is the whole integration surface: + + .voice_state idle | listening | thinking | speaking + .voice_waveform JSON {ts, samples: [64 floats]} while audio plays + .voice_loading_pid exists while the thinking sound is playing + .voice_heartbeat unix time as text, rewritten every ~2 s while the + voice line is alive (stale or missing = dead/hung) + .voice_activity JSON {ts, turn_started, line} — what the agent is + doing RIGHT NOW during a turn ("Read: foo.py"); + deleted when the turn ends + +Written to signals_dir (default: the repo root). Visualizers built on +this contract just work. The heartbeat and activity files are how a face +tells "thinking hard" from "process dead": state says thinking either +way, only the heartbeat keeps ticking in the first case. + +THE BAREHANDS SEAM: set barehands_state_dir in backtalk.json to a +barehands checkout's state/ folder and the same signals are mirrored in +its format (state/state as a bare word, state/wave.json normalized +0..1) — the on-screen ring becomes your agent's face with zero glue. + +Every write is wrapped: the bus must never crash the voice line. +""" +import asyncio +import json +import os +import subprocess +import sys +import threading +import time + +import numpy as np + +from backtalk.config import CFG + +_DIR = CFG["signals_dir"] +_STATE_FILE = os.path.join(_DIR, ".voice_state") +_WAVEFORM_FILE = os.path.join(_DIR, ".voice_waveform") +_LOADING_PID_FILE = os.path.join(_DIR, ".voice_loading_pid") +_HEARTBEAT_FILE = os.path.join(_DIR, ".voice_heartbeat") +_ACTIVITY_FILE = os.path.join(_DIR, ".voice_activity") + +_BH = CFG.get("barehands_state_dir") or "" +_BH_STATE = os.path.join(_BH, "state") if _BH else "" +_BH_WAVE = os.path.join(_BH, "wave.json") if _BH else "" + +_THINKING_SOUND = CFG.get("thinking_sound") or "" + +_WAVEFORM_MIN_INTERVAL = 1.0 / 15 # ~15 writes/sec is plenty for 60fps reads +_last_waveform_write = 0.0 +_static_proc: subprocess.Popen | None = None + + +def set_state(name: str): + """Write the state. Never raises — the show must go on.""" + try: + with open(_STATE_FILE, "w") as f: + f.write(name) + except OSError: + pass + if _BH_STATE: + try: + with open(_BH_STATE, "w") as f: + f.write(name) + except OSError: + pass + + +def feed_waveform(pcm: np.ndarray): + """Feed one PCM block (int16) — throttled, downsampled to 64 points. + + Also re-asserts state="speaking" on the same throttle: this only runs + while the mouth is audibly playing, so the bus self-heals within + ~70ms if a stray writer stomps the state mid-speech. (That self-heal + rule once closed a bug that took a whole evening to find.)""" + global _last_waveform_write + if pcm.size == 0: + return + now = time.time() + if now - _last_waveform_write < _WAVEFORM_MIN_INTERVAL: + return + _last_waveform_write = now + try: + idx = np.linspace(0, pcm.size - 1, 64).astype(int) + raw = pcm[idx].astype(float) + with open(_WAVEFORM_FILE, "w") as f: + f.write(json.dumps({"ts": now, "samples": raw.tolist()})) + if _BH_WAVE: + norm = np.clip(np.abs(raw) / 32768.0, 0.0, 1.0) + with open(_BH_WAVE, "w") as f: + f.write(json.dumps({"ts": now, "samples": norm.tolist()})) + except (OSError, ValueError): + pass + set_state("speaking") + + +def _player_cmd(path: str) -> list[str] | None: + if sys.platform == "darwin": + return ["afplay", "-v", "0.35", path] + for cand in ("ffplay", "aplay", "paplay"): + from shutil import which + if which(cand): + if cand == "ffplay": + return ["ffplay", "-nodisp", "-autoexit", "-loglevel", + "quiet", "-volume", "35", path] + return [cand, path] + return None + + +def static_start(): + """Optional thinking sound — plays while the brain works.""" + global _static_proc + if not _THINKING_SOUND or not os.path.exists(_THINKING_SOUND): + return + static_stop() + cmd = _player_cmd(_THINKING_SOUND) + if not cmd: + return + try: + _static_proc = subprocess.Popen( + cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + with open(_LOADING_PID_FILE, "w") as f: + f.write(str(_static_proc.pid)) + except OSError: + _static_proc = None + + +def static_stop(): + global _static_proc + if _static_proc is not None: + try: + _static_proc.terminate() + except OSError: + pass + _static_proc = None + try: + os.remove(_LOADING_PID_FILE) + except OSError: + pass + + +# ---- THE HEARTBEAT: .voice_heartbeat +# Content: the current unix time as text (e.g. "1787702400.12"), rewritten +# every ~2 s for as long as the voice line is alive. A face that sees it +# older than a few seconds (ai-visualizer uses 6) shows the link as LOST. +# It rides the asyncio loop ON PURPOSE: a loop wedged in a sync call +# stops beating, so "hung" reads as dead — that is the signal we want. +# Falls back to a daemon thread only if no loop is running. Removed on +# heartbeat_stop() so a clean exit reads LOST at once, not 6 s later. +_HEARTBEAT_INTERVAL = 2.0 +_hb_task: asyncio.Task | None = None +_hb_thread: threading.Thread | None = None +_hb_stop_evt = threading.Event() + + +def _beat(): try: - with open(_STATE_FILE, "w") as f: - f.write(name) + with open(_HEARTBEAT_FILE, "w") as f: + f.write(f"{time.time():.2f}") except OSError: pass - if _BH_STATE: - try: - with open(_BH_STATE, "w") as f: - f.write(name) - except OSError: - pass - -def feed_waveform(pcm: np.ndarray): - """Feed one PCM block (int16) — throttled, downsampled to 64 points. - Also re-asserts state="speaking" on the same throttle: this only runs - while the mouth is audibly playing, so the bus self-heals within - ~70ms if a stray writer stomps the state mid-speech. (That self-heal - rule once closed a bug that took a whole evening to find.)""" - global _last_waveform_write - if pcm.size == 0: - return - now = time.time() - if now - _last_waveform_write < _WAVEFORM_MIN_INTERVAL: - return - _last_waveform_write = now +async def _heartbeat_loop(): try: - idx = np.linspace(0, pcm.size - 1, 64).astype(int) - raw = pcm[idx].astype(float) - with open(_WAVEFORM_FILE, "w") as f: - f.write(json.dumps({"ts": now, "samples": raw.tolist()})) - if _BH_WAVE: - norm = np.clip(np.abs(raw) / 32768.0, 0.0, 1.0) - with open(_BH_WAVE, "w") as f: - f.write(json.dumps({"ts": now, "samples": norm.tolist()})) - except (OSError, ValueError): + while True: + _beat() + await asyncio.sleep(_HEARTBEAT_INTERVAL) + except asyncio.CancelledError: pass - set_state("speaking") - - -def _player_cmd(path: str) -> list[str] | None: - if sys.platform == "darwin": - return ["afplay", "-v", "0.35", path] - for cand in ("ffplay", "aplay", "paplay"): - from shutil import which - if which(cand): - if cand == "ffplay": - return ["ffplay", "-nodisp", "-autoexit", "-loglevel", - "quiet", "-volume", "35", path] - return [cand, path] - return None - - -def static_start(): - """Optional thinking sound — plays while the brain works.""" - global _static_proc - if not _THINKING_SOUND or not os.path.exists(_THINKING_SOUND): - return - static_stop() - cmd = _player_cmd(_THINKING_SOUND) - if not cmd: + + +def _heartbeat_thread(): + while not _hb_stop_evt.wait(_HEARTBEAT_INTERVAL): + _beat() + + +def heartbeat_start(): + """Start the pulse. Never raises; calling it twice is a no-op.""" + global _hb_task, _hb_thread + if (_hb_task and not _hb_task.done()) or \ + (_hb_thread and _hb_thread.is_alive()): return + _beat() try: - _static_proc = subprocess.Popen( - cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) - with open(_LOADING_PID_FILE, "w") as f: - f.write(str(_static_proc.pid)) - except OSError: - _static_proc = None + _hb_task = asyncio.get_running_loop().create_task( + _heartbeat_loop()) + except RuntimeError: # no loop here: thread it + _hb_stop_evt.clear() + _hb_thread = threading.Thread(target=_heartbeat_thread, + daemon=True) + _hb_thread.start() + except Exception: + pass -def static_stop(): - global _static_proc - if _static_proc is not None: +def heartbeat_stop(): + """Stop the pulse and remove the file. Never raises.""" + global _hb_task, _hb_thread + if _hb_task is not None: + try: + _hb_task.cancel() + except Exception: + pass + _hb_task = None + if _hb_thread is not None: + _hb_stop_evt.set() + _hb_thread = None + for path in (_HEARTBEAT_FILE, _ACTIVITY_FILE): try: - _static_proc.terminate() + os.remove(path) except OSError: pass - _static_proc = None + + +# ---- THE ACTIVITY LINE: .voice_activity +# Content: JSON {"ts": , "turn_started": , +# "line": ""} +# Exists only while a turn is in flight. turn_begin() opens it (line +# "working"), activity("Read: foo.py") rewrites the line per tool call +# keeping turn_started, turn_end() deletes it. A face shows the line as +# a ticker and turn_started as an elapsed clock; the server-side age of +# ts says how long the agent has sat on one tool call. +_turn_started = 0.0 + + +def _write_activity(line: str): + try: + with open(_ACTIVITY_FILE, "w") as f: + f.write(json.dumps({"ts": time.time(), + "turn_started": _turn_started, + "line": line})) + except (OSError, ValueError, TypeError): + pass + + +def turn_begin(): + """A user turn starts: reset the elapsed clock, open the file.""" + global _turn_started + _turn_started = time.time() + _write_activity("working") + + +def activity(line: str): + """What the agent is doing right now (one short line, no prefix). + Outside a turn (warmup, resume recap) the clock starts here.""" + global _turn_started + if not _turn_started: + _turn_started = time.time() + line = " ".join(str(line or "").split())[:120] + _write_activity(line) + + +def turn_end(): + """The turn is over (answered, cancelled, or failed): clear it.""" + global _turn_started + _turn_started = 0.0 try: - os.remove(_LOADING_PID_FILE) + os.remove(_ACTIVITY_FILE) except OSError: pass diff --git a/tests/test_mouth_lookahead.py b/tests/test_mouth_lookahead.py new file mode 100644 index 0000000..5de9dd1 --- /dev/null +++ b/tests/test_mouth_lookahead.py @@ -0,0 +1,153 @@ +# backtalk: talk to your Claude Code agent out loud. +# SPDX-License-Identifier: AGPL-3.0-or-later +"""Lookahead mouth check, no speakers needed: a fake synth (0.3s to first +audio, then streamed blocks) and a fake output device (sleeps in real +time, records every write) stand in for Kokoro and sounddevice. + +Asserts: + 1. chunk boundaries play back-to-back (gap < 80ms) — the point of the + synth-ahead thread; the old one-thread mouth paid ~1s per boundary + 2. a chunk shorter than the prebuffer still plays and completes + 3. barge-in drops everything queued/rendering, nothing stale plays, + and the mouth speaks fresh text afterwards + +Run: .venv/Scripts/python tests/test_mouth_lookahead.py +""" +import sys +import time +import types +from pathlib import Path + +import numpy as np + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import backtalk # noqa: E402 + +# Stub the signal bus and the ducker: file writes and OS mixer calls +# have no place in a unit check. +_sig = types.SimpleNamespace(static_stop=lambda: None, + set_state=lambda s: None, + feed_waveform=lambda p: None) +sys.modules["backtalk.signals"] = _sig +backtalk.signals = _sig + + +class _Ducker: + def speech_start(self): pass + def speech_end(self): pass + def restore_now(self): pass + + +_duck = types.SimpleNamespace(Ducker=_Ducker) +sys.modules["backtalk.ducking"] = _duck +backtalk.ducking = _duck + +import backtalk.mouth as mouth # noqa: E402 + +RATE = 24000 +writes: list = [] # (t_start, value, seconds) + + +class FakeOut: + def __init__(self, samplerate, channels, dtype): + self.rate = samplerate + self.active = False + + def start(self): + self.active = True + + def close(self, ignore_errors=False): + self.active = False + + def write(self, pcm): + writes.append((time.perf_counter(), int(pcm[0]) if len(pcm) else 0, + len(pcm) / self.rate)) + time.sleep(len(pcm) / self.rate) + + +_seq = [0] + + +def fake_synth(text, timeout=30.0): + """Each chunk renders as a distinct constant sample value, so the + write log tells chunks apart. Short text -> one block shorter than + the prebuffer; otherwise 4 streamed blocks (1.2s of audio).""" + _seq[0] += 1 + v = _seq[0] + time.sleep(0.3) # time-to-first-audio + blocks = 1 if len(text) < 6 else 4 + for i in range(blocks): + if i: + time.sleep(0.1) # streaming render + yield RATE, np.full(int(RATE * 0.3), v, dtype=np.int16) + + +mouth.sd = types.SimpleNamespace(OutputStream=FakeOut) +mouth.synth_stream = fake_synth + + +def _spans(): + """value -> (first write start, last write end), ignoring the zero + padding _cut() writes.""" + spans: dict = {} + for t, v, d in writes: + if v == 0: + continue + first, end = spans.get(v, (t, t + d)) + spans[v] = (min(first, t), max(end, t + d)) + return spans + + +def main(): + m = mouth.Mouth() + + # 1. back-to-back chunks + writes.clear() + m.say("Number one. Number two. Number three. Number four.") + m.wait_done(timeout=30) + spans = _spans() + assert sorted(spans) == [1, 2, 3, 4], spans + order = [v for _, v, _ in writes if v] + assert order == sorted(order), "chunks played out of order" + gaps = [spans[v + 1][0] - spans[v][1] for v in (1, 2, 3)] + print(f"boundary gaps: {[f'{g*1000:.0f}ms' for g in gaps]}") + assert all(g < 0.08 for g in gaps), gaps + assert m._pending == 0 and not m.speaking + + # 2. short chunk (under the prebuffer) completes + writes.clear() + m.say("Hi.") + m.wait_done(timeout=10) + assert _spans() and list(_spans()) == [5], _spans() + assert m._pending == 0 and not m.speaking + print("short chunk: ok") + + # 3. barge-in + writes.clear() + m.say("Number six. Number seven. Number eight. Number nine.") + time.sleep(1.2) # chunk 6 is playing, 7 rendering + m.shut_up() + t_cut = time.perf_counter() + time.sleep(2.0) + late = [(round(t - t_cut, 2), v) for t, v, d in writes + if v and t > t_cut + 0.15] + assert not late, f"stale audio played after barge-in: {late}" + assert set(_spans()) == {6}, _spans() + assert m._pending == 0 and not m.speaking + print(f"barge-in: cut mid-chunk, nothing stale played") + + writes.clear() + m.say("Ten.") + m.wait_done(timeout=10) + # its synth number is whatever came after the aborted chunk 7 — + # chunks 8 and 9 were dropped before they ever reached the synth + vals = list(_spans()) + assert len(vals) == 1 and vals[0] > 6, _spans() + assert m._pending == 0 and not m.speaking + print("speaks fresh text after barge-in: ok") + print("ALL OK") + + +if __name__ == "__main__": + main()