solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4"""Tests for the speaker verdict replay harness."""
5
6from __future__ import annotations
7
8import json
9import socket
10from pathlib import Path
11from typing import Any
12
13import numpy as np
14import pytest
15
16from solstone.apps.speakers import encoder_config
17from solstone.observe.transcribe import diarize
18from tests import verify_speaker_differential as harness
19from tests import verify_speaker_verdict as verdict
20from tests._repo_inventory import assert_inventory_unchanged, repository_inventory
21from tests._speaker_differential_fixtures import EMBEDDING_MAX_ABS_TOLERANCE
22from tests.test_speaker_differential import _emit_model_free_bundle
23
24
25def _eps(*values: float) -> float:
26 positives = [abs(float(value)) for value in values if float(value) > 0]
27 return min(positives) / 10
28
29
30def _unit_vector(components: list[float]) -> np.ndarray:
31 remainder = 1.0 - sum(float(value) * float(value) for value in components)
32 assert remainder > 0
33 return np.array([*components, np.sqrt(remainder)], dtype=np.float32)
34
35
36def _axis(index: int, width: int) -> np.ndarray:
37 vector = np.zeros(width, dtype=np.float32)
38 vector[index] = 1.0
39 return vector
40
41
42def _statement_bundle(
43 rows: list[np.ndarray], ids: list[int] | None = None
44) -> harness.Bundle:
45 bundle = harness.new_bundle(producer="test")
46 statement_ids = np.array(
47 ids if ids is not None else list(range(1, len(rows) + 1)),
48 dtype=np.int32,
49 )
50 harness._set_array(
51 bundle,
52 harness.STATEMENT_EMBEDDING_IDS,
53 statement_ids,
54 "statement_embeddings",
55 )
56 harness._set_array(
57 bundle,
58 harness.STATEMENT_EMBEDDINGS,
59 np.asarray(rows, dtype=np.float32),
60 "statement_embeddings",
61 )
62 return bundle
63
64
65def _prediction_bundle(
66 rows: list[tuple[int, float, float, int | np.integer]],
67) -> harness.Bundle:
68 bundle = harness.new_bundle(producer="test")
69 harness._set_array(
70 bundle,
71 harness.INPUT_DIARIZATION_IDS,
72 np.array([row[0] for row in rows], dtype=np.int32),
73 "inputs.diarization",
74 )
75 harness._set_array(
76 bundle,
77 harness.INPUT_DIARIZATION_SPANS,
78 np.array([[row[1], row[2]] for row in rows], dtype=np.float32),
79 "inputs.diarization",
80 )
81 harness._set_array(
82 bundle,
83 harness.DIARIZATION_STATEMENT_LABELS,
84 np.array([row[3] for row in rows], dtype=np.int32),
85 "statement_labels",
86 )
87 return bundle
88
89
90def _cluster_bundle(
91 raw_embeddings: np.ndarray, silhouette_k: int, effective_k: int
92) -> harness.Bundle:
93 bundle = harness.new_bundle(producer="test")
94 harness._set_array(
95 bundle,
96 harness.DIARIZATION_INTERVAL_EMBEDDINGS,
97 raw_embeddings.astype(np.float32),
98 "interval_embeddings",
99 )
100 harness._set_scalar(
101 bundle,
102 harness.DIARIZATION_SILHOUETTE_K,
103 silhouette_k,
104 "clustering",
105 )
106 harness._set_scalar(
107 bundle,
108 harness.DIARIZATION_EFFECTIVE_K,
109 effective_k,
110 "clustering",
111 )
112 return bundle
113
114
115def _refs(
116 *,
117 width: int,
118 owner_index: int,
119 entity_indices: list[int] | None = None,
120 owner_margin: float | None = encoder_config.OWNER_MARGIN_MIN,
121) -> verdict.ReferenceCentroids:
122 entity_indices = entity_indices or []
123 return verdict.ReferenceCentroids(
124 owner_centroid=_axis(owner_index, width),
125 owner_threshold=encoder_config.OWNER_THRESHOLD,
126 owner_threshold_source="constant_default",
127 owner_margin=owner_margin,
128 owner_margin_source="constant_default",
129 entity_ids=tuple(f"entity_{index}" for index in entity_indices),
130 entity_centroids=tuple(_axis(index, width) for index in entity_indices),
131 entity_usable=tuple(True for _index in entity_indices),
132 )
133
134
135def _reference_turns(*turns: tuple[float, float, str]) -> verdict.ReferenceTurns:
136 return verdict.ReferenceTurns(
137 "present",
138 tuple(
139 verdict.ReferenceTurn(start, end, speaker) for start, end, speaker in turns
140 ),
141 )
142
143
144def _write_centroids(
145 path: Path,
146 *,
147 owner: np.ndarray | None,
148 entities: list[tuple[str, np.ndarray, bool]],
149 owner_meta: dict[str, Any] | None = None,
150) -> None:
151 manifest_owner = {
152 "state": harness.PRESENT if owner is not None else harness.ABSENT_NONE
153 }
154 if owner is not None:
155 manifest_owner["array"] = "owner.centroid"
156 if owner_meta:
157 manifest_owner.update(owner_meta)
158 manifest = {
159 "schema": verdict.REFERENCE_CENTROIDS_SCHEMA,
160 "schema_version": verdict.SCHEMA_VERSION,
161 "owner": manifest_owner,
162 "entities": [
163 {"id": entity_id, "usable": usable}
164 for entity_id, _centroid, usable in entities
165 ],
166 }
167 arrays: dict[str, np.ndarray] = {
168 verdict.REFERENCE_CENTROIDS_MANIFEST_KEY: np.array(
169 json.dumps(manifest, sort_keys=True)
170 ),
171 }
172 if owner is not None:
173 arrays["owner.centroid"] = np.asarray(owner, dtype=np.float32)
174 if entities:
175 arrays["entities.centroids"] = np.asarray(
176 [centroid for _entity_id, centroid, _usable in entities],
177 dtype=np.float32,
178 )
179 with path.open("wb") as fh:
180 np.savez(fh, **arrays)
181
182
183def test_family1_model_free_replay_matches_recorded_scalars(
184 monkeypatch: pytest.MonkeyPatch,
185) -> None:
186 _case, bundle, _reset = _emit_model_free_bundle(monkeypatch)
187
188 side = verdict._replay_cluster_side(bundle)
189
190 assert side["evaluated"] is True
191 assert side["recorded_mismatches"] == []
192 assert side["replayed_silhouette_k"] == harness._scalar(
193 bundle, harness.DIARIZATION_SILHOUETTE_K
194 )
195 assert side["replayed_effective_k"] == harness._scalar(
196 bundle, harness.DIARIZATION_EFFECTIVE_K
197 )
198
199
200def test_silhouette_improvement_flip_and_no_flip(
201 monkeypatch: pytest.MonkeyPatch,
202) -> None:
203 eps = _eps(diarize.SILHOUETTE_IMPROVEMENT)
204
205 def fake_ahc(embs_n: np.ndarray, k: int) -> np.ndarray:
206 return np.arange(len(embs_n), dtype=np.int32) % int(k)
207
208 def fake_silhouette(embs_n: np.ndarray, labels: np.ndarray) -> float:
209 k = len(np.unique(labels))
210 if k == 2:
211 return 0.0
212 if k == 3 and embs_n[0, 0] > embs_n[0, 1]:
213 return diarize.SILHOUETTE_IMPROVEMENT + eps
214 if k == 3:
215 return diarize.SILHOUETTE_IMPROVEMENT - eps
216 raise AssertionError(f"unexpected silhouette k {k}")
217
218 monkeypatch.setattr(diarize, "_ahc", fake_ahc)
219 monkeypatch.setattr(diarize, "_silhouette", fake_silhouette)
220 left = _cluster_bundle(
221 np.array([[0.0, 1.0], [0.0, 0.9], [0.1, 0.9], [0.2, 0.8]]),
222 2,
223 2,
224 )
225 right = _cluster_bundle(
226 np.array([[1.0, 0.0], [0.9, 0.0], [0.9, 0.1], [0.8, 0.2]]),
227 3,
228 3,
229 )
230
231 flip_report = verdict._family1_report(left, right)
232 no_flip_report = verdict._family1_report(left, harness.copy_bundle(left))
233
234 assert flip_report["flip_count"] == 1
235 assert flip_report["classification"] == harness.UNEXPECTED_DIFFERS
236 assert no_flip_report["flip_count"] == 0
237 assert no_flip_report["classification"] == harness.EQUAL
238
239
240def test_max_k_cap_binds_and_does_not_bind(monkeypatch: pytest.MonkeyPatch) -> None:
241 def fake_ahc(embs_n: np.ndarray, k: int) -> np.ndarray:
242 return np.arange(len(embs_n), dtype=np.int32) % int(k)
243
244 def fake_silhouette(_embs_n: np.ndarray, labels: np.ndarray) -> float:
245 return float(len(np.unique(labels)))
246
247 monkeypatch.setattr(diarize, "_ahc", fake_ahc)
248 monkeypatch.setattr(diarize, "_silhouette", fake_silhouette)
249 capped = _cluster_bundle(
250 np.eye(diarize.MAX_K + 1, dtype=np.float32),
251 diarize.MAX_K,
252 diarize.MAX_K,
253 )
254 uncapped_rows = max(3, diarize.MAX_K - 1)
255 if uncapped_rows - 1 >= diarize.MAX_K:
256 pytest.skip("MAX_K leaves no uncapped multi-cluster case")
257 uncapped = _cluster_bundle(
258 np.eye(uncapped_rows, dtype=np.float32),
259 1,
260 1,
261 )
262
263 capped_side = verdict._replay_cluster_side(capped)
264 uncapped_side = verdict._replay_cluster_side(uncapped)
265
266 assert capped_side["replayed_effective_k"] == diarize.MAX_K
267 assert uncapped_side["replayed_effective_k"] < diarize.MAX_K
268
269
270def test_owner_threshold_flip_and_no_flip() -> None:
271 eps = _eps(encoder_config.OWNER_THRESHOLD)
272 refs = _refs(width=2, owner_index=0)
273 left = _statement_bundle([_unit_vector([encoder_config.OWNER_THRESHOLD - eps])])
274 right = _statement_bundle([_unit_vector([encoder_config.OWNER_THRESHOLD + eps])])
275 no_flip_left = _statement_bundle(
276 [_unit_vector([encoder_config.OWNER_THRESHOLD + eps])]
277 )
278 no_flip_right = _statement_bundle(
279 [_unit_vector([encoder_config.OWNER_THRESHOLD + eps * 2])]
280 )
281
282 flip_report, *_ = verdict._family2_report(left, right, refs)
283 no_flip_report, *_ = verdict._family2_report(no_flip_left, no_flip_right, refs)
284
285 assert flip_report["flip_count"] == 1
286 assert flip_report["classification"] == harness.UNEXPECTED_DIFFERS
287 assert no_flip_report["flip_count"] == 0
288 assert no_flip_report["classification"] == harness.EQUAL
289
290
291def test_owner_margin_flip_and_no_flip() -> None:
292 eps = _eps(encoder_config.OWNER_MARGIN_MIN)
293 owner_score = encoder_config.OWNER_THRESHOLD + encoder_config.OWNER_MARGIN_MIN + eps
294 entity_fail = owner_score - encoder_config.OWNER_MARGIN_MIN + eps
295 entity_pass = owner_score - encoder_config.OWNER_MARGIN_MIN - eps
296 refs = _refs(width=3, owner_index=0, entity_indices=[1])
297 left = _statement_bundle([_unit_vector([owner_score, entity_fail])])
298 right = _statement_bundle([_unit_vector([owner_score, entity_pass])])
299 no_flip_right = _statement_bundle([_unit_vector([owner_score + eps, entity_pass])])
300
301 flip_report, *_ = verdict._family2_report(left, right, refs)
302 no_flip_report, *_ = verdict._family2_report(right, no_flip_right, refs)
303
304 assert flip_report["flip_count"] == 1
305 assert flip_report["flips"][0]["underlying_values"]["left_owner_margin_declined"]
306 assert flip_report["classification"] == harness.UNEXPECTED_DIFFERS
307 assert no_flip_report["flip_count"] == 0
308 assert no_flip_report["classification"] == harness.EQUAL
309
310
311def test_asymmetric_owner_evaluability_is_not_a_flip() -> None:
312 refs = _refs(width=2, owner_index=0)
313 left = _statement_bundle([np.zeros(2, dtype=np.float32)])
314 right = _statement_bundle([_unit_vector([encoder_config.OWNER_THRESHOLD])])
315
316 report, *_ = verdict._family2_report(left, right, refs)
317
318 assert report["classification"] == harness.UNEXPECTED_DIFFERS
319 assert report["reason"] == "asymmetric_evaluability"
320 assert report["flip_count"] == 0
321 assert report["asymmetric_evaluability"] == [1]
322
323
324def test_functionally_equal_rollup_keeps_zero_flip_answer_visible(
325 monkeypatch: pytest.MonkeyPatch,
326) -> None:
327 _case, left, _reset = _emit_model_free_bundle(monkeypatch)
328 right = harness.copy_bundle(left)
329 embeddings = harness._array(right, harness.STATEMENT_EMBEDDINGS).copy()
330 embeddings[0, 0] += EMBEDDING_MAX_ABS_TOLERANCE / 2
331 harness.replace_array(right, harness.STATEMENT_EMBEDDINGS, embeddings)
332
333 report = verdict.compare_verdicts(left, right)
334
335 assert report["classification"] == harness.FUNCTIONALLY_EQUAL
336 assert (
337 report["components"]["bundle_comparator"]["components"]["statement_embeddings"][
338 "classification"
339 ]
340 == harness.FUNCTIONALLY_EQUAL
341 )
342 assert report["components"]["decision_flips"]["flip_count"] == 0
343
344
345def test_acoustic_medium_threshold_flip_and_no_flip() -> None:
346 eps = _eps(encoder_config.ACOUSTIC_MEDIUM)
347 refs = _refs(width=3, owner_index=1, entity_indices=[0])
348 left = _statement_bundle(
349 [_unit_vector([encoder_config.ACOUSTIC_MEDIUM - eps, 0.0])]
350 )
351 right = _statement_bundle(
352 [_unit_vector([encoder_config.ACOUSTIC_MEDIUM + eps, 0.0])]
353 )
354 no_flip_right = _statement_bundle(
355 [_unit_vector([encoder_config.ACOUSTIC_MEDIUM + eps * 2, 0.0])]
356 )
357
358 left_owner = verdict._family2_report(left, right, refs)
359 flip_report = verdict._family3_report(
360 left, right, refs, left_owner[0], *left_owner[1:]
361 )
362 right_owner = verdict._family2_report(right, no_flip_right, refs)
363 no_flip_report = verdict._family3_report(
364 right, no_flip_right, refs, right_owner[0], *right_owner[1:]
365 )
366
367 assert flip_report["flip_count"] == 1
368 assert flip_report["classification"] == harness.UNEXPECTED_DIFFERS
369 assert no_flip_report["flip_count"] == 0
370 assert no_flip_report["classification"] == harness.EQUAL
371
372
373def test_acoustic_high_threshold_flip_and_no_flip() -> None:
374 eps = _eps(encoder_config.ACOUSTIC_HIGH)
375 refs = _refs(width=3, owner_index=1, entity_indices=[0])
376 left = _statement_bundle([_unit_vector([encoder_config.ACOUSTIC_HIGH - eps, 0.0])])
377 right = _statement_bundle([_unit_vector([encoder_config.ACOUSTIC_HIGH + eps, 0.0])])
378 no_flip_right = _statement_bundle(
379 [_unit_vector([encoder_config.ACOUSTIC_HIGH + eps * 2, 0.0])]
380 )
381
382 left_owner = verdict._family2_report(left, right, refs)
383 flip_report = verdict._family3_report(
384 left, right, refs, left_owner[0], *left_owner[1:]
385 )
386 right_owner = verdict._family2_report(right, no_flip_right, refs)
387 no_flip_report = verdict._family3_report(
388 right, no_flip_right, refs, right_owner[0], *right_owner[1:]
389 )
390
391 assert flip_report["flip_count"] == 1
392 assert flip_report["flips"][0]["left_outcome"]["tier"] == "medium"
393 assert flip_report["flips"][0]["right_outcome"]["tier"] == "high"
394 assert no_flip_report["flip_count"] == 0
395 assert no_flip_report["classification"] == harness.EQUAL
396
397
398def test_acoustic_margin_flip_and_no_flip() -> None:
399 eps = _eps(encoder_config.ACOUSTIC_MARGIN_MIN)
400 best = encoder_config.ACOUSTIC_HIGH + eps
401 runner_fail = best - encoder_config.ACOUSTIC_MARGIN_MIN + eps
402 runner_pass = best - encoder_config.ACOUSTIC_MARGIN_MIN - eps
403 refs = _refs(width=4, owner_index=2, entity_indices=[0, 1])
404 left = _statement_bundle([_unit_vector([best, runner_fail, 0.0])])
405 right = _statement_bundle([_unit_vector([best, runner_pass, 0.0])])
406 no_flip_right = _statement_bundle([_unit_vector([best + eps, runner_pass, 0.0])])
407
408 left_owner = verdict._family2_report(left, right, refs)
409 flip_report = verdict._family3_report(
410 left, right, refs, left_owner[0], *left_owner[1:]
411 )
412 right_owner = verdict._family2_report(right, no_flip_right, refs)
413 no_flip_report = verdict._family3_report(
414 right, no_flip_right, refs, right_owner[0], *right_owner[1:]
415 )
416
417 assert flip_report["flip_count"] == 1
418 assert flip_report["flips"][0]["left_outcome"]["demotion_causes"] == [
419 "acoustic_margin_declined"
420 ]
421 assert no_flip_report["flip_count"] == 0
422 assert no_flip_report["classification"] == harness.EQUAL
423
424
425def test_owner_margin_decline_demotes_acoustic_high() -> None:
426 eps = _eps(encoder_config.OWNER_THRESHOLD, encoder_config.OWNER_MARGIN_MIN)
427 owner_declined_score = encoder_config.OWNER_THRESHOLD + eps
428 best = max(
429 encoder_config.ACOUSTIC_HIGH + eps,
430 owner_declined_score - encoder_config.OWNER_MARGIN_MIN + eps,
431 )
432 owner_clear_score = encoder_config.OWNER_THRESHOLD - eps
433 refs = _refs(width=3, owner_index=1, entity_indices=[0])
434 left = _statement_bundle([_unit_vector([best, owner_declined_score])])
435 right = _statement_bundle([_unit_vector([best, owner_clear_score])])
436
437 owner_report = verdict._family2_report(left, right, refs)
438 acoustic_report = verdict._family3_report(
439 left, right, refs, owner_report[0], *owner_report[1:]
440 )
441
442 assert owner_report[0]["left_margin_declined_sids"] == [1]
443 assert acoustic_report["flip_count"] == 1
444 assert acoustic_report["flips"][0]["left_outcome"]["demotion_causes"] == [
445 "owner_margin_declined"
446 ]
447 assert acoustic_report["flips"][0]["right_outcome"]["tier"] == "high"
448
449
450@pytest.mark.parametrize(
451 ("bundle", "reference_turns", "expected_der", "expected_breakdown"),
452 [
453 (
454 _prediction_bundle([(2, 5.0, 10.0, 2), (1, 0.0, 5.0, 1)]),
455 _reference_turns((0.0, 5.0, "a"), (5.0, 10.0, "b")),
456 0.0,
457 {"missed": 0.0, "false_alarm": 0.0, "confusion": 0.0, "denominator": 10.0},
458 ),
459 (
460 _prediction_bundle([(1, 10.0, 20.0, 1)]),
461 _reference_turns((0.0, 10.0, "a")),
462 2.0,
463 {
464 "missed": 10.0,
465 "false_alarm": 10.0,
466 "confusion": 0.0,
467 "denominator": 10.0,
468 },
469 ),
470 (
471 _prediction_bundle([(1, 0.0, 4.0, 1)]),
472 _reference_turns((0.0, 10.0, "a")),
473 0.6,
474 {"missed": 6.0, "false_alarm": 0.0, "confusion": 0.0, "denominator": 10.0},
475 ),
476 (
477 _prediction_bundle([(1, 4.0, 10.0, 1), (2, 0.0, 4.0, 2)]),
478 _reference_turns((0.0, 4.0, "a"), (4.0, 10.0, "b")),
479 0.0,
480 {"missed": 0.0, "false_alarm": 0.0, "confusion": 0.0, "denominator": 10.0},
481 ),
482 (
483 _prediction_bundle([(1, 0.0, 15.0, 1)]),
484 _reference_turns((0.0, 10.0, "a"), (5.0, 15.0, "b")),
485 0.5,
486 {"missed": 5.0, "false_alarm": 0.0, "confusion": 5.0, "denominator": 20.0},
487 ),
488 ],
489)
490def test_der_hand_computed_cases(
491 bundle: harness.Bundle,
492 reference_turns: verdict.ReferenceTurns,
493 expected_der: float,
494 expected_breakdown: dict[str, float],
495) -> None:
496 score = verdict.score_der(bundle, reference_turns)
497
498 assert score["status"] == harness.PRESENT
499 assert score["der"] == pytest.approx(expected_der)
500 for key, value in expected_breakdown.items():
501 assert score["breakdown"][key] == pytest.approx(value)
502
503
504def test_der_ignores_nan_and_null_predicted_spans() -> None:
505 bundle = _prediction_bundle(
506 [
507 (1, 0.0, 10.0, harness.LABEL_NULL_SENTINEL),
508 (2, float("nan"), float("nan"), 2),
509 ]
510 )
511
512 score = verdict.score_der(bundle, _reference_turns((0.0, 10.0, "a")))
513
514 assert score["der"] == pytest.approx(1.0)
515 assert score["breakdown"]["missed"] == pytest.approx(10.0)
516
517
518def test_der_counts_distinct_speakers_not_overlapping_turns() -> None:
519 bundle = _prediction_bundle([(1, 0.0, 15.0, 1)])
520
521 score = verdict.score_der(
522 bundle,
523 _reference_turns((0.0, 10.0, "a"), (5.0, 15.0, "a")),
524 )
525
526 assert score["der"] == pytest.approx(0.0)
527 assert score["breakdown"]["denominator"] == pytest.approx(15.0)
528
529
530def test_der_reference_absent_empty_and_zero_denominator(tmp_path: Path) -> None:
531 bundle = _prediction_bundle([(1, 0.0, 1.0, 1)])
532 empty_path = tmp_path / "empty-reference-turns.json"
533 empty_path.write_text("[]", encoding="utf-8")
534
535 absent = verdict._der_report(bundle, bundle, verdict.load_reference_turns(None))
536 empty = verdict._der_report(
537 bundle, bundle, verdict.load_reference_turns(empty_path)
538 )
539 zero = verdict.score_der(bundle, verdict.ReferenceTurns("present", ()))
540
541 assert absent["reason"] == "reference_turns_absent"
542 assert empty["reason"] == "reference_turns_empty"
543 assert zero["reason"] == "zero_reference_speech"
544
545
546def test_der_predicted_labels_field_state_gates_component() -> None:
547 bundle = harness.new_bundle(producer="test")
548 harness._set_array(
549 bundle,
550 harness.INPUT_DIARIZATION_IDS,
551 np.array([1], dtype=np.int32),
552 "inputs.diarization",
553 )
554 harness._set_array(
555 bundle,
556 harness.INPUT_DIARIZATION_SPANS,
557 np.array([[0.0, 1.0]], dtype=np.float32),
558 "inputs.diarization",
559 )
560 harness._set_state(
561 bundle,
562 harness.DIARIZATION_STATEMENT_LABELS,
563 harness.NOT_EVALUATED,
564 "statement_labels",
565 )
566
567 score = verdict.score_der(bundle, _reference_turns((0.0, 1.0, "a")))
568
569 assert score["status"] == harness.NOT_EVALUATED
570 assert score["reason"] == "predicted_labels_not_present"
571
572
573def test_reference_turn_reader_rejects_unknown_text_key(tmp_path: Path) -> None:
574 path = tmp_path / "reference-turns.json"
575 path.write_text(
576 json.dumps(
577 [{"start_s": 0.0, "end_s": 1.0, "speaker": "a", "text": "redacted"}]
578 ),
579 encoding="utf-8",
580 )
581
582 with pytest.raises(harness.HarnessError, match="text"):
583 verdict.load_reference_turns(path)
584
585
586@pytest.mark.parametrize(
587 "payload",
588 [
589 {"start_s": 0.0, "end_s": 1.0, "speaker": "a"},
590 [{"start_s": -1.0, "end_s": 1.0, "speaker": "a"}],
591 [{"start_s": 1.0, "end_s": 1.0, "speaker": "a"}],
592 [{"start_s": "0", "end_s": 1.0, "speaker": "a"}],
593 [{"start_s": 0.0, "end_s": 1.0, "speaker": ""}],
594 [
595 {"start_s": 2.0, "end_s": 3.0, "speaker": "a"},
596 {"start_s": 0.0, "end_s": 1.0, "speaker": "b"},
597 ],
598 ],
599)
600def test_reference_turn_reader_rejects_invalid_shapes(
601 tmp_path: Path,
602 payload: object,
603) -> None:
604 path = tmp_path / "reference-turns.json"
605 path.write_text(json.dumps(payload), encoding="utf-8")
606
607 with pytest.raises(harness.HarnessError):
608 verdict.load_reference_turns(path)
609
610
611def test_reference_centroid_file_manifest_is_single_source_of_truth(
612 tmp_path: Path,
613) -> None:
614 eps = _eps(encoder_config.OWNER_THRESHOLD)
615 path = tmp_path / "centroids.npz"
616 _write_centroids(
617 path,
618 owner=_axis(0, 2),
619 entities=[("entity_1", _axis(1, 2), True)],
620 owner_meta={"threshold": encoder_config.OWNER_THRESHOLD + eps, "margin": None},
621 )
622
623 refs = verdict.load_reference_centroids(path)
624 with np.load(path, allow_pickle=False) as payload:
625 assert set(payload.files) == {
626 verdict.REFERENCE_CENTROIDS_MANIFEST_KEY,
627 "owner.centroid",
628 "entities.centroids",
629 }
630 threshold_block = verdict._owner_thresholds(refs)
631
632 assert refs is not None
633 assert refs.owner_threshold_source == "supplied"
634 assert refs.owner_margin is None
635 assert refs.owner_margin_source == "supplied"
636 assert threshold_block["owner_margin"]["effective_value"] is None
637
638
639def test_comparator_failure_short_circuits_verdict() -> None:
640 left = harness.new_bundle(producer="left")
641 right = harness.new_bundle(producer="right")
642
643 report = verdict.compare_verdicts(left, right)
644
645 assert report["classification"] == harness.NOT_EVALUATED
646 assert report["failure"] == report["components"]["bundle_comparator"]["failure"]
647 assert set(report["components"]) == {"bundle_comparator"}
648
649
650def test_cli_does_not_use_network_or_write_inside_repository(
651 tmp_path: Path,
652 monkeypatch: pytest.MonkeyPatch,
653 capsys: pytest.CaptureFixture[str],
654) -> None:
655 connect_calls: list[object] = []
656
657 def fail_connect(_self: socket.socket, address: object) -> None:
658 connect_calls.append(address)
659 raise AssertionError("network call attempted")
660
661 monkeypatch.setattr(socket.socket, "connect", fail_connect)
662 _case, bundle, _reset = _emit_model_free_bundle(monkeypatch)
663 output_dir = tmp_path / "speaker-verdict"
664 output_dir.mkdir()
665 left_path = output_dir / "left.npz"
666 right_path = output_dir / "right.npz"
667 centroids_path = output_dir / "centroids.npz"
668 reference_path = output_dir / "reference-turns.json"
669 report_path = output_dir / "report.json"
670 harness.write_bundle(bundle, left_path)
671 harness.write_bundle(harness.copy_bundle(bundle), right_path)
672 owner = np.zeros(harness._array(bundle, harness.STATEMENT_EMBEDDINGS).shape[1])
673 owner[0] = 1.0
674 _write_centroids(centroids_path, owner=owner, entities=[])
675 spans = harness._array(bundle, harness.INPUT_DIARIZATION_SPANS)
676 labels = harness._array(bundle, harness.DIARIZATION_STATEMENT_LABELS)
677 reference_turns = [
678 {"start_s": float(start), "end_s": float(end), "speaker": str(int(label))}
679 for (start, end), label in zip(spans, labels)
680 if int(label) != int(harness.LABEL_NULL_SENTINEL)
681 and np.isfinite(start)
682 and np.isfinite(end)
683 and end > start
684 ]
685 reference_path.write_text(json.dumps(reference_turns), encoding="utf-8")
686
687 before_inventory = repository_inventory(harness.ROOT)
688 code = verdict.main(
689 [
690 str(left_path),
691 str(right_path),
692 "--reference-centroids",
693 str(centroids_path),
694 "--reference-turns",
695 str(reference_path),
696 "--report",
697 str(report_path),
698 ]
699 )
700 after_inventory = repository_inventory(harness.ROOT)
701
702 captured = capsys.readouterr()
703 assert code == 0
704 assert json.loads(report_path.read_text(encoding="utf-8"))["classification"] in {
705 harness.EQUAL,
706 harness.FUNCTIONALLY_EQUAL,
707 }
708 assert json.loads(captured.out)["schema"] == verdict.REPORT_SCHEMA
709 assert connect_calls == []
710 assert_inventory_unchanged(before_inventory, after_inventory)