diff --git a/README.md b/README.md
index e8f2def..d408741 100644
--- a/README.md
+++ b/README.md
@@ -54,6 +54,22 @@ Two engines, and the setup wizard offers you both instead of quietly defaulting.
**ElevenLabs, the natural one.** The human-sounding voice most people actually want, on your own API key. The free tier is enough to audition it; day-to-day talking runs on the paid starter plan. The wizard walks the whole thing with you: account, key into the keychain, then an audition of real voices through backtalk's own mouth until one fits. Want the exact voice from my videos? It's called **Tarquin** in the ElevenLabs voice library: search it by name and you're done hunting. Under the hood it is: set `elevenlabs.enabled` and your `voice_id` in the config, and have `ffmpeg` installed. **The key never goes in a file.** On macOS, seed it into the Keychain once with `security add-generic-password -a "$USER" -s backtalk-elevenlabs -T /usr/bin/security -w` (it prompts for the secret) and backtalk reads it from there. Linux: `secret-tool store --label backtalk service backtalk-elevenlabs`. The `ELEVENLABS_API_KEY` environment variable works as a last resort, but an export in a shell profile is a plaintext key on disk; the keychain is the grown-up path. Kokoro stays wired in as the automatic fallback, so if the cloud fails the voice degrades instead of going mute, and `logs/backtalk.log` records why.
+## Local AI Brain (Optional)
+
+While backtalk defaults to Claude Code via the Agent SDK, it also supports local OpenAI-compatible inference servers (such as `llama-server`, Ollama, vLLM, or Aphrodite) for a 100% offline, private voice assistant.
+
+To switch to a local model, set `"brain": "local"` in your `backtalk.json`:
+
+```json
+{
+ "brain": "local",
+ "api_base": "http://127.0.0.1:8080/v1",
+ "model": "default"
+}
+```
+
+The local brain maintains multi-step tool execution (recursive directory inspection, file reading, command execution, and web browsing) and streams partial sentences directly to the voice pipeline with sentence boundary chunking. See `backtalk.local.json.example`.
+
## 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.
diff --git a/backtalk.local.json.example b/backtalk.local.json.example
new file mode 100644
index 0000000..8c90a9c
--- /dev/null
+++ b/backtalk.local.json.example
@@ -0,0 +1,13 @@
+{
+ "brain": "local",
+ "api_base": "http://127.0.0.1:8080/v1",
+ "model": "default",
+ "agent_dir": "~",
+ "name": "Assistant",
+ "ptt_key": "home",
+ "voice": "bm_lewis",
+ "tts_device": "cpu",
+ "stt_model": "small.en",
+ "stt_device": "auto",
+ "extra_dirs": []
+}
diff --git a/backtalk/brain.py b/backtalk/brain.py
index 4ac7c25..9b422d8 100644
--- a/backtalk/brain.py
+++ b/backtalk/brain.py
@@ -1,5 +1,5 @@
-# backtalk: talk to your Claude Code agent out loud.
-# Copyright (C) 2026 Jared Rhodenizer
+# backtalk: talk to your agent out loud.
+# Copyright (C) 2026 Jared Rhodenizer, AnZym contributors
#
# 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
@@ -15,354 +15,34 @@
# 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.
+"""The warm brain router — dispatches to the configured brain provider.
-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.
+Supports:
+- "claude" (default): Claude Agent SDK session (brain_claude.py)
+- "local" / "openai": Local OpenAI-compatible server (brain_local.py)
"""
-import asyncio
import os
-import re
-import warnings
-from datetime import datetime
-
-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
+from backtalk.config import CFG
-_SENTENCE_END = re.compile(r"(?<=[.!?])\s")
+SESSION_FILE = os.path.join(CFG.get("signals_dir", "/tmp/signals"), ".backtalk_session")
-SESSION_FILE = os.path.join(CFG["signals_dir"], ".backtalk_session")
+def get_brain_class():
+ """Resolve the brain implementation class based on configuration."""
+ brain_type = str(CFG.get("brain", "claude")).strip().lower()
+ if brain_type in ("local", "openai", "llama", "vllm", "ollama"):
+ from backtalk.brain_local import LocalWarmBrain
+ return LocalWarmBrain
+ from backtalk.brain_claude import ClaudeWarmBrain
+ return ClaudeWarmBrain
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"],
- skills=CFG["visible_skills"],
- 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 _pull_rate_limits(self):
- """Ask the CLI outright how much of the plan is spent.
-
- A DIRECT QUERY, not the RateLimitEvent stream. The event fires
- rarely and usually arrives carrying resets_at with no utilization
- at all, so a listener built on it reports nothing most of the
- time -- which is exactly how this feature looked broken for its
- whole life. (Community fix, ai-visualizer issue #1.)
-
- THIS REACHES PAST THE SDK'S PUBLIC SURFACE ON PURPOSE, and a
- reader should know it rather than discover it. `get_usage` is a
- control request the bundled CLI answers but the SDK never wraps,
- so there is no supported call to make. The supported-looking
- alternative is a dead end and was tested as one: the terminal
- status line never fires in a headless session, so its numbers
- are unreachable from here.
-
- Which means this can stop working without anyone doing anything
- wrong, and the containment is the point. Every failure is
- swallowed and the readout simply goes quiet. It must never cost
- a turn, so it is also bounded -- an unanswered control request
- would otherwise hang the voice line mid-conversation."""
- if not CFG.get("show_usage"):
- return
- try:
- usage = await asyncio.wait_for(
- self._client._query._send_control_request(
- {"subtype": "get_usage"}), 5)
- for window in ("five_hour", "seven_day"):
- w = (usage.get("rate_limits") or {}).get(window)
- if not w:
- continue
- # Two spellings accepted deliberately: this shape is not
- # documented anywhere, so the cheap tolerance is worth
- # more than the tidiness. Both are percentages, and the
- # rest of the pipeline wants a 0..1 fraction.
- pct = w.get("utilization")
- if pct is None:
- pct = w.get("used_percentage")
- pct = pct / 100 if pct is not None else None
- resets = w.get("resets_at")
- if isinstance(resets, str):
- resets = int(datetime.fromisoformat(resets).timestamp())
- signals.set_rate_limit(window, pct, resets)
- 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)
- await self._pull_rate_limits()
- break
- tail = buf.strip()
- if tail:
- yield tail
-
+ """WarmBrain dispatcher: instantiates the configured brain backend."""
-if __name__ == "__main__":
- import time
+ def __new__(cls, *args, **kwargs):
+ target_cls = get_brain_class()
+ return target_cls(*args, **kwargs)
- 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())
+__all__ = ["WarmBrain", "SESSION_FILE", "get_brain_class"]
diff --git a/backtalk/brain_claude.py b/backtalk/brain_claude.py
new file mode 100644
index 0000000..7488f48
--- /dev/null
+++ b/backtalk/brain_claude.py
@@ -0,0 +1,370 @@
+# 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 datetime import datetime
+
+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 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"],
+ skills=CFG["visible_skills"],
+ 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 _pull_rate_limits(self):
+ """Ask the CLI outright how much of the plan is spent.
+
+ A DIRECT QUERY, not the RateLimitEvent stream. The event fires
+ rarely and usually arrives carrying resets_at with no utilization
+ at all, so a listener built on it reports nothing most of the
+ time -- which is exactly how this feature looked broken for its
+ whole life. (Community fix, ai-visualizer issue #1.)
+
+ THIS REACHES PAST THE SDK'S PUBLIC SURFACE ON PURPOSE, and a
+ reader should know it rather than discover it. `get_usage` is a
+ control request the bundled CLI answers but the SDK never wraps,
+ so there is no supported call to make. The supported-looking
+ alternative is a dead end and was tested as one: the terminal
+ status line never fires in a headless session, so its numbers
+ are unreachable from here.
+
+ Which means this can stop working without anyone doing anything
+ wrong, and the containment is the point. Every failure is
+ swallowed and the readout simply goes quiet. It must never cost
+ a turn, so it is also bounded -- an unanswered control request
+ would otherwise hang the voice line mid-conversation."""
+ if not CFG.get("show_usage"):
+ return
+ try:
+ usage = await asyncio.wait_for(
+ self._client._query._send_control_request(
+ {"subtype": "get_usage"}), 5)
+ for window in ("five_hour", "seven_day"):
+ w = (usage.get("rate_limits") or {}).get(window)
+ if not w:
+ continue
+ # Two spellings accepted deliberately: this shape is not
+ # documented anywhere, so the cheap tolerance is worth
+ # more than the tidiness. Both are percentages, and the
+ # rest of the pipeline wants a 0..1 fraction.
+ pct = w.get("utilization")
+ if pct is None:
+ pct = w.get("used_percentage")
+ pct = pct / 100 if pct is not None else None
+ resets = w.get("resets_at")
+ if isinstance(resets, str):
+ resets = int(datetime.fromisoformat(resets).timestamp())
+ signals.set_rate_limit(window, pct, resets)
+ 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)
+ await self._pull_rate_limits()
+ break
+ tail = buf.strip()
+ if tail:
+ yield tail
+
+ClaudeWarmBrain = WarmBrain
+
+
+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/brain_local.py b/backtalk/brain_local.py
new file mode 100644
index 0000000..df1067a
--- /dev/null
+++ b/backtalk/brain_local.py
@@ -0,0 +1,546 @@
+# backtalk: talk to your local AI agent out loud.
+# Copyright (C) 2026 Jared Rhodenizer, AnZym contributors
+#
+# 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 local brain — connects to a local OpenAI-compatible inference server
+(such as llama-server, Ollama, vLLM, or Aphrodite), executing multi-step
+filesystem and web tools and streaming spoken output to the mouth.
+"""
+import asyncio
+import json
+import os
+import re
+import subprocess
+from pathlib import Path
+
+import httpx
+
+try:
+ from bs4 import BeautifulSoup
+except ImportError:
+ BeautifulSoup = None
+
+try:
+ from ddgs import DDGS
+except ImportError:
+ try:
+ from duckduckgo_search import DDGS
+ except ImportError:
+ DDGS = None
+
+from backtalk import signals
+from backtalk.config import CFG, DISCIPLINE
+from backtalk.vlog import log
+
+_SENTENCE_END = re.compile(r"(?<=[.!?])\s")
+_SPECIAL_TOKENS = re.compile(r"<\|im_end\|>|<\|im_start\|>|<\|endoftext\|>|<\|[^|]+?\|>")
+_THINK_BLOCK = re.compile(r"[\s\S]*?|\[THINK\][\s\S]*?\[/THINK\]")
+_TOOL_TAGS = re.compile(r"[\s\S]*?|]+>[\s\S]*?|]+>[\s\S]*?")
+_CODE_BLOCK = re.compile(r"```[\s\S]*?```", re.DOTALL)
+_INLINE_SCRIPT = re.compile(r"<<\s*['\"]?[A-Za-z0-9_]+['\"]?[\s\S]*", re.DOTALL)
+_RAW_CODE_LINE = re.compile(r"^(?:import |from |def |class |with open|cat >|\s*return |\s*#|\s*if __name__).*$", re.MULTILINE)
+
+SESSION_FILE = os.path.join(CFG.get("signals_dir", "/tmp/signals"), ".backtalk_session")
+
+
+def _clean_text(text: str) -> str:
+ """Strip special tokens, think blocks, code blocks, and tool tags for speech output."""
+ t = _THINK_BLOCK.sub("", text)
+ t = _TOOL_TAGS.sub("", t)
+ t = _CODE_BLOCK.sub("", t)
+ t = _INLINE_SCRIPT.sub("", t)
+ t = _SPECIAL_TOKENS.sub("", t)
+ t = _RAW_CODE_LINE.sub("", t)
+ t = re.sub(r"^#+\s*", "", t, flags=re.MULTILINE)
+ t = t.replace("**", "").replace("*", "").replace("`", "")
+ lines = [
+ line.strip()
+ for line in t.split("\n")
+ if line.strip() and not line.strip().startswith(("#", "//", "/*", "*", "def ", "import "))
+ ]
+ return " ".join(lines).strip()
+
+
+def parse_tool_call(text: str) -> tuple[str, dict] | None:
+ """Extract tool name and arguments from JSON or model XML format."""
+ # Format 1: JSON
+ m1 = re.search(r"([\s\S]*?)", text)
+ if m1:
+ raw = m1.group(1).strip()
+ try:
+ d = json.loads(raw)
+ return d.get("tool") or d.get("name"), d.get("parameters") or d.get("arguments") or d
+ except Exception:
+ pass
+
+ # Format 2: VALUE
+ m2 = re.search(r"([\s\S]*?)", text)
+ if m2:
+ tool_name = m2.group(1).strip()
+ params = {}
+ for pm in re.finditer(r"([\s\S]*?)", m2.group(2)):
+ params[pm.group(1).strip()] = pm.group(2).strip()
+ return tool_name, params
+
+ # Format 3: JSON markdown code block
+ m3 = re.search(r'```(?:json)?\s*(\{\s*"(?:tool|name)"[\s\S]*?\})\s*```', text)
+ if m3:
+ try:
+ d = json.loads(m3.group(1))
+ return d.get("tool") or d.get("name"), d.get("parameters") or d.get("arguments") or d
+ except Exception:
+ pass
+
+ return None
+
+
+def resolve_project_path(target: str, current_cwd: str) -> str:
+ """Map alias or relative name to absolute filesystem directory."""
+ aliases = CFG.get("project_aliases", {})
+ cleaned = target.strip().lower().replace("_", " ").replace("-", " ")
+ for alias, p in aliases.items():
+ if alias.lower() in cleaned or cleaned == alias.lower():
+ return os.path.expanduser(p)
+
+ for extra in CFG.get("extra_dirs", []):
+ expanded_extra = os.path.expanduser(extra)
+ if os.path.basename(expanded_extra).lower() == cleaned:
+ return expanded_extra
+
+ expanded = os.path.expanduser(target.strip())
+ if os.path.isabs(expanded) and os.path.exists(expanded):
+ return expanded
+ rel = os.path.join(current_cwd, target.strip())
+ if os.path.exists(rel):
+ return rel
+ return expanded
+
+
+def execute_tool(tool_name: str, args: dict, brain_ref=None) -> str:
+ """Execute a local, web, or workspace tool safely."""
+ default_ws = CFG.get("agent_dir", os.getcwd())
+ cwd = brain_ref.active_project_dir if brain_ref else default_ws
+ try:
+ signals.set_state("thinking")
+
+ if tool_name == "switch_workspace":
+ target = args.get("path") or args.get("project") or args.get("name", "")
+ resolved = resolve_project_path(target, cwd)
+ log(f"[brain-local] switching workspace to: {resolved}")
+ if not os.path.exists(resolved):
+ return f"Error: Workspace path {resolved} does not exist."
+ if brain_ref:
+ brain_ref.active_project_dir = resolved
+
+ tree_lines = []
+ for root, dirs, files in os.walk(resolved):
+ dirs[:] = [d for d in dirs if not d.startswith(".") and d not in ("__pycache__", "node_modules", "build")]
+ rel = os.path.relpath(root, resolved)
+ depth = rel.count(os.sep)
+ if depth > 2:
+ continue
+ indent = " " * depth
+ if rel != ".":
+ tree_lines.append(f"{indent}📂 {os.path.basename(root)}/")
+ for f in files[:8]:
+ if not f.startswith("."):
+ tree_lines.append(f"{indent} 📄 {f}")
+ if len(tree_lines) > 30:
+ break
+
+ readme_text = ""
+ readme_p = os.path.join(resolved, "README.md")
+ if os.path.exists(readme_p):
+ try:
+ with open(readme_p, "r", encoding="utf-8", errors="ignore") as f:
+ readme_text = f.read(1500)
+ except Exception:
+ pass
+
+ return (
+ f"--- Switched Active Workspace to: {resolved} ---\n"
+ f"Workspace File Tree:\n" + "\n".join(tree_lines[:25]) + "\n"
+ + (f"README Overview:\n{readme_text}\n" if readme_text else "")
+ )
+
+ elif tool_name == "search_web":
+ if not DDGS:
+ return "Error: Web search library (ddgs / duckduckgo_search) is not installed."
+ query = args.get("query", "")
+ if not query:
+ return "Error: No search query provided."
+ log(f"[brain-local] searching web for: {query}")
+ results = []
+ with DDGS() as ddgs:
+ for r in ddgs.text(query, max_results=int(args.get("max_results", 4))):
+ title = r.get("title", "")
+ snippet = r.get("body", "")
+ href = r.get("href", "")
+ results.append(f"Title: {title}\nSnippet: {snippet}\nURL: {href}\n")
+ return f"--- Web Search Results for '{query}' ---\n" + ("\n".join(results) if results else "No results found.")
+
+ elif tool_name == "read_web_page":
+ if not BeautifulSoup:
+ return "Error: BeautifulSoup (bs4) is not installed."
+ url = args.get("url", "")
+ if not url:
+ return "Error: No URL provided."
+ log(f"[brain-local] fetching web page: {url}")
+ headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}
+ resp = httpx.get(url, headers=headers, timeout=12.0, follow_redirects=True)
+ soup = BeautifulSoup(resp.text, "html.parser")
+ for tag in soup(["script", "style", "nav", "footer", "header", "aside", "form"]):
+ tag.decompose()
+ text = " ".join(soup.stripped_strings)
+ max_chars = int(args.get("max_chars", 3000))
+ return f"--- Web Page Content ({url}) ---\n{text[:max_chars]}"
+
+ elif tool_name == "list_dir":
+ raw_p = args.get("path", ".")
+ p = raw_p if os.path.isabs(raw_p) else os.path.join(cwd, raw_p)
+ p = os.path.expanduser(p)
+ log(f"[brain-local] inspecting directory: {p}")
+ if not os.path.exists(p):
+ return f"Error: Directory {p} does not exist."
+
+ tree_items = []
+ for root, dirs, files in os.walk(p):
+ dirs[:] = [d for d in dirs if not d.startswith(".") and d not in ("__pycache__", "build", "install", "node_modules", ".git")]
+ rel = os.path.relpath(root, p)
+ depth = rel.count(os.sep)
+ if depth > 2:
+ continue
+ prefix = " " * depth
+ if rel != ".":
+ tree_items.append(f"{prefix}📂 {os.path.basename(root)}/")
+ for f in files:
+ if not f.startswith("."):
+ full_rel = os.path.join(rel, f) if rel != "." else f
+ tree_items.append(f"{prefix} 📄 {full_rel}")
+ if len(tree_items) > 50:
+ break
+
+ return f"Recursive File Tree of {p}:\n" + "\n".join(tree_items[:45])
+
+ elif tool_name == "read_file":
+ raw_p = args.get("path", "")
+ p = raw_p if os.path.isabs(raw_p) else os.path.join(cwd, raw_p)
+ p = os.path.expanduser(p)
+ log(f"[brain-local] reading file: {p}")
+ if not os.path.exists(p):
+ found = list(Path(cwd).glob(f"**/{os.path.basename(raw_p)}"))
+ if found:
+ p = str(found[0])
+ else:
+ return f"Error: File {raw_p} does not exist in {cwd}."
+ max_lines = int(args.get("max_lines", 120))
+ with open(p, "r", encoding="utf-8", errors="ignore") as f:
+ lines = [f.readline() for _ in range(max_lines)]
+ return f"--- File Content: {p} ---\n" + "".join(lines)
+
+ elif tool_name == "search_files":
+ raw_p = args.get("path", ".")
+ base = raw_p if os.path.isabs(raw_p) else os.path.join(cwd, raw_p)
+ base = os.path.expanduser(base)
+ query = args.get("query", "")
+ log(f"[brain-local] searching files for: {query} in {base}")
+ matches = []
+ for root, _, filenames in os.walk(base):
+ for fn in filenames:
+ if query.lower() in fn.lower():
+ matches.append(os.path.join(root, fn))
+ if len(matches) >= 20:
+ break
+ if len(matches) >= 20:
+ break
+ return f"Found matching files for '{query}':\n" + "\n".join(matches)
+
+ elif tool_name == "run_command":
+ cmd = args.get("cmd", "")
+ target_cwd = args.get("cwd", cwd)
+ target_cwd = target_cwd if os.path.isabs(target_cwd) else os.path.join(cwd, target_cwd)
+ target_cwd = os.path.expanduser(target_cwd)
+ log(f"[brain-local] running command: {cmd} (in {target_cwd})")
+ res = subprocess.run(cmd, shell=True, cwd=target_cwd, capture_output=True, text=True, timeout=15)
+ out = res.stdout if res.stdout else res.stderr
+ return f"Command output ($ {cmd} in {target_cwd}):\n{out[:2000]}"
+
+ return f"Unknown tool: {tool_name}"
+ except Exception as e:
+ return f"Tool execution failed: {e}"
+
+
+TOOL_PROMPT = """
+### OPERATIONAL DIRECTIVE: MULTI-STEP TOOLS & SPOKEN ANSWERS
+You are a voice-interactive assistant connected directly to audio input and speech synthesis.
+
+RULES:
+1. When you need to inspect directories, read code files, or search the web, invoke tools using {"tool": "name", ...}.
+2. `list_dir` returns the recursive file tree, so you can locate and read target files immediately.
+3. As soon as you have inspected the necessary code or files, STOP calling tools and deliver your spoken answer directly.
+4. Keep spoken responses clear, concise, and natural for text-to-speech audio. Avoid markdown tables, URLs, or long unpronounceable code blocks.
+
+Available Tools:
+- switch_workspace(path)
+- list_dir(path)
+- read_file(path, max_lines)
+- search_files(path, query)
+- run_command(cmd, cwd)
+- search_web(query)
+- read_web_page(url)
+"""
+
+
+class LocalWarmBrain:
+ """A persistent local LLM session via an OpenAI-compatible endpoint (llama-server, Ollama, etc.)."""
+
+ def __init__(self, model: str | None = None, can_use_tool=None, resume_id: str | None = None):
+ self.api_base = CFG.get("api_base", "http://127.0.0.1:8080/v1").rstrip("/")
+ self.model = model or CFG.get("model", "default")
+ self.active_project_dir = CFG.get("agent_dir", os.getcwd())
+ self._can_use_tool = can_use_tool
+ self.session = {"turns": 0, "out_tokens": 0, "in_tokens": 0, "cost": 0.0}
+ self.messages = []
+ self._dirty = False
+ self._interrupted = False
+ self.system_prompt = self._load_system_prompt()
+ self.permission_mode = CFG.get("permission_mode", "ask")
+
+ def _get_workspace_snapshot(self) -> str:
+ try:
+ cwd = self.active_project_dir
+ if not os.path.exists(cwd):
+ return ""
+ entries = os.listdir(cwd)
+ dirs = [f"{e}/" for e in sorted(entries) if os.path.isdir(os.path.join(cwd, e)) and not e.startswith(".")]
+ files = [e for e in sorted(entries) if os.path.isfile(os.path.join(cwd, e)) and not e.startswith(".")]
+ return (
+ f"\n[Active Workspace: {cwd}]\n"
+ f"Directories: {', '.join(dirs[:15])}\n"
+ f"Files: {', '.join(files[:15])}"
+ )
+ except Exception:
+ return ""
+
+ def _load_system_prompt(self) -> str:
+ agent_dir = Path(os.path.expanduser(CFG.get("agent_dir", os.getcwd())))
+ prompt_parts = [DISCIPLINE, TOOL_PROMPT]
+
+ for filename in ("AGENT.md", "CLAUDE.md", "SYSTEM.md"):
+ p = agent_dir / filename
+ if p.exists():
+ try:
+ prompt_parts.append(p.read_text(encoding="utf-8"))
+ log(f"[brain-local] loaded persona from {p}")
+ break
+ except Exception as e:
+ log(f"[brain-local] error reading {p}: {e}")
+
+ for extra in CFG.get("extra_dirs", []):
+ extra_path = Path(os.path.expanduser(extra))
+ idx = extra_path / "VAULT-INDEX.md"
+ if idx.exists():
+ try:
+ prompt_parts.append(f"## Context ({extra})\n{idx.read_text(encoding='utf-8')[:3000]}")
+ log(f"[brain-local] loaded index from {idx}")
+ except Exception:
+ pass
+ readme = extra_path / "README.md"
+ if readme.exists():
+ try:
+ prompt_parts.append(f"## Workspace Overview ({extra})\n{readme.read_text(encoding='utf-8')[:4000]}")
+ log(f"[brain-local] loaded workspace summary from {readme}")
+ except Exception:
+ pass
+
+ return "\n\n".join(prompt_parts)
+
+ async def start(self):
+ full_prompt = self.system_prompt + self._get_workspace_snapshot()
+ self.messages = [{"role": "system", "content": full_prompt}]
+ self._dirty = False
+ self._interrupted = False
+ log(f"[brain-local] connected to {self.api_base} (model={self.model}, cwd={self.active_project_dir})")
+
+ async def stop(self):
+ self.messages.clear()
+ self._dirty = False
+
+ async def interrupt(self):
+ self._interrupted = True
+ self._dirty = False
+
+ async def reset_turn(self, timeout: float = 8.0):
+ self._dirty = False
+ self._interrupted = False
+
+ async def set_permission_mode(self, mode: str):
+ self.permission_mode = mode
+ log(f"[brain-local] permission mode set to: {mode}")
+
+ async def context_usage(self):
+ return {
+ "total_tokens": self.session["in_tokens"] + self.session["out_tokens"],
+ "turns": self.session["turns"],
+ }
+
+ async def command(self, cmd: str) -> str:
+ parts = cmd.strip().split()
+ if not parts:
+ return ""
+ verb = parts[0].lower()
+ if verb == "/clear":
+ full_prompt = self.system_prompt + self._get_workspace_snapshot()
+ self.messages = [{"role": "system", "content": full_prompt}]
+ self.session = {"turns": 0, "out_tokens": 0, "in_tokens": 0, "cost": 0.0}
+ return "Conversation history cleared."
+ elif verb == "/compact":
+ if len(self.messages) > 9:
+ self.messages = [self.messages[0]] + self.messages[-8:]
+ return "Session compacted."
+ elif verb == "/model" and len(parts) > 1:
+ self.model = parts[1]
+ return f"Switched model to {self.model}."
+ elif verb == "/effort":
+ return "Effort level updated."
+ elif verb in ("/cd", "/workspace") and len(parts) > 1:
+ res = execute_tool("switch_workspace", {"path": parts[1]}, self)
+ return res
+ return f"Command acknowledged: {cmd}"
+
+ async def _query_llm(self, messages: list) -> str:
+ payload = {
+ "model": self.model,
+ "messages": messages,
+ "stream": False,
+ "temperature": 0.3,
+ "max_tokens": 2048,
+ }
+ async with httpx.AsyncClient(timeout=120.0) as client:
+ resp = await client.post(f"{self.api_base}/chat/completions", json=payload)
+ if resp.status_code == 200:
+ data = resp.json()
+ return data["choices"][0]["message"]["content"]
+ log(f"[brain-local] LLM returned status {resp.status_code}: {resp.text[:200]}")
+ return ""
+
+ async def ask_stream(self, utterance: str):
+ self._dirty = True
+ self._interrupted = False
+
+ snapshot = self._get_workspace_snapshot()
+ user_msg = utterance
+ if snapshot:
+ user_msg = f"{utterance}\n\n[Active Workspace Telemetry]:{snapshot}"
+ self.messages.append({"role": "user", "content": user_msg})
+
+ max_tool_turns = 5
+ for turn_idx in range(max_tool_turns):
+ reply = await self._query_llm(self.messages)
+ parsed = parse_tool_call(reply)
+ if not parsed:
+ break
+
+ tool_name, args = parsed
+ log(f"[brain-local] tool call ({turn_idx+1}/{max_tool_turns}): {tool_name} with args {args}")
+
+ tool_result = execute_tool(tool_name, args, self)
+ log(f"[brain-local] tool output ({len(tool_result)} chars):\n{tool_result[:300]}...")
+
+ self.messages.append({"role": "assistant", "content": reply})
+
+ if turn_idx == max_tool_turns - 1:
+ self.messages.append({
+ "role": "user",
+ "content": f"[Tool Result from {tool_name}]:\n{tool_result}\n\n"
+ f"You now have all necessary data. Deliver your final spoken answer now (do NOT call any tools)."
+ })
+ else:
+ self.messages.append({
+ "role": "user",
+ "content": f"[Tool Result from {tool_name}]:\n{tool_result}\n\n"
+ f"If you need to read a specific code file, call read_file. Otherwise, deliver your spoken answer directly."
+ })
+
+ buf = ""
+ full_reply = ""
+ payload = {
+ "model": self.model,
+ "messages": self.messages,
+ "stream": True,
+ "temperature": 0.7,
+ "max_tokens": 2048,
+ }
+
+ try:
+ async with httpx.AsyncClient(timeout=120.0) as client:
+ async with client.stream("POST", f"{self.api_base}/chat/completions", json=payload) as response:
+ if response.status_code != 200:
+ err_text = await response.aread()
+ log(f"[brain-local] server error {response.status_code}: {err_text.decode('utf-8', errors='ignore')}")
+ yield "I had trouble connecting to the local inference server."
+ self._dirty = False
+ return
+
+ async for line in response.aiter_lines():
+ if self._interrupted:
+ log("[brain-local] stream interrupted")
+ break
+ if not line.startswith("data: "):
+ continue
+ data_str = line[6:].strip()
+ if data_str == "[DONE]":
+ break
+ try:
+ chunk = json.loads(data_str)
+ delta = chunk["choices"][0]["delta"].get("content", "")
+ if delta:
+ if "<|im_end|>" in delta or "<|endoftext|>" in delta:
+ delta = _SPECIAL_TOKENS.sub("", delta)
+ buf += delta
+ full_reply += delta
+ break
+ buf += delta
+ full_reply += delta
+ while True:
+ m_sent = _SENTENCE_END.search(buf)
+ if not m_sent:
+ break
+ sentence, buf = buf[:m_sent.end()].strip(), buf[m_sent.end():]
+ cleaned = _clean_text(sentence)
+ if cleaned:
+ yield cleaned
+ except Exception:
+ continue
+ except Exception as e:
+ log(f"[brain-local] error querying {self.api_base}: {e}")
+ yield "Sorry, I lost connection to the local model."
+
+ tail = _clean_text(buf)
+ if tail and not self._interrupted:
+ yield tail
+
+ if full_reply:
+ self.messages.append({"role": "assistant", "content": _clean_text(full_reply)})
+ self.session["turns"] += 1
+ self.session["out_tokens"] += len(full_reply.split()) * 2
+ self.session["in_tokens"] += len(utterance.split()) * 2
+
+ self._dirty = False
+ self._interrupted = False
+
+
+WarmBrain = LocalWarmBrain
diff --git a/backtalk/config.py b/backtalk/config.py
index d59f7f2..29ffb13 100644
--- a/backtalk/config.py
+++ b/backtalk/config.py
@@ -42,7 +42,13 @@
# Display name, used in logs and to build the quit phrases
# ("goodbye " hangs up). Match your agent's actual name.
"name": "Assistant",
- # The brain. Full model id ON PURPOSE — never a bare alias like
+ # The brain backend: "claude" (default, uses Claude Agent SDK) or
+ # "local" (connects to an OpenAI-compatible endpoint such as llama-server,
+ # Ollama, vLLM, Aphrodite, etc.).
+ "brain": "claude",
+ # Endpoint base URL for local OpenAI-compatible brain.
+ "api_base": "http://127.0.0.1:8080/v1",
+ # The brain model. When using "claude", specify full model id on purpose
# "sonnet": the SDK resolves aliases through its own bundled CLI and
# can silently land on an older model. The fast tier is most of the
# speed difference people ask about; a deep-work model makes every
@@ -131,6 +137,8 @@
# language pipeline (a=American, b=British, e/f/h/i/j/p/z = other
# languages), so keep voice and accent matched.
"voice": "bm_lewis",
+ # Device for local Kokoro TTS pipeline: "cpu" (default) or "cuda"
+ "tts_device": "cpu",
# Speech recognition (faster-whisper, local, free).
# Models: tiny.en / base.en / small.en / medium.en — small.en is the
# accuracy/speed sweet spot on a normal machine.
diff --git a/backtalk/ears.py b/backtalk/ears.py
index ad52343..6cb9b66 100644
--- a/backtalk/ears.py
+++ b/backtalk/ears.py
@@ -289,12 +289,13 @@ def warm():
verbose=None)
_model, _backend = repo, "mlx"
else:
- from faster_whisper import WhisperModel
want = CFG["stt_device"]
+ compute = CFG["stt_compute"]
log(f"[ears] loading {CFG['stt_model']} "
- f"({want}/{CFG['stt_compute']})...")
+ f"({want}/{compute})...")
+ from faster_whisper import WhisperModel
_model = WhisperModel(CFG["stt_model"], device=want,
- compute_type=CFG["stt_compute"])
+ compute_type=compute)
# PROVE the device before the greeting, not at the first
# spoken sentence. WhisperModel CONSTRUCTS perfectly well
# against a GPU it cannot actually use: "auto" picks CUDA
@@ -418,6 +419,12 @@ def record_held(is_held, max_s: float = 60.0, min_s: float = 0.25) -> str | None
frames.append(block[:, 0].copy())
if len(frames) * FRAME_MS / 1000 < min_s:
return None
+ try:
+ from backtalk.signals import set_state
+ set_state("thinking")
+ except Exception:
+ pass
+ print("[ptt] transcribing...", flush=True)
return transcribe(np.concatenate(frames))
diff --git a/backtalk/main.py b/backtalk/main.py
index a7b85d1..51f37d7 100644
--- a/backtalk/main.py
+++ b/backtalk/main.py
@@ -679,7 +679,7 @@ async def amain():
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)
+ warm_ears_fut = 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
@@ -699,14 +699,24 @@ async def _warmup():
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.")
+ if str(CFG.get("brain", "claude")).lower() in ("local", "openai", "llama", "vllm", "ollama"):
+ mouth.say("Bad news. The voice and the face are fine, but I "
+ f"couldn't reach my local brain at {CFG.get('api_base')}. "
+ "Check if your local inference server is running.")
+ else:
+ 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")
+ try:
+ await asyncio.wait_for(asyncio.shield(warm_ears_fut), 60)
+ log("[backtalk] ears warm and ready")
+ except Exception as e:
+ log(f"[ears] warmup error: {e!r}")
# 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
diff --git a/backtalk/mouth.py b/backtalk/mouth.py
index c3bc4f5..b7428f7 100644
--- a/backtalk/mouth.py
+++ b/backtalk/mouth.py
@@ -171,8 +171,9 @@ def warm():
lang = (CFG["voice"] or "bm_lewis")[0]
log(f"[mouth] loading kokoro (lang '{lang}', "
f"voice {CFG['voice']})...")
- _pipe = KPipeline(lang_code=lang)
- log("[mouth] voice ready")
+ device = CFG.get("tts_device", "cpu")
+ _pipe = KPipeline(lang_code=lang, device=device)
+ log(f"[mouth] voice ready ({device})")
return _pipe