solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4from __future__ import annotations
5
6import asyncio
7import json
8from concurrent.futures import Future
9
10from solstone.think.voice import sideband
11
12
13class _FakeEvent:
14 def __init__(
15 self, event_type: str, *, name: str = "", arguments: str = "", call_id: str = ""
16 ):
17 self.type = event_type
18 self.name = name
19 self.arguments = arguments
20 self.call_id = call_id
21
22
23class _FakeConversationItem:
24 def __init__(self) -> None:
25 self.items: list[dict] = []
26
27 async def create(self, *, item):
28 self.items.append(item)
29
30
31class _FakeResponse:
32 def __init__(self) -> None:
33 self.count = 0
34
35 async def create(self):
36 self.count += 1
37
38
39class _FakeConversation:
40 def __init__(self) -> None:
41 self.item = _FakeConversationItem()
42
43
44class _FakeConn:
45 def __init__(self, events):
46 self._events = iter(events)
47 self.conversation = _FakeConversation()
48 self.response = _FakeResponse()
49
50 def __aiter__(self):
51 return self
52
53 async def __anext__(self):
54 try:
55 return next(self._events)
56 except StopIteration:
57 raise StopAsyncIteration
58
59
60def test_sideband_loop_dispatches_function_calls(monkeypatch):
61 conn = _FakeConn(
62 [
63 _FakeEvent("session.created"),
64 _FakeEvent(
65 "response.function_call_arguments.done",
66 name="journal.get_day",
67 arguments='{"day":"2026-03-04"}',
68 call_id="call-1",
69 ),
70 ]
71 )
72 seen: list[tuple[str, str, str]] = []
73
74 async def fake_dispatch(name, arguments, call_id, app):
75 seen.append((name, arguments, call_id))
76 return json.dumps({"day": "2026-03-04"})
77
78 monkeypatch.setattr(sideband, "dispatch_tool_call", fake_dispatch)
79
80 asyncio.run(sideband._sideband_loop(conn, "call-1", object()))
81
82 assert seen == [("journal.get_day", '{"day":"2026-03-04"}', "call-1")]
83 assert conn.conversation.item.items == [
84 {
85 "type": "function_call_output",
86 "call_id": "call-1",
87 "output": '{"day": "2026-03-04"}',
88 }
89 ]
90 assert conn.response.count == 1
91
92
93def test_sideband_loop_ignores_non_tool_events(monkeypatch):
94 conn = _FakeConn([_FakeEvent("session.created")])
95 called = False
96
97 async def fake_dispatch(name, arguments, call_id, app):
98 nonlocal called
99 called = True
100 return "{}"
101
102 monkeypatch.setattr(sideband, "dispatch_tool_call", fake_dispatch)
103
104 asyncio.run(sideband._sideband_loop(conn, "call-1", object()))
105
106 assert called is False
107
108
109def test_register_voice_task_tracks_and_prunes():
110 class DummyApp:
111 def __init__(self):
112 self.voice_tasks = set()
113
114 app = DummyApp()
115 future: Future[None] = Future()
116
117 sideband.register_voice_task(app, future)
118 assert future in app.voice_tasks
119
120 future.set_result(None)
121 assert future not in app.voice_tasks