solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4"""Tests for NDJSON-only input in think.talents."""
5
6import asyncio
7import json
8import sys
9from datetime import datetime, timezone
10from io import StringIO
11from types import SimpleNamespace
12from unittest.mock import MagicMock, patch
13
14import pytest
15
16from solstone.think.models import GPT_5
17from solstone.think.providers.brain_state import (
18 begin_brain_refresh,
19 finish_brain_refresh,
20 inspect_brain_state,
21)
22
23NOW = datetime(2026, 1, 2, 3, 4, 5, tzinfo=timezone.utc)
24
25
26@pytest.fixture
27def mock_journal(tmp_path, monkeypatch):
28 """Set up a temporary journal directory."""
29 journal_path = tmp_path / "journal"
30 journal_path.mkdir()
31 agents_path = journal_path / "talents"
32 agents_path.mkdir()
33
34 monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal_path))
35 return journal_path
36
37
38async def mock_run_cogitate(config, on_event=None):
39 """Mock run_cogitate function for testing."""
40 prompt = config.get("prompt", "")
41 provider = config.get("provider", "")
42 model = config.get("model", "")
43 name = config.get("name", "chat")
44
45 if on_event:
46 on_event(
47 {
48 "event": "start",
49 "prompt": prompt,
50 "provider": provider,
51 "model": model,
52 "name": name,
53 "ts": 1234567890,
54 }
55 )
56 on_event(
57 {
58 "event": "finish",
59 "result": f"Response to: {prompt}",
60 "ts": 1234567891,
61 }
62 )
63 return f"Response to: {prompt}"
64
65
66def mock_prepare_config(request: dict) -> dict:
67 """Mock prepare_config that passes through request with minimal additions."""
68 config = dict(request)
69 # Add required fields if not present
70 if "name" not in config:
71 config["name"] = "chat"
72 if "provider" not in config:
73 config["provider"] = "google"
74 if "model" not in config:
75 config["model"] = "gpt-5-mini"
76 if "type" not in config:
77 config["type"] = "cogitate"
78 # Add empty meta for hooks
79 config["meta"] = {}
80 return config
81
82
83def _ok_component() -> dict[str, str]:
84 return {
85 "status": "ok",
86 "observed_at": NOW.isoformat(),
87 "expires_at": datetime(2026, 1, 3, 3, 4, 5, tzinfo=timezone.utc).isoformat(),
88 }
89
90
91def _write_ready_brain(journal_path):
92 config_path = journal_path / "config" / "journal.json"
93 config_path.parent.mkdir(parents=True, exist_ok=True)
94 config_path.write_text(
95 json.dumps(
96 {
97 "providers": {
98 "active": {
99 "provider": "google",
100 "model": "gemini-3.5-flash",
101 }
102 },
103 "env": {"GOOGLE_API_KEY": "test-key"},
104 }
105 ),
106 encoding="utf-8",
107 )
108 permit = begin_brain_refresh(NOW, journal_path=journal_path)
109 assert permit is not None
110 finish_brain_refresh(
111 permit,
112 {
113 "configuration": _ok_component(),
114 "lane_prerequisites": _ok_component(),
115 "generate": _ok_component(),
116 "cogitate": _ok_component(),
117 },
118 NOW,
119 journal_path=journal_path,
120 )
121
122
123def mock_all_providers(monkeypatch):
124 """Mock the registered cogitate provider module with mock_run_cogitate.
125
126 Cloud providers route through the OpenHands facade for cogitate, so one mock
127 covers openai, anthropic, and google registry lookups.
128 """
129 mock_module = MagicMock()
130 mock_module.run_cogitate = mock_run_cogitate
131 monkeypatch.setitem(sys.modules, "solstone.think.providers.openhands", mock_module)
132
133 monkeypatch.setitem(sys.modules, "agents", MagicMock())
134
135 # Mock prepare_config to avoid needing real agent configs
136 monkeypatch.setattr("solstone.think.talents.prepare_config", mock_prepare_config)
137
138
139def test_ndjson_single_request(mock_journal, monkeypatch, capsys):
140 """Test processing a single NDJSON request from stdin."""
141 ndjson_input = json.dumps(
142 {
143 "prompt": "What is 2+2?",
144 "provider": "openai",
145 "name": "chat",
146 "model": GPT_5,
147 "max_output_tokens": 100,
148 }
149 )
150
151 monkeypatch.setattr("sys.stdin", StringIO(ndjson_input))
152
153 mock_args = MagicMock()
154 mock_args.verbose = False
155 mock_args.dry_run = False
156
157 mock_all_providers(monkeypatch)
158
159 from solstone.think.talents import main_async
160
161 with patch("solstone.think.talents.setup_cli", return_value=mock_args):
162 asyncio.run(main_async())
163
164 captured = capsys.readouterr()
165 lines = captured.out.strip().split("\n")
166
167 events = [json.loads(line) for line in lines if line]
168
169 assert events
170
171 start_event = events[0]
172 assert start_event["event"] == "start"
173 # Prompt includes system instruction prepended during enrichment
174 assert "What is 2+2?" in start_event["prompt"]
175 assert start_event["provider"] == "openai"
176 assert start_event["model"] == GPT_5
177
178 finish_events = [e for e in events if e["event"] == "finish"]
179 assert finish_events
180
181
182def test_ndjson_cogitate_model_not_found_records_runtime_failure(
183 mock_journal, monkeypatch, capsys
184):
185 from litellm.exceptions import NotFoundError
186
187 _write_ready_brain(mock_journal)
188
189 async def run_cogitate(config, on_event=None):
190 del config, on_event
191 raise NotFoundError("model not found", model="m", llm_provider="gemini")
192
193 monkeypatch.setattr(
194 "solstone.think.providers.get_provider_module",
195 lambda _provider: SimpleNamespace(run_cogitate=run_cogitate),
196 )
197 monkeypatch.setattr("solstone.think.talents.prepare_config", mock_prepare_config)
198 monkeypatch.setattr("sys.stdin", StringIO(json.dumps({"prompt": "use tools"})))
199 mock_args = MagicMock()
200 mock_args.verbose = False
201 mock_args.dry_run = False
202
203 from solstone.think.talents import main_async
204
205 with patch("solstone.think.talents.setup_cli", return_value=mock_args):
206 asyncio.run(main_async())
207
208 captured = capsys.readouterr()
209 events = [json.loads(line) for line in captured.out.strip().split("\n") if line]
210 error_events = [event for event in events if event["event"] == "error"]
211 assert len(error_events) == 1
212 assert error_events[0]["reason_code"] == "model_not_found"
213 assert error_events[0]["provider"] == "google"
214
215 record = inspect_brain_state(NOW, journal_path=mock_journal)["record"]
216 assert record is not None
217 assert record["reason_code"] == "model_not_found"
218 assert record["evidence"]["cogitate"]["reason_code"] == "model_not_found"
219
220
221def test_ndjson_cogitate_generic_tool_404_stays_unknown(
222 mock_journal, monkeypatch, capsys
223):
224 class ToolLookupError(Exception):
225 status_code = 404
226
227 _write_ready_brain(mock_journal)
228
229 async def run_cogitate(config, on_event=None):
230 del config, on_event
231 raise ToolLookupError("tool endpoint missing")
232
233 monkeypatch.setattr(
234 "solstone.think.providers.get_provider_module",
235 lambda _provider: SimpleNamespace(run_cogitate=run_cogitate),
236 )
237 monkeypatch.setattr("solstone.think.talents.prepare_config", mock_prepare_config)
238 monkeypatch.setattr("sys.stdin", StringIO(json.dumps({"prompt": "use tools"})))
239 mock_args = MagicMock()
240 mock_args.verbose = False
241 mock_args.dry_run = False
242
243 from solstone.think.talents import main_async
244
245 with patch("solstone.think.talents.setup_cli", return_value=mock_args):
246 asyncio.run(main_async())
247
248 captured = capsys.readouterr()
249 events = [json.loads(line) for line in captured.out.strip().split("\n") if line]
250 error_events = [event for event in events if event["event"] == "error"]
251 assert len(error_events) == 1
252 assert error_events[0]["reason_code"] == "unknown"
253 assert error_events[0]["provider"] == "google"
254
255 record = inspect_brain_state(NOW, journal_path=mock_journal)["record"]
256 assert record is not None
257 assert record["reason_code"] is None
258 assert record["evidence"]["cogitate"].get("reason_code") is None
259
260
261def test_ndjson_multiple_requests(mock_journal, monkeypatch, capsys):
262 """Test processing multiple NDJSON requests from stdin."""
263 requests = [
264 {
265 "prompt": "First question",
266 "provider": "openai",
267 },
268 {
269 "prompt": "Second question",
270 "provider": "anthropic",
271 "model": "claude-3",
272 },
273 {
274 "prompt": "Third question",
275 "provider": "google",
276 "name": "technical",
277 },
278 ]
279
280 ndjson_input = "\n".join(json.dumps(r) for r in requests)
281
282 monkeypatch.setattr("sys.stdin", StringIO(ndjson_input))
283
284 mock_args = MagicMock()
285 mock_args.verbose = False
286 mock_args.dry_run = False
287
288 mock_all_providers(monkeypatch)
289
290 from solstone.think.talents import main_async
291
292 with patch("solstone.think.talents.setup_cli", return_value=mock_args):
293 asyncio.run(main_async())
294
295 captured = capsys.readouterr()
296 lines = [line for line in captured.out.strip().split("\n") if line]
297
298 assert len(lines) >= 6
299
300 events = [json.loads(line) for line in lines]
301 start_events = [e for e in events if e["event"] == "start"]
302
303 assert len(start_events) == 3
304 # Prompts include system instruction prepended during enrichment
305 assert "First question" in start_events[0]["prompt"]
306 assert "Second question" in start_events[1]["prompt"]
307 assert start_events[1]["provider"] == "anthropic"
308 assert "Third question" in start_events[2]["prompt"]
309 assert start_events[2]["name"] == "technical"
310
311
312def test_ndjson_invalid_json(mock_journal, monkeypatch, capsys):
313 """Test handling of invalid JSON in NDJSON input."""
314 ndjson_input = """{"prompt": "Valid request", "provider": "openai"}
315not valid json
316{"prompt": "Another valid request", "provider": "openai"}"""
317
318 monkeypatch.setattr("sys.stdin", StringIO(ndjson_input))
319
320 mock_args = MagicMock()
321 mock_args.verbose = False
322 mock_args.dry_run = False
323
324 mock_all_providers(monkeypatch)
325
326 from solstone.think.talents import main_async
327
328 with patch("solstone.think.talents.setup_cli", return_value=mock_args):
329 asyncio.run(main_async())
330
331 captured = capsys.readouterr()
332 lines = [line for line in captured.out.strip().split("\n") if line]
333
334 events = [json.loads(line) for line in lines]
335
336 error_events = [e for e in events if e["event"] == "error"]
337 assert len(error_events) == 1
338 assert "Invalid JSON" in error_events[0]["error"]
339
340 start_events = [e for e in events if e["event"] == "start"]
341 assert len(start_events) == 2
342
343
344def test_ndjson_missing_prompt(mock_journal, monkeypatch, capsys):
345 """Test handling of NDJSON request without required 'prompt' field."""
346 ndjson_input = json.dumps(
347 {
348 "provider": "openai",
349 "model": GPT_5,
350 }
351 )
352
353 monkeypatch.setattr("sys.stdin", StringIO(ndjson_input))
354
355 mock_args = MagicMock()
356 mock_args.verbose = False
357 mock_args.dry_run = False
358
359 mock_all_providers(monkeypatch)
360
361 from solstone.think.talents import main_async
362
363 with patch("solstone.think.talents.setup_cli", return_value=mock_args):
364 asyncio.run(main_async())
365
366 captured = capsys.readouterr()
367 lines = [line for line in captured.out.strip().split("\n") if line]
368
369 assert len(lines) >= 1
370 error_event = json.loads(lines[0])
371 assert error_event["event"] == "error"
372 assert "prompt" in error_event["error"].lower() # Error mentions prompt
373
374
375def test_ndjson_empty_lines(mock_journal, monkeypatch, capsys):
376 """Test that empty lines in NDJSON input are ignored."""
377 ndjson_input = """{"prompt": "First", "provider": "openai"}
378
379{"prompt": "Second", "provider": "openai"}
380
381"""
382
383 monkeypatch.setattr("sys.stdin", StringIO(ndjson_input))
384
385 mock_args = MagicMock()
386 mock_args.verbose = False
387 mock_args.dry_run = False
388
389 mock_all_providers(monkeypatch)
390
391 from solstone.think.talents import main_async
392
393 with patch("solstone.think.talents.setup_cli", return_value=mock_args):
394 asyncio.run(main_async())
395
396 captured = capsys.readouterr()
397 lines = [line for line in captured.out.strip().split("\n") if line]
398
399 events = [json.loads(line) for line in lines]
400 start_events = [e for e in events if e["event"] == "start"]
401
402 assert len(start_events) == 2