5d45d1d3bd
STREAM_BROWSER becomes the broadcast *capability* master flag; the live screen-share state (new ToolContext.broadcasting, passed per turn by the bot) decides the backend: - master off -> broadcast disabled, always Gemini - master on + live on -> on-screen Chrome (visible on the stream) - master on + live off -> Gemini context.broadcasting is None outside the voice path (evals, text entry, older bot) and falls back to the master flag, so current behaviour is unchanged. This is the brain-side foundation; bot-side wiring (bridge passes broadcast state, auto-broadcast on voice join, voice on/off toggle) follows. Specs + docs/llm_contexts.md updated. Covered by tests/test_web_search_broadcast_routing.py. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
76 lines
2.6 KiB
Python
76 lines
2.6 KiB
Python
"""Behaviour tests for real-time search routing by live broadcast state.
|
|
|
|
Routing contract (docs/stream_browser_modes.md):
|
|
- master flag cfg.stream_browser (env STREAM_BROWSER) = broadcast capability
|
|
- context.broadcasting = live screen-share state for the turn (set by the bot)
|
|
|
|
master live(ctx) -> backend
|
|
------- --------- -------
|
|
True True on-screen Chrome (browser_search)
|
|
True False Gemini
|
|
True None Chrome (no signal -> fall back to master)
|
|
False True/None Gemini (broadcast disabled, never Chrome)
|
|
"""
|
|
|
|
from types import SimpleNamespace
|
|
|
|
import jarvis.tools.builtin.realtime_search as rs
|
|
from jarvis.tools.base import ToolContext
|
|
from jarvis.tools.builtin.web_search import WebSearchTool
|
|
|
|
|
|
def _ctx(stream_browser, broadcasting):
|
|
cfg = SimpleNamespace(
|
|
web_search_enabled=True,
|
|
stream_browser=stream_browser,
|
|
gemini_auth="oauth",
|
|
gemini_api_key="",
|
|
gemini_model="gemini-2.0-flash",
|
|
)
|
|
return ToolContext(
|
|
db=None,
|
|
cfg=cfg,
|
|
system_prompt="",
|
|
original_prompt="",
|
|
redacted_text="",
|
|
max_retries=0,
|
|
user_print=lambda *a, **k: None,
|
|
language=None,
|
|
broadcasting=broadcasting,
|
|
)
|
|
|
|
|
|
def _route(monkeypatch, stream_browser, broadcasting):
|
|
"""Run webSearch and return which backend handled it ('browser'/'gemini')."""
|
|
monkeypatch.setattr(rs, "browser_search", lambda *a, **k: "BROWSER_RESULT")
|
|
monkeypatch.setattr(rs, "gemini_cli_search", lambda *a, **k: "GEMINI_RESULT")
|
|
monkeypatch.setattr(rs, "gemini_search", lambda *a, **k: "GEMINI_RESULT")
|
|
res = WebSearchTool().run({"search_query": "현재 서울 날씨"}, _ctx(stream_browser, broadcasting))
|
|
if res.reply_text == "BROWSER_RESULT":
|
|
return "browser"
|
|
if res.reply_text == "GEMINI_RESULT":
|
|
return "gemini"
|
|
return "fallthrough"
|
|
|
|
|
|
def test_master_on_live_on_uses_browser(monkeypatch):
|
|
assert _route(monkeypatch, True, True) == "browser"
|
|
|
|
|
|
def test_master_on_live_off_uses_gemini(monkeypatch):
|
|
assert _route(monkeypatch, True, False) == "gemini"
|
|
|
|
|
|
def test_master_on_no_signal_defaults_to_browser(monkeypatch):
|
|
# None = no live signal (evals / text entry / older bot) -> behave as master
|
|
assert _route(monkeypatch, True, None) == "browser"
|
|
|
|
|
|
def test_master_off_never_uses_browser_even_if_live(monkeypatch):
|
|
# STREAM_BROWSER=false disables broadcast entirely: always Gemini.
|
|
assert _route(monkeypatch, False, True) == "gemini"
|
|
|
|
|
|
def test_master_off_no_signal_uses_gemini(monkeypatch):
|
|
assert _route(monkeypatch, False, None) == "gemini"
|