diff --git a/bot/src/bridge.ts b/bot/src/bridge.ts index 44c518e..b6721de 100644 --- a/bot/src/bridge.ts +++ b/bot/src/bridge.ts @@ -4,11 +4,15 @@ */ import { config } from "./config.ts"; +export type BroadcastAction = "start" | "stop"; + export interface ConverseResult { transcript: string; language?: string | null; reply: string; 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 */ audio_b64?: string | null; } @@ -16,12 +20,16 @@ export interface ConverseResult { export interface TextResult { reply: string; error?: string | null; + broadcast_action?: BroadcastAction | null; audio_b64?: string | null; } -/** Full voice turn: WAV in -> {transcript, reply, reply audio}. */ -export async function converse(wav: Buffer): Promise { - const res = await fetch(`${config.bridgeUrl}/converse`, { +/** Full voice turn: WAV in -> {transcript, reply, reply audio}. ``broadcasting`` + * is the current live screen-share state so the brain routes search (Chrome + * while live, Gemini when off). */ +export async function converse(wav: Buffer, broadcasting?: boolean): Promise { + const qs = broadcasting === undefined ? "" : `?broadcasting=${broadcasting ? "1" : "0"}`; + const res = await fetch(`${config.bridgeUrl}/converse${qs}`, { method: "POST", headers: { "content-type": "audio/wav" }, body: wav, diff --git a/bot/src/index.ts b/bot/src/index.ts index 8daf65c..6367170 100644 --- a/bot/src/index.ts +++ b/bot/src/index.ts @@ -81,6 +81,30 @@ async function handleJoin(i: ChatInputCommandInteraction) { const session = await joinChannel(channel); session.onTurn = ({ transcript, 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}' 채널에 접속했습니다. 말씀하세요.`); } diff --git a/bot/src/voice.ts b/bot/src/voice.ts index e97a7bc..35c8028 100644 --- a/bot/src/voice.ts +++ b/bot/src/voice.ts @@ -71,6 +71,11 @@ export class VoiceSession { private playQueue: Buffer[] = []; /** Optional callback to surface transcripts/replies to a text channel. */ 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; constructor(channel: VoiceBasedChannel) { this.guildId = channel.guild.id; @@ -123,10 +128,19 @@ export class VoiceSession { const wav = pcm16MonoToWav(mono, DISCORD_RATE); try { - const result = await converse(wav); + const result = await converse(wav, this.getBroadcasting?.()); if (result.transcript) { 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); if (audio) this.play(audio); } catch (err) { diff --git a/bridge/server.py b/bridge/server.py index 3418c62..f017788 100644 --- a/bridge/server.py +++ b/bridge/server.py @@ -165,8 +165,14 @@ def transcribe(wav_bytes: bytes) -> dict: return {"text": text, "language": getattr(info, "language", None)} -def think(text: str, language: Optional[str] = None) -> dict: - """Run the Jarvis reply engine on a piece of text.""" +def think(text: str, language: Optional[str] = None, broadcasting: Optional[bool] = None) -> dict: + """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: return {"reply": text, "error": "brain disabled (JARVIS_BRAIN_ENABLED=0)"} _ensure_brain() @@ -174,6 +180,10 @@ def think(text: str, language: Optional[str] = None) -> dict: return {"reply": "", "error": _brain_error or "brain unavailable"} try: 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 # 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() if 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 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]: """Synthesize text to a 16-bit PCM WAV using Piper. Returns None if TTS off.""" if not TTS_ENABLED or not text.strip(): @@ -234,7 +253,7 @@ def http_text(): text = (data.get("text") or "").strip() if not text: 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", "")) if audio: result["audio_b64"] = base64.b64encode(audio).decode("ascii") @@ -263,7 +282,8 @@ def http_converse(): transcript = stt.get("text", "") if not transcript: 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", "")) return jsonify( { @@ -271,6 +291,7 @@ def http_converse(): "language": stt.get("language"), "reply": result.get("reply", ""), "error": result.get("error"), + "broadcast_action": result.get("broadcast_action"), "audio_b64": base64.b64encode(audio).decode("ascii") if audio else None, } ) diff --git a/docs/stream_browser_modes.md b/docs/stream_browser_modes.md index edbfa9d..c811de0 100644 --- a/docs/stream_browser_modes.md +++ b/docs/stream_browser_modes.md @@ -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 | | 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 | -| Live broadcast state in tool ctx | `tools/base.py` `ToolContext.broadcasting` | done | -| Bridge passes broadcast state | bot `bridge.ts` request + `bridge/server.py` -> ToolContext | TODO | -| Auto-broadcast on voice join (true mode) | `bot/src/index.ts` `handleJoin()` starts streamer | TODO | -| Voice broadcast on/off toggle | brain tool -> reply directive -> bot `streamer.start/stop` | TODO | -| `/stream` + start refused (false mode) | `bot/src/index.ts` `handleStream()` already gated | done | +| Live broadcast state in tool ctx | `tools/base.py` `ToolContext.broadcasting` from `turn_state` | done | +| Per-turn state channel | `reply/turn_state.py` (thread-local: broadcasting in, directive out) | done | +| Bridge passes broadcast state | `bridge.ts` `?broadcasting=` + `server.py` `think(broadcasting=)` | done | +| Auto-broadcast on voice join (true mode) | `bot/src/index.ts` `handleJoin()` starts streamer + wires hooks | 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 | ## Design decisions diff --git a/src/jarvis/reply/turn_state.py b/src/jarvis/reply/turn_state.py new file mode 100644 index 0000000..7af4ace --- /dev/null +++ b/src/jarvis/reply/turn_state.py @@ -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) diff --git a/src/jarvis/tools/base.py b/src/jarvis/tools/base.py index 3e1f584..ab9456d 100644 --- a/src/jarvis/tools/base.py +++ b/src/jarvis/tools/base.py @@ -111,6 +111,14 @@ class Tool(ABC): This method creates the context and calls the tool's run 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( db=db, cfg=cfg, @@ -120,5 +128,6 @@ class Tool(ABC): max_retries=max_retries, user_print=user_print, language=language, + broadcasting=broadcasting, ) return self.run(tool_args, context) diff --git a/src/jarvis/tools/builtin/set_broadcast.py b/src/jarvis/tools/builtin/set_broadcast.py new file mode 100644 index 0000000..ea66677 --- /dev/null +++ b/src/jarvis/tools/builtin/set_broadcast.py @@ -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 "방송을 종료합니다.", + ) diff --git a/src/jarvis/tools/builtin/set_broadcast.spec.md b/src/jarvis/tools/builtin/set_broadcast.spec.md new file mode 100644 index 0000000..b6cf4af --- /dev/null +++ b/src/jarvis/tools/builtin/set_broadcast.spec.md @@ -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. diff --git a/src/jarvis/tools/registry.py b/src/jarvis/tools/registry.py index dfe9da9..2616348 100644 --- a/src/jarvis/tools/registry.py +++ b/src/jarvis/tools/registry.py @@ -21,6 +21,7 @@ from .builtin.weather import WeatherTool from .builtin.stop import StopTool from .builtin.tool_search import ToolSearchTool from .builtin.browse_and_play import BrowseAndPlayTool +from .builtin.set_broadcast import SetBroadcastTool from .types import ToolExecutionResult from ..config import Settings from .external.mcp_client import MCPClient @@ -41,6 +42,7 @@ BUILTIN_TOOLS = { "stop": StopTool(), "toolSearchTool": ToolSearchTool(), "browseAndPlay": BrowseAndPlayTool(), + "setBroadcast": SetBroadcastTool(), } # Global MCP tools cache diff --git a/tests/test_set_broadcast.py b/tests/test_set_broadcast.py new file mode 100644 index 0000000..b0ad2da --- /dev/null +++ b/tests/test_set_broadcast.py @@ -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