diff --git a/solstone/apps/speakers/_overlap.py b/solstone/apps/speakers/_overlap.py index 88897124d..7826d4704 100644 --- a/solstone/apps/speakers/_overlap.py +++ b/solstone/apps/speakers/_overlap.py @@ -8,6 +8,9 @@ from __future__ import annotations import json import logging from pathlib import Path +from typing import NamedTuple + +from solstone.apps.speakers.encoder_config import SPEAKER_EVIDENCE_VERSION logger = logging.getLogger(__name__) @@ -31,3 +34,53 @@ def _read_segment_overlap_fraction(jsonl_path: Path) -> float: return float(value) except (TypeError, ValueError): return 0.0 + + +class SegmentSpeakerEvidence(NamedTuple): + speaker_evidence: str + multi_fraction: float | None + version: str | None + + +UNKNOWN_SPEAKER_EVIDENCE = SegmentSpeakerEvidence( + speaker_evidence="unknown", + multi_fraction=None, + version=None, +) + + +def _read_segment_speaker_evidence(jsonl_path: Path) -> SegmentSpeakerEvidence: + """Return speaker-evidence metadata, or explicit unknown on absent/corrupt input.""" + try: + with jsonl_path.open(encoding="utf-8") as f: + line = f.readline() + if not line: + return UNKNOWN_SPEAKER_EVIDENCE + header = json.loads(line) + except FileNotFoundError: + return UNKNOWN_SPEAKER_EVIDENCE + except (OSError, json.JSONDecodeError) as exc: + logger.info("speaker evidence header read failed at %s: %s", jsonl_path, exc) + return UNKNOWN_SPEAKER_EVIDENCE + + if not isinstance(header, dict): + return UNKNOWN_SPEAKER_EVIDENCE + + speaker_evidence = header.get("speaker_evidence") + version = header.get("speaker_evidence_version") + if ( + speaker_evidence not in {"none", "single", "multi"} + or version != SPEAKER_EVIDENCE_VERSION + ): + return UNKNOWN_SPEAKER_EVIDENCE + + try: + multi_fraction = float(header["speaker_evidence_multi_fraction"]) + except (KeyError, TypeError, ValueError): + return UNKNOWN_SPEAKER_EVIDENCE + + return SegmentSpeakerEvidence( + speaker_evidence=speaker_evidence, + multi_fraction=multi_fraction, + version=version, + ) diff --git a/solstone/apps/speakers/candidate_tracker.py b/solstone/apps/speakers/candidate_tracker.py index d20323b45..e25222600 100644 --- a/solstone/apps/speakers/candidate_tracker.py +++ b/solstone/apps/speakers/candidate_tracker.py @@ -17,6 +17,14 @@ from solstone.apps.speakers.attribution import ( _load_integer_speaker_labels, segment_path, ) +from solstone.apps.speakers.encoder_config import ( + CONFIRM_MIN_DURATION_S, + CONFIRM_MIN_INTERVALS, + CONFIRM_MIN_SEGMENTS, + MERGE_THRESHOLD, + SPLIT_THRESHOLD, + STABILITY_THRESHOLD, +) from solstone.think.journal_io import ( MalformedPolicy, atomic_replace, @@ -25,13 +33,6 @@ from solstone.think.journal_io import ( ) from solstone.think.utils import get_journal -MERGE_THRESHOLD = 0.72 -SPLIT_THRESHOLD = 0.55 -STABILITY_THRESHOLD = 0.25 -CONFIRM_MIN_SEGMENTS = 2 -CONFIRM_MIN_INTERVALS = 5 -CONFIRM_MIN_DURATION_S = 25.0 - @dataclass class CandidateProfile: diff --git a/solstone/apps/speakers/encoder_config.py b/solstone/apps/speakers/encoder_config.py index fadedb45e..4856bff26 100644 --- a/solstone/apps/speakers/encoder_config.py +++ b/solstone/apps/speakers/encoder_config.py @@ -1,7 +1,7 @@ # SPDX-License-Identifier: AGPL-3.0-only # Copyright (c) 2026 sol pbc -"""Encoder-specific calibration constants. Source of truth for owner-separation, acoustic-match, and owner-bootstrap quality thresholds. Changes require CPO spec revision; see cpo/specs/in-flight/speaker-attribution-wespeaker.md.""" +"""Encoder-specific calibration constants governed by operator spec signoff.""" import math @@ -25,7 +25,19 @@ OWNER_BOOTSTRAP_MIN_INTRA_COSINE_P25: float = 0.30 OWNER_BOOTSTRAP_PROVISIONAL_GUARD_MIN_TAGS: int = 5 NOISY_FLYWHEEL_OVERLAP_MAX: float = 0.10 +SLOT_ACTIVE_MIN_SHARE: float = 0.10 +SPEAKER_EVIDENCE_MULTI_MIN: float = 0.05 +SPEAKER_EVIDENCE_SINGLE_MAX: float = 0.05 +DIARIZE_MIN_OVERLAP: float = 0.05 +SPEAKER_EVIDENCE_VERSION: str = "windowed-slots-v1" OVERLAP_DETECTOR_ID: str = "pyannote-segmentation-3.0-onnx" OVERLAP_DETECTOR_SHA256: str = ( "057ee564753071c0b09b5b611648b50ac188d50846bff5f01e9f7bbf1591ea25" ) + +MERGE_THRESHOLD = 0.72 +SPLIT_THRESHOLD = 0.55 +STABILITY_THRESHOLD = 0.25 +CONFIRM_MIN_SEGMENTS = 2 +CONFIRM_MIN_INTERVALS = 5 +CONFIRM_MIN_DURATION_S = 25.0 diff --git a/solstone/apps/speakers/tests/test_candidate_tracker.py b/solstone/apps/speakers/tests/test_candidate_tracker.py index 26a3d9ff6..5068a489f 100644 --- a/solstone/apps/speakers/tests/test_candidate_tracker.py +++ b/solstone/apps/speakers/tests/test_candidate_tracker.py @@ -11,15 +11,17 @@ from pathlib import Path import numpy as np from solstone.apps.speakers.candidate_tracker import ( + CandidateTracker, +) +from solstone.apps.speakers.encoder_config import ( CONFIRM_MIN_DURATION_S, CONFIRM_MIN_INTERVALS, CONFIRM_MIN_SEGMENTS, + ENCODER_ID, MERGE_THRESHOLD, SPLIT_THRESHOLD, STABILITY_THRESHOLD, - CandidateTracker, ) -from solstone.apps.speakers.encoder_config import ENCODER_ID from solstone.apps.speakers.owner import OWNER_THRESHOLD from solstone.think.entities import save_voiceprints_batch diff --git a/solstone/apps/speakers/tests/test_encoder_config.py b/solstone/apps/speakers/tests/test_encoder_config.py index b8ac2bc91..1cad20e32 100644 --- a/solstone/apps/speakers/tests/test_encoder_config.py +++ b/solstone/apps/speakers/tests/test_encoder_config.py @@ -2,9 +2,11 @@ # Copyright (c) 2026 sol pbc """Test the constant-import contract for encoder_config.""" +import ast import math +from pathlib import Path -from solstone.apps.speakers import attribution, encoder_config, owner +from solstone.apps.speakers import attribution, candidate_tracker, encoder_config, owner from solstone.observe.transcribe.main import ( OVERLAP_DETECTOR_ID as MAIN_OVERLAP_DETECTOR_ID, ) @@ -26,8 +28,46 @@ def test_locked_constants(): assert encoder_config.OWNER_BOOTSTRAP_MIN_INTRA_COSINE_P25 == 0.30 assert encoder_config.OWNER_BOOTSTRAP_PROVISIONAL_GUARD_MIN_TAGS == 5 assert encoder_config.NOISY_FLYWHEEL_OVERLAP_MAX == 0.10 + assert encoder_config.SLOT_ACTIVE_MIN_SHARE == 0.10 + assert encoder_config.SPEAKER_EVIDENCE_MULTI_MIN == 0.05 + assert encoder_config.SPEAKER_EVIDENCE_SINGLE_MAX == 0.05 + assert encoder_config.DIARIZE_MIN_OVERLAP == 0.05 + assert encoder_config.SPEAKER_EVIDENCE_VERSION == "windowed-slots-v1" assert encoder_config.OVERLAP_DETECTOR_ID == MAIN_OVERLAP_DETECTOR_ID assert encoder_config.OVERLAP_DETECTOR_SHA256 == PYANNOTE_OVERLAP_MODEL_SHA256 + assert encoder_config.MERGE_THRESHOLD == 0.72 + assert encoder_config.SPLIT_THRESHOLD == 0.55 + assert encoder_config.STABILITY_THRESHOLD == 0.25 + assert encoder_config.CONFIRM_MIN_SEGMENTS == 2 + assert encoder_config.CONFIRM_MIN_INTERVALS == 5 + assert encoder_config.CONFIRM_MIN_DURATION_S == 25.0 + + +def _module_assignment_targets(path: Path) -> set[str]: + tree = ast.parse(path.read_text(encoding="utf-8")) + targets: set[str] = set() + for node in tree.body: + if isinstance(node, ast.Assign): + for target in node.targets: + if isinstance(target, ast.Name): + targets.add(target.id) + elif isinstance(node, ast.AnnAssign) and isinstance(node.target, ast.Name): + targets.add(node.target.id) + return targets + + +def test_candidate_tracker_constants_are_not_assigned_in_tracker_module(): + moved_constants = { + "MERGE_THRESHOLD", + "SPLIT_THRESHOLD", + "STABILITY_THRESHOLD", + "CONFIRM_MIN_SEGMENTS", + "CONFIRM_MIN_INTERVALS", + "CONFIRM_MIN_DURATION_S", + } + targets = _module_assignment_targets(Path(candidate_tracker.__file__)) + + assert moved_constants.isdisjoint(targets) def test_attribution_imports_acoustic_constants(): diff --git a/solstone/apps/speakers/tests/test_overlap.py b/solstone/apps/speakers/tests/test_overlap.py new file mode 100644 index 000000000..027152216 --- /dev/null +++ b/solstone/apps/speakers/tests/test_overlap.py @@ -0,0 +1,68 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +from __future__ import annotations + +import json + +from solstone.apps.speakers._overlap import ( + UNKNOWN_SPEAKER_EVIDENCE, + _read_segment_overlap_fraction, + _read_segment_speaker_evidence, +) +from solstone.apps.speakers.encoder_config import SPEAKER_EVIDENCE_VERSION + + +def test_read_segment_speaker_evidence_returns_header_fields(tmp_path): + path = tmp_path / "audio.jsonl" + path.write_text( + json.dumps( + { + "speaker_evidence": "multi", + "speaker_evidence_multi_fraction": 0.125, + "speaker_evidence_version": SPEAKER_EVIDENCE_VERSION, + } + ) + + "\n", + encoding="utf-8", + ) + + result = _read_segment_speaker_evidence(path) + + assert result.speaker_evidence == "multi" + assert result.multi_fraction == 0.125 + assert result.version == SPEAKER_EVIDENCE_VERSION + + +def test_read_segment_speaker_evidence_unknown_for_absent_or_corrupt_header(tmp_path): + missing = tmp_path / "missing.jsonl" + corrupt = tmp_path / "corrupt.jsonl" + corrupt.write_text("{not-json}\n", encoding="utf-8") + absent = tmp_path / "absent.jsonl" + absent.write_text(json.dumps({"raw": "audio.flac"}) + "\n", encoding="utf-8") + wrong_version = tmp_path / "wrong-version.jsonl" + wrong_version.write_text( + json.dumps( + { + "speaker_evidence": "multi", + "speaker_evidence_multi_fraction": 0.5, + "speaker_evidence_version": "other", + } + ) + + "\n", + encoding="utf-8", + ) + + assert _read_segment_speaker_evidence(missing) == UNKNOWN_SPEAKER_EVIDENCE + assert _read_segment_speaker_evidence(corrupt) == UNKNOWN_SPEAKER_EVIDENCE + assert _read_segment_speaker_evidence(absent) == UNKNOWN_SPEAKER_EVIDENCE + assert _read_segment_speaker_evidence(wrong_version) == UNKNOWN_SPEAKER_EVIDENCE + + +def test_read_segment_overlap_fraction_legacy_absent_and_corrupt_return_zero(tmp_path): + missing = tmp_path / "missing.jsonl" + corrupt = tmp_path / "corrupt.jsonl" + corrupt.write_text("{not-json}\n", encoding="utf-8") + + assert _read_segment_overlap_fraction(missing) == 0.0 + assert _read_segment_overlap_fraction(corrupt) == 0.0 diff --git a/solstone/observe/transcribe/audio.schema.json b/solstone/observe/transcribe/audio.schema.json index c7b305f5c..dd9ca4f61 100644 --- a/solstone/observe/transcribe/audio.schema.json +++ b/solstone/observe/transcribe/audio.schema.json @@ -22,6 +22,20 @@ "imported": {"type": "object", "additionalProperties": true}, "facet": {"type": "string"}, "id": {"type": "string"}, + "speaker_evidence": { + "type": "string", + "enum": ["single", "multi", "none"], + "description": "Windowed slot evidence decision for local diarization engagement. This records engagement, not clustering outcome: a multi value with no integer speaker labels means diarization engaged and clustering produced nothing or failed." + }, + "speaker_evidence_multi_fraction": { + "type": "number", + "description": "Fraction of speech-bearing pyannote windows with multi-speaker slot evidence. This records engagement evidence, not clustering outcome." + }, + "speaker_evidence_version": { + "type": "string", + "const": "windowed-slots-v1", + "description": "Version of the windowed slot evidence decision contract. This records engagement, not clustering outcome: a multi value with no integer speaker labels means diarization engaged and clustering produced nothing or failed." + }, "sound_tags": { "type": "object", "additionalProperties": true, diff --git a/solstone/observe/transcribe/main.py b/solstone/observe/transcribe/main.py index 7e969c770..72fee9db5 100644 --- a/solstone/observe/transcribe/main.py +++ b/solstone/observe/transcribe/main.py @@ -63,6 +63,7 @@ from solstone.apps.settings.install_copy import ( from solstone.apps.speakers.encoder_config import ( OVERLAP_DETECTOR_ID, OVERLAP_DETECTOR_SHA256, + SPEAKER_EVIDENCE_VERSION, ) from solstone.observe.exit_codes import EXIT_PROVIDER_BLOCKED from solstone.observe.model_assets import resolve_wespeaker_model @@ -121,6 +122,7 @@ if TYPE_CHECKING: import numpy as np import onnxruntime as ort + from solstone.observe.transcribe.overlap import SpeakerEvidenceDecision from solstone.observe.vad import AudioReduction, VadResult # Re-export defaults for backwards compatibility @@ -673,6 +675,7 @@ def _statements_to_jsonl( *, overlap_fraction: float | None = None, overlap_detector: str | None = None, + speaker_evidence: SpeakerEvidenceDecision | None = None, processing_record: dict | None = None, sound_tags: dict | None = None, ) -> list[str]: @@ -691,6 +694,7 @@ def _statements_to_jsonl( backend: Optional STT backend name (e.g., "parakeet") overlap_fraction: Optional fraction of speech containing overlapping speakers overlap_detector: Optional overlap detector identifier + speaker_evidence: Optional local diarization engagement decision processing_record: Optional _solstone_processing record sound_tags: Optional ambient sound-tag metadata @@ -724,6 +728,12 @@ def _statements_to_jsonl( if overlap_fraction is not None and overlap_detector is not None: metadata["overlap_fraction"] = round(float(overlap_fraction), 4) metadata["overlap_detector"] = overlap_detector + if speaker_evidence is not None: + metadata["speaker_evidence"] = speaker_evidence.speaker_evidence + metadata["speaker_evidence_multi_fraction"] = round( + float(speaker_evidence.multi_window_fraction), 4 + ) + metadata["speaker_evidence_version"] = SPEAKER_EVIDENCE_VERSION # Add segment metadata (from SEGMENT_META env var) if segment_meta: @@ -978,12 +988,19 @@ def process_audio( # Use reduced audio buffer if available for consistent timestamps with timings.time("embed"): embeddings_data = _embed_statements(stt_buffer, statements, SAMPLE_RATE) - from solstone.observe.transcribe.overlap import compute_overlap_and_logprobs + from solstone.observe.transcribe.overlap import ( + compute_overlap_and_logprobs, + decide_speaker_evidence, + ) with timings.time("overlap"): - overlap_fraction_value, pyannote_logprobs = compute_overlap_and_logprobs( - audio_buffer + overlap_result = compute_overlap_and_logprobs(audio_buffer) + speaker_evidence = decide_speaker_evidence( + overlap_result.overlap_fraction, + overlap_result.window_stats, ) + overlap_fraction_value = overlap_result.overlap_fraction + pyannote_logprobs = overlap_result.avg_log_probs # Restore original timestamps if audio was reduced. if reduction: @@ -996,15 +1013,15 @@ def process_audio( ) # Local speaker diarization for backends that produce no speaker labels. - # Skip when overlap is near zero — the recording is effectively solo - # speech and diarization adds no value. Otherwise reuse the pyannote - # log-probs computed above so the diarizer skips its own pyannote pass. - _DIARIZE_MIN_OVERLAP = 0.05 - if overlap_fraction_value < _DIARIZE_MIN_OVERLAP: + # Reuse the pyannote log-probs computed above so the diarizer skips its + # own pyannote pass when the speaker-evidence gate engages it. + if speaker_evidence.speaker_evidence != "multi": logging.info( - " Skipping diarization: overlap=%.2f (threshold %.2f)", + " Skipping diarization: speaker_evidence=%s overlap=%.2f " + "multi_window_fraction=%.2f", + speaker_evidence.speaker_evidence, overlap_fraction_value, - _DIARIZE_MIN_OVERLAP, + speaker_evidence.multi_window_fraction, ) else: try: @@ -1053,6 +1070,7 @@ def process_audio( resolved_backend, overlap_fraction=overlap_fraction_value, overlap_detector=OVERLAP_DETECTOR_ID, + speaker_evidence=speaker_evidence, processing_record=processing_record, sound_tags=sound_tags, ) diff --git a/solstone/observe/transcribe/overlap.py b/solstone/observe/transcribe/overlap.py index d5fa2fc64..e637a8893 100644 --- a/solstone/observe/transcribe/overlap.py +++ b/solstone/observe/transcribe/overlap.py @@ -8,8 +8,14 @@ from __future__ import annotations import logging import time from pathlib import Path -from typing import TYPE_CHECKING - +from typing import TYPE_CHECKING, NamedTuple, Sequence + +from solstone.apps.speakers.encoder_config import ( + DIARIZE_MIN_OVERLAP, + SLOT_ACTIVE_MIN_SHARE, + SPEAKER_EVIDENCE_MULTI_MIN, + SPEAKER_EVIDENCE_SINGLE_MAX, +) from solstone.observe.utils import SAMPLE_RATE if TYPE_CHECKING: @@ -29,6 +35,97 @@ _DIARIZE_STRIDE_S = 2 _overlap_session: ort.InferenceSession | None = None +class SpeakerWindowStats(NamedTuple): + speech_frames: int + active_slot_count: int + overlap_frames: int + + +class SpeakerEvidenceDecision(NamedTuple): + speaker_evidence: str + multi_window_fraction: float + mean_window_overlap_share: float + + +class OverlapInferenceResult(NamedTuple): + overlap_fraction: float + avg_log_probs: np.ndarray + window_stats: tuple[SpeakerWindowStats, ...] + + +def _speaker_window_stats(log_probs: np.ndarray) -> SpeakerWindowStats: + """Derive raw speaker-evidence counts from one truncated pyannote window.""" + import numpy as np + + argmax = log_probs.argmax(axis=-1) + counts = np.bincount(argmax, minlength=7) + speech_frames = int(counts[1:].sum()) + if speech_frames == 0: + return SpeakerWindowStats( + speech_frames=0, + active_slot_count=0, + overlap_frames=0, + ) + + active_slot_count = int( + (counts[1:4] / speech_frames >= SLOT_ACTIVE_MIN_SHARE).sum() + ) + return SpeakerWindowStats( + speech_frames=speech_frames, + active_slot_count=active_slot_count, + overlap_frames=int(counts[list(OVERLAP_CLASSES)].sum()), + ) + + +def decide_speaker_evidence( + overlap_fraction: float, window_stats: Sequence[SpeakerWindowStats] +) -> SpeakerEvidenceDecision: + """Decide whether local diarization should engage for this segment. + + ``window overlap share`` is overlap-class frames divided by speech frames + within a single window, from that window's raw argmax. It is explicitly not + the segment-level ``overlap_fraction``, which argmaxes averaged log-probs + across overlapping windows. + + ``mean_window_overlap_share`` is averaged over speech-bearing windows only. + ``DIARIZE_MIN_OVERLAP`` governs branch 3's overlap term. Branches 2 and 3 + are deliberately not mutually simplified: the constants are separate + re-calibration controls, and the gap between them is the ambiguous band. + """ + speech_windows = [row for row in window_stats if row.speech_frames > 0] + if not speech_windows: + return SpeakerEvidenceDecision( + speaker_evidence="none", + multi_window_fraction=0.0, + mean_window_overlap_share=0.0, + ) + + multi_window_count = sum(1 for row in speech_windows if row.active_slot_count > 1) + multi_window_fraction = multi_window_count / len(speech_windows) + mean_window_overlap_share = sum( + row.overlap_frames / row.speech_frames for row in speech_windows + ) / len(speech_windows) + + if ( + multi_window_fraction >= SPEAKER_EVIDENCE_MULTI_MIN + or overlap_fraction >= DIARIZE_MIN_OVERLAP + ): + speaker_evidence = "multi" + elif ( + multi_window_fraction < SPEAKER_EVIDENCE_SINGLE_MAX + and mean_window_overlap_share < DIARIZE_MIN_OVERLAP + ): + speaker_evidence = "single" + else: + speaker_evidence = "multi" + + return SpeakerEvidenceDecision( + speaker_evidence=speaker_evidence, + multi_window_fraction=multi_window_fraction, + mean_window_overlap_share=mean_window_overlap_share, + ) + + def _get_overlap_session() -> ort.InferenceSession: """Return a cached ONNX InferenceSession for the pyannote overlap model.""" global _overlap_session @@ -123,7 +220,7 @@ def compute_overlap_fraction( def compute_overlap_and_logprobs( audio: np.ndarray, sample_rate: int = SAMPLE_RATE -) -> tuple[float, np.ndarray]: +) -> OverlapInferenceResult: """Compute overlap fraction and return the pyannote log-probs for reuse. Uses a 2-second stride (vs. the 5-second stride in compute_overlap_fraction) @@ -162,6 +259,7 @@ def compute_overlap_and_logprobs( num_frames = int(np.ceil(len(audio_padded) / samples_per_frame)) accum = np.zeros((num_frames, 7), dtype=np.float64) counts = np.zeros((num_frames,), dtype=np.int32) + window_stats: list[SpeakerWindowStats] = [] for start_sample in starts: chunk = audio_padded[start_sample : start_sample + window_samples][ @@ -173,6 +271,7 @@ def compute_overlap_and_logprobs( if frame_end > num_frames: frame_end = num_frames log_probs = log_probs[: frame_end - frame_start] + window_stats.append(_speaker_window_stats(log_probs)) accum[frame_start:frame_end] += log_probs.astype(np.float64) counts[frame_start:frame_end] += 1 @@ -182,10 +281,14 @@ def compute_overlap_and_logprobs( argmax = avg_log_probs.argmax(axis=-1) speech_count = int((argmax >= 1).sum()) if speech_count == 0: - return 0.0, avg_log_probs + return OverlapInferenceResult(0.0, avg_log_probs, tuple(window_stats)) overlap_count = int(np.isin(argmax, OVERLAP_CLASSES).sum()) - return float(overlap_count / speech_count), avg_log_probs + return OverlapInferenceResult( + float(overlap_count / speech_count), + avg_log_probs, + tuple(window_stats), + ) def compute_overlap_fraction_for_wav(path: Path) -> float: diff --git a/solstone/talent/journal/contract/bundle.json b/solstone/talent/journal/contract/bundle.json index e53bde554..4a554b8c0 100644 --- a/solstone/talent/journal/contract/bundle.json +++ b/solstone/talent/journal/contract/bundle.json @@ -122,6 +122,24 @@ ], "type": "object" }, + "speaker_evidence": { + "description": "Windowed slot evidence decision for local diarization engagement. This records engagement, not clustering outcome: a multi value with no integer speaker labels means diarization engaged and clustering produced nothing or failed.", + "enum": [ + "single", + "multi", + "none" + ], + "type": "string" + }, + "speaker_evidence_multi_fraction": { + "description": "Fraction of speech-bearing pyannote windows with multi-speaker slot evidence. This records engagement evidence, not clustering outcome.", + "type": "number" + }, + "speaker_evidence_version": { + "const": "windowed-slots-v1", + "description": "Version of the windowed slot evidence decision contract. This records engagement, not clustering outcome: a multi value with no integer speaker labels means diarization engaged and clustering produced nothing or failed.", + "type": "string" + }, "topics": { "type": "array" }, diff --git a/tests/test_journal_contract.py b/tests/test_journal_contract.py index 5c620e74f..711e9fc1d 100644 --- a/tests/test_journal_contract.py +++ b/tests/test_journal_contract.py @@ -107,6 +107,20 @@ def test_contract_validator_accepts_audio_sound_tags_header() -> None: assert journal.validate_contract_file("audio.jsonl", audio, audio_schema) == [] +def test_contract_validator_accepts_audio_speaker_evidence_header() -> None: + bundle = journal.build_bundle() + audio_schema = bundle["schemas"]["audio-jsonl"]["schema"] + header = { + "raw": "audio.flac", + "speaker_evidence": "multi", + "speaker_evidence_multi_fraction": 0.125, + "speaker_evidence_version": "windowed-slots-v1", + } + audio = json.dumps(header).encode("utf-8") + b'\n{"start":"00:00:00","text":"hi"}\n' + + assert journal.validate_contract_file("audio.jsonl", audio, audio_schema) == [] + + def test_contract_validator_still_rejects_non_raw_floor_violations() -> None: bundle = journal.build_bundle() screen_schema = bundle["schemas"]["screen-jsonl"]["schema"] diff --git a/tests/test_transcribe.py b/tests/test_transcribe.py index 493986e92..4065e3c25 100644 --- a/tests/test_transcribe.py +++ b/tests/test_transcribe.py @@ -24,12 +24,30 @@ from solstone.observe.transcribe import ( build_statements_from_acoustic, ) from solstone.observe.transcribe.main import EMBEDDER_NAME, _statements_to_jsonl +from solstone.observe.transcribe.overlap import ( + OverlapInferenceResult, + SpeakerEvidenceDecision, + SpeakerWindowStats, +) from solstone.observe.utils import SAMPLE_RATE, AudioDecodeError, load_audio from solstone.observe.vad import VadResult from solstone.think.journal_io.errors import MalformedDataError from solstone.think.journal_io.npz import load_npz from solstone.think.media import AUDIO_EXTENSIONS +CLEAN_SINGLE_STATS = (SpeakerWindowStats(589, 1, 0),) +MULTI_STATS = (SpeakerWindowStats(589, 2, 300),) + + +def _overlap_result( + overlap_fraction: float, + avg_log_probs: np.ndarray | None = None, + window_stats: tuple[SpeakerWindowStats, ...] = CLEAN_SINGLE_STATS, +) -> OverlapInferenceResult: + if avg_log_probs is None: + avg_log_probs = np.zeros((589, 7), dtype=np.float32) + return OverlapInferenceResult(overlap_fraction, avg_log_probs, window_stats) + class TestBuildStatementsFromAcoustic: """Test building statements from acoustic segments.""" @@ -531,7 +549,7 @@ def test_process_audio_failed_embeddings_write_emits_failed_event(tmp_path): ), patch( "solstone.observe.transcribe.overlap.compute_overlap_and_logprobs", - return_value=(0.0, np.zeros((589, 7), dtype=np.float32)), + return_value=_overlap_result(0.0), ), patch("solstone.observe.transcribe.main.callosum_send") as mock_send, patch( @@ -600,7 +618,7 @@ def test_process_audio_embeddings_write_round_trips_without_lock(tmp_path): ), patch( "solstone.observe.transcribe.overlap.compute_overlap_and_logprobs", - return_value=(0.0, np.zeros((589, 7), dtype=np.float32)), + return_value=_overlap_result(0.0), ), patch("solstone.observe.transcribe.main.callosum_send") as mock_send, ): @@ -672,7 +690,7 @@ def test_process_audio_records_analyzed_processing(tmp_path): ), patch( "solstone.observe.transcribe.overlap.compute_overlap_and_logprobs", - return_value=(0.0, np.zeros((589, 7), dtype=np.float32)), + return_value=_overlap_result(0.0), ), patch( "solstone.observe.processing_record.now_iso_utc", @@ -791,7 +809,7 @@ def test_process_audio_diarizer_failure_is_fail_soft(tmp_path): patch("solstone.observe.transcribe.main._embed_statements", return_value=None), patch( "solstone.observe.transcribe.overlap.compute_overlap_and_logprobs", - return_value=(0.5, np.zeros((589, 7), dtype=np.float32)), + return_value=_overlap_result(0.5, window_stats=MULTI_STATS), ), patch( "solstone.observe.transcribe.diarize.diarize_auto_k", @@ -852,7 +870,7 @@ def test_process_audio_diarizes_parakeet_cpp_when_overlap_meets_threshold(tmp_pa patch("solstone.observe.transcribe.main._embed_statements", return_value=None), patch( "solstone.observe.transcribe.overlap.compute_overlap_and_logprobs", - return_value=(0.5, logprobs), + return_value=_overlap_result(0.5, logprobs, MULTI_STATS), ), patch( "solstone.observe.transcribe.diarize.diarize_auto_k", @@ -957,6 +975,22 @@ class TestJSONLFormat: # transcriber must keep emitting it. assert metadata["raw"] == "audio.flac" + def test_statements_to_jsonl_writes_speaker_evidence_decision_fields(self): + lines = _statements_to_jsonl( + [{"start": 1.0, "end": 2.0, "text": "Hello"}], + "audio.flac", + datetime(2026, 5, 22, 9, 0, 0), + {"model": "unit", "device": "cpu", "compute_type": "int8"}, + speaker_evidence=SpeakerEvidenceDecision("none", 0.0, 0.25), + ) + + metadata = json.loads(lines[0]) + + assert metadata["speaker_evidence"] == "none" + assert metadata["speaker_evidence_multi_fraction"] == 0.0 + assert metadata["speaker_evidence_version"] == "windowed-slots-v1" + assert "speaker_evidence_mean_window_overlap_share" not in metadata + def test_metadata_first_line(self): """First line should be metadata with 'raw' field.""" lines = [ diff --git a/tests/test_transcribe_empty_result.py b/tests/test_transcribe_empty_result.py index 1cdbeb4f3..ca05e1cea 100644 --- a/tests/test_transcribe_empty_result.py +++ b/tests/test_transcribe_empty_result.py @@ -18,6 +18,10 @@ from solstone.observe.processing_record import ( SCHEMA, STATE_EMPTY, ) +from solstone.observe.transcribe.overlap import ( + OverlapInferenceResult, + SpeakerWindowStats, +) from solstone.observe.utils import SAMPLE_RATE from solstone.observe.vad import VadResult from solstone.think.data_state import ( @@ -35,6 +39,16 @@ SOUND_TAGS = { "windows": 1, "tags": {"Music": 0.201, "Silence": 0.5}, } +CLEAN_SINGLE_STATS = (SpeakerWindowStats(589, 1, 0),) + + +def _overlap_result() -> OverlapInferenceResult: + return OverlapInferenceResult( + 0.0, + np.zeros((589, 7), dtype=np.float32), + CLEAN_SINGLE_STATS, + ) + SILENCE_SOUND_TAGS = { "engine": "ced.cpp v0.1.0", @@ -137,7 +151,7 @@ def test_process_audio_speech_writes_sound_tags_and_keeps_audio( patch("solstone.observe.transcribe.main._embed_statements", return_value=None), patch( "solstone.observe.transcribe.overlap.compute_overlap_and_logprobs", - return_value=(0.0, np.zeros((589, 7), dtype=np.float32)), + return_value=_overlap_result(), ), patch("solstone.observe.transcribe.main.callosum_send") as mock_send, ): diff --git a/tests/test_transcribe_overlap.py b/tests/test_transcribe_overlap.py index 49e90b76e..29a6dd2d8 100644 --- a/tests/test_transcribe_overlap.py +++ b/tests/test_transcribe_overlap.py @@ -20,14 +20,27 @@ class _Input: class _StubSession: - def __init__(self, log_probs: np.ndarray): - self._log_probs = log_probs.astype(np.float32) + def __init__(self, log_probs: np.ndarray | list[np.ndarray]): + if isinstance(log_probs, list): + self._log_probs = [item.astype(np.float32) for item in log_probs] + self._repeat = False + else: + self._log_probs = [log_probs.astype(np.float32)] + self._repeat = True + self._idx = 0 def get_inputs(self): return [_Input("input_values")] def run(self, _outputs, _inputs): - return [self._log_probs[None, :, :]] + if self._idx >= len(self._log_probs): + if not self._repeat: + raise AssertionError("unexpected pyannote run") + idx = len(self._log_probs) - 1 + else: + idx = self._idx + self._idx += 1 + return [self._log_probs[idx][None, :, :]] def _dominant_log_probs(classes: np.ndarray) -> np.ndarray: @@ -162,10 +175,106 @@ def test_compute_overlap_and_logprobs_returns_fraction_and_logprobs(monkeypatch) lambda: _StubSession(_dominant_log_probs(classes)), ) - result, log_probs = overlap.compute_overlap_and_logprobs( + result = overlap.compute_overlap_and_logprobs( np.zeros(10 * SAMPLE_RATE, dtype=np.float32) ) - assert result == pytest.approx(100 / 400) - assert log_probs.shape == (589, 7) - assert log_probs.dtype == np.float32 + assert result.overlap_fraction == pytest.approx(100 / 400) + assert result.avg_log_probs.shape == (589, 7) + assert result.avg_log_probs.dtype == np.float32 + assert result.window_stats == (overlap.SpeakerWindowStats(400, 1, 100),) + + +def test_decide_speaker_evidence_solo_one_slot_returns_single(): + from solstone.observe.transcribe import overlap + + decision = overlap.decide_speaker_evidence( + 0.0, + (overlap.SpeakerWindowStats(100, 1, 0),), + ) + + assert decision.speaker_evidence == "single" + assert decision.multi_window_fraction == 0.0 + + +def test_decide_speaker_evidence_slot_permuted_windows_return_single(monkeypatch): + from solstone.observe.transcribe import overlap + + monkeypatch.setattr( + overlap, + "_get_overlap_session", + lambda: _StubSession( + [ + _dominant_log_probs(np.full(589, 1, dtype=np.int64)), + _dominant_log_probs(np.full(589, 2, dtype=np.int64)), + ] + ), + ) + + result = overlap.compute_overlap_and_logprobs( + np.zeros(12 * SAMPLE_RATE, dtype=np.float32) + ) + decision = overlap.decide_speaker_evidence( + result.overlap_fraction, + result.window_stats, + ) + + assert len(result.window_stats) == 2 + assert {row.active_slot_count for row in result.window_stats} == {1} + assert decision.speaker_evidence == "single" + + +def test_decide_speaker_evidence_turn_taking_returns_multi(): + from solstone.observe.transcribe import overlap + + decision = overlap.decide_speaker_evidence( + 0.0, + (overlap.SpeakerWindowStats(100, 2, 0),), + ) + + assert decision.speaker_evidence == "multi" + + +def test_decide_speaker_evidence_overlap_heavy_returns_multi(): + from solstone.observe.transcribe import overlap + + decision = overlap.decide_speaker_evidence( + 0.8, + (overlap.SpeakerWindowStats(100, 1, 80),), + ) + + assert decision.speaker_evidence == "multi" + + +def test_decide_speaker_evidence_all_silence_returns_none(): + from solstone.observe.transcribe import overlap + + decision = overlap.decide_speaker_evidence( + 0.0, + (overlap.SpeakerWindowStats(0, 0, 0),), + ) + + assert decision.speaker_evidence == "none" + assert decision.multi_window_fraction == 0.0 + + +def test_decide_speaker_evidence_overlap_fraction_term_engages_multi(): + from solstone.observe.transcribe import overlap + + decision = overlap.decide_speaker_evidence( + 0.5, + (overlap.SpeakerWindowStats(100, 1, 0),), + ) + + assert decision.speaker_evidence == "multi" + + +def test_decide_speaker_evidence_branch_four_overlap_ambiguity_returns_multi(): + from solstone.observe.transcribe import overlap + + decision = overlap.decide_speaker_evidence( + 0.0, + (overlap.SpeakerWindowStats(100, 1, 5),), + ) + + assert decision.speaker_evidence == "multi" diff --git a/tests/test_transcribe_telemetry.py b/tests/test_transcribe_telemetry.py index 872073ac6..ba1a25528 100644 --- a/tests/test_transcribe_telemetry.py +++ b/tests/test_transcribe_telemetry.py @@ -17,6 +17,10 @@ from unittest.mock import MagicMock, patch import numpy as np import pytest +from solstone.observe.transcribe.overlap import ( + OverlapInferenceResult, + SpeakerWindowStats, +) from solstone.observe.utils import SAMPLE_RATE from solstone.observe.vad import VadResult from solstone.think.providers.parakeet_server import ParakeetServerNotReady @@ -24,6 +28,15 @@ from solstone.think.providers.parakeet_server import ParakeetServerNotReady # A string that exists nowhere but in the (mocked) transcript. If it shows up in a # serialized event, transcript content leaked into telemetry. TRANSCRIPT_SENTINEL = "zzq-secret-utterance-do-not-leak" +CLEAN_SINGLE_STATS = (SpeakerWindowStats(589, 1, 0),) + + +def _overlap_result() -> OverlapInferenceResult: + return OverlapInferenceResult( + 0.0, + np.zeros((589, 7), dtype=np.float32), + CLEAN_SINGLE_STATS, + ) @pytest.fixture @@ -91,7 +104,7 @@ def _run_success( patch("solstone.observe.transcribe.main._embed_statements", return_value=None), patch( "solstone.observe.transcribe.overlap.compute_overlap_and_logprobs", - return_value=(0.0, np.zeros((589, 7), dtype=np.float32)), + return_value=_overlap_result(), ), patch("solstone.observe.transcribe.main.callosum_send") as mock_send, ):