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
9
10from solstone.talent import pulse
11from solstone.think.day_accumulator import read_latest
12
13
14@pytest.fixture
15def journal(tmp_path, monkeypatch):
16 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
17 return tmp_path
18
19
20def _select_local_provider(journal_path):
21 config_dir = journal_path / "config"
22 config_dir.mkdir(parents=True, exist_ok=True)
23 (config_dir / "journal.json").write_text(
24 json.dumps(
25 {
26 "providers": {
27 "active": {"provider": "local", "model": "local/qwen3.5-4b"},
28 }
29 }
30 ),
31 encoding="utf-8",
32 )
33
34
35def _pulse_payload(**overrides):
36 payload = {
37 "title": "Focused morning",
38 "one_sentence": "The morning is centered on a launch review.",
39 "full_details": "The owner has a coherent launch-review block in motion.",
40 "needs_you": ["Review the launch checklist."],
41 }
42 payload.update(overrides)
43 return payload
44
45
46def test_post_process_persists_valid_pulse_record(journal):
47 config = {
48 "day": "20260611",
49 "model": "test-model",
50 "_pulse_window_note": {
51 "segments": 1,
52 "activities": 0,
53 "input_segments": 1,
54 "input_activities": 0,
55 "since_ms": 123,
56 "gaps": [],
57 },
58 }
59
60 returned = pulse.post_process(json.dumps(_pulse_payload()), config)
61
62 summary = json.loads(returned)
63 assert summary == _pulse_payload()
64 record = read_latest("20260611", "pulse", lookback_days=0)
65 assert record is not None
66 for key, value in _pulse_payload().items():
67 assert record[key] == value
68 assert record["model"] == "test-model"
69 assert record["generated_at"].endswith("Z")
70 assert isinstance(record["ts"], int)
71 assert record["window"] == config["_pulse_window_note"]
72
73
74@pytest.mark.parametrize("result", ["not json", json.dumps({"title": "Incomplete"})])
75def test_post_process_falls_back_for_malformed_model_output(journal, result):
76 default = _pulse_payload(
77 title="Fallback",
78 one_sentence="Fallback sentence.",
79 full_details="Fallback details.",
80 needs_you=["Use the deterministic fallback."],
81 )
82 config = {"day": "20260611", "model": "test-model", "_pulse_default": default}
83
84 returned = pulse.post_process(result, config)
85
86 summary = json.loads(returned)
87 assert summary == default
88 record = read_latest("20260611", "pulse", lookback_days=0)
89 assert record is not None
90 for key, value in default.items():
91 assert record[key] == value
92 assert record["model"] == "test-model"
93 assert "generated_at" in record
94 assert "ts" in record
95 assert "window" in record
96
97
98def test_normalize_pulse_coerces_and_clamps_fields():
99 default = _pulse_payload(
100 title="Fallback",
101 one_sentence="Fallback sentence.",
102 full_details="Fallback details.",
103 needs_you=["Fallback need."],
104 )
105 raw = {
106 "title": "T" * 100,
107 "one_sentence": "S" * 260,
108 "full_details": "D" * 1900,
109 "needs_you": [
110 "one",
111 42,
112 None,
113 "",
114 "x" * 300,
115 "five",
116 "six",
117 "seven",
118 "eight",
119 "nine",
120 ],
121 "ignored": "dropped",
122 }
123
124 summary = pulse._normalize_pulse(raw, default)
125
126 assert set(summary) == {"title", "one_sentence", "full_details", "needs_you"}
127 assert summary["title"] == "T" * pulse._TITLE_MAX
128 assert summary["one_sentence"] == "S" * pulse._SENTENCE_MAX
129 assert summary["full_details"] == "D" * pulse._DETAILS_MAX
130 assert summary["needs_you"] == [
131 "one",
132 "42",
133 "x" * pulse._NEED_MAX,
134 "five",
135 "six",
136 "seven",
137 "eight",
138 ]
139
140 assert (
141 pulse._normalize_pulse(
142 {
143 "title": "",
144 "one_sentence": " ",
145 "full_details": "",
146 "needs_you": ["ignored"],
147 },
148 default,
149 )
150 == default
151 )
152
153
154def test_pre_process_includes_segment_timeline_and_missing_gap(journal, monkeypatch):
155 day = "20260611"
156 segment = "101500_300"
157 seg_dir = journal / "chronicle" / day / "desktop" / segment
158 seg_dir.mkdir(parents=True)
159 (seg_dir / "timeline.json").write_text(
160 json.dumps(
161 {
162 "title": "Launch review",
163 "description": "The segment focused on launch readiness.",
164 }
165 ),
166 encoding="utf-8",
167 )
168 (journal / "identity").mkdir()
169 (journal / "identity" / "partner.md").write_text(
170 "Partner context", encoding="utf-8"
171 )
172
173 monkeypatch.setattr(pulse, "get_current", lambda: {"attention": "clear"})
174 monkeypatch.setattr(pulse, "get_imports", lambda: {"pending": []})
175 monkeypatch.setattr(pulse, "get_facets", lambda: [])
176 monkeypatch.setattr(pulse, "load_recent_entity_names", lambda limit=12: ["Alice"])
177
178 result = pulse.pre_process(
179 {
180 "day": day,
181 "cadence_window": {
182 "since_ms": 1000,
183 "segments": [
184 {"stream": "desktop", "segment": segment, "ts": 2000},
185 {"stream": None, "segment": "missing_300", "ts": 1000},
186 ],
187 "activities": [],
188 },
189 }
190 )
191
192 assert result is not None
193 template_vars = result["template_vars"]
194 assert "Launch review" in template_vars["completed_since"]
195 assert (
196 "The segment focused on launch readiness." in template_vars["completed_since"]
197 )
198 assert "no timeline.json found for segment missing_300" in template_vars["gaps"]
199 assert "Partner context" == template_vars["partner_profile"]
200 assert pulse._compact_json(["Alice"]) == template_vars["recent_entities"]
201
202
203def test_pre_process_total_failure_returns_skip_reason(monkeypatch):
204 monkeypatch.setattr(
205 pulse,
206 "_completed_since",
207 lambda day, config, gaps: (_ for _ in ()).throw(RuntimeError("boom")),
208 )
209
210 result = pulse.pre_process({"day": "20260611"})
211
212 assert result is not None
213 assert result["skip_reason"] == "pulse pre-hook failed: boom"
214
215
216def test_accumulate_suppresses_single_file_output_path(journal):
217 from solstone.think.talent import get_talent, get_talent_configs
218 from solstone.think.talents import prepare_config
219 from solstone.think.thinking import _apply_output_persistence
220
221 raw_config = get_talent_configs()["pulse"]
222 pulse_config = get_talent("pulse")
223 assert raw_config["output"] == "json"
224 assert raw_config["schema"] == "pulse.schema.json"
225 assert pulse_config["output"] == "json"
226 assert pulse_config["json_schema"]["required"] == [
227 "title",
228 "one_sentence",
229 "full_details",
230 "needs_you",
231 ]
232 assert pulse_config["accumulate"] is True
233
234 request_config = {}
235 _apply_output_persistence(request_config, pulse_config, force_refresh=False)
236
237 assert "output" not in request_config
238 assert "refresh" not in request_config
239
240 _select_local_provider(journal)
241 prepared = prepare_config({"name": "pulse", "day": "20260611"})
242 assert "output_path" not in prepared