solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4"""Tests for the Sense output splitter."""
5
6import json
7from pathlib import Path
8
9
10def _make_sense_output(**overrides):
11 """Build a complete Sense output dict with sensible defaults."""
12 base = {
13 "density": "active",
14 "content_type": "coding",
15 "activity_summary": "Writing unit tests for the API module.",
16 "entities": [
17 {
18 "type": "Project",
19 "name": "SolAPI",
20 "role": "mentioned",
21 "source": "screen",
22 "context": "main project",
23 },
24 ],
25 "facets": [
26 {"facet": "work", "activity": "coding", "level": "high"},
27 ],
28 "meeting_detected": False,
29 "speakers": [],
30 "recommend": {
31 "screen_record": False,
32 "speaker_attribution": False,
33 },
34 "emotional_register": "neutral",
35 }
36 base.update(overrides)
37 return base
38
39
40class TestWriteSenseOutputs:
41 def test_writes_all_standard_files(self, tmp_path):
42 from solstone.think.sense_splitter import write_sense_outputs
43
44 seg_dir = Path(tmp_path) / "20260304" / "default" / "090000_300"
45 sense_json = _make_sense_output()
46
47 write_sense_outputs(sense_json, seg_dir)
48
49 agents_dir = seg_dir / "talents"
50 assert (agents_dir / "activity.md").exists()
51 assert (agents_dir / "facets.json").exists()
52 assert (agents_dir / "density.json").exists()
53 assert (agents_dir / "sense.json").exists()
54 assert not (agents_dir / "speakers.json").exists()
55
56 assert (agents_dir / "activity.md").read_text(encoding="utf-8") == (
57 "Writing unit tests for the API module."
58 )
59 assert json.loads((agents_dir / "facets.json").read_text(encoding="utf-8")) == [
60 {"facet": "work", "activity": "coding", "level": "high"}
61 ]
62
63 density = json.loads((agents_dir / "density.json").read_text(encoding="utf-8"))
64 assert set(density.keys()) == {
65 "classification",
66 "timestamp",
67 }
68 assert density["classification"] == "active"
69
70 assert json.loads((agents_dir / "sense.json").read_text(encoding="utf-8")) == (
71 sense_json
72 )
73
74 def test_preserves_raw_payload_with_extra_keys_for_defensive_replay(self, tmp_path):
75 from solstone.think.sense_splitter import write_sense_outputs
76
77 seg_dir = Path(tmp_path) / "20260304" / "default" / "090000_300"
78 # Advisory validation means unexpected keys can still reach the splitter.
79 sense_json = _make_sense_output(foo="bar")
80
81 write_sense_outputs(sense_json, seg_dir)
82
83 stored = json.loads((seg_dir / "talents" / "sense.json").read_text("utf-8"))
84 assert stored["foo"] == "bar"
85 assert stored == sense_json
86
87 def test_writes_sense_markdown_when_entities_exist(self, tmp_path):
88 from solstone.think.sense_splitter import write_sense_outputs
89
90 seg_dir = Path(tmp_path) / "20260304" / "default" / "090000_300"
91 sense_json = _make_sense_output(
92 entities=[
93 {
94 "type": "Project",
95 "name": "SolAPI",
96 "role": "mentioned",
97 "source": "screen",
98 "context": "main project",
99 },
100 {
101 "type": "Person",
102 "name": "John Borthwick",
103 "role": "attendee",
104 "source": "voice",
105 "context": "active meeting participant",
106 },
107 ]
108 )
109
110 write_sense_outputs(sense_json, seg_dir)
111
112 sense_md = (seg_dir / "talents" / "sense.md").read_text(encoding="utf-8")
113 assert sense_md == (
114 "# Sense Entities\n\n"
115 "- Project — SolAPI (role=mentioned, source=screen) — main project\n"
116 "- Person — John Borthwick (role=attendee, source=voice) "
117 "— active meeting participant"
118 )
119
120 def test_skips_sense_markdown_when_entities_empty(self, tmp_path):
121 from solstone.think.sense_splitter import write_sense_outputs
122
123 seg_dir = Path(tmp_path) / "20260304" / "default" / "090000_300"
124
125 write_sense_outputs(_make_sense_output(entities=[]), seg_dir)
126
127 assert not (seg_dir / "talents" / "sense.md").exists()
128
129
130class TestMeetingDetection:
131 def test_writes_speakers_when_meeting_detected(self, tmp_path):
132 from solstone.think.sense_splitter import write_sense_outputs
133
134 seg_dir = Path(tmp_path) / "20260304" / "default" / "090000_300"
135 sense_json = _make_sense_output(
136 meeting_detected=True,
137 speakers=["Alice", "Bob"],
138 )
139
140 write_sense_outputs(sense_json, seg_dir)
141
142 speakers_path = seg_dir / "talents" / "speakers.json"
143 assert speakers_path.exists()
144 assert json.loads(speakers_path.read_text(encoding="utf-8")) == ["Alice", "Bob"]
145
146 def test_no_speakers_when_not_meeting(self, tmp_path):
147 from solstone.think.sense_splitter import write_sense_outputs
148
149 seg_dir = Path(tmp_path) / "20260304" / "default" / "090000_300"
150
151 write_sense_outputs(_make_sense_output(meeting_detected=False), seg_dir)
152
153 assert not (seg_dir / "talents" / "speakers.json").exists()
154
155 def test_meeting_with_no_speakers_writes_empty_array(self, tmp_path):
156 from solstone.think.sense_splitter import write_sense_outputs
157
158 seg_dir = Path(tmp_path) / "20260304" / "default" / "090000_300"
159 sense_json = _make_sense_output(meeting_detected=True, speakers=None)
160
161 write_sense_outputs(sense_json, seg_dir)
162
163 speakers_path = seg_dir / "talents" / "speakers.json"
164 assert speakers_path.exists()
165 assert json.loads(speakers_path.read_text(encoding="utf-8")) == []
166
167
168class TestEdgeCases:
169 def test_empty_activity_summary(self, tmp_path):
170 from solstone.think.sense_splitter import write_sense_outputs
171
172 seg_dir = Path(tmp_path) / "20260304" / "default" / "090000_300"
173
174 write_sense_outputs(_make_sense_output(activity_summary=""), seg_dir)
175
176 assert (seg_dir / "talents" / "activity.md").read_text(encoding="utf-8") == ""
177
178
179class TestMultipleFacets:
180 def test_multiple_facets_as_flat_array(self, tmp_path):
181 from solstone.think.sense_splitter import write_sense_outputs
182
183 seg_dir = Path(tmp_path) / "20260304" / "default" / "090000_300"
184 facets = [
185 {"facet": "work", "activity": "coding", "level": "high"},
186 {"facet": "personal", "activity": "reading", "level": "low"},
187 ]
188
189 write_sense_outputs(_make_sense_output(facets=facets), seg_dir)
190
191 assert json.loads((seg_dir / "talents" / "facets.json").read_text("utf-8")) == (
192 facets
193 )
194
195
196class TestWriteIdleStubs:
197 def test_writes_density_only(self, tmp_path):
198 from solstone.think.sense_splitter import write_idle_stubs
199
200 seg_dir = Path(tmp_path) / "20260304" / "default" / "090000_300"
201
202 write_idle_stubs(seg_dir)
203
204 agents_dir = seg_dir / "talents"
205 assert (agents_dir / "density.json").exists()
206 density = json.loads((agents_dir / "density.json").read_text(encoding="utf-8"))
207 assert set(density.keys()) == {"classification", "timestamp"}
208 assert density["classification"] == "idle"
209 assert not (agents_dir / "activity.md").exists()
210 assert not (agents_dir / "facets.json").exists()
211 assert not (agents_dir / "speakers.json").exists()
212 assert not (agents_dir / "sense.json").exists()
213
214
215class TestWriteChangeDetection:
216 def test_round_trip(self, tmp_path):
217 from solstone.think.sense_splitter import write_change_detection
218
219 seg_dir = Path(tmp_path) / "20260304" / "default" / "090000_300"
220 result = {
221 "timestamp": "2026-03-04T09:00:00+00:00",
222 "predecessor": None,
223 "change_class": "active",
224 "changed_sensors": ["screen"],
225 "sensors": {
226 "screen": {
227 "monitors": {
228 "center:DP-3": {
229 "first_hash": "0000000000000000",
230 "last_hash": "0000000000000001",
231 "qualified_count": 2,
232 }
233 }
234 },
235 "transcript": {
236 "present": False,
237 "word_count": 0,
238 "content_hash": None,
239 },
240 },
241 }
242
243 write_change_detection(seg_dir, result)
244
245 stored = json.loads(
246 (seg_dir / "talents" / "change.json").read_text(encoding="utf-8")
247 )
248 assert stored == result