solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4import json
5import logging
6from pathlib import Path
7
8import frontmatter
9
10SENSE_PATH = Path(__file__).resolve().parents[1] / "solstone" / "talent" / "sense.md"
11
12
13def _role_section() -> str:
14 content = frontmatter.load(SENSE_PATH).content
15 start = content.index("#### role")
16 end = content.index("#### source", start)
17 return content[start:end]
18
19
20def test_sense_role_section_contains_contamination_guard():
21 role_section = _role_section()
22
23 assert "tool or product names visible on screen" in role_section
24 assert "`source: screen`" in role_section
25 assert "`role: mentioned`" in role_section
26 assert "Google Meet" in role_section
27 assert "Zoom" in role_section
28
29
30def test_sense_role_section_has_screen_and_mentioned_guidance_for_tools_and_apps():
31 role_section = _role_section()
32
33 assert "screen" in role_section
34 assert "mentioned" in role_section
35 assert "tool" in role_section
36 assert "Video-conference app names" in role_section
37
38
39def test_sense_role_section_gates_attendee_on_meeting_detected():
40 role_section = _role_section()
41
42 assert "`meeting_detected: true`" in role_section
43 assert "`meeting_detected: false`" in role_section
44 assert "`role: attendee`" in role_section
45
46
47def _write_detected_entities(tmp_path, facet: str, day: str, rows: list[dict]) -> None:
48 entities_path = tmp_path / "facets" / facet / "entities" / f"{day}.jsonl"
49 entities_path.parent.mkdir(parents=True, exist_ok=True)
50 entities_path.write_text(
51 "".join(json.dumps(row, ensure_ascii=False) + "\n" for row in rows),
52 encoding="utf-8",
53 )
54
55
56def _write_sense_json(
57 tmp_path,
58 day: str,
59 stream: str,
60 segment_key: str,
61 payload: dict | None,
62) -> None:
63 talents_dir = tmp_path / "chronicle" / day / stream / segment_key / "talents"
64 talents_dir.mkdir(parents=True, exist_ok=True)
65 if payload is not None:
66 (talents_dir / "sense.json").write_text(json.dumps(payload), encoding="utf-8")
67
68
69def _sense_payload(*, meeting_detected: bool) -> dict:
70 return {
71 "density": "idle",
72 "content_type": "idle",
73 "activity_summary": "Idle segment.",
74 "entities": [],
75 "facets": [],
76 "meeting_detected": meeting_detected,
77 "speakers": [],
78 "recommend": {
79 "screen_record": False,
80 "speaker_attribution": False,
81 },
82 "emotional_register": "neutral",
83 }
84
85
86def _activity_record(segments: list[str]) -> dict:
87 return {
88 "id": "meeting_090000_300",
89 "activity": "meeting",
90 "segments": segments,
91 "level_avg": 1.0,
92 "description": "Team sync",
93 "active_entities": ["Guest Speaker"],
94 "created_at": 1,
95 }
96
97
98def _participation_result(role: str) -> str:
99 return json.dumps(
100 {
101 "participation": [
102 {
103 "name": "Guest Speaker",
104 "role": role,
105 "source": "voice",
106 "confidence": 0.98,
107 "context": "Spoke during the session",
108 "entity_id": None,
109 }
110 ]
111 }
112 )
113
114
115def test_participation_clamps_attendees_when_all_segments_are_non_meetings(
116 tmp_path, monkeypatch, caplog
117):
118 from solstone.talent.participation import post_process
119 from solstone.think.activities import append_activity_record, load_activity_records
120
121 facet = "work"
122 day = "20260418"
123 stream = "default"
124 segments = ["090000_300", "090500_300"]
125 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
126
127 _write_detected_entities(
128 tmp_path,
129 facet,
130 day,
131 [{"id": "guest_speaker", "type": "Person", "name": "Guest Speaker"}],
132 )
133 for segment_key in segments:
134 _write_sense_json(
135 tmp_path, day, stream, segment_key, _sense_payload(meeting_detected=False)
136 )
137
138 activity = _activity_record(segments)
139 append_activity_record(facet, day, activity)
140
141 with caplog.at_level(logging.WARNING, logger="solstone.talent.participation"):
142 post_process(
143 _participation_result("attendee"),
144 {"activity": activity, "facet": facet, "day": day},
145 )
146
147 record = load_activity_records(facet, day)[0]
148 assert record["participation"][0]["role"] == "mentioned"
149 warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
150 assert len(warnings) == 1
151 assert (
152 warnings[0].getMessage()
153 == "participation hook: clamped 1 attendee entries to mentioned on activity meeting_090000_300 (facet=work day=20260418); no contributing sense segment had meeting_detected=true"
154 )
155
156
157def test_participation_preserves_attendees_when_any_segment_is_meeting(
158 tmp_path, monkeypatch, caplog
159):
160 from solstone.talent.participation import post_process
161 from solstone.think.activities import append_activity_record, load_activity_records
162
163 facet = "work"
164 day = "20260418"
165 stream = "default"
166 segments = ["090000_300", "090500_300"]
167 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
168
169 _write_detected_entities(
170 tmp_path,
171 facet,
172 day,
173 [{"id": "guest_speaker", "type": "Person", "name": "Guest Speaker"}],
174 )
175 _write_sense_json(
176 tmp_path, day, stream, segments[0], _sense_payload(meeting_detected=False)
177 )
178 _write_sense_json(
179 tmp_path, day, stream, segments[1], _sense_payload(meeting_detected=True)
180 )
181 (
182 tmp_path
183 / "chronicle"
184 / day
185 / stream
186 / segments[1]
187 / "talents"
188 / "speakers.json"
189 ).write_text(json.dumps(["Guest Speaker"]), encoding="utf-8")
190
191 activity = _activity_record(segments)
192 append_activity_record(facet, day, activity)
193
194 with caplog.at_level(logging.WARNING, logger="solstone.talent.participation"):
195 post_process(
196 _participation_result("attendee"),
197 {"activity": activity, "facet": facet, "day": day},
198 )
199
200 record = load_activity_records(facet, day)[0]
201 assert record["participation"][0]["role"] == "attendee"
202 warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
203 assert warnings == []
204
205
206def test_participation_clamp_is_idempotent_on_second_pass(
207 tmp_path, monkeypatch, caplog
208):
209 from solstone.talent.participation import post_process
210 from solstone.think.activities import append_activity_record, load_activity_records
211
212 facet = "work"
213 day = "20260418"
214 stream = "default"
215 segments = ["090000_300", "090500_300"]
216 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
217
218 _write_detected_entities(
219 tmp_path,
220 facet,
221 day,
222 [{"id": "guest_speaker", "type": "Person", "name": "Guest Speaker"}],
223 )
224 for segment_key in segments:
225 _write_sense_json(
226 tmp_path, day, stream, segment_key, _sense_payload(meeting_detected=False)
227 )
228
229 activity = _activity_record(segments)
230 append_activity_record(facet, day, activity)
231
232 with caplog.at_level(logging.WARNING, logger="solstone.talent.participation"):
233 post_process(
234 _participation_result("attendee"),
235 {"activity": activity, "facet": facet, "day": day},
236 )
237 first_warning_count = len(
238 [r for r in caplog.records if r.levelno == logging.WARNING]
239 )
240 assert first_warning_count == 1
241
242 record = load_activity_records(facet, day)[0]
243 second_result = json.dumps({"participation": record["participation"]})
244
245 with caplog.at_level(logging.WARNING, logger="solstone.talent.participation"):
246 post_process(
247 second_result,
248 {"activity": record, "facet": facet, "day": day},
249 )
250
251 updated = load_activity_records(facet, day)[0]
252 assert updated["participation"] == record["participation"]
253 warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
254 assert len(warnings) == 1
255
256
257def test_participation_treats_missing_sense_json_as_non_meeting(
258 tmp_path, monkeypatch, caplog
259):
260 from solstone.talent.participation import post_process
261 from solstone.think.activities import append_activity_record, load_activity_records
262
263 facet = "work"
264 day = "20260418"
265 stream = "default"
266 segments = ["090000_300", "090500_300"]
267 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
268
269 _write_detected_entities(
270 tmp_path,
271 facet,
272 day,
273 [{"id": "guest_speaker", "type": "Person", "name": "Guest Speaker"}],
274 )
275 _write_sense_json(tmp_path, day, stream, segments[0], None)
276 _write_sense_json(
277 tmp_path, day, stream, segments[1], _sense_payload(meeting_detected=False)
278 )
279
280 activity = _activity_record(segments)
281 append_activity_record(facet, day, activity)
282
283 with caplog.at_level(logging.WARNING, logger="solstone.talent.participation"):
284 post_process(
285 _participation_result("attendee"),
286 {"activity": activity, "facet": facet, "day": day},
287 )
288
289 record = load_activity_records(facet, day)[0]
290 assert record["participation"][0]["role"] == "mentioned"
291 warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
292 assert len(warnings) == 1