solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4"""Tests for mutation-safe entity resolution ambiguities."""
5
6from __future__ import annotations
7
8import ast
9import hashlib
10import inspect
11import json
12from contextlib import contextmanager
13from importlib import import_module
14from pathlib import Path
15from typing import Iterator
16
17import pytest
18
19import solstone.think.entities.ambiguities as ambiguities
20import solstone.think.entities.history as entity_history
21from solstone.think.entities import (
22 EntityAmbiguityError,
23 EntityResolutionOutcome,
24 MatchTier,
25 ResolutionOrigin,
26 ResolutionScope,
27 delete_journal_entity,
28 find_matching_entity,
29 load_ambiguities,
30 record_ambiguity_choice,
31 record_entity_resolution,
32)
33from solstone.think.journal_io import LockTimeout
34
35
36@pytest.fixture
37def journal(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
38 """Use an isolated journal for ambiguity tests."""
39 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
40 return tmp_path
41
42
43def _entity(
44 name: str,
45 entity_id: str | None = None,
46 *,
47 aka: list[str] | None = None,
48 emails: list[str] | None = None,
49 blocked: bool = False,
50) -> dict:
51 entity = {"id": entity_id or name.lower().replace(" ", "_"), "name": name}
52 if aka is not None:
53 entity["aka"] = aka
54 if emails is not None:
55 entity["emails"] = emails
56 if blocked:
57 entity["blocked"] = True
58 return entity
59
60
61def _origin(lane: str = "test", invocation_id: str = "run-1") -> ResolutionOrigin:
62 return ResolutionOrigin(lane=lane, field="entity", invocation_id=invocation_id)
63
64
65def _ambiguity_file(journal: Path) -> Path:
66 return journal / "entities" / "ambiguities.jsonl"
67
68
69def _tree_snapshot(root: Path) -> dict[str, str]:
70 snapshot: dict[str, str] = {}
71 for path in sorted(root.rglob("*")):
72 if path.name.endswith(".lock") or path.is_dir():
73 continue
74 rel = path.relative_to(root).as_posix()
75 if path.is_file():
76 snapshot[rel] = hashlib.sha256(path.read_bytes()).hexdigest()
77 return snapshot
78
79
80def _record(
81 query: str,
82 entities: list[dict],
83 *,
84 scope: ResolutionScope | None = None,
85 origin: ResolutionOrigin | None = None,
86):
87 return record_entity_resolution(
88 query,
89 entities,
90 scope=scope or ResolutionScope.journal(),
91 origin=origin or _origin(),
92 )
93
94
95def test_high_confidence_tiers_resolve_without_creating_ambiguities(
96 journal: Path,
97) -> None:
98 cases = [
99 ("Alice Smith", [_entity("Alice Smith")], MatchTier.EXACT),
100 ("alice smith", [_entity("Alice Smith")], MatchTier.CASE_INSENSITIVE),
101 (
102 "alice@example.com",
103 [_entity("Alice Smith", emails=["alice@example.com"])],
104 MatchTier.EMAIL,
105 ),
106 (
107 "Alice Smith",
108 [_entity("Alice Jones", entity_id="alice_smith")],
109 MatchTier.SLUG,
110 ),
111 ]
112
113 for query, entities, expected_tier in cases:
114 resolution = _record(query, entities, origin=_origin(invocation_id=query))
115 assert resolution.outcome == EntityResolutionOutcome.RESOLVED
116 assert resolution.entity is not None
117 assert resolution.tier == expected_tier
118 assert resolution.ambiguity_id is None
119
120 assert not _ambiguity_file(journal).exists()
121
122
123def test_tier_five_multi_candidate_legacy_none_becomes_ambiguous(
124 journal: Path,
125) -> None:
126 entities = [_entity("Sarah Connor"), _entity("Sarah Lee")]
127
128 assert find_matching_entity("Sarah", entities) is None
129 resolution = _record("Sarah", entities)
130
131 assert resolution.outcome == EntityResolutionOutcome.AMBIGUOUS
132 assert resolution.entity is None
133 assert resolution.tier == MatchTier.FIRST_WORD
134 assert resolution.ambiguity_id
135 assert {candidate.id for candidate in resolution.candidates} == {
136 "sarah_connor",
137 "sarah_lee",
138 }
139 assert list(resolution.candidates) == sorted(
140 resolution.candidates,
141 key=lambda candidate: (-candidate.score, candidate.name, candidate.id),
142 )
143
144 row = load_ambiguities()[0]
145 assert row["observed_tier"] == int(MatchTier.FIRST_WORD)
146 assert len(row["ranked_candidates"]) == 2
147
148
149def test_record_entity_resolution_read_only_skips_trust_lock_and_ambiguity_observation(
150 journal: Path,
151) -> None:
152 entities = [_entity("Sarah Connor"), _entity("Sarah Lee")]
153 trust_lock = journal / "health" / "locks" / "entity-trust.lock"
154
155 resolution = record_entity_resolution(
156 "Sarah",
157 entities,
158 scope=ResolutionScope.journal(),
159 origin=_origin(),
160 read_only=True,
161 )
162
163 assert resolution.outcome == EntityResolutionOutcome.AMBIGUOUS
164 assert resolution.ambiguity_id == ""
165 assert {candidate.id for candidate in resolution.candidates} == {
166 "sarah_connor",
167 "sarah_lee",
168 }
169 assert not _ambiguity_file(journal).exists()
170 assert not trust_lock.exists()
171
172
173def test_tier_six_multi_candidate_legacy_none_becomes_ambiguous(
174 journal: Path,
175) -> None:
176 entities = [_entity("Josh Jones Dilworth"), _entity("Mary Jones Dilworth")]
177
178 assert find_matching_entity("Jones Dilworth", entities) is None
179 resolution = _record("Jones Dilworth", entities)
180
181 assert resolution.outcome == EntityResolutionOutcome.AMBIGUOUS
182 assert resolution.entity is None
183 assert resolution.tier == MatchTier.TOKEN_SUBSET
184 assert {candidate.id for candidate in resolution.candidates} == {
185 "josh_jones_dilworth",
186 "mary_jones_dilworth",
187 }
188
189
190def test_tier_seven_multi_candidate_legacy_none_becomes_ambiguous(
191 journal: Path,
192) -> None:
193 entities = [_entity("Jonathan Dilton"), _entity("Jonas Diltmore")]
194
195 assert find_matching_entity("Jona Dilt", entities) is None
196 resolution = _record("Jona Dilt", entities)
197
198 assert resolution.outcome == EntityResolutionOutcome.AMBIGUOUS
199 assert resolution.entity is None
200 assert resolution.tier == MatchTier.PREFIX
201 assert {candidate.id for candidate in resolution.candidates} == {
202 "jonathan_dilton",
203 "jonas_diltmore",
204 }
205
206
207def test_tier_eight_fuzzy_is_ambiguous(journal: Path) -> None:
208 entities = [_entity("Robert Johnson")]
209
210 resolution = _record("Robert Jonson", entities)
211
212 assert resolution.outcome == EntityResolutionOutcome.AMBIGUOUS
213 assert resolution.entity is None
214 assert resolution.tier == MatchTier.FUZZY
215 assert [candidate.id for candidate in resolution.candidates] == ["robert_johnson"]
216
217
218def test_true_no_match_does_not_create_row(journal: Path) -> None:
219 resolution = _record("Xy", [_entity("Alice Smith")])
220
221 assert resolution.outcome == EntityResolutionOutcome.NO_MATCH
222 assert resolution.entity is None
223 assert resolution.ambiguity_id is None
224 assert not _ambiguity_file(journal).exists()
225
226
227def test_one_open_row_per_scope_query_with_deduplicated_origins(
228 journal: Path,
229 monkeypatch: pytest.MonkeyPatch,
230) -> None:
231 times = iter(
232 [
233 "2026-01-01T00:00:00Z",
234 "2026-01-01T00:00:01Z",
235 "2026-01-01T00:00:02Z",
236 "2026-01-01T00:00:03Z",
237 ]
238 )
239 monkeypatch.setattr(ambiguities, "utc_now_iso", lambda: next(times))
240 entities = [_entity("Sarah Connor"), _entity("Sarah Lee")]
241 origin1 = _origin("participation", "run-1")
242 origin2 = _origin("participation", "run-2")
243 origin3 = _origin("schedule", "run-3")
244
245 first = _record("Sarah", entities, origin=origin1)
246 _record(" sarah ", entities, origin=origin1)
247 second = _record("SARAH", entities, origin=origin2)
248 third = _record("Sarah", entities, origin=origin3)
249
250 rows = load_ambiguities()
251 assert len(rows) == 1
252 row = rows[0]
253 assert first.ambiguity_id == second.ambiguity_id == third.ambiguity_id
254 assert row["first_seen"] == "2026-01-01T00:00:00Z"
255 assert row["last_seen"] == "2026-01-01T00:00:03Z"
256 assert row["occurrence_count"] == 3
257 assert len(row["origins"]) == 3
258
259
260def test_normalization_collapses_case_whitespace_and_unicode_not_punctuation(
261 journal: Path,
262) -> None:
263 assert ambiguities.normalize_resolution_query(" Sarah\tLee ") == "sarah lee"
264 assert ambiguities.normalize_resolution_query("Ｓａｒａｈ") == "sarah"
265 assert ambiguities.normalize_resolution_query("Sarah.") == "sarah."
266
267 entities = [_entity("Sarah Connor"), _entity("Sarah Lee")]
268 punct_entities = [_entity("Sarah. Connor"), _entity("Sarah. Lee")]
269
270 first = _record(" Sarah ", entities, origin=_origin("test", "run-1"))
271 second = _record("sarah", entities, origin=_origin("test", "run-2"))
272 third = _record("Sarah.", punct_entities, origin=_origin("test", "run-3"))
273
274 rows = load_ambiguities()
275 assert len(rows) == 2
276 assert first.ambiguity_id == second.ambiguity_id
277 assert third.ambiguity_id != first.ambiguity_id
278
279
280def test_scope_discriminator_separates_facet_and_journal(journal: Path) -> None:
281 entities = [_entity("Sarah Connor"), _entity("Sarah Lee")]
282
283 journal_resolution = _record("Sarah", entities, scope=ResolutionScope.journal())
284 facet_resolution = _record(
285 "Sarah",
286 entities,
287 scope=ResolutionScope.facet_scope("work"),
288 origin=_origin("test", "facet-run"),
289 )
290
291 rows = load_ambiguities()
292 assert len(rows) == 2
293 assert journal_resolution.ambiguity_id != facet_resolution.ambiguity_id
294 assert {json.dumps(row["scope"], sort_keys=True) for row in rows} == {
295 json.dumps({"kind": "journal"}, sort_keys=True),
296 json.dumps({"kind": "facet", "facet": "work"}, sort_keys=True),
297 }
298
299
300def test_record_ambiguity_choice_returns_choice_without_reopening(
301 journal: Path,
302) -> None:
303 entities = [_entity("Sarah Connor"), _entity("Sarah Lee")]
304 scope = ResolutionScope.journal()
305
306 ambiguous = _record("Sarah", entities, scope=scope)
307 row = record_ambiguity_choice("Sarah", "sarah_lee", entities, scope=scope)
308
309 assert row["ambiguity_id"] == ambiguous.ambiguity_id
310 assert row["status"] == "resolved"
311 assert row["resolved_entity_id"] == "sarah_lee"
312
313 resolved = _record("Sarah", entities, scope=scope, origin=_origin("test", "run-2"))
314 rows = load_ambiguities()
315 assert resolved.outcome == EntityResolutionOutcome.RESOLVED
316 assert resolved.entity is not None
317 assert resolved.entity["id"] == "sarah_lee"
318 assert rows[0]["occurrence_count"] == 1
319 assert rows[0]["status"] == "resolved"
320
321
322def test_invalid_ambiguity_choice_fails_with_row_byte_unchanged(
323 journal: Path,
324) -> None:
325 entities = [_entity("Sarah Connor"), _entity("Sarah Lee")]
326 scope = ResolutionScope.journal()
327 _record("Sarah", entities, scope=scope)
328 path = _ambiguity_file(journal)
329 before = path.read_bytes()
330
331 with pytest.raises(EntityAmbiguityError):
332 record_ambiguity_choice("Sarah", "missing", entities, scope=scope)
333 assert path.read_bytes() == before
334
335 with pytest.raises(EntityAmbiguityError):
336 record_ambiguity_choice(
337 "Sarah",
338 "blocked_sarah",
339 [*entities, _entity("Blocked Sarah", "blocked_sarah", blocked=True)],
340 scope=scope,
341 )
342 assert path.read_bytes() == before
343
344 with pytest.raises(EntityAmbiguityError):
345 record_ambiguity_choice(
346 "Sarah",
347 "sarah_lee",
348 entities,
349 scope=ResolutionScope.facet_scope("work"),
350 )
351 assert path.read_bytes() == before
352
353
354def test_reresolve_retains_prior_choice_in_audit(journal: Path) -> None:
355 entities = [_entity("Sarah Connor"), _entity("Sarah Lee")]
356 scope = ResolutionScope.journal()
357
358 _record("Sarah", entities, scope=scope)
359 record_ambiguity_choice("Sarah", "sarah_connor", entities, scope=scope)
360 row = record_ambiguity_choice(
361 "Sarah",
362 "sarah_lee",
363 entities,
364 scope=scope,
365 origin=_origin("repair", "repair-run"),
366 )
367
368 assert row["resolved_entity_id"] == "sarah_lee"
369 prior_choices = row["audit"]["prior_choices"]
370 assert len(prior_choices) == 1
371 assert prior_choices[0]["resolved_entity_id"] == "sarah_connor"
372 assert prior_choices[0]["replaced_by_origin"]["lane"] == "repair"
373
374
375def test_stale_resolved_choice_fails_loudly_without_fallback(journal: Path) -> None:
376 entities = [_entity("Sarah Connor"), _entity("Sarah Lee")]
377 scope = ResolutionScope.journal()
378
379 _record("Sarah", entities, scope=scope)
380 record_ambiguity_choice("Sarah", "sarah_lee", entities, scope=scope)
381
382 with pytest.raises(EntityAmbiguityError):
383 _record("Sarah", [_entity("Sarah Connor")], scope=scope)
384
385 with pytest.raises(EntityAmbiguityError):
386 _record(
387 "Sarah",
388 [_entity("Sarah Lee", entity_id="sarah_lee", blocked=True)],
389 scope=scope,
390 )
391
392
393def test_resolved_choice_with_empty_entity_set_fails_loudly(
394 journal: Path,
395) -> None:
396 entities = [_entity("Sarah Connor"), _entity("Sarah Lee")]
397 scope = ResolutionScope.journal()
398
399 _record("Sarah", entities, scope=scope)
400 record_ambiguity_choice("Sarah", "sarah_lee", entities, scope=scope)
401
402 with pytest.raises(EntityAmbiguityError):
403 resolution = _record("Sarah", [], scope=scope)
404 assert resolution.outcome != EntityResolutionOutcome.NO_MATCH
405
406 assert not (journal / "entities" / "sarah" / "entity.json").exists()
407
408
409def test_lock_timeout_during_persistence_fails_closed(
410 journal: Path,
411 monkeypatch: pytest.MonkeyPatch,
412) -> None:
413 def raise_timeout(path: Path):
414 raise LockTimeout(path, 0.01)
415
416 monkeypatch.setattr(ambiguities, "hold_lock", raise_timeout)
417 before = _tree_snapshot(journal)
418
419 with pytest.raises(LockTimeout):
420 _record("Sarah", [_entity("Sarah Connor"), _entity("Sarah Lee")])
421
422 assert _tree_snapshot(journal) == before
423 assert not _ambiguity_file(journal).exists()
424
425
426def test_resolution_and_choice_use_trust_then_ambiguity_lock_order(
427 journal: Path,
428 monkeypatch: pytest.MonkeyPatch,
429) -> None:
430 entities = [_entity("Sarah Connor"), _entity("Sarah Lee")]
431 scope = ResolutionScope.journal()
432 events: list[str] = []
433
434 @contextmanager
435 def trust_lock(path: Path) -> Iterator[None]:
436 events.append("trust-enter")
437 try:
438 yield
439 finally:
440 events.append("trust-exit")
441
442 @contextmanager
443 def ambiguity_lock(path: Path) -> Iterator[None]:
444 events.append("ambiguity-enter")
445 try:
446 yield
447 finally:
448 events.append("ambiguity-exit")
449
450 monkeypatch.setattr(entity_history, "hold_lock", trust_lock)
451 monkeypatch.setattr(ambiguities, "hold_lock", ambiguity_lock)
452
453 _record("Sarah", entities, scope=scope)
454 assert events == [
455 "trust-enter",
456 "ambiguity-enter",
457 "ambiguity-exit",
458 "trust-exit",
459 ]
460
461 events.clear()
462 record_ambiguity_choice("Sarah", "sarah_lee", entities, scope=scope)
463 assert events == [
464 "trust-enter",
465 "ambiguity-enter",
466 "ambiguity-exit",
467 "trust-exit",
468 ]
469
470
471def test_entity_delete_uses_trust_boundary(
472 journal: Path,
473 monkeypatch: pytest.MonkeyPatch,
474) -> None:
475 entity_dir = journal / "entities" / "sarah_lee"
476 entity_dir.mkdir(parents=True)
477 (entity_dir / "entity.json").write_text(
478 json.dumps(_entity("Sarah Lee")),
479 encoding="utf-8",
480 )
481 events: list[str] = []
482
483 @contextmanager
484 def trust_lock(path: Path) -> Iterator[None]:
485 events.append("trust-enter")
486 try:
487 yield
488 finally:
489 events.append("trust-exit")
490
491 monkeypatch.setattr(entity_history, "hold_lock", trust_lock)
492
493 result = delete_journal_entity("sarah_lee")
494
495 assert result == {"success": True, "facets_deleted": []}
496 assert events == ["trust-enter", "trust-exit"]
497 assert not entity_dir.exists()
498
499
500@pytest.mark.parametrize(
501 "corrupt_line",
502 [
503 "{not-json",
504 "[]",
505 json.dumps({"schema_version": 1}),
506 ],
507)
508@pytest.mark.parametrize("position", ["before", "after", "only"])
509def test_mutation_resolution_fails_closed_on_corrupt_store_rows(
510 journal: Path,
511 corrupt_line: str,
512 position: str,
513) -> None:
514 entities = [_entity("Sarah Connor"), _entity("Sarah Lee")]
515 _record("Sarah", entities)
516 path = _ambiguity_file(journal)
517 valid = path.read_text(encoding="utf-8").rstrip("\n")
518 if position == "before":
519 content = f"{corrupt_line}\n{valid}\n"
520 elif position == "after":
521 content = f"{valid}\n{corrupt_line}\n"
522 else:
523 content = f"{corrupt_line}\n"
524 path.write_text(content, encoding="utf-8")
525 before = path.read_bytes()
526
527 with pytest.raises(EntityAmbiguityError):
528 _record("Sarah", entities, origin=_origin("retry", "retry-1"))
529
530 assert path.read_bytes() == before
531
532
533def test_read_only_list_remains_lenient_for_malformed_lines(journal: Path) -> None:
534 entities = [_entity("Sarah Connor"), _entity("Sarah Lee")]
535 _record("Sarah", entities)
536 path = _ambiguity_file(journal)
537 valid = path.read_text(encoding="utf-8")
538 path.write_text(f"{{not-json\n[]\n{valid}", encoding="utf-8")
539 before = path.read_bytes()
540
541 rows = load_ambiguities()
542
543 assert len(rows) == 1
544 assert rows[0]["normalized_query"] == "sarah"
545 assert path.read_bytes() == before
546
547
548def test_serialization_failure_is_typed_and_preserves_store(
549 journal: Path,
550 monkeypatch: pytest.MonkeyPatch,
551) -> None:
552 entities = [_entity("Sarah Connor"), _entity("Sarah Lee")]
553 scope = ResolutionScope.journal()
554 _record("Sarah", entities, scope=scope)
555 path = _ambiguity_file(journal)
556 before = path.read_bytes()
557
558 def fail_serialization(*args, **kwargs):
559 raise TypeError("injected serialization failure")
560
561 monkeypatch.setattr(ambiguities.json, "dumps", fail_serialization)
562
563 with pytest.raises(EntityAmbiguityError, match="cannot write"):
564 record_ambiguity_choice("Sarah", "sarah_lee", entities, scope=scope)
565
566 assert path.read_bytes() == before
567
568
569def test_strict_choice_read_oserror_is_typed_and_preserves_store(
570 journal: Path,
571 monkeypatch: pytest.MonkeyPatch,
572) -> None:
573 entities = [_entity("Sarah Connor"), _entity("Sarah Lee")]
574 path = _ambiguity_file(journal)
575 path.parent.mkdir(parents=True)
576 path.write_text("sentinel\n", encoding="utf-8")
577 before = path.read_bytes()
578
579 def fail_read(*args, **kwargs):
580 raise PermissionError("injected read failure")
581
582 monkeypatch.setattr(ambiguities, "open", fail_read, raising=False)
583
584 with pytest.raises(EntityAmbiguityError, match="cannot read"):
585 _record("Sarah", entities)
586
587 assert path.read_bytes() == before
588
589
590def test_trust_lock_timeout_fails_before_ambiguity_mutation(
591 journal: Path,
592 monkeypatch: pytest.MonkeyPatch,
593) -> None:
594 def raise_timeout(path: Path):
595 raise LockTimeout(path, 0.01)
596
597 monkeypatch.setattr(entity_history, "hold_lock", raise_timeout)
598 before = _tree_snapshot(journal)
599
600 with pytest.raises(LockTimeout):
601 _record("Sarah", [_entity("Sarah Connor"), _entity("Sarah Lee")])
602
603 assert _tree_snapshot(journal) == before
604 assert not _ambiguity_file(journal).exists()
605
606
607def test_per_invocation_memoization_keeps_occurrence_count_one(
608 journal: Path,
609) -> None:
610 entities = [_entity("Sarah Connor"), _entity("Sarah Lee")]
611 origin = _origin("participation", "one-talent-run")
612
613 for _ in range(5):
614 resolution = _record("Sarah", entities, origin=origin)
615 assert resolution.outcome == EntityResolutionOutcome.AMBIGUOUS
616
617 rows = load_ambiguities()
618 assert len(rows) == 1
619 assert rows[0]["occurrence_count"] == 1
620 assert len(rows[0]["origins"]) == 1
621
622
623def _function_calls_record_entity_resolution(
624 module_name: str, function_name: str
625) -> bool:
626 module = import_module(module_name)
627 tree = ast.parse(inspect.getsource(module))
628 for node in ast.walk(tree):
629 if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
630 continue
631 if node.name != function_name:
632 continue
633 for child in ast.walk(node):
634 if isinstance(child, ast.Call):
635 func = child.func
636 if isinstance(func, ast.Name) and func.id == "record_entity_resolution":
637 return True
638 if (
639 isinstance(func, ast.Attribute)
640 and func.attr == "record_entity_resolution"
641 ):
642 return True
643 return False
644
645
646def test_audited_mutation_owners_use_record_entity_resolution() -> None:
647 owners = [
648 ("solstone.talent.participation", "post_process"),
649 ("solstone.talent.schedule", "post_process"),
650 ("solstone.talent.story", "_resolve_entity_id"),
651 ("solstone.talent.speaker_attribution", "post_process"),
652 ("solstone.apps.activities.routes", "_resolve_participation_entity_ids"),
653 ("solstone.apps.entities.routes", "detect_entity_route"),
654 ("solstone.apps.entities.talent.entity_observer", "_clean_relation"),
655 ("solstone.apps.import.ingest", "ingest_entities"),
656 ("solstone.apps.speakers.attribution", "attribute_segment"),
657 ("solstone.apps.speakers.bootstrap", "bootstrap_voiceprints"),
658 ("solstone.apps.speakers.bootstrap", "merge_names"),
659 ("solstone.apps.speakers.bootstrap", "seed_from_imports"),
660 ("solstone.apps.speakers.discovery", "_plan_identify"),
661 ("solstone.think.entities.seeding", "seed_entities"),
662 ("solstone.think.merge", "_merge_entities"),
663 ]
664
665 missing = [
666 f"{module_name}:{function_name}"
667 for module_name, function_name in owners
668 if not _function_calls_record_entity_resolution(module_name, function_name)
669 ]
670 assert missing == []