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
7from pathlib import Path
8
9import pytest
10
11import solstone.think.speaker_identify_operations as ledger
12
13
14@pytest.fixture
15def op_journal(monkeypatch, tmp_path):
16 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
17 import solstone.think.utils as think_utils
18
19 think_utils._journal_path_cache = None
20 return Path(tmp_path)
21
22
23def _members() -> list[dict[str, object]]:
24 return [
25 {
26 "day": "20260101",
27 "stream": "test",
28 "segment_key": "090000_300",
29 "source": "mic_audio",
30 "sentence_id": 1,
31 },
32 {
33 "day": "20260101",
34 "stream": "test",
35 "segment_key": "090000_300",
36 "source": "mic_audio",
37 "sentence_id": 2,
38 },
39 ]
40
41
42def _prepared_event(request_id: str = "req-1") -> dict[str, object]:
43 operation_id = ledger.operation_id_for_request(request_id)
44 members = _members()
45 fingerprint = ledger.request_fingerprint(
46 cluster_members=members,
47 target_entity_id="alice_test",
48 will_create=True,
49 entity_type="Person",
50 reviewed_near_match_entity_ids=["bob_test"],
51 )
52 plan = {
53 "plan_schema_version": 1,
54 "operation_id": operation_id,
55 "request_id": request_id,
56 "planned_at": "2026-07-20T12:00:00Z",
57 "request": {
58 "cluster_id": 7,
59 "name": "Alice Test",
60 "entity_id": "alice_test",
61 "resolve_only": False,
62 "create_new": True,
63 "entity_type": "Person",
64 "reviewed_near_match_entity_ids": ["bob_test"],
65 },
66 "cluster": {"cluster_id": 7, "member_count": len(members), "members": members},
67 "target": {
68 "entity_id": "alice_test",
69 "entity_name": "Alice Test",
70 "entity_type": "Person",
71 "will_create": True,
72 },
73 "entity_identity": {
74 "prior_identity": None,
75 "intended_identity": {
76 "id": "alice_test",
77 "name": "Alice Test",
78 "type": "Person",
79 "created_at": 1,
80 },
81 "expected_history_operation": {
82 "operation_kind": "speaker_identify",
83 "operation_id": operation_id,
84 },
85 },
86 "direct_voiceprints": {"preexisting_keys": [], "entries_to_add": []},
87 "segments": [],
88 "retro_confirm": {
89 "matched": False,
90 "match_score": None,
91 "candidate_id": None,
92 "candidate_before": None,
93 "candidate_after": None,
94 "preexisting_voiceprint_keys": [],
95 "voiceprints_to_add": [],
96 },
97 "sentinel": {
98 "cluster_key": "7",
99 "prior_entry": None,
100 "intended_entry": {
101 "entity_id": "alice_test",
102 "label": "Alice Test",
103 "ts": "2026-07-20T12:00:00Z",
104 },
105 },
106 "keep_separate_assertions": [],
107 }
108 return {
109 "schema_version": ledger.IDENTIFY_OPERATION_SCHEMA_VERSION,
110 "event_id": f"{operation_id}:prepared",
111 "operation_id": operation_id,
112 "request_id": request_id,
113 "event_kind": "prepared",
114 "ts": "2026-07-20T12:00:00Z",
115 "caller": "test",
116 "actor": None,
117 "request_fingerprint": fingerprint,
118 "prepared_plan": plan,
119 }
120
121
122def _checkpoint_event(
123 operation_id: str, request_id: str, phase: str
124) -> dict[str, object]:
125 return {
126 "schema_version": ledger.IDENTIFY_OPERATION_SCHEMA_VERSION,
127 "event_id": f"{operation_id}:checkpoint:{phase}",
128 "operation_id": operation_id,
129 "request_id": request_id,
130 "event_kind": "checkpoint",
131 "ts": "2026-07-20T12:00:01Z",
132 "caller": "test",
133 "actor": None,
134 "phase": phase,
135 "checkpoint": {
136 "phase_status": "complete",
137 "completed_at": "2026-07-20T12:00:01Z",
138 "counts": {"saved_count": 2},
139 "skipped_reasons": {},
140 "saved_count": 2,
141 "skipped_existing_count": 0,
142 "saved_keys": [
143 {
144 "day": "20260101",
145 "segment_key": "090000_300",
146 "source": "mic_audio",
147 "sentence_id": 1,
148 }
149 ],
150 },
151 }
152
153
154def test_operation_id_for_request_is_deterministic():
155 assert ledger.operation_id_for_request("abc") == ledger.operation_id_for_request(
156 "abc"
157 )
158 assert ledger.operation_id_for_request("abc").startswith("idop_")
159 assert ledger.operation_id_for_request("abc") != ledger.operation_id_for_request(
160 "abcd"
161 )
162
163
164def test_request_fingerprint_changes_for_each_identity_input():
165 base = ledger.request_fingerprint(
166 cluster_members=_members(),
167 target_entity_id="alice_test",
168 will_create=True,
169 entity_type="Person",
170 reviewed_near_match_entity_ids=["bob_test"],
171 )
172 changed_member = ledger.request_fingerprint(
173 cluster_members=[{**_members()[0], "sentence_id": 99}],
174 target_entity_id="alice_test",
175 will_create=True,
176 entity_type="Person",
177 reviewed_near_match_entity_ids=["bob_test"],
178 )
179 changed_target = ledger.request_fingerprint(
180 cluster_members=_members(),
181 target_entity_id="carol_test",
182 will_create=True,
183 entity_type="Person",
184 reviewed_near_match_entity_ids=["bob_test"],
185 )
186 changed_create = ledger.request_fingerprint(
187 cluster_members=_members(),
188 target_entity_id="alice_test",
189 will_create=False,
190 entity_type="Person",
191 reviewed_near_match_entity_ids=["bob_test"],
192 )
193 changed_type = ledger.request_fingerprint(
194 cluster_members=_members(),
195 target_entity_id="alice_test",
196 will_create=True,
197 entity_type="Project",
198 reviewed_near_match_entity_ids=["bob_test"],
199 )
200 changed_reviewed = ledger.request_fingerprint(
201 cluster_members=_members(),
202 target_entity_id="alice_test",
203 will_create=True,
204 entity_type="Person",
205 reviewed_near_match_entity_ids=["carol_test"],
206 )
207
208 assert len(base) == 64
209 assert (
210 len(
211 {
212 base,
213 changed_member,
214 changed_target,
215 changed_create,
216 changed_type,
217 changed_reviewed,
218 }
219 )
220 == 6
221 )
222 assert base not in {
223 changed_member,
224 changed_target,
225 changed_create,
226 changed_type,
227 changed_reviewed,
228 }
229
230
231def test_append_and_fold_prepared_resume_state(op_journal):
232 prepared = _prepared_event()
233 checkpoint = _checkpoint_event(
234 str(prepared["operation_id"]),
235 str(prepared["request_id"]),
236 "direct_voiceprints",
237 )
238
239 ledger.append_event(prepared)
240 ledger.append_event(checkpoint)
241
242 state = ledger.fold_operation(str(prepared["operation_id"]))
243 assert state is not None
244 assert state.operation_id == prepared["operation_id"]
245 assert state.request_fingerprint == prepared["request_fingerprint"]
246 assert state.cluster_member_set == {
247 ("20260101", "test", "090000_300", "mic_audio", 1),
248 ("20260101", "test", "090000_300", "mic_audio", 2),
249 }
250 assert state.target_entity_id == "alice_test"
251 assert state.target_entity_name == "Alice Test"
252 assert state.will_create is True
253 assert state.entity_type == "Person"
254 assert state.reviewed_near_match_entity_ids == ("bob_test",)
255 assert state.completed_phases == ("direct_voiceprints",)
256 assert state.pending_phases == (
257 "entity",
258 "keep_separate",
259 "corrections",
260 "labels",
261 "retro_tracker",
262 "sentinel",
263 )
264 assert state.terminal_status == "in_progress"
265 assert state.phase_checkpoints["direct_voiceprints"]["counts"]["saved_count"] == 2
266
267
268def test_read_paths_do_not_create_speakers_dir(op_journal):
269 assert ledger.load_operations() == []
270 assert ledger.fold_operation("idop_missing") is None
271 assert ledger.fold_all_operations() == []
272 assert not (op_journal / "speakers").exists()
273
274
275def test_committed_fold_returns_stored_result(op_journal):
276 prepared = _prepared_event()
277 operation_id = str(prepared["operation_id"])
278 committed = {
279 "schema_version": ledger.IDENTIFY_OPERATION_SCHEMA_VERSION,
280 "event_id": f"{operation_id}:committed",
281 "operation_id": operation_id,
282 "request_id": prepared["request_id"],
283 "event_kind": "committed",
284 "ts": "2026-07-20T12:00:02Z",
285 "caller": "test",
286 "actor": None,
287 "result": {"status": "identified", "operation_id": operation_id},
288 }
289
290 ledger.append_event(prepared)
291 ledger.append_event(committed)
292
293 state = ledger.fold_operation(operation_id)
294 assert state is not None
295 assert state.terminal_status == "committed"
296 assert state.result == {"status": "identified", "operation_id": operation_id}
297 assert state.pending_phases == ()
298
299
300def test_undo_started_folds_as_undoing_not_committed(op_journal):
301 prepared = _prepared_event()
302 operation_id = str(prepared["operation_id"])
303 committed = {
304 "schema_version": ledger.IDENTIFY_OPERATION_SCHEMA_VERSION,
305 "event_id": f"{operation_id}:committed",
306 "operation_id": operation_id,
307 "request_id": prepared["request_id"],
308 "event_kind": "committed",
309 "ts": "2026-07-20T12:00:02Z",
310 "caller": "test",
311 "actor": None,
312 "result": {"status": "identified", "operation_id": operation_id},
313 }
314 undo_prepared = {
315 "schema_version": ledger.IDENTIFY_OPERATION_SCHEMA_VERSION,
316 "event_id": f"{operation_id}:undo_prepared",
317 "operation_id": operation_id,
318 "request_id": prepared["request_id"],
319 "event_kind": "undo_prepared",
320 "ts": "2026-07-20T12:00:03Z",
321 "caller": "test",
322 "actor": None,
323 "undo_started_at": "2026-07-20T12:00:03Z",
324 }
325
326 ledger.append_event(prepared)
327 ledger.append_event(committed)
328 ledger.append_event(undo_prepared)
329
330 state = ledger.fold_operation(operation_id)
331 assert state is not None
332 assert state.terminal_status == "undoing"
333 assert state.pending_phases == ledger.UNDO_PHASE_ORDER
334
335
336def test_identical_duplicate_event_id_folds_once(op_journal):
337 prepared = _prepared_event()
338 path = ledger.identify_operations_path(create=True)
339 path.write_text(
340 json.dumps(prepared) + "\n" + json.dumps(prepared) + "\n",
341 encoding="utf-8",
342 )
343
344 state = ledger.fold_operation(str(prepared["operation_id"]))
345 assert state is not None
346 assert state.terminal_status == "in_progress"
347
348
349def test_non_identical_duplicate_event_id_raises(op_journal):
350 prepared = _prepared_event()
351 changed = dict(prepared)
352 changed["ts"] = "2026-07-20T12:00:09Z"
353 path = ledger.identify_operations_path(create=True)
354 path.write_text(
355 json.dumps(prepared) + "\n" + json.dumps(changed) + "\n",
356 encoding="utf-8",
357 )
358
359 with pytest.raises(ledger.IdentifyOperationError):
360 ledger.fold_operation(str(prepared["operation_id"]))
361
362
363def test_strict_malformed_row_raises(op_journal):
364 ledger.identify_operations_path(create=True).write_text(
365 "not-json\n",
366 encoding="utf-8",
367 )
368
369 with pytest.raises(ledger.IdentifyOperationError):
370 ledger.load_operations()