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 json
7
8import pytest
9from flask import Flask
10
11from solstone.apps.chat.copy import (
12 CHAT_DEFERRED_NOT_ANALYZED,
13 CHAT_THINKING_ENGINE_NOT_CHOSEN,
14)
15from solstone.convey.chat import chat_bp, compose_honest_degradation
16from solstone.convey.chat_stream import read_chat_events
17from solstone.think.pipeline_health import SegmentBacklog, SegmentCompletion
18from solstone.think.processing import (
19 format_awaiting_analysis,
20 parse_processing_settings,
21)
22
23
24def _setup_journal(tmp_path, monkeypatch):
25 journal = tmp_path / "journal"
26 journal.mkdir()
27 monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))
28 return journal
29
30
31def _reset_chat_state(chat_module) -> None:
32 chat_module.stop_all_chat_runtime()
33 with chat_module._state_lock:
34 chat_module._current_chat_use_id = None
35 chat_module._current_chat_state = None
36 chat_module._queued_triggers.clear()
37 chat_module._active_talents.clear()
38 chat_module._reserved_use_ids.clear()
39 chat_module._thinking_buffers.clear()
40 chat_module._thinking_providers.clear()
41 for timer in chat_module._watchdog_timers.values():
42 timer.cancel()
43 chat_module._watchdog_timers.clear()
44 chat_module._last_use_id = 0
45
46
47@pytest.fixture(autouse=True)
48def _default_thinking_engine_selected(monkeypatch):
49 monkeypatch.setattr(
50 "solstone.convey.chat._no_thinking_engine_chosen", lambda: False
51 )
52
53
54def _set_current_chat(chat_module, logical_use_id: str, raw_use_id: str | None) -> None:
55 with chat_module._state_lock:
56 chat_module._current_chat_use_id = logical_use_id
57 chat_module._current_chat_state = {
58 "raw_use_id": raw_use_id,
59 "raw_use_ids_seen": {raw_use_id} if raw_use_id else set(),
60 "trigger": {"type": "owner_message", "message": "help"},
61 "location": {"app": "sol", "path": "/app/sol", "facet": "work"},
62 "retry_count": 0,
63 }
64
65
66@pytest.fixture
67def chat_client(tmp_path, monkeypatch):
68 import solstone.convey.chat as chat
69
70 _setup_journal(tmp_path, monkeypatch)
71 _reset_chat_state(chat)
72
73 app = Flask(__name__)
74 app.config["TESTING"] = True
75 app.register_blueprint(chat_bp)
76 return app.test_client()
77
78
79def _settings(mode: str):
80 return parse_processing_settings({"mode": mode}, strict=False)
81
82
83def _backlog(
84 day: str,
85 *,
86 not_sensed: int,
87 not_thought: int,
88 errors: tuple[str, ...] = (),
89) -> SegmentBacklog:
90 total = not_sensed + not_thought
91 completion = SegmentCompletion(
92 blockers=[],
93 not_sensed=not_sensed,
94 not_thought=not_thought,
95 total=total,
96 capped=0,
97 exhausted=(),
98 )
99 return SegmentBacklog(
100 days=(day,),
101 not_thought=not_thought,
102 not_sensed=not_sensed,
103 total=total,
104 per_day={day: completion},
105 errors=errors,
106 )
107
108
109def _empty_backlog(day: str) -> SegmentBacklog:
110 return SegmentBacklog(
111 days=(day,),
112 not_thought=0,
113 not_sensed=0,
114 total=0,
115 per_day={},
116 errors=(),
117 )
118
119
120def _patch_backlog_reads(
121 monkeypatch,
122 *,
123 mode: str,
124 backlog: SegmentBacklog,
125) -> None:
126 monkeypatch.setattr(
127 "solstone.convey.chat.load_processing_settings",
128 lambda: _settings(mode),
129 )
130 monkeypatch.setattr(
131 "solstone.convey.chat.read_segment_backlog",
132 lambda: backlog,
133 )
134
135
136def _patch_finish_side_effects(monkeypatch):
137 emitted: dict[str, list] = {"finish": [], "error": [], "cortex": []}
138 monkeypatch.setattr(
139 "solstone.convey.chat._emit_cortex_event",
140 lambda *args, **kwargs: emitted["cortex"].append((args, kwargs)),
141 )
142 monkeypatch.setattr(
143 "solstone.convey.chat._emit_finish",
144 lambda *args: emitted["finish"].append(args),
145 )
146 monkeypatch.setattr(
147 "solstone.convey.chat._emit_error",
148 lambda *args, **kwargs: emitted["error"].append((args, kwargs)),
149 )
150 monkeypatch.setattr("solstone.convey.chat._run_next_action", lambda _action: None)
151 return emitted
152
153
154def _finish_chat(chat_module, message_text: str) -> None:
155 _set_current_chat(chat_module, "logical-chat", "raw-chat")
156 chat_module._on_cortex_finish(
157 {
158 "use_id": "raw-chat",
159 "result": json.dumps(
160 {
161 "message": message_text,
162 "notes": "ok",
163 "talent_request": None,
164 }
165 ),
166 }
167 )
168
169
170def _events_of_kind(day: str, kind: str) -> list[dict]:
171 return [event for event in read_chat_events(day) if event["kind"] == kind]
172
173
174def test_thinking_engine_not_chosen_copy_is_pinned_byte_for_byte():
175 assert (
176 CHAT_THINKING_ENGINE_NOT_CHOSEN
177 == "no thinking engine is chosen yet. choose one in thinking so i can "
178 "answer from your journal."
179 )
180
181
182def test_compose_honest_degradation_fires_for_deferred_today_pending():
183 import solstone.convey.chat as chat
184
185 today = chat._today_day()
186
187 result = compose_honest_degradation(
188 _settings("deferred"),
189 _backlog(today, not_sensed=2, not_thought=1),
190 )
191
192 assert result is not None
193 assert CHAT_DEFERRED_NOT_ANALYZED in result
194 assert format_awaiting_analysis(3) in result
195
196
197def test_compose_honest_degradation_ignores_untracked_queried_day():
198 import solstone.convey.chat as chat
199
200 today = chat._today_day()
201
202 result = compose_honest_degradation(
203 _settings("deferred"),
204 _backlog(today, not_sensed=2, not_thought=1),
205 queried_day="20260101",
206 )
207
208 assert result is None
209
210
211def test_compose_honest_degradation_ignores_empty_backlog():
212 import solstone.convey.chat as chat
213
214 today = chat._today_day()
215
216 result = compose_honest_degradation(
217 _settings("deferred"),
218 _backlog(today, not_sensed=0, not_thought=0),
219 )
220
221 assert result is None
222
223
224def test_compose_honest_degradation_ignores_realtime_mode():
225 import solstone.convey.chat as chat
226
227 today = chat._today_day()
228
229 result = compose_honest_degradation(
230 _settings("realtime"),
231 _backlog(today, not_sensed=2, not_thought=1),
232 )
233
234 assert result is None
235
236
237def test_compose_honest_degradation_ignores_indeterminate_backlog():
238 import solstone.convey.chat as chat
239
240 today = chat._today_day()
241
242 result = compose_honest_degradation(
243 _settings("deferred"),
244 _backlog(today, not_sensed=2, not_thought=1, errors=(today,)),
245 )
246
247 assert result is None
248
249
250def test_compose_honest_degradation_fires_for_no_engine_realtime(monkeypatch):
251 import solstone.convey.chat as chat
252
253 monkeypatch.setattr(chat, "_no_thinking_engine_chosen", lambda: True)
254 today = chat._today_day()
255
256 result = compose_honest_degradation(
257 _settings("realtime"),
258 _backlog(today, not_sensed=2, not_thought=1),
259 )
260
261 assert result is not None
262 assert CHAT_THINKING_ENGINE_NOT_CHOSEN in result
263 assert format_awaiting_analysis(3) in result
264
265
266def test_empty_chat_finish_substitutes_honest_degradation(chat_client, monkeypatch):
267 import solstone.convey.chat as chat
268
269 today = chat._today_day()
270 _patch_backlog_reads(
271 monkeypatch,
272 mode="deferred",
273 backlog=_backlog(today, not_sensed=2, not_thought=1),
274 )
275 emitted = _patch_finish_side_effects(monkeypatch)
276
277 _finish_chat(chat, "")
278
279 sol_message = _events_of_kind(today, "sol_message")[0]
280 assert CHAT_DEFERRED_NOT_ANALYZED in sol_message["text"]
281 assert format_awaiting_analysis(3) in sol_message["text"]
282 assert [
283 event
284 for event in read_chat_events(today)
285 if event["kind"] == "chat_error"
286 and event["reason"] == "provider_response_invalid"
287 ] == []
288 assert emitted["finish"] == [("logical-chat", sol_message["text"])]
289 assert emitted["error"] == []
290
291
292def test_empty_chat_finish_remains_invalid_in_realtime(chat_client, monkeypatch):
293 import solstone.convey.chat as chat
294
295 today = chat._today_day()
296 _patch_backlog_reads(
297 monkeypatch,
298 mode="realtime",
299 backlog=_backlog(today, not_sensed=2, not_thought=1),
300 )
301 emitted = _patch_finish_side_effects(monkeypatch)
302
303 _finish_chat(chat, "")
304
305 sol_message = _events_of_kind(today, "sol_message")[0]
306 assert sol_message["text"] == ""
307 chat_errors = _events_of_kind(today, "chat_error")
308 assert chat_errors[0]["reason"] == "provider_response_invalid"
309 assert emitted["finish"] == []
310 assert emitted["error"] == [(("logical-chat", "provider_response_invalid"), {})]
311
312
313def test_empty_chat_finish_remains_invalid_with_empty_backlog(chat_client, monkeypatch):
314 import solstone.convey.chat as chat
315
316 today = chat._today_day()
317 _patch_backlog_reads(
318 monkeypatch,
319 mode="deferred",
320 backlog=_empty_backlog(today),
321 )
322 emitted = _patch_finish_side_effects(monkeypatch)
323
324 _finish_chat(chat, "")
325
326 sol_message = _events_of_kind(today, "sol_message")[0]
327 assert sol_message["text"] == ""
328 assert _events_of_kind(today, "chat_error")[0]["reason"] == (
329 "provider_response_invalid"
330 )
331 assert emitted["finish"] == []
332 assert emitted["error"] == [(("logical-chat", "provider_response_invalid"), {})]
333
334
335def test_empty_chat_finish_remains_invalid_with_backlog_errors(
336 chat_client, monkeypatch
337):
338 import solstone.convey.chat as chat
339
340 today = chat._today_day()
341 _patch_backlog_reads(
342 monkeypatch,
343 mode="deferred",
344 backlog=_backlog(today, not_sensed=2, not_thought=1, errors=(today,)),
345 )
346 emitted = _patch_finish_side_effects(monkeypatch)
347
348 _finish_chat(chat, "")
349
350 sol_message = _events_of_kind(today, "sol_message")[0]
351 assert sol_message["text"] == ""
352 assert _events_of_kind(today, "chat_error")[0]["reason"] == (
353 "provider_response_invalid"
354 )
355 assert emitted["finish"] == []
356 assert emitted["error"] == [(("logical-chat", "provider_response_invalid"), {})]
357
358
359def test_non_empty_chat_finish_is_not_replaced(chat_client, monkeypatch):
360 import solstone.convey.chat as chat
361
362 today = chat._today_day()
363 _patch_backlog_reads(
364 monkeypatch,
365 mode="deferred",
366 backlog=_backlog(today, not_sensed=2, not_thought=1),
367 )
368 emitted = _patch_finish_side_effects(monkeypatch)
369
370 _finish_chat(chat, "Here is your answer.")
371
372 sol_message = _events_of_kind(today, "sol_message")[0]
373 assert sol_message["text"] == "Here is your answer."
374 assert emitted["finish"] == [("logical-chat", "Here is your answer.")]
375 assert emitted["error"] == []