solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4"""Unit tests for talent.speaker_attribution pre_process stub-writing behavior."""
5
6import json
7import logging
8from unittest.mock import patch
9
10
11def _run_pre_process(context, seg_dir, attribute_result):
12 """Helper: run pre_process with mocked dependencies."""
13 with (
14 patch(
15 "solstone.apps.speakers.attribution.attribute_segment",
16 return_value=attribute_result,
17 ),
18 patch(
19 "solstone.think.utils.segment_path",
20 return_value=seg_dir,
21 ),
22 ):
23 from solstone.talent.speaker_attribution import pre_process
24
25 return pre_process(context)
26
27
28CONTEXT = {"day": "20260318", "segment": "100000_300", "stream": "default"}
29
30
31class TestPreProcessStub:
32 def test_error_with_npz_writes_stub(self, tmp_path):
33 """no_owner_centroid error + .npz present -> stub written."""
34 (tmp_path / "audio.npz").write_bytes(b"x")
35 result = _run_pre_process(
36 CONTEXT,
37 tmp_path,
38 {"error": "no_owner_centroid"},
39 )
40 stub_path = tmp_path / "talents" / "speaker_labels.json"
41 assert stub_path.exists()
42 data = json.loads(stub_path.read_text())
43 assert data == {"labels": [], "skipped": True, "reason": "no_owner_centroid"}
44 assert result == {"skip_reason": "no_owner_centroid"}
45
46 def test_error_without_npz_no_stub(self, tmp_path):
47 """no_owner_centroid error + no .npz -> no stub written."""
48 result = _run_pre_process(
49 CONTEXT,
50 tmp_path,
51 {"error": "no_owner_centroid"},
52 )
53 stub_path = tmp_path / "talents" / "speaker_labels.json"
54 assert not stub_path.exists()
55 assert result == {"skip_reason": "no_owner_centroid"}
56
57 def test_empty_labels_with_npz_writes_stub(self, tmp_path):
58 """Empty labels (loaded-but-empty .npz) + .npz present -> stub written."""
59 (tmp_path / "audio.npz").write_bytes(b"x")
60 result = _run_pre_process(
61 CONTEXT,
62 tmp_path,
63 {"labels": []},
64 )
65 stub_path = tmp_path / "talents" / "speaker_labels.json"
66 assert stub_path.exists()
67 data = json.loads(stub_path.read_text())
68 assert data == {"labels": [], "skipped": True, "reason": "no_embeddings"}
69 assert result == {"skip_reason": "no_embeddings"}
70
71 def test_empty_labels_without_npz_no_stub(self, tmp_path):
72 """Empty labels + no .npz -> no stub written."""
73 result = _run_pre_process(
74 CONTEXT,
75 tmp_path,
76 {"labels": []},
77 )
78 stub_path = tmp_path / "talents" / "speaker_labels.json"
79 assert not stub_path.exists()
80 assert result == {"skip_reason": "no_embeddings"}
81
82 def test_no_segment_context_no_stub(self, tmp_path):
83 """Missing day/segment -> returns early before any stub logic."""
84 (tmp_path / "audio.npz").write_bytes(b"x")
85 with patch("solstone.think.utils.segment_path", return_value=tmp_path):
86 from solstone.talent.speaker_attribution import pre_process
87
88 result = pre_process({"stream": "default"})
89 stub_path = tmp_path / "talents" / "speaker_labels.json"
90 assert not stub_path.exists()
91 assert result == {"skip_reason": "no_segment_context"}
92
93 def test_layer4_returns_template_vars(self):
94 """Unmatched sentences return template vars without changing transcript."""
95 result = _run_pre_process(
96 {
97 **CONTEXT,
98 "transcript": "some transcript",
99 "meta": {},
100 },
101 None,
102 {
103 "labels": [
104 {
105 "sentence_id": 0,
106 "speaker": "owner",
107 "confidence": "high",
108 "method": "embedding",
109 },
110 {
111 "sentence_id": 1,
112 "speaker": None,
113 "confidence": None,
114 "method": None,
115 },
116 {
117 "sentence_id": 2,
118 "speaker": None,
119 "confidence": None,
120 "method": None,
121 },
122 ],
123 "unmatched": [1, 2],
124 "unmatched_texts": {1: "Hello everyone", 2: "Let me explain"},
125 "candidates": ["Alice Johnson", "Bob Smith"],
126 "metadata": {"total": 3, "resolved": 1},
127 "source": "audio",
128 },
129 )
130
131 assert "meta" in result
132 assert "attribution_result" in result["meta"]
133 assert "template_vars" in result
134 assert "unmatched_context" in result["template_vars"]
135 assert "transcript" not in result
136
137 unmatched_context = result["template_vars"]["unmatched_context"]
138 assert "Alice Johnson" in unmatched_context
139 assert "Bob Smith" in unmatched_context
140 assert "Hello everyone" in unmatched_context
141 assert "Let me explain" in unmatched_context
142 assert "Sentence 1" in unmatched_context
143 assert "Sentence 2" in unmatched_context
144
145
146def _post_process_context():
147 attribution_result = {
148 "labels": [
149 {"sentence_id": 1, "speaker": None, "confidence": None, "method": None},
150 {
151 "sentence_id": 2,
152 "speaker": "bob",
153 "confidence": "high",
154 "method": "owner",
155 },
156 ],
157 "metadata": {},
158 "source": None,
159 }
160 return {
161 "day": "20260419",
162 "segment": "000000",
163 "stream": "default",
164 "meta": {"attribution_result": attribution_result},
165 }
166
167
168class TestPostProcess:
169 def test_bare_list_merges_layer4_attributions(self, tmp_path):
170 result = json.dumps(
171 [{"sentence_id": 1, "speaker": "Alice", "reasoning": "said her name"}]
172 )
173 context = _post_process_context()
174
175 with (
176 patch(
177 "solstone.apps.speakers.attribution.save_speaker_labels"
178 ) as save_mock,
179 patch(
180 "solstone.apps.speakers.attribution.accumulate_voiceprints"
181 ) as accumulate_mock,
182 patch(
183 "solstone.think.entities.journal.load_all_journal_entities",
184 return_value={"alice": {"id": "alice", "name": "Alice"}},
185 ),
186 patch("solstone.think.utils.segment_path", return_value=tmp_path),
187 ):
188 from solstone.talent.speaker_attribution import post_process
189
190 post_process(result, context)
191
192 saved_labels = save_mock.call_args[0][1]
193 assert saved_labels[0] == {
194 "sentence_id": 1,
195 "speaker": "alice",
196 "confidence": "medium",
197 "method": "contextual",
198 }
199 assert saved_labels[1] == {
200 "sentence_id": 2,
201 "speaker": "bob",
202 "confidence": "high",
203 "method": "owner",
204 }
205 accumulate_mock.assert_not_called()
206
207 def test_wrapped_attributions_merge_layer4_attributions(self, tmp_path):
208 result = json.dumps(
209 {
210 "attributions": [
211 {
212 "sentence_id": 1,
213 "speaker": "Alice",
214 "reasoning": "said her name",
215 }
216 ]
217 }
218 )
219 context = _post_process_context()
220
221 with (
222 patch(
223 "solstone.apps.speakers.attribution.save_speaker_labels"
224 ) as save_mock,
225 patch(
226 "solstone.apps.speakers.attribution.accumulate_voiceprints"
227 ) as accumulate_mock,
228 patch(
229 "solstone.think.entities.journal.load_all_journal_entities",
230 return_value={"alice": {"id": "alice", "name": "Alice"}},
231 ),
232 patch("solstone.think.utils.segment_path", return_value=tmp_path),
233 ):
234 from solstone.talent.speaker_attribution import post_process
235
236 post_process(result, context)
237
238 saved_labels = save_mock.call_args[0][1]
239 assert saved_labels[0] == {
240 "sentence_id": 1,
241 "speaker": "alice",
242 "confidence": "medium",
243 "method": "contextual",
244 }
245 assert saved_labels[1] == {
246 "sentence_id": 2,
247 "speaker": "bob",
248 "confidence": "high",
249 "method": "owner",
250 }
251 accumulate_mock.assert_not_called()
252
253 def test_ambiguous_layer4_attribution_leaves_speaker_unmatched(
254 self, tmp_path, monkeypatch
255 ):
256 from solstone.think.entities import load_ambiguities
257
258 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
259 import solstone.think.utils as think_utils
260
261 think_utils._journal_path_cache = None
262 result = json.dumps(
263 [{"sentence_id": 1, "speaker": "Sarah", "reasoning": "said her name"}]
264 )
265 context = _post_process_context()
266
267 with (
268 patch(
269 "solstone.apps.speakers.attribution.save_speaker_labels"
270 ) as save_mock,
271 patch(
272 "solstone.apps.speakers.attribution.accumulate_voiceprints"
273 ) as accumulate_mock,
274 patch(
275 "solstone.think.entities.journal.load_all_journal_entities",
276 return_value={
277 "sarah_connor": {"id": "sarah_connor", "name": "Sarah Connor"},
278 "sarah_lee": {"id": "sarah_lee", "name": "Sarah Lee"},
279 },
280 ),
281 patch("solstone.think.utils.segment_path", return_value=tmp_path),
282 ):
283 from solstone.talent.speaker_attribution import post_process
284
285 post_process(result, context)
286
287 saved_labels = save_mock.call_args[0][1]
288 assert saved_labels[0] == {
289 "sentence_id": 1,
290 "speaker": None,
291 "confidence": None,
292 "method": None,
293 }
294 accumulate_mock.assert_not_called()
295 rows = load_ambiguities()
296 assert len(rows) == 1
297 assert rows[0]["normalized_query"] == "sarah"
298
299 def test_non_list_non_dict_yields_zero_merges_and_warns(self, tmp_path, caplog):
300 context = _post_process_context()
301
302 with (
303 patch(
304 "solstone.apps.speakers.attribution.save_speaker_labels"
305 ) as save_mock,
306 patch("solstone.apps.speakers.attribution.accumulate_voiceprints"),
307 patch(
308 "solstone.think.entities.journal.load_all_journal_entities",
309 return_value={"alice": {"id": "alice", "name": "Alice"}},
310 ) as load_mock,
311 patch("solstone.think.utils.segment_path", return_value=tmp_path),
312 caplog.at_level(logging.WARNING),
313 ):
314 from solstone.talent.speaker_attribution import post_process
315
316 post_process("42", context)
317
318 saved_labels = save_mock.call_args[0][1]
319 assert saved_labels[0] == {
320 "sentence_id": 1,
321 "speaker": None,
322 "confidence": None,
323 "method": None,
324 }
325 assert saved_labels[1] == {
326 "sentence_id": 2,
327 "speaker": "bob",
328 "confidence": "high",
329 "method": "owner",
330 }
331 assert "expected JSON array, got int" in caplog.text
332 load_mock.assert_not_called()