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
6from pathlib import Path
7from typing import Any
8
9import numpy as np
10import pytest
11
12import solstone.think.curation as curation
13from solstone.apps.speakers.candidate_tracker import (
14 CandidateProfile,
15 CandidateTracker,
16 canonical_candidate_anchor,
17)
18from solstone.think.curation import (
19 KIND_ENTITY_AMBIGUITY,
20 KIND_SPEAKER_CANDIDATE_PAIR,
21 KIND_SPEAKER_NAME_VARIANT,
22 accept_entity_candidate,
23 accept_entity_candidate_batch,
24 accept_facet_candidate,
25 accept_speaker_candidate_pair,
26 dismiss_entity_candidate,
27 dismiss_entity_candidate_batch,
28 dismiss_facet_candidate,
29 dismiss_speaker_candidate_pair,
30 load_open_items,
31 merge_preview_fields,
32)
33from solstone.think.entities import (
34 EntityResolutionOutcome,
35 ResolutionOrigin,
36 ResolutionScope,
37 record_entity_resolution,
38 undo_entity_merge,
39)
40from solstone.think.entities.journal import load_journal_entity, save_journal_entity
41from solstone.think.entities.review_candidates import (
42 load_candidates as load_entity_candidates,
43)
44from solstone.think.entities.review_candidates import (
45 save_candidates as save_entity_candidates,
46)
47from solstone.think.facet_review_candidates import (
48 load_candidates as load_facet_candidates,
49)
50from solstone.think.facet_review_candidates import (
51 save_candidates as save_facet_candidates,
52)
53from solstone.think.indexer.edges import insert_edges
54from solstone.think.indexer.journal import get_journal_index
55from solstone.think.journal_io import LockTimeout
56from solstone.think.speaker_candidate_pair_review_candidates import (
57 load_candidates as load_pair_candidates,
58)
59from solstone.think.speaker_candidate_pair_review_candidates import (
60 record_candidate_pair,
61)
62from solstone.think.speaker_keep_separate import (
63 pair_key,
64 record_keep_separate_assertion,
65 remove_operation_sources,
66)
67from solstone.think.speaker_review_candidates import record_name_variant_candidate
68
69
70@pytest.fixture
71def curation_journal(monkeypatch, tmp_path) -> Path:
72 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
73 import solstone.think.utils as think_utils
74
75 think_utils._journal_path_cache = None
76 return Path(tmp_path)
77
78
79def _seed_entities() -> None:
80 save_journal_entity(
81 {
82 "id": "kognova_inc",
83 "name": "Kognova Inc",
84 "type": "Company",
85 "aka": ["Kognova Incorporated"],
86 }
87 )
88 save_journal_entity(
89 {
90 "id": "kognova",
91 "name": "Kognova",
92 "type": "Company",
93 "aka": [],
94 }
95 )
96
97
98def test_open_ambiguity_is_a_ranked_curation_item(curation_journal):
99 save_journal_entity(
100 {"id": "sarah_connor", "name": "Sarah Connor", "type": "Person"}
101 )
102 save_journal_entity({"id": "sarah_lee", "name": "Sarah Lee", "type": "Person"})
103 resolution = record_entity_resolution(
104 "Sarah",
105 [load_journal_entity("sarah_connor"), load_journal_entity("sarah_lee")],
106 scope=ResolutionScope.journal(),
107 origin=ResolutionOrigin(lane="test.curation", field="entity"),
108 )
109
110 assert resolution.outcome == EntityResolutionOutcome.AMBIGUOUS
111 item = next(
112 item for item in load_open_items() if item.kind == KIND_ENTITY_AMBIGUITY
113 )
114 assert item.key == resolution.ambiguity_id
115 assert item.name == "Sarah"
116 assert {row["id"] for row in item.evidence["ranked_candidates"]} == {
117 "sarah_connor",
118 "sarah_lee",
119 }
120 assert item.evidence["origins"] == [{"lane": "test.curation", "field": "entity"}]
121
122
123def _entity_candidate_row(
124 source_slug: str = "kognova_inc",
125 target_slug: str = "kognova",
126 *,
127 source: str = "Kognova Inc",
128 target: str = "Kognova",
129 status: str = "open",
130 detection_count: int = 4,
131) -> dict[str, Any]:
132 return {
133 "facet": "work",
134 "source": source,
135 "source_slug": source_slug,
136 "target": target,
137 "target_slug": target_slug,
138 "status": status,
139 "evidence": {
140 "basis": "name-variant",
141 "summary": f"{source} / {target}",
142 "detection_count": detection_count,
143 "needs": 0,
144 },
145 }
146
147
148def _seed_entity_candidates(rows: list[dict[str, Any]]) -> None:
149 save_entity_candidates(rows)
150
151
152def _insert_edges(journal: Path, rows: list[dict[str, Any]]) -> None:
153 conn, _ = get_journal_index(str(journal))
154 try:
155 insert_edges(conn, rows)
156 conn.commit()
157 finally:
158 conn.close()
159
160
161def _edge(
162 src: str,
163 dst: str,
164 path: str,
165 *,
166 day: str | None = "20260601",
167) -> dict[str, Any]:
168 return {
169 "src": src,
170 "dst": dst,
171 "kind": "works-with",
172 "day": day,
173 "facet": "work",
174 "source": "curation-test",
175 "path": path,
176 "weight": 1,
177 }
178
179
180def _tree_snapshot(root: Path) -> dict[str, bytes]:
181 return {
182 str(path.relative_to(root)): path.read_bytes()
183 for path in sorted(root.rglob("*"))
184 if path.is_file()
185 }
186
187
188def _seed_entity_candidate(status: str = "open", detection_count: int = 4) -> None:
189 _seed_entity_candidates(
190 [_entity_candidate_row(status=status, detection_count=detection_count)]
191 )
192
193
194def _seed_entity_pair(
195 source_slug: str,
196 source_name: str,
197 target_slug: str,
198 target_name: str,
199) -> None:
200 save_journal_entity(
201 {
202 "id": source_slug,
203 "name": source_name,
204 "type": "Company",
205 "aka": [],
206 }
207 )
208 save_journal_entity(
209 {
210 "id": target_slug,
211 "name": target_name,
212 "type": "Company",
213 "aka": [],
214 }
215 )
216
217
218def _assert_folded(source_slug: str, target_slug: str, source_name: str) -> None:
219 assert load_journal_entity(source_slug) is None
220 target = load_journal_entity(target_slug)
221 assert target is not None
222 assert source_name in target["aka"]
223
224
225def _unit(vector: list[float]) -> np.ndarray:
226 embedding = np.array(vector + [0.0] * (256 - len(vector)), dtype=np.float32)
227 return embedding / np.linalg.norm(embedding)
228
229
230def _source_segment(day: str, cluster_label: int = 1) -> dict[str, object]:
231 return {
232 "day": day,
233 "segment_key": "090000_300",
234 "stream": "test",
235 "source": "mic_audio",
236 "cluster_label": cluster_label,
237 }
238
239
240def _candidate_profile(
241 cand_id: int,
242 centroid: np.ndarray,
243 *,
244 status: str = "pending",
245 confirmed_entity: str | None = None,
246) -> CandidateProfile:
247 source_segment = _source_segment(f"2026010{cand_id}", cand_id)
248 return CandidateProfile(
249 cand_id=cand_id,
250 centroid=centroid,
251 n_segments=1,
252 n_intervals=30,
253 total_duration_s=30.0,
254 source_segments=[source_segment],
255 status=status,
256 confirmed_entity=confirmed_entity,
257 )
258
259
260def _seed_candidate_tracker(
261 journal: Path,
262 candidates: list[CandidateProfile],
263) -> CandidateTracker:
264 tracker = CandidateTracker(journal / "awareness" / "speaker_candidates.json")
265 tracker._candidates = {candidate.cand_id: candidate for candidate in candidates}
266 tracker._next_id = (
267 max((candidate.cand_id for candidate in candidates), default=0) + 1
268 )
269 tracker.save()
270 return CandidateTracker(journal / "awareness" / "speaker_candidates.json")
271
272
273def _record_pair_for_candidates(
274 left: CandidateProfile,
275 right: CandidateProfile,
276 *,
277 similarity: float = 0.70,
278) -> tuple[str, str]:
279 anchor_a = canonical_candidate_anchor(left)
280 anchor_b = canonical_candidate_anchor(right)
281 record_candidate_pair(
282 source_anchor=anchor_a,
283 target_anchor=anchor_b,
284 source_anchors={anchor_a},
285 target_anchors={anchor_b},
286 similarity=similarity,
287 source_intervals=left.n_intervals,
288 target_intervals=right.n_intervals,
289 source_samples=[],
290 target_samples=[],
291 )
292 return anchor_a, anchor_b
293
294
295def test_load_open_items_normalizes_and_orders(curation_journal):
296 save_facet_candidates(
297 [
298 {
299 "name": "Home Reno",
300 "name_key": "home reno",
301 "status": "open",
302 "count": 3,
303 "window_days": 14,
304 "evidence": {"samples": [{"day": "20260602"}]},
305 },
306 {
307 "name": "Done",
308 "name_key": "done",
309 "status": "dismissed",
310 "count": 9,
311 },
312 ]
313 )
314 _seed_entity_candidate(detection_count=5)
315
316 items = load_open_items()
317
318 assert [item.key for item in items] == ["work|kognova_inc|kognova", "home reno"]
319 assert items[0].kind == "entity_merge"
320 assert items[0].strength == 5
321 assert items[1].kind == "facet_candidate"
322 assert items[1].evidence["count"] == 3
323
324
325def test_load_open_items_includes_speaker_name_variant(curation_journal):
326 record_name_variant_candidate(
327 source_id="alice",
328 source_label="Alice",
329 target_id="alice_johnson",
330 target_label="Alice Johnson",
331 similarity=0.934,
332 )
333
334 items = load_open_items()
335
336 assert len(items) == 1
337 item = items[0]
338 assert item.kind == KIND_SPEAKER_NAME_VARIANT
339 assert item.key == "alice|alice_johnson"
340 assert item.name is None
341 assert item.facet is None
342 assert item.source == "Alice"
343 assert item.source_slug == "alice"
344 assert item.target == "Alice Johnson"
345 assert item.target_slug == "alice_johnson"
346 assert item.evidence["similarity"] == 0.934
347 assert item.evidence["readiness"] == "ready"
348 assert item.strength == 93
349
350
351def test_load_open_items_filters_keep_separate_speaker_name_variant(curation_journal):
352 record_name_variant_candidate(
353 source_id="alice",
354 source_label="Alice",
355 target_id="alice_johnson",
356 target_label="Alice Johnson",
357 similarity=0.934,
358 )
359 record_keep_separate_assertion(
360 "alice",
361 "alice_johnson",
362 source_kind="explicit_create_near_match",
363 operation_id="idop_test",
364 detection_count=1,
365 )
366
367 items = load_open_items()
368
369 assert [item for item in items if item.kind == KIND_SPEAKER_NAME_VARIANT] == []
370
371
372def test_load_open_items_resurfaces_expired_keep_separate_suppression(
373 curation_journal,
374):
375 operation_id = "idop_test"
376 record_keep_separate_assertion(
377 "alice",
378 "alice_johnson",
379 source_kind="explicit_create_near_match",
380 operation_id=operation_id,
381 detection_count=1,
382 )
383 row, created, suppressed = record_name_variant_candidate(
384 source_id="alice",
385 source_label="Alice",
386 target_id="alice_johnson",
387 target_label="Alice Johnson",
388 similarity=0.934,
389 )
390
391 assert created is True
392 assert suppressed is True
393 assert row["status"] == "suppressed"
394 assert row["suppressed_by_keep_separate"] is True
395 assert [
396 item for item in load_open_items() if item.kind == KIND_SPEAKER_NAME_VARIANT
397 ] == []
398
399 remove_operation_sources(operation_id, [pair_key("alice", "alice_johnson")])
400
401 items = [
402 item for item in load_open_items() if item.kind == KIND_SPEAKER_NAME_VARIANT
403 ]
404 assert len(items) == 1
405 assert items[0].key == "alice|alice_johnson"
406
407
408def test_load_open_items_includes_speaker_candidate_pair(curation_journal):
409 left = _candidate_profile(1, _unit([1.0, 0.0]))
410 right = _candidate_profile(2, _unit([0.62, np.sqrt(1.0 - 0.62**2)]))
411 anchor_a, anchor_b = _record_pair_for_candidates(left, right, similarity=0.62)
412
413 items = load_open_items()
414
415 assert len(items) == 1
416 item = items[0]
417 assert item.kind == KIND_SPEAKER_CANDIDATE_PAIR
418 assert item.key == curation._speaker_candidate_pair_key(anchor_a, anchor_b)
419 assert item.source_slug == anchor_a
420 assert item.target_slug == anchor_b
421 assert item.source == "candidate A"
422 assert item.target == "candidate B"
423 assert item.evidence["similarity"] == 0.62
424 assert item.evidence["source_intervals"] == 30
425 assert item.evidence["target_intervals"] == 30
426 assert item.strength == 62
427
428
429def test_entity_merge_equal_strength_uses_shared_neighborhood(curation_journal):
430 _seed_entity_candidates(
431 [
432 _entity_candidate_row(
433 source_slug="alpha_source",
434 target_slug="alpha_target",
435 source="Alpha Source",
436 target="Alpha Target",
437 detection_count=5,
438 ),
439 _entity_candidate_row(
440 source_slug="zeta_source",
441 target_slug="zeta_target",
442 source="Zeta Source",
443 target="Zeta Target",
444 detection_count=5,
445 ),
446 ]
447 )
448 _insert_edges(
449 curation_journal,
450 [
451 _edge("zeta_source", "shared_one", "curation/zeta-shared-one-a"),
452 _edge("zeta_target", "shared_one", "curation/zeta-shared-one-b"),
453 _edge("zeta_source", "shared_two", "curation/zeta-shared-two-a"),
454 _edge("zeta_target", "shared_two", "curation/zeta-shared-two-b"),
455 _edge("zeta_source", "source_only", "curation/zeta-source-only"),
456 _edge("zeta_target", "target_only", "curation/zeta-target-only"),
457 ],
458 )
459
460 items = load_open_items()
461
462 assert [item.key for item in items] == [
463 "work|zeta_source|zeta_target",
464 "work|alpha_source|alpha_target",
465 ]
466 assert items[0].strength == 5
467 assert items[0].evidence["shared_neighbors"] == ["shared_one", "shared_two"]
468 assert items[0].evidence["neighborhood_similarity"] == pytest.approx(0.5)
469 assert items[0].evidence["composite"] == pytest.approx(5.125)
470 assert items[0].composite == pytest.approx(5.125)
471 assert items[0].to_dict()["composite"] == pytest.approx(5.125)
472
473
474def test_entity_merge_detection_strength_stays_primary(curation_journal):
475 _seed_entity_candidates(
476 [
477 _entity_candidate_row(
478 source_slug="alpha_high",
479 target_slug="alpha_target",
480 source="Alpha High",
481 target="Alpha Target",
482 detection_count=6,
483 ),
484 _entity_candidate_row(
485 source_slug="zeta_low",
486 target_slug="zeta_target",
487 source="Zeta Low",
488 target="Zeta Target",
489 detection_count=5,
490 ),
491 ]
492 )
493 _insert_edges(
494 curation_journal,
495 [
496 _edge("zeta_low", "shared_one", "curation/low-shared-one-a"),
497 _edge("zeta_target", "shared_one", "curation/low-shared-one-b"),
498 _edge("zeta_low", "shared_two", "curation/low-shared-two-a"),
499 _edge("zeta_target", "shared_two", "curation/low-shared-two-b"),
500 ],
501 )
502
503 items = load_open_items()
504
505 assert [item.key for item in items] == [
506 "work|alpha_high|alpha_target",
507 "work|zeta_low|zeta_target",
508 ]
509 assert items[0].strength == 6
510 assert items[0].composite == 6.0
511 assert items[1].strength == 5
512 assert items[1].composite == pytest.approx(5.25)
513
514
515def test_load_open_items_missing_edge_index_degrades_without_mutation(
516 curation_journal,
517):
518 _seed_entity_candidates(
519 [
520 _entity_candidate_row(
521 source_slug="zeta_source",
522 target_slug="zeta_target",
523 source="Zeta Source",
524 target="Zeta Target",
525 detection_count=3,
526 ),
527 _entity_candidate_row(
528 source_slug="alpha_source",
529 target_slug="alpha_target",
530 source="Alpha Source",
531 target="Alpha Target",
532 detection_count=3,
533 ),
534 _entity_candidate_row(
535 source_slug="middle_source",
536 target_slug="middle_target",
537 source="Middle Source",
538 target="Middle Target",
539 detection_count=4,
540 ),
541 ]
542 )
543 before = _tree_snapshot(curation_journal)
544
545 items = load_open_items()
546
547 assert _tree_snapshot(curation_journal) == before
548 assert [item.key for item in items] == [
549 "work|middle_source|middle_target",
550 "work|alpha_source|alpha_target",
551 "work|zeta_source|zeta_target",
552 ]
553 assert [item.key for item in items] == [
554 item.key for item in sorted(items, key=lambda item: (-item.strength, item.key))
555 ]
556 assert [item.composite for item in items] == [4.0, 3.0, 3.0]
557
558
559def test_accept_facet_candidate_creates_facet_then_marks_accepted(curation_journal):
560 save_facet_candidates(
561 [
562 {
563 "name": "Home Reno",
564 "name_key": "home reno",
565 "status": "open",
566 "count": 3,
567 }
568 ]
569 )
570
571 result = accept_facet_candidate("home reno")
572
573 assert result["status"] == "accepted"
574 assert result["facet_slug"] == "home-reno"
575 assert (curation_journal / "facets" / "home-reno" / "facet.json").exists()
576 assert load_facet_candidates()[0]["status"] == "accepted"
577
578
579def test_accept_facet_candidate_is_idempotent_when_already_accepted(curation_journal):
580 save_facet_candidates(
581 [
582 {
583 "name": "Home Reno",
584 "name_key": "home reno",
585 "status": "accepted",
586 "count": 3,
587 }
588 ]
589 )
590
591 result = accept_facet_candidate("home reno")
592
593 assert result["status"] == "already_accepted"
594 assert not (curation_journal / "facets" / "home-reno").exists()
595
596
597def test_accept_facet_candidate_duplicate_keeps_candidate_open(curation_journal):
598 save_facet_candidates(
599 [
600 {
601 "name": "Home Reno",
602 "name_key": "home reno",
603 "status": "open",
604 "count": 3,
605 }
606 ]
607 )
608 first = accept_facet_candidate("home reno")
609 save_facet_candidates(
610 [
611 {
612 "name": "Home Reno",
613 "name_key": "home reno",
614 "status": "open",
615 "count": 3,
616 }
617 ]
618 )
619
620 second = accept_facet_candidate("home reno")
621
622 assert first["status"] == "accepted"
623 assert second["status"] == "error"
624 assert "already exists" in second["error"]
625 assert load_facet_candidates()[0]["status"] == "open"
626
627
628def test_dismiss_facet_candidate_sets_watermark_and_is_idempotent(curation_journal):
629 save_facet_candidates(
630 [
631 {
632 "name": "Home Reno",
633 "name_key": "home reno",
634 "status": "open",
635 "count": 4,
636 }
637 ]
638 )
639
640 first = dismiss_facet_candidate("home reno")
641 second = dismiss_facet_candidate("home reno")
642
643 assert first["status"] == "dismissed"
644 assert second["status"] == "already_dismissed"
645 assert load_facet_candidates()[0]["dismissed_count"] == 4
646
647
648def test_entity_preview_is_read_only(curation_journal):
649 _seed_entities()
650 _seed_entity_candidate()
651
652 result = accept_entity_candidate(
653 "work",
654 "kognova_inc",
655 "kognova",
656 commit=False,
657 )
658
659 assert result["status"] == "preview"
660 assert result["merge"]["would_identity"]["akas_added"] == [
661 "Kognova Inc",
662 "Kognova Incorporated",
663 ]
664 assert load_entity_candidates()[0]["status"] == "open"
665
666
667def test_accept_entity_candidate_commits_then_marks_accepted(curation_journal):
668 _seed_entities()
669 _seed_entity_candidate()
670
671 result = accept_entity_candidate(
672 "work",
673 "kognova_inc",
674 "kognova",
675 commit=True,
676 )
677
678 assert result["status"] == "accepted"
679 assert result["merge"]["merged"] is True
680 assert load_entity_candidates()[0]["status"] == "accepted"
681
682
683def test_accept_entity_candidate_error_keeps_status_open(curation_journal):
684 _seed_entity_candidate()
685
686 result = accept_entity_candidate(
687 "work",
688 "kognova_inc",
689 "kognova",
690 commit=True,
691 )
692
693 assert result["status"] == "error"
694 assert "Source entity not found" in result["error"]
695 assert load_entity_candidates()[0]["status"] == "open"
696
697
698def test_repair_required_merge_state_survives_single_and_batch_results(
699 curation_journal,
700 monkeypatch: pytest.MonkeyPatch,
701):
702 _seed_entity_candidate()
703 repair = {
704 "error": {"code": "repair_required", "message": "rollback failed"},
705 "operation_state": "repair_required",
706 "mutation_applied": True,
707 "source_state": {"exists": True},
708 "target_state": {"exists": True},
709 "safe_remediation": "Inspect before retrying.",
710 }
711 monkeypatch.setattr(curation, "merge_entity", lambda *args, **kwargs: repair)
712
713 single = accept_entity_candidate("work", "kognova_inc", "kognova", commit=True)
714 batch = accept_entity_candidate_batch(
715 [
716 {
717 "facet": "work",
718 "source_slug": "kognova_inc",
719 "target_slug": "kognova",
720 }
721 ]
722 )
723
724 assert single["error"] == "rollback failed"
725 assert single["operation_state"] == "repair_required"
726 assert single["mutation_applied"] is True
727 assert single["safe_remediation"] == "Inspect before retrying."
728 assert batch["results"][0]["operation_state"] == "repair_required"
729 assert batch["results"][0]["safe_remediation"] == "Inspect before retrying."
730
731
732def test_dismiss_entity_candidate_sets_watermark_and_is_idempotent(curation_journal):
733 _seed_entity_candidate()
734
735 first = dismiss_entity_candidate("work", "kognova_inc", "kognova")
736 second = dismiss_entity_candidate("work", "kognova_inc", "kognova")
737
738 assert first["status"] == "dismissed"
739 assert second["status"] == "already_dismissed"
740 assert load_entity_candidates()[0]["dismissed_detection_count"] == 4
741
742
743def test_accept_entity_candidate_batch_accepts_many_in_order(curation_journal):
744 _seed_entity_pair("kognova_inc", "Kognova Inc", "kognova", "Kognova")
745 _seed_entity_pair("octo_labs_inc", "Octo Labs Inc", "octo_labs", "Octo Labs")
746 _seed_entity_candidates(
747 [
748 _entity_candidate_row(),
749 _entity_candidate_row(
750 "octo_labs_inc",
751 "octo_labs",
752 source="Octo Labs Inc",
753 target="Octo Labs",
754 detection_count=3,
755 ),
756 ]
757 )
758
759 result = accept_entity_candidate_batch(
760 [
761 {"facet": "work", "source_slug": "kognova_inc", "target_slug": "kognova"},
762 {
763 "facet": "work",
764 "source_slug": "octo_labs_inc",
765 "target_slug": "octo_labs",
766 },
767 ]
768 )
769
770 assert result["accepted"] == 2
771 assert result["failed"] == 0
772 assert len(result["results"]) == 2
773 assert [row["source_slug"] for row in result["results"]] == [
774 "kognova_inc",
775 "octo_labs_inc",
776 ]
777 assert all(
778 set(row)
779 == {
780 "facet",
781 "source_slug",
782 "target_slug",
783 "status",
784 "error",
785 "merge_id",
786 "undo",
787 }
788 for row in result["results"]
789 )
790 assert all(row["merge_id"].startswith("em_") for row in result["results"])
791 assert all(row["undo"]["available"] for row in result["results"])
792 assert all(
793 "merge" not in row and "candidate" not in row for row in result["results"]
794 )
795 _assert_folded("kognova_inc", "kognova", "Kognova Inc")
796 _assert_folded("octo_labs_inc", "octo_labs", "Octo Labs Inc")
797
798
799def test_batch_merge_undo_one_preserves_later_success_on_same_target(
800 curation_journal,
801):
802 save_journal_entity(
803 {
804 "id": "alpha_inc",
805 "name": "Alpha Inc",
806 "type": "Company",
807 "aka": ["Alpha Alias"],
808 }
809 )
810 save_journal_entity(
811 {
812 "id": "beta_inc",
813 "name": "Beta Inc",
814 "type": "Company",
815 "aka": ["Beta Alias"],
816 }
817 )
818 save_journal_entity(
819 {"id": "anchor", "name": "Anchor", "type": "Company", "aka": []}
820 )
821 _seed_entity_candidates(
822 [
823 _entity_candidate_row(
824 source_slug="alpha_inc",
825 target_slug="anchor",
826 source="Alpha Inc",
827 target="Anchor",
828 ),
829 _entity_candidate_row(
830 source_slug="beta_inc",
831 target_slug="anchor",
832 source="Beta Inc",
833 target="Anchor",
834 ),
835 ]
836 )
837
838 batch = accept_entity_candidate_batch(
839 [
840 {
841 "facet": "work",
842 "source_slug": "alpha_inc",
843 "target_slug": "anchor",
844 },
845 {
846 "facet": "work",
847 "source_slug": "beta_inc",
848 "target_slug": "anchor",
849 },
850 ]
851 )
852 undone = undo_entity_merge(batch["results"][0]["merge_id"], caller="test.curation")
853
854 assert undone["undone"] is True
855 assert load_journal_entity("alpha_inc")["name"] == "Alpha Inc"
856 assert load_journal_entity("beta_inc") is None
857 anchor = load_journal_entity("anchor")
858 assert "Beta Inc" in anchor["aka"]
859 assert "Beta Alias" in anchor["aka"]
860 assert "Alpha Inc" not in anchor["aka"]
861 assert "Alpha Alias" not in anchor["aka"]
862
863
864def test_accept_entity_candidate_batch_malformed_first_continues(curation_journal):
865 _seed_entities()
866 _seed_entity_candidate()
867
868 result = accept_entity_candidate_batch(
869 [
870 {"facet": "work", "source_slug": "missing_target"},
871 {"facet": "work", "source_slug": "kognova_inc", "target_slug": "kognova"},
872 ]
873 )
874
875 assert result["accepted"] == 1
876 assert result["failed"] == 1
877 assert len(result["results"]) == 2
878 assert result["results"][0] == {
879 "facet": "work",
880 "source_slug": "missing_target",
881 "target_slug": "",
882 "status": "error",
883 "error": "candidate is missing facet, source_slug, or target_slug",
884 }
885 _assert_folded("kognova_inc", "kognova", "Kognova Inc")
886
887
888def test_accept_entity_candidate_batch_lock_timeout_continues(
889 curation_journal,
890 monkeypatch,
891):
892 _seed_entities()
893 _seed_entity_candidates(
894 [
895 _entity_candidate_row(
896 "busy_inc",
897 "busy",
898 source="Busy Inc",
899 target="Busy",
900 ),
901 _entity_candidate_row(),
902 ]
903 )
904 real_merge = curation.merge_entity
905
906 def raise_busy_for_one(source_id: str, target_id: str, **kwargs):
907 if source_id == "busy_inc":
908 raise LockTimeout(Path("busy"), 0.01)
909 return real_merge(source_id, target_id, **kwargs)
910
911 monkeypatch.setattr(curation, "merge_entity", raise_busy_for_one)
912
913 result = accept_entity_candidate_batch(
914 [
915 {"facet": "work", "source_slug": "busy_inc", "target_slug": "busy"},
916 {"facet": "work", "source_slug": "kognova_inc", "target_slug": "kognova"},
917 ]
918 )
919
920 assert result["accepted"] == 1
921 assert result["failed"] == 1
922 assert result["results"][0]["status"] == "error"
923 assert (
924 result["results"][0]["error"] == "entity merge candidates are busy; try again"
925 )
926 _assert_folded("kognova_inc", "kognova", "Kognova Inc")
927
928
929def test_accept_entity_candidate_batch_counts_already_accepted(curation_journal):
930 _seed_entities()
931 _seed_entity_candidates(
932 [
933 _entity_candidate_row(
934 "accepted_inc",
935 "accepted",
936 source="Accepted Inc",
937 target="Accepted",
938 status="accepted",
939 ),
940 _entity_candidate_row(),
941 ]
942 )
943
944 result = accept_entity_candidate_batch(
945 [
946 {
947 "facet": "work",
948 "source_slug": "accepted_inc",
949 "target_slug": "accepted",
950 },
951 {"facet": "work", "source_slug": "kognova_inc", "target_slug": "kognova"},
952 ]
953 )
954
955 assert result["accepted"] == 2
956 assert result["failed"] == 0
957 assert [row["status"] for row in result["results"]] == [
958 "already_accepted",
959 "accepted",
960 ]
961 _assert_folded("kognova_inc", "kognova", "Kognova Inc")
962
963
964def test_dismiss_entity_candidate_batch_dismisses_many(curation_journal):
965 _seed_entity_candidates(
966 [
967 _entity_candidate_row(),
968 _entity_candidate_row(
969 "octo_labs_inc",
970 "octo_labs",
971 source="Octo Labs Inc",
972 target="Octo Labs",
973 ),
974 ]
975 )
976
977 result = dismiss_entity_candidate_batch(
978 [
979 {"facet": "work", "source_slug": "kognova_inc", "target_slug": "kognova"},
980 {
981 "facet": "work",
982 "source_slug": "octo_labs_inc",
983 "target_slug": "octo_labs",
984 },
985 ]
986 )
987
988 assert result["dismissed"] == 2
989 assert result["failed"] == 0
990 assert len(result["results"]) == 2
991 assert all(
992 set(row) == {"facet", "source_slug", "target_slug", "status", "error"}
993 for row in result["results"]
994 )
995 assert [row["status"] for row in load_entity_candidates()] == [
996 "dismissed",
997 "dismissed",
998 ]
999
1000
1001def test_dismiss_entity_candidate_batch_malformed_first_continues(curation_journal):
1002 _seed_entity_candidate()
1003
1004 result = dismiss_entity_candidate_batch(
1005 [
1006 {"facet": "work", "source_slug": "missing_target"},
1007 {"facet": "work", "source_slug": "kognova_inc", "target_slug": "kognova"},
1008 ]
1009 )
1010
1011 assert result["dismissed"] == 1
1012 assert result["failed"] == 1
1013 assert result["results"][0]["status"] == "error"
1014 assert result["results"][0]["error"] == (
1015 "candidate is missing facet, source_slug, or target_slug"
1016 )
1017 assert load_entity_candidates()[0]["status"] == "dismissed"
1018
1019
1020def test_dismiss_entity_candidate_batch_counts_already_dismissed(curation_journal):
1021 _seed_entity_candidates(
1022 [
1023 _entity_candidate_row(
1024 "dismissed_inc",
1025 "dismissed",
1026 source="Dismissed Inc",
1027 target="Dismissed",
1028 status="dismissed",
1029 ),
1030 _entity_candidate_row(),
1031 ]
1032 )
1033
1034 result = dismiss_entity_candidate_batch(
1035 [
1036 {
1037 "facet": "work",
1038 "source_slug": "dismissed_inc",
1039 "target_slug": "dismissed",
1040 },
1041 {"facet": "work", "source_slug": "kognova_inc", "target_slug": "kognova"},
1042 ]
1043 )
1044
1045 assert result["dismissed"] == 2
1046 assert result["failed"] == 0
1047 assert [row["status"] for row in result["results"]] == [
1048 "already_dismissed",
1049 "dismissed",
1050 ]
1051
1052
1053def test_accept_speaker_candidate_pair_merges_tracker_without_entity_merge(
1054 curation_journal,
1055 monkeypatch,
1056):
1057 calls: list[tuple[tuple[Any, ...], dict[str, Any]]] = []
1058
1059 def merge_entity_spy(*args: Any, **kwargs: Any) -> dict[str, Any]:
1060 calls.append((args, kwargs))
1061 raise AssertionError("speaker candidate-pair accept must not merge entities")
1062
1063 monkeypatch.setattr(curation, "merge_entity", merge_entity_spy)
1064 left = _candidate_profile(1, _unit([1.0, 0.0]))
1065 right = _candidate_profile(
1066 2,
1067 _unit([0.70, np.sqrt(1.0 - 0.70**2)]),
1068 status="confirmed",
1069 confirmed_entity="alice_test",
1070 )
1071 tracker = _seed_candidate_tracker(curation_journal, [left, right])
1072 anchor_a, anchor_b = _record_pair_for_candidates(
1073 tracker.load_all_candidates()[0],
1074 tracker.load_all_candidates()[1],
1075 )
1076
1077 result = accept_speaker_candidate_pair(anchor_b, anchor_a)
1078
1079 assert calls == []
1080 assert result["status"] == "accepted"
1081 assert result["kind"] == KIND_SPEAKER_CANDIDATE_PAIR
1082 assert result["merge"]["status"] == "merged"
1083 assert result["undo"] == {
1084 "available": False,
1085 "merge_id": None,
1086 "reason": "Speaker candidate-pair merges cannot be undone.",
1087 }
1088 survivor = CandidateTracker(
1089 curation_journal / "awareness" / "speaker_candidates.json"
1090 ).load_all_candidates()
1091 assert len(survivor) == 1
1092 assert survivor[0].cand_id == 1
1093 assert survivor[0].status == "confirmed"
1094 assert survivor[0].confirmed_entity == "alice_test"
1095 assert load_pair_candidates()[0]["status"] == "accepted"
1096
1097
1098def test_accept_speaker_candidate_pair_stale_tracker_error_keeps_row_open(
1099 curation_journal,
1100 monkeypatch,
1101):
1102 calls: list[tuple[tuple[Any, ...], dict[str, Any]]] = []
1103 monkeypatch.setattr(
1104 curation,
1105 "merge_entity",
1106 lambda *args, **kwargs: calls.append((args, kwargs)),
1107 )
1108 left = _candidate_profile(1, _unit([1.0, 0.0]))
1109 right = _candidate_profile(2, _unit([0.70, np.sqrt(1.0 - 0.70**2)]))
1110 anchor_a, anchor_b = _record_pair_for_candidates(left, right)
1111
1112 result = accept_speaker_candidate_pair(anchor_a, anchor_b)
1113
1114 assert calls == []
1115 assert result == {
1116 "status": "error",
1117 "kind": KIND_SPEAKER_CANDIDATE_PAIR,
1118 "key": curation._speaker_candidate_pair_key(anchor_a, anchor_b),
1119 "error": "candidate anchor not found",
1120 }
1121 assert load_pair_candidates()[0]["status"] == "open"
1122 assert (
1123 CandidateTracker(
1124 curation_journal / "awareness" / "speaker_candidates.json"
1125 ).load_all_candidates()
1126 == []
1127 )
1128
1129
1130def test_dismiss_speaker_candidate_pair_sets_status_and_removes_open_item(
1131 curation_journal,
1132):
1133 left = _candidate_profile(1, _unit([1.0, 0.0]))
1134 right = _candidate_profile(2, _unit([0.62, np.sqrt(1.0 - 0.62**2)]))
1135 anchor_a, anchor_b = _record_pair_for_candidates(left, right, similarity=0.62)
1136
1137 result = dismiss_speaker_candidate_pair(anchor_b, anchor_a)
1138
1139 assert result["status"] == "dismissed"
1140 assert result["kind"] == KIND_SPEAKER_CANDIDATE_PAIR
1141 assert load_pair_candidates()[0]["status"] == "dismissed"
1142 assert [
1143 item for item in load_open_items() if item.kind == KIND_SPEAKER_CANDIDATE_PAIR
1144 ] == []
1145
1146
1147def test_merge_preview_fields_returns_compact_summary():
1148 fields = merge_preview_fields(
1149 {
1150 "would_identity": {
1151 "akas_added": ["Kognova Inc"],
1152 "emails_added_count": 1,
1153 },
1154 "would_facets": {
1155 "moved_count": 2,
1156 "merged_count": 3,
1157 "observations_appended": 4,
1158 },
1159 "would_segments": {
1160 "labels_rewritten": 5,
1161 "corrections_rewritten": 6,
1162 "errors": [{"message": "bad"}],
1163 },
1164 "would_voiceprints": {
1165 "added": 7,
1166 "target_total": 8,
1167 },
1168 }
1169 )
1170
1171 assert fields == {
1172 "akas_added": ["Kognova Inc"],
1173 "emails_added_count": 1,
1174 "facet_moved_count": 2,
1175 "facet_merged_count": 3,
1176 "observations_appended": 4,
1177 "labels_rewritten": 5,
1178 "corrections_rewritten": 6,
1179 "segment_errors": [{"message": "bad"}],
1180 "voiceprints_added": 7,
1181 "voiceprints_target_total": 8,
1182 }