solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4"""Tests for Convey Callosum SSE route and bridge fan-out."""
5
6from __future__ import annotations
7
8import json
9import time
10from collections.abc import Callable, Iterator
11
12import pytest
13
14import solstone.convey.bridge as convey_bridge
15from solstone.think.spl.health import LINK_HEALTH_EVENT
16
17
18@pytest.fixture(autouse=True)
19def clear_sse_subscribers() -> Iterator[None]:
20 with convey_bridge._SSE_LOCK:
21 convey_bridge._SSE_SUBSCRIBERS_BY_KEY.clear()
22 convey_bridge._SSE_LAST_CHAT_REQUEST_AT_BY_KEY.clear()
23 convey_bridge._STATE_CACHE["link_health"] = None
24 yield
25 with convey_bridge._SSE_LOCK:
26 convey_bridge._SSE_SUBSCRIBERS_BY_KEY.clear()
27 convey_bridge._SSE_LAST_CHAT_REQUEST_AT_BY_KEY.clear()
28 convey_bridge._STATE_CACHE["link_health"] = None
29
30
31def _next_chunk(response) -> str:
32 chunk = next(iter(response.response))
33 if isinstance(chunk, bytes):
34 return chunk.decode("utf-8")
35 return str(chunk)
36
37
38def _parse_sse_data(chunk: str) -> dict:
39 for line in chunk.splitlines():
40 if line.startswith("data: "):
41 return json.loads(line[len("data: ") :])
42 raise AssertionError(f"No data line found in chunk: {chunk!r}")
43
44
45def _next_data(response) -> dict:
46 for chunk in response.response:
47 text = chunk.decode("utf-8") if isinstance(chunk, bytes) else str(chunk)
48 if "data: " in text:
49 return _parse_sse_data(text)
50 raise AssertionError("SSE stream ended before a data frame was received")
51
52
53def _wait_until(predicate: Callable[[], bool], timeout: float = 1.0) -> None:
54 deadline = time.monotonic() + timeout
55 while time.monotonic() < deadline:
56 if predicate():
57 return
58 time.sleep(0.01)
59 raise AssertionError("timed out waiting for condition")
60
61
62def test_callosum_sse_success_headers(convey_env):
63 env = convey_env()
64
65 resp = env.client.get("/sse/events", buffered=False)
66 try:
67 assert resp.status_code == 200
68 assert resp.content_type.startswith("text/event-stream")
69 assert resp.headers["Cache-Control"] == "no-cache"
70 assert resp.headers["X-Accel-Buffering"] == "no"
71 finally:
72 resp.close()
73
74
75def test_callosum_sse_setup_complete_serves_with_proxy_header(convey_env):
76 env = convey_env()
77
78 resp = env.client.get(
79 "/sse/events",
80 headers={"X-Forwarded-For": "1.2.3.4"},
81 buffered=False,
82 )
83 try:
84 assert resp.status_code == 200
85 assert resp.content_type.startswith("text/event-stream")
86 finally:
87 resp.close()
88
89
90def test_callosum_sse_round_trip_payload(convey_env):
91 env = convey_env()
92 resp = env.client.get("/sse/events", buffered=False)
93 try:
94 assert resp.status_code == 200
95 assert _next_chunk(resp) == ": heartbeat\n\n"
96 assert convey_bridge.subscription_count("convey-ui") == 1
97 message = {"tract": "test", "event": "ping", "ts": 0, "extra": "value"}
98 convey_bridge._broadcast_callosum_event(message)
99
100 parsed = _next_data(resp)
101 assert parsed == message
102 finally:
103 resp.close()
104
105
106def _link_health_message(
107 *,
108 generation: int,
109 state: str = "connected",
110 ts: int = 1_700_000_000_000,
111) -> dict:
112 return {
113 "tract": "link",
114 "event": LINK_HEALTH_EVENT,
115 "state": state,
116 "listen_generation": generation,
117 "last_successful_relay_tunnel_at": 1_700_000_000_100,
118 "last_relay_tunnel_error": None,
119 "last_relay_tunnel_error_at": None,
120 "relay_tunnel_error_status": None,
121 "ts": ts,
122 }
123
124
125def test_bridge_caches_structured_link_health() -> None:
126 convey_bridge._broadcast_callosum_event(_link_health_message(generation=7))
127
128 assert convey_bridge.get_cached_state()["link_health"] == {
129 "state": "connected",
130 "listen_generation": 7,
131 "last_successful_relay_tunnel_at": 1_700_000_000_100,
132 "last_relay_tunnel_error": None,
133 "last_relay_tunnel_error_at": None,
134 "relay_tunnel_error_status": None,
135 "ts": 1_700_000_000_000,
136 }
137
138
139def test_bridge_ignores_coarse_link_events_for_health_cache() -> None:
140 for event in (
141 "connecting",
142 "connected",
143 "disconnect",
144 "tunnel_pair",
145 "tunnel_close",
146 ):
147 convey_bridge._broadcast_callosum_event({"tract": "link", "event": event})
148 assert convey_bridge.get_cached_state()["link_health"] is None
149
150
151def test_bridge_drops_older_link_health_generation() -> None:
152 convey_bridge._broadcast_callosum_event(
153 _link_health_message(generation=3, state="connected")
154 )
155 convey_bridge._broadcast_callosum_event(
156 _link_health_message(generation=2, state="reconnecting")
157 )
158
159 health = convey_bridge.get_cached_state()["link_health"]
160 assert health["listen_generation"] == 3
161 assert health["state"] == "connected"
162
163
164def test_bridge_overwrites_equal_or_newer_link_health_generation() -> None:
165 convey_bridge._broadcast_callosum_event(
166 _link_health_message(generation=3, state="connected")
167 )
168 convey_bridge._broadcast_callosum_event(
169 _link_health_message(generation=3, state="connecting")
170 )
171 assert convey_bridge.get_cached_state()["link_health"]["state"] == "connecting"
172
173 convey_bridge._broadcast_callosum_event(
174 _link_health_message(generation=4, state="reconnecting")
175 )
176 health = convey_bridge.get_cached_state()["link_health"]
177 assert health["listen_generation"] == 4
178 assert health["state"] == "reconnecting"
179
180
181def test_callosum_sse_multi_client_fanout(convey_env):
182 env = convey_env()
183 first_client = env.app.test_client()
184 second_client = env.app.test_client()
185 first = first_client.get("/sse/events", buffered=False)
186 second = second_client.get("/sse/events", buffered=False)
187 try:
188 assert _next_chunk(first) == ": heartbeat\n\n"
189 assert _next_chunk(second) == ": heartbeat\n\n"
190 assert convey_bridge.subscription_count("convey-ui") == 2
191
192 message = {"tract": "test", "event": "fanout", "ts": 1}
193 convey_bridge._broadcast_callosum_event(message)
194
195 assert _next_data(first) == message
196 assert _next_data(second) == message
197 finally:
198 second.close()
199 first.close()
200
201
202# test_slow_sse_subscriber_is_dropped_without_blocking_healthy_subscriber was
203# dropped here 2026-07-24: duplicated verbatim by
204# solstone/apps/observer/tests/test_callosum_sse.py, which CI already collects,
205# and it carried this module's only wall-clock assertion (elapsed < 0.5).
206
207
208def test_callosum_sse_dropped_subscriber_breaks_stream(convey_env):
209 env = convey_env()
210 resp = env.client.get("/sse/events", buffered=False)
211 extra = None
212 try:
213 assert _next_chunk(resp) == ": heartbeat\n\n"
214 extra = convey_bridge.register_sse_subscriber("convey-ui")
215 with convey_bridge._SSE_LOCK:
216 subscribers = set(convey_bridge._SSE_SUBSCRIBERS_BY_KEY["convey-ui"])
217 route_handles = [handle for handle in subscribers if handle is not extra]
218 assert len(route_handles) == 1
219 route_handle = route_handles[0]
220
221 for i in range(convey_bridge._SSE_QUEUE_MAXSIZE + 1):
222 convey_bridge._broadcast_callosum_event(
223 {"tract": "test", "event": "overflow", "ts": i}
224 )
225
226 assert route_handle.dropped.is_set()
227 assert route_handle not in convey_bridge._SSE_SUBSCRIBERS_BY_KEY.get(
228 "convey-ui", set()
229 )
230 with pytest.raises(StopIteration):
231 next(iter(resp.response))
232 finally:
233 if extra is not None:
234 convey_bridge.unregister_sse_subscriber(extra)
235 resp.close()
236
237
238def test_callosum_sse_clean_disconnect_unregisters_subscriber(convey_env):
239 env = convey_env()
240 resp = env.client.get("/sse/events", buffered=False)
241
242 assert _next_chunk(resp) == ": heartbeat\n\n"
243 assert convey_bridge.subscription_count("convey-ui") == 1
244
245 resp.close()
246
247 _wait_until(lambda: convey_bridge.subscription_count("convey-ui") == 0)
248
249
250def test_callosum_sse_heartbeat(convey_env, monkeypatch):
251 env = convey_env()
252 monkeypatch.setattr(convey_bridge, "_SSE_HEARTBEAT_SECONDS", 0.01)
253
254 resp = env.client.get("/sse/events", buffered=False)
255 try:
256 assert resp.status_code == 200
257 assert _next_chunk(resp) == ": heartbeat\n\n"
258 assert _next_chunk(resp) == ": heartbeat\n\n"
259 finally:
260 resp.close()