feat: couple broadcast to voice + voice-controlled broadcast toggle

Completes the STREAM_BROWSER=true behaviour:
- handleJoin auto-starts the broadcast on voice join and wires the session to
  the guild streamer; each turn reports the live state to the brain so search
  routes Chrome (live) vs Gemini (off).
- New setBroadcast tool lets the user toggle the broadcast by voice ("방송
  켜줘/꺼줘") via the LLM (no hardcoded phrases); it refuses when
  STREAM_BROWSER=false. The directive flows brain -> bridge (broadcast_action)
  -> bot streamer.start/stop, guarded by isActive() so it's idempotent.
- Per-turn IPC uses a thread-local (reply/turn_state.py) instead of threading
  params through the whole engine chain: bridge sets broadcasting in, tool
  records the directive out; Tool.execute exposes broadcasting on ToolContext.

Bot typecheck clean; brain covered by tests/test_set_broadcast.py (+ existing
routing tests). Specs + docs updated.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
javis-bot
2026-06-11 10:08:30 +09:00
parent 35e754d6ee
commit ca86390407
11 changed files with 316 additions and 14 deletions

View File

@@ -4,11 +4,15 @@
*/ */
import { config } from "./config.ts"; import { config } from "./config.ts";
export type BroadcastAction = "start" | "stop";
export interface ConverseResult { export interface ConverseResult {
transcript: string; transcript: string;
language?: string | null; language?: string | null;
reply: string; reply: string;
error?: string | null; error?: string | null;
/** Broadcast directive the brain requested this turn (setBroadcast tool). */
broadcast_action?: BroadcastAction | null;
/** base64-encoded 16-bit PCM WAV of the spoken reply, or null if TTS off */ /** base64-encoded 16-bit PCM WAV of the spoken reply, or null if TTS off */
audio_b64?: string | null; audio_b64?: string | null;
} }
@@ -16,12 +20,16 @@ export interface ConverseResult {
export interface TextResult { export interface TextResult {
reply: string; reply: string;
error?: string | null; error?: string | null;
broadcast_action?: BroadcastAction | null;
audio_b64?: string | null; audio_b64?: string | null;
} }
/** Full voice turn: WAV in -> {transcript, reply, reply audio}. */ /** Full voice turn: WAV in -> {transcript, reply, reply audio}. ``broadcasting``
export async function converse(wav: Buffer): Promise<ConverseResult> { * is the current live screen-share state so the brain routes search (Chrome
const res = await fetch(`${config.bridgeUrl}/converse`, { * while live, Gemini when off). */
export async function converse(wav: Buffer, broadcasting?: boolean): Promise<ConverseResult> {
const qs = broadcasting === undefined ? "" : `?broadcasting=${broadcasting ? "1" : "0"}`;
const res = await fetch(`${config.bridgeUrl}/converse${qs}`, {
method: "POST", method: "POST",
headers: { "content-type": "audio/wav" }, headers: { "content-type": "audio/wav" },
body: wav, body: wav,

View File

@@ -81,6 +81,30 @@ async function handleJoin(i: ChatInputCommandInteraction) {
const session = await joinChannel(channel); const session = await joinChannel(channel);
session.onTurn = ({ transcript, reply }) => session.onTurn = ({ transcript, reply }) =>
console.log(`🗣️ ${transcript}\n🤖 ${reply}`); console.log(`🗣️ ${transcript}\n🤖 ${reply}`);
// Screen-share mode (STREAM_BROWSER=true): the broadcast is coupled to the
// voice session. Auto-start it on join, report its live state to the brain
// each turn (search routes Chrome while live / Gemini when off), and let the
// brain toggle it via voice ("방송 켜줘 / 꺼줘"). In voice-only mode
// (STREAM_BROWSER=false) none of this runs and the broadcast stays off.
if (config.screenBrowser) {
const streamer = await getStreamer(i.guildId!);
const ctx: StreamContext = { guildId: i.guildId!, voiceChannelId: channel.id };
session.getBroadcasting = () => streamer.isActive();
session.onBroadcastAction = async (action) => {
if (action === "start") {
if (!streamer.isActive()) await streamer.start(ctx);
} else if (streamer.isActive()) {
await streamer.stop();
}
};
try {
if (!streamer.isActive()) await streamer.start(ctx);
} catch (e) {
console.error("[join] auto-broadcast failed:", e);
}
}
await i.editReply(`🎙️ '${channel.name}' 채널에 접속했습니다. 말씀하세요.`); await i.editReply(`🎙️ '${channel.name}' 채널에 접속했습니다. 말씀하세요.`);
} }

View File

@@ -71,6 +71,11 @@ export class VoiceSession {
private playQueue: Buffer[] = []; private playQueue: Buffer[] = [];
/** Optional callback to surface transcripts/replies to a text channel. */ /** Optional callback to surface transcripts/replies to a text channel. */
onTurn?: (info: { user: string; transcript: string; reply: string }) => void; onTurn?: (info: { user: string; transcript: string; reply: string }) => void;
/** Live screen-share state, sent with each turn so the brain routes search
* (Chrome while broadcasting, Gemini when off). */
getBroadcasting?: () => boolean;
/** Apply a broadcast directive the brain requested (start/stop the stream). */
onBroadcastAction?: (action: "start" | "stop") => void | Promise<void>;
constructor(channel: VoiceBasedChannel) { constructor(channel: VoiceBasedChannel) {
this.guildId = channel.guild.id; this.guildId = channel.guild.id;
@@ -123,10 +128,19 @@ export class VoiceSession {
const wav = pcm16MonoToWav(mono, DISCORD_RATE); const wav = pcm16MonoToWav(mono, DISCORD_RATE);
try { try {
const result = await converse(wav); const result = await converse(wav, this.getBroadcasting?.());
if (result.transcript) { if (result.transcript) {
this.onTurn?.({ user: userId, transcript: result.transcript, reply: result.reply }); this.onTurn?.({ user: userId, transcript: result.transcript, reply: result.reply });
} }
// Apply any broadcast directive the brain requested (e.g. user said
// "방송 켜줘 / 꺼줘") before playing the reply.
if (result.broadcast_action && this.onBroadcastAction) {
try {
await this.onBroadcastAction(result.broadcast_action);
} catch (e) {
console.error("[voice] broadcast action failed:", e);
}
}
const audio = decodeWav(result.audio_b64); const audio = decodeWav(result.audio_b64);
if (audio) this.play(audio); if (audio) this.play(audio);
} catch (err) { } catch (err) {

View File

@@ -165,8 +165,14 @@ def transcribe(wav_bytes: bytes) -> dict:
return {"text": text, "language": getattr(info, "language", None)} return {"text": text, "language": getattr(info, "language", None)}
def think(text: str, language: Optional[str] = None) -> dict: def think(text: str, language: Optional[str] = None, broadcasting: Optional[bool] = None) -> dict:
"""Run the Jarvis reply engine on a piece of text.""" """Run the Jarvis reply engine on a piece of text.
``broadcasting`` is the bot's live screen-share state for this turn; it is
stashed in request-scoped state so the search routing can pick Chrome vs
Gemini. If the reply engine calls the setBroadcast tool, the recorded
directive is returned as ``broadcast_action`` for the bot to act on.
"""
if not BRAIN_ENABLED: if not BRAIN_ENABLED:
return {"reply": text, "error": "brain disabled (JARVIS_BRAIN_ENABLED=0)"} return {"reply": text, "error": "brain disabled (JARVIS_BRAIN_ENABLED=0)"}
_ensure_brain() _ensure_brain()
@@ -174,6 +180,10 @@ def think(text: str, language: Optional[str] = None) -> dict:
return {"reply": "", "error": _brain_error or "brain unavailable"} return {"reply": "", "error": _brain_error or "brain unavailable"}
try: try:
from jarvis.reply.engine import run_reply_engine from jarvis.reply.engine import run_reply_engine
from jarvis.reply import turn_state
turn_state.reset()
turn_state.set_broadcasting(broadcasting)
# tts=None: we do our own Discord-side synthesis, the engine must not # tts=None: we do our own Discord-side synthesis, the engine must not
# try to speak to a local speaker that doesn't exist in this process. # try to speak to a local speaker that doesn't exist in this process.
@@ -183,11 +193,20 @@ def think(text: str, language: Optional[str] = None) -> dict:
reply = (reply or "").strip() reply = (reply or "").strip()
if reply: if reply:
_dialogue_memory.add_interaction(text, reply) _dialogue_memory.add_interaction(text, reply)
return {"reply": reply} return {"reply": reply, "broadcast_action": turn_state.get_broadcast_action()}
except Exception as e: # pragma: no cover except Exception as e: # pragma: no cover
return {"reply": "", "error": f"{type(e).__name__}: {e}"} return {"reply": "", "error": f"{type(e).__name__}: {e}"}
def _coerce_bool(value) -> Optional[bool]:
"""Parse a broadcasting flag from JSON (bool) or a query string."""
if value is None:
return None
if isinstance(value, bool):
return value
return str(value).strip().lower() in ("1", "true", "yes", "on")
def synthesize(text: str) -> Optional[bytes]: def synthesize(text: str) -> Optional[bytes]:
"""Synthesize text to a 16-bit PCM WAV using Piper. Returns None if TTS off.""" """Synthesize text to a 16-bit PCM WAV using Piper. Returns None if TTS off."""
if not TTS_ENABLED or not text.strip(): if not TTS_ENABLED or not text.strip():
@@ -234,7 +253,7 @@ def http_text():
text = (data.get("text") or "").strip() text = (data.get("text") or "").strip()
if not text: if not text:
return jsonify({"error": "missing 'text'"}), 400 return jsonify({"error": "missing 'text'"}), 400
result = think(text, data.get("language")) result = think(text, data.get("language"), _coerce_bool(data.get("broadcasting")))
audio = synthesize(result.get("reply", "")) audio = synthesize(result.get("reply", ""))
if audio: if audio:
result["audio_b64"] = base64.b64encode(audio).decode("ascii") result["audio_b64"] = base64.b64encode(audio).decode("ascii")
@@ -263,7 +282,8 @@ def http_converse():
transcript = stt.get("text", "") transcript = stt.get("text", "")
if not transcript: if not transcript:
return jsonify({"transcript": "", "reply": "", "audio_b64": None}) return jsonify({"transcript": "", "reply": "", "audio_b64": None})
result = think(transcript, stt.get("language")) broadcasting = _coerce_bool(request.args.get("broadcasting"))
result = think(transcript, stt.get("language"), broadcasting)
audio = synthesize(result.get("reply", "")) audio = synthesize(result.get("reply", ""))
return jsonify( return jsonify(
{ {
@@ -271,6 +291,7 @@ def http_converse():
"language": stt.get("language"), "language": stt.get("language"),
"reply": result.get("reply", ""), "reply": result.get("reply", ""),
"error": result.get("error"), "error": result.get("error"),
"broadcast_action": result.get("broadcast_action"),
"audio_b64": base64.b64encode(audio).decode("ascii") if audio else None, "audio_b64": base64.b64encode(audio).decode("ascii") if audio else None,
} }
) )

View File

@@ -35,11 +35,12 @@ entry) and falls back to the master flag so behaviour is unchanged.
| `browseAndSearch` tool (true) | `src/jarvis/tools/builtin/browse_and_search.py` -> subprocess the Node core | TODO | | `browseAndSearch` tool (true) | `src/jarvis/tools/builtin/browse_and_search.py` -> subprocess the Node core | TODO |
| Gemini search (false) | `realtime_search.py` `gemini_cli_search()` (oauth, CLI) / `gemini_search()` (apikey, REST) | done | | Gemini search (false) | `realtime_search.py` `gemini_cli_search()` (oauth, CLI) / `gemini_search()` (apikey, REST) | done |
| Search routing by live broadcast | `web_search.py` uses `ToolContext.broadcasting` + master flag | done | | Search routing by live broadcast | `web_search.py` uses `ToolContext.broadcasting` + master flag | done |
| Live broadcast state in tool ctx | `tools/base.py` `ToolContext.broadcasting` | done | | Live broadcast state in tool ctx | `tools/base.py` `ToolContext.broadcasting` from `turn_state` | done |
| Bridge passes broadcast state | bot `bridge.ts` request + `bridge/server.py` -> ToolContext | TODO | | Per-turn state channel | `reply/turn_state.py` (thread-local: broadcasting in, directive out) | done |
| Auto-broadcast on voice join (true mode) | `bot/src/index.ts` `handleJoin()` starts streamer | TODO | | Bridge passes broadcast state | `bridge.ts` `?broadcasting=` + `server.py` `think(broadcasting=)` | done |
| Voice broadcast on/off toggle | brain tool -> reply directive -> bot `streamer.start/stop` | TODO | | Auto-broadcast on voice join (true mode) | `bot/src/index.ts` `handleJoin()` starts streamer + wires hooks | done |
| `/stream` + start refused (false mode) | `bot/src/index.ts` `handleStream()` already gated | done | | Voice broadcast on/off toggle | `setBroadcast` tool -> `broadcast_action` -> bot `streamer.start/stop` | done |
| `/stream` + start refused (false mode) | `handleStream()` gated + `setBroadcast` refuses | done |
| Specs + `docs/llm_contexts.md` | alongside each tool | done | | Specs + `docs/llm_contexts.md` | alongside each tool | done |
## Design decisions ## Design decisions

View File

@@ -0,0 +1,48 @@
"""Per-turn, request-scoped state shared between the bridge, the reply engine
and tools — without threading extra parameters through the whole engine call
chain (run_reply_engine -> run_tool_with_retries -> execute_tool -> execute).
Flask serves each HTTP request on its own thread and the reply engine + tools
run synchronously on that thread, so a ``threading.local`` cleanly scopes a
single turn. Anything that can't see the thread-local (a worker thread, an eval,
a unit test) just reads ``None`` and the caller falls back to its default.
Fields
------
broadcasting : Optional[bool]
The live screen-share (Go-Live) state the bot sends with the request. The
webSearch routing reads it to pick on-screen Chrome (live) vs Gemini (off).
broadcast_action : Optional[str]
A directive ("start"/"stop") recorded by the setBroadcast tool for the
bridge to return to the bot, which owns the actual broadcast.
"""
from __future__ import annotations
import threading
from typing import Optional
_state = threading.local()
def reset() -> None:
"""Clear per-turn state. Call at the start of each request."""
_state.broadcasting = None
_state.broadcast_action = None
def set_broadcasting(value: Optional[bool]) -> None:
_state.broadcasting = value
def get_broadcasting() -> Optional[bool]:
return getattr(_state, "broadcasting", None)
def request_broadcast(action: str) -> None:
"""Record a broadcast directive. ``action`` is "start" or "stop"."""
_state.broadcast_action = action
def get_broadcast_action() -> Optional[str]:
return getattr(_state, "broadcast_action", None)

View File

@@ -111,6 +111,14 @@ class Tool(ABC):
This method creates the context and calls the tool's run method. This method creates the context and calls the tool's run method.
Tools should implement run(), not this method. Tools should implement run(), not this method.
""" """
# Live broadcast state for this turn is carried in request-scoped
# thread-local state (set by the bridge) rather than threaded through
# the whole engine call chain. Absent (eval / unit test) -> None.
try:
from jarvis.reply.turn_state import get_broadcasting
broadcasting = get_broadcasting()
except Exception:
broadcasting = None
context = ToolContext( context = ToolContext(
db=db, db=db,
cfg=cfg, cfg=cfg,
@@ -120,5 +128,6 @@ class Tool(ABC):
max_retries=max_retries, max_retries=max_retries,
user_print=user_print, user_print=user_print,
language=language, language=language,
broadcasting=broadcasting,
) )
return self.run(tool_args, context) return self.run(tool_args, context)

View File

@@ -0,0 +1,75 @@
"""Start or stop the live screen-share broadcast (Go-Live) on request.
The brain cannot drive the Discord broadcast itself — the bot owns it. This
tool records a directive ("start"/"stop") in the per-turn state; the bridge
returns it to the bot, which performs the actual start/stop. Only meaningful in
screen-share mode (``STREAM_BROWSER`` true); when broadcasting is disabled the
tool refuses so the user is told it cannot be turned on.
"""
from __future__ import annotations
from typing import Dict, Any, Optional
from ..base import Tool, ToolContext
from ..types import ToolExecutionResult
from ...debug import debug_log
class SetBroadcastTool(Tool):
"""Turn the live screen-share broadcast on or off."""
@property
def name(self) -> str:
return "setBroadcast"
@property
def description(self) -> str:
return (
"Start or stop the live screen-share broadcast (Go-Live / showing your "
"screen on the call). Use when the user asks to turn the broadcast or "
"screen on or off (show/hide your screen, go live, stop streaming). "
"Only available in screen-share mode."
)
@property
def inputSchema(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"action": {
"type": "string",
"enum": ["start", "stop"],
"description": "'start' to begin the broadcast, 'stop' to end it.",
}
},
"required": ["action"],
}
def run(self, args: Optional[Dict[str, Any]], context: ToolContext) -> ToolExecutionResult:
cfg = context.cfg
if not getattr(cfg, "stream_browser", True):
# Broadcast capability is disabled (STREAM_BROWSER=false): refuse so
# the user is clearly told it can't be turned on in this mode.
return ToolExecutionResult(
success=False,
reply_text="방송(화면 공유) 기능이 꺼져 있어 켤 수 없습니다 (STREAM_BROWSER=false).",
)
action = ""
if args and isinstance(args, dict):
action = str(args.get("action", "")).strip().lower()
if action not in ("start", "stop"):
return ToolExecutionResult(
success=False,
reply_text="Specify action='start' or action='stop'.",
)
# Record the directive for the bridge to return to the bot.
from ...reply.turn_state import request_broadcast
request_broadcast(action)
debug_log(f" 📡 setBroadcast requested: {action}", "stream")
return ToolExecutionResult(
success=True,
reply_text="방송을 시작합니다." if action == "start" else "방송을 종료합니다.",
)

View File

@@ -0,0 +1,30 @@
# setBroadcast tool
Lets the user turn the live screen-share broadcast (Go-Live) on or off by voice
or text, e.g. "방송 켜줘 / 화면 보여줘" or "방송 꺼줘". The LLM picks the intent
and calls the tool (no hardcoded phrase matching), so it works in any language.
## Contract
- Input: `{ "action": "start" | "stop" }`.
- The brain cannot drive the Discord broadcast itself (the bot owns it). The
tool records the directive in request-scoped state
(`jarvis/reply/turn_state.request_broadcast`); `bridge/server.py` returns it as
`broadcast_action` in the `/converse` (and `/text`) response, and the bot
(`voice.ts` -> `index.ts` `onBroadcastAction`) calls `streamer.start/stop`.
- Mode gate: when `cfg.stream_browser` is false (`STREAM_BROWSER=false`)
broadcasting is disabled — the tool refuses with a spoken explanation and
records no directive. This is the same capability flag that makes `/stream`
refuse and forces real-time search to Gemini.
## Principles
- Tool returns raw intent/result; it performs no broadcast itself and makes no
LLM call.
- Idempotent at the bot edge: `start` is a no-op if already live, `stop` a no-op
if already off (`streamer.isActive()` guard).
- Fail-safe: a missing/invalid action records nothing; the bot only acts on an
explicit `"start"`/`"stop"` directive.
See `docs/stream_browser_modes.md` for how this fits the broadcast-coupled
search routing.

View File

@@ -21,6 +21,7 @@ from .builtin.weather import WeatherTool
from .builtin.stop import StopTool from .builtin.stop import StopTool
from .builtin.tool_search import ToolSearchTool from .builtin.tool_search import ToolSearchTool
from .builtin.browse_and_play import BrowseAndPlayTool from .builtin.browse_and_play import BrowseAndPlayTool
from .builtin.set_broadcast import SetBroadcastTool
from .types import ToolExecutionResult from .types import ToolExecutionResult
from ..config import Settings from ..config import Settings
from .external.mcp_client import MCPClient from .external.mcp_client import MCPClient
@@ -41,6 +42,7 @@ BUILTIN_TOOLS = {
"stop": StopTool(), "stop": StopTool(),
"toolSearchTool": ToolSearchTool(), "toolSearchTool": ToolSearchTool(),
"browseAndPlay": BrowseAndPlayTool(), "browseAndPlay": BrowseAndPlayTool(),
"setBroadcast": SetBroadcastTool(),
} }
# Global MCP tools cache # Global MCP tools cache

View File

@@ -0,0 +1,70 @@
"""Behaviour tests for the setBroadcast tool + per-turn directive plumbing."""
from types import SimpleNamespace
from jarvis.reply import turn_state
from jarvis.tools.base import ToolContext
from jarvis.tools.builtin.set_broadcast import SetBroadcastTool
def _ctx(stream_browser):
cfg = SimpleNamespace(stream_browser=stream_browser, voice_debug=True)
return ToolContext(
db=None, cfg=cfg, system_prompt="", original_prompt="",
redacted_text="", max_retries=0, user_print=lambda *a, **k: None,
)
def setup_function():
turn_state.reset()
def test_start_records_directive_in_true_mode():
res = SetBroadcastTool().run({"action": "start"}, _ctx(True))
assert res.success is True
assert turn_state.get_broadcast_action() == "start"
def test_stop_records_directive_in_true_mode():
res = SetBroadcastTool().run({"action": "stop"}, _ctx(True))
assert res.success is True
assert turn_state.get_broadcast_action() == "stop"
def test_false_mode_refuses_and_records_nothing():
res = SetBroadcastTool().run({"action": "start"}, _ctx(False))
assert res.success is False
assert turn_state.get_broadcast_action() is None
def test_invalid_action_records_nothing():
res = SetBroadcastTool().run({"action": "sideways"}, _ctx(True))
assert res.success is False
assert turn_state.get_broadcast_action() is None
def test_reset_clears_directive():
SetBroadcastTool().run({"action": "start"}, _ctx(True))
assert turn_state.get_broadcast_action() == "start"
turn_state.reset()
assert turn_state.get_broadcast_action() is None
def test_execute_threads_broadcasting_from_turn_state_into_context():
# Tool.execute should read the per-turn broadcasting flag and expose it on
# the ToolContext so webSearch routing can see the live broadcast state.
captured = {}
class _Probe(SetBroadcastTool):
def run(self, args, context):
captured["broadcasting"] = context.broadcasting
return super().run(args, context)
turn_state.reset()
turn_state.set_broadcasting(True)
_Probe().execute(
db=None, cfg=SimpleNamespace(stream_browser=True, voice_debug=True),
tool_args={"action": "stop"}, system_prompt="", original_prompt="",
redacted_text="", max_retries=0, user_print=lambda *a, **k: None,
)
assert captured["broadcasting"] is True