solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4import asyncio
5import logging
6from pathlib import Path
7from types import SimpleNamespace
8
9import pytest
10
11from solstone.think import talent_provenance, talents
12
13
14def _set_journal(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
15 journal = tmp_path / "journal"
16 journal.mkdir()
17 monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal.resolve()))
18 return journal
19
20
21def _minimal_config(output_path: Path, *, day: str = "20260101") -> dict:
22 return {
23 "name": "daily",
24 "type": "cogitate",
25 "provider": "google",
26 "model": "gemini-test",
27 "output": "md",
28 "output_path": str(output_path),
29 "day": day,
30 "schedule": "daily",
31 "prompt": "prompt",
32 "user_instruction": "",
33 "system_instruction": "",
34 "sources": {},
35 "health_stale": False,
36 }
37
38
39def test_weekly_reflection_provenance_path(tmp_path, monkeypatch):
40 journal = _set_journal(tmp_path, monkeypatch)
41 output_path = journal / "reflections" / "weekly" / "20260622.md"
42
43 assert talent_provenance._day_and_logical_output(output_path) == (
44 "20260622",
45 Path("reflections", "weekly", "20260622.md"),
46 )
47 assert talent_provenance.provenance_path_for_output(output_path) == (
48 journal
49 / "chronicle"
50 / "20260622"
51 / "health"
52 / "talent-provenance"
53 / "reflections"
54 / "weekly"
55 / "20260622.md.json"
56 )
57
58
59def test_facet_news_provenance_path(tmp_path, monkeypatch):
60 journal = _set_journal(tmp_path, monkeypatch)
61 output_path = journal / "facets" / "work" / "news" / "20260622.md"
62
63 assert talent_provenance._day_and_logical_output(output_path) == (
64 "20260622",
65 Path("facets", "work", "news", "20260622.md"),
66 )
67 assert talent_provenance.provenance_path_for_output(output_path) == (
68 journal
69 / "chronicle"
70 / "20260622"
71 / "health"
72 / "talent-provenance"
73 / "facets"
74 / "work"
75 / "news"
76 / "20260622.md.json"
77 )
78
79
80@pytest.mark.parametrize(
81 "relative",
82 [
83 Path("reflections", "weekly", "draft.md"),
84 Path("facets", "work", "news", "draft.md"),
85 ],
86)
87def test_stem_day_requires_date_without_day_dir_side_effect(
88 tmp_path,
89 monkeypatch,
90 relative,
91):
92 journal = _set_journal(tmp_path, monkeypatch)
93
94 with pytest.raises(talent_provenance.UnsupportedProvenancePath):
95 talent_provenance.provenance_path_for_output(journal / relative)
96
97 assert not (journal / "chronicle" / "draft").exists()
98
99
100def test_existing_chronicle_mapping_unchanged(tmp_path, monkeypatch):
101 journal = _set_journal(tmp_path, monkeypatch)
102 output_path = journal / "chronicle" / "20260101" / "talents" / "flow.md"
103
104 assert talent_provenance._day_and_logical_output(output_path) == (
105 "20260101",
106 Path("talents", "flow.md"),
107 )
108
109
110def test_existing_activity_mapping_unchanged(tmp_path, monkeypatch):
111 journal = _set_journal(tmp_path, monkeypatch)
112 output_path = (
113 journal / "facets" / "work" / "activities" / "20260101" / "abc" / "summary.md"
114 )
115
116 assert talent_provenance._day_and_logical_output(output_path) == (
117 "20260101",
118 Path("facets", "work", "activities", "abc", "summary.md"),
119 )
120
121
122def test_write_clean_provenance_skips_unmapped_path_without_error(
123 tmp_path,
124 monkeypatch,
125 caplog,
126):
127 journal = _set_journal(tmp_path, monkeypatch)
128 output_path = journal / "apps" / "chat" / "talents" / "support.md"
129 output_path.parent.mkdir(parents=True)
130 output_path.write_text("support", encoding="utf-8")
131 caplog.set_level(logging.WARNING, logger="solstone.think.talents")
132
133 talents._write_clean_provenance(
134 _minimal_config(output_path),
135 output_path,
136 "support",
137 None,
138 123,
139 )
140
141 assert any(record.levelno == logging.WARNING for record in caplog.records)
142 assert not any(record.levelno >= logging.ERROR for record in caplog.records)
143
144
145def test_write_clean_provenance_logs_unexpected_write_failure(
146 tmp_path,
147 monkeypatch,
148 caplog,
149):
150 journal = _set_journal(tmp_path, monkeypatch)
151 output_path = journal / "chronicle" / "20260101" / "talents" / "daily.md"
152 output_path.parent.mkdir(parents=True)
153 output_path.write_text("daily", encoding="utf-8")
154
155 def raise_os_error(*args, **kwargs):
156 raise OSError("disk full")
157
158 monkeypatch.setattr(talents, "write_provenance", raise_os_error)
159 caplog.set_level(logging.WARNING, logger="solstone.think.talents")
160
161 talents._write_clean_provenance(
162 _minimal_config(output_path),
163 output_path,
164 "daily",
165 None,
166 123,
167 )
168
169 assert any(record.levelno == logging.ERROR for record in caplog.records)
170 assert not any(record.levelno == logging.WARNING for record in caplog.records)
171
172
173def test_execute_with_tools_weekly_reflection_finish_writes_sidecar(
174 tmp_path,
175 monkeypatch,
176):
177 journal = _set_journal(tmp_path, monkeypatch)
178 output_path = journal / "reflections" / "weekly" / "20260622.md"
179 result = "Full reflection text."
180 events: list[dict] = []
181
182 async def run_cogitate(config, on_event):
183 on_event({"event": "finish", "result": result})
184 return ""
185
186 monkeypatch.setattr(
187 "solstone.think.providers.get_provider_module",
188 lambda _provider: SimpleNamespace(run_cogitate=run_cogitate),
189 )
190
191 config = _minimal_config(output_path, day="20260622")
192 asyncio.run(talents._execute_with_tools(config, events.append))
193
194 assert any(event.get("event") == "finish" for event in events)
195 assert not any(event.get("event") == "error" for event in events)
196 assert output_path.read_text(encoding="utf-8") == result
197 assert (
198 journal
199 / "chronicle"
200 / "20260622"
201 / "health"
202 / "talent-provenance"
203 / "reflections"
204 / "weekly"
205 / "20260622.md.json"
206 ).exists()
207
208
209def test_execute_with_tools_unmapped_output_still_finishes(
210 tmp_path,
211 monkeypatch,
212):
213 journal = _set_journal(tmp_path, monkeypatch)
214 output_path = journal / "apps" / "chat" / "talents" / "support.md"
215 result = "Support response."
216 events: list[dict] = []
217
218 async def run_cogitate(config, on_event):
219 on_event({"event": "finish", "result": result})
220 return ""
221
222 monkeypatch.setattr(
223 "solstone.think.providers.get_provider_module",
224 lambda _provider: SimpleNamespace(run_cogitate=run_cogitate),
225 )
226
227 config = _minimal_config(output_path, day="20260622")
228 asyncio.run(talents._execute_with_tools(config, events.append))
229
230 assert any(event.get("event") == "finish" for event in events)
231 assert not any(event.get("event") == "error" for event in events)
232 assert output_path.read_text(encoding="utf-8") == result
233 assert not any((journal / "chronicle").glob("*/health/talent-provenance/**/*.json"))