ca86390407
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>
71 lines
2.3 KiB
Python
71 lines
2.3 KiB
Python
"""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
|