diff --git a/solstone/apps/speakers/candidate_tracker.py b/solstone/apps/speakers/candidate_tracker.py index e563fe437..d20323b45 100644 --- a/solstone/apps/speakers/candidate_tracker.py +++ b/solstone/apps/speakers/candidate_tracker.py @@ -138,6 +138,10 @@ class CandidateTracker: json.dumps(data, indent=2, sort_keys=True) + "\n", ) + def load_all_candidates(self) -> list[CandidateProfile]: + """Return all tracked speaker candidates without mutating state.""" + return sorted(self._candidates.values(), key=lambda item: item.cand_id) + def _new_id(self) -> int: cand_id = self._next_id self._next_id += 1 diff --git a/solstone/apps/speakers/owner.py b/solstone/apps/speakers/owner.py index c3af02931..682fbf707 100644 --- a/solstone/apps/speakers/owner.py +++ b/solstone/apps/speakers/owner.py @@ -7,6 +7,7 @@ from __future__ import annotations import json import logging +import os from dataclasses import dataclass from datetime import UTC, datetime from pathlib import Path @@ -21,7 +22,7 @@ from solstone.apps.speakers.encoder_config import ( OWNER_BOOTSTRAP_PROVISIONAL_GUARD_MIN_TAGS, OWNER_THRESHOLD, ) -from solstone.think.awareness import update_state +from solstone.think.awareness import get_current, update_state from solstone.think.entities.journal import ( ensure_journal_entity_memory, get_journal_principal, @@ -29,15 +30,18 @@ from solstone.think.entities.journal import ( ) from solstone.think.entities.voiceprints import load_entity_voiceprints_file from solstone.think.journal_io.errors import LockTimeout -from solstone.think.journal_io.npz import load_npz, save_npz -from solstone.think.utils import day_dirs, get_journal, segment_path +from solstone.think.journal_io.npz import load_npz, load_npz_row_count, save_npz +from solstone.think.utils import day_dirs, get_journal, segment_parse, segment_path if TYPE_CHECKING: import numpy as np + from solstone.apps.speakers.candidate_tracker import CandidateProfile + logger = logging.getLogger(__name__) -MAX_EMBEDDINGS = 30000 +OWNER_CANDIDATE_SOURCE = "candidate_pool" +OWNER_CANDIDATE_EXPANSION_MAX_EMBEDDINGS = 3000 LOW_QUALITY_REASON_TOO_FEW_STMTS = "too_few_stmts" LOW_QUALITY_REASON_MEDIAN_DURATION_TOO_SHORT = "median_duration_too_short" LOW_QUALITY_REASON_CLUSTER_TOO_DIFFUSE = "cluster_too_diffuse" @@ -59,6 +63,19 @@ class OwnerCentroid: streams: list[str] +@dataclass(frozen=True) +class OwnerCandidateExpansion: + """Expanded candidate-pool rows and journal-touch diagnostics.""" + + embeddings: np.ndarray + durations: list[float] + provenance: list[dict[str, Any]] + segments_checked: int + segments_available: int + embeddings_available: int + skipped: dict[str, int] + + def _mark_no_cluster(segment_count: int) -> None: """Record that detection ran but did not produce a usable cluster.""" update_state( @@ -169,10 +186,11 @@ def _routes_helpers(): return _load_embeddings_file, _normalize_embedding, _scan_segment_embeddings -def _owner_candidate_path() -> Path: +def _owner_candidate_path(*, create: bool = False) -> Path: """Return the temporary owner candidate NPZ path.""" awareness_dir = Path(get_journal()) / "awareness" - awareness_dir.mkdir(parents=True, exist_ok=True) + if create: + awareness_dir.mkdir(parents=True, exist_ok=True) return awareness_dir / "owner_candidate.npz" @@ -400,29 +418,67 @@ def load_manual_tag_stats(principal_id: str) -> dict[str, int]: } +def _safe_scandir(path: Path) -> list[os.DirEntry[str]]: + try: + return list(os.scandir(path)) + except OSError: + return [] + + +def _iter_audio_embedding_sources() -> list[tuple[str, str, str, Path, list[str]]]: + """Return segment/source candidates for inventory without opening speakers.json.""" + rows: list[tuple[str, str, str, Path, list[str]]] = [] + for day, day_path in day_dirs().items(): + for stream_entry in sorted( + _safe_scandir(day_path), key=lambda entry: entry.name + ): + if not stream_entry.is_dir(): + continue + stream = stream_entry.name + for segment_entry in sorted( + _safe_scandir(Path(stream_entry.path)), key=lambda entry: entry.name + ): + if not segment_entry.is_dir(): + continue + segment_key = segment_entry.name + if segment_parse(segment_key)[0] is None: + continue + segment_dir = Path(segment_entry.path) + sources = sorted( + npz_path.stem + for npz_path in segment_dir.glob("*.npz") + if npz_path.stem == "audio" or npz_path.stem.endswith("_audio") + ) + if sources: + rows.append((day, stream, segment_key, segment_dir, sources)) + return rows + + def load_owner_embedding_inventory() -> dict[str, int]: """Return journal-wide segment and embedding availability for owner bootstrap.""" - load_embeddings_file, _, scan_segment_embeddings = _routes_helpers() - segment_count = 0 embeddings_count = 0 overlap_cache: dict[Path, float] = {} - for day in day_dirs().keys(): - for segment in scan_segment_embeddings(day): - segment_count += 1 - segment_dir = segment_path(day, segment["key"], segment["stream"]) - for source in segment["sources"]: - jsonl_path = segment_dir / f"{source}.jsonl" - overlap = overlap_cache.setdefault( - jsonl_path, - _read_segment_overlap_fraction(jsonl_path), - ) - if overlap > NOISY_FLYWHEEL_OVERLAP_MAX: - continue - emb_data = load_embeddings_file(segment_dir / f"{source}.npz") - if emb_data is None: - continue - embeddings_count += int(len(emb_data[0])) + for ( + _day, + _stream, + _segment_key, + segment_dir, + sources, + ) in _iter_audio_embedding_sources(): + segment_count += 1 + for source in sources: + jsonl_path = segment_dir / f"{source}.jsonl" + overlap = overlap_cache.setdefault( + jsonl_path, + _read_segment_overlap_fraction(jsonl_path), + ) + if overlap > NOISY_FLYWHEEL_OVERLAP_MAX: + continue + row_count = load_npz_row_count(segment_dir / f"{source}.npz", "embeddings") + if row_count is None: + continue + embeddings_count += row_count return { "segments_available": segment_count, @@ -651,221 +707,263 @@ def load_owner_provisional_centroid(principal_id: str) -> np.ndarray | None: return centroid -def count_segments_with_embeddings() -> int: - """Count all journal segments that contain audio embedding files.""" - return load_owner_embedding_inventory()["segments_available"] +def _confirmed_owner_payload(centroid: OwnerCentroid) -> dict[str, Any]: + return { + "status": "confirmed", + "recommendation": "confirmed", + "cluster_size": centroid.cluster_size, + "streams_represented": len(centroid.streams), + "samples": [], + } -def _subsample_embeddings( - embeddings: np.ndarray, provenance: list[dict[str, Any]] -) -> tuple[np.ndarray, list[dict[str, Any]]]: - """Subsample embeddings proportionally across streams when over the limit.""" - import numpy as np +def _candidate_payload_from_awareness(voiceprint: dict[str, Any]) -> dict[str, Any]: + samples = voiceprint.get("samples") + return { + "status": "candidate", + "cluster_size": int(voiceprint.get("cluster_size") or 0), + "streams_represented": int(voiceprint.get("streams_represented") or 0), + "recommendation": str(voiceprint.get("recommendation") or "single_stream"), + "samples": samples if isinstance(samples, list) else [], + } - total = len(embeddings) - if total <= MAX_EMBEDDINGS: - return embeddings, provenance - - rng = np.random.default_rng(42) - stream_indices: dict[str, list[int]] = {} - for idx, record in enumerate(provenance): - stream_indices.setdefault(record["stream"], []).append(idx) - - allocations: dict[str, int] = {} - remainders: list[tuple[float, str]] = [] - allocated = 0 - - for stream, indices in stream_indices.items(): - count = len(indices) - proportional = MAX_EMBEDDINGS * count / total - allocation = min(count, int(proportional)) - allocations[stream] = allocation - allocated += allocation - remainders.append((proportional - allocation, stream)) - - remaining = MAX_EMBEDDINGS - allocated - for _, stream in sorted(remainders, reverse=True): - if remaining <= 0: - break - available = len(stream_indices[stream]) - allocations[stream] - if available <= 0: - continue - allocations[stream] += 1 - remaining -= 1 - selected_indices: list[int] = [] - for stream, indices in stream_indices.items(): - take = allocations[stream] - if take <= 0: - continue - if take >= len(indices): - selected_indices.extend(indices) - continue - sampled = rng.choice(indices, size=take, replace=False) - selected_indices.extend(int(idx) for idx in sampled) +def _candidate_no_cluster( + reason: str, + *, + segments_checked: int, + segments_available: int, + embeddings_available: int, +) -> dict[str, Any]: + _mark_no_cluster(segments_checked) + return { + "status": "no_cluster", + "reason": reason, + "segments_checked": int(segments_checked), + "segments_available": int(segments_available), + "embeddings_available": int(embeddings_available), + "recommendation": "no_cluster", + } + + +def _select_owner_candidate( + principal_id: str | None, +) -> tuple[CandidateProfile | None, str | None]: + from solstone.apps.speakers.candidate_tracker import CandidateTracker + + tracker = CandidateTracker() + pool_exists = tracker.store_path.exists() + candidates = tracker.load_all_candidates() + if not pool_exists: + return None, "pool_missing" + if not candidates: + return None, "pool_empty" + + eligible = [ + candidate + for candidate in candidates + if candidate.status != "rejected" + and ( + candidate.confirmed_entity is None + or candidate.confirmed_entity == principal_id + ) + ] + if not eligible: + return None, "no_eligible_candidate" + return ( + sorted( + eligible, + key=lambda item: ( + -item.n_intervals, + -item.total_duration_s, + item.cand_id, + ), + )[0], + None, + ) - selected_indices.sort() - sampled_embeddings = embeddings[selected_indices] - sampled_provenance = [provenance[idx] for idx in selected_indices] - return sampled_embeddings, sampled_provenance +def _round_robin_source_segments( + source_segments: list[dict[str, Any]], +) -> list[dict[str, Any]]: + grouped: dict[str, list[dict[str, Any]]] = {} + for source_segment in source_segments: + stream = str(source_segment.get("stream") or "") + grouped.setdefault(stream, []).append(source_segment) + + ordered: list[dict[str, Any]] = [] + positions = {stream: 0 for stream in grouped} + while True: + made_progress = False + for stream in sorted(grouped): + position = positions[stream] + if position >= len(grouped[stream]): + continue + ordered.append(grouped[stream][position]) + positions[stream] = position + 1 + made_progress = True + if not made_progress: + return ordered -def detect_owner_candidate() -> dict[str, Any]: - """Detect a likely owner voice centroid from journal embeddings.""" + +def _expand_owner_candidate( + candidate: CandidateProfile, + *, + max_embeddings: int = OWNER_CANDIDATE_EXPANSION_MAX_EMBEDDINGS, +) -> OwnerCandidateExpansion: + """Expand one candidate pool profile into usable owner bootstrap rows.""" import numpy as np - from sklearn.cluster import HDBSCAN - load_embeddings_file, normalize_embedding, scan_segment_embeddings = ( + from solstone.apps.speakers.attribution import _load_integer_speaker_labels + + load_embeddings_file, normalize_embedding, _scan_segment_embeddings = ( _routes_helpers() ) - segment_count = count_segments_with_embeddings() - - embedding_chunks: list[np.ndarray] = [] + embeddings_cache: dict[ + Path, tuple[np.ndarray, np.ndarray, np.ndarray | None] | None + ] = {} + labels_cache: dict[tuple[Path, str], dict[int, int]] = {} + fallback_cache: dict[Path, dict[int, float | None]] = {} overlap_cache: dict[Path, float] = {} + skipped: dict[str, int] = {} + seen_segments: set[tuple[str, str, str]] = set() + embedding_rows: list[np.ndarray] = [] + durations: list[float] = [] provenance: list[dict[str, Any]] = [] - for day in day_dirs().keys(): - for segment in scan_segment_embeddings(day): - stream = segment["stream"] - segment_key = segment["key"] - segment_dir = segment_path(day, segment_key, stream) + def skip(reason: str) -> None: + skipped[reason] = skipped.get(reason, 0) + 1 - for source in segment["sources"]: - jsonl_path = segment_dir / f"{source}.jsonl" - overlap = overlap_cache.setdefault( - jsonl_path, _read_segment_overlap_fraction(jsonl_path) - ) - if overlap > NOISY_FLYWHEEL_OVERLAP_MAX: - logger.info( - "owner bootstrap skip: overlap=%.3f exceeds %.2f at %s/%s/%s", - overlap, - NOISY_FLYWHEEL_OVERLAP_MAX, - day, - segment_key, - source, - ) - continue + for source_segment in _round_robin_source_segments(candidate.source_segments): + if len(embedding_rows) >= max_embeddings: + break - emb_data = load_embeddings_file(segment_dir / f"{source}.npz") - if emb_data is None: - continue + try: + day = str(source_segment["day"]) + stream = str(source_segment["stream"]) + segment_key = str(source_segment["segment_key"]) + source = str(source_segment["source"]) + cluster_label = int(source_segment["cluster_label"]) + except (KeyError, TypeError, ValueError): + skip("invalid_source_segment") + continue - embeddings, statement_ids, durations_data = emb_data - if len(embeddings) == 0: - continue + segment_triplet = (day, stream, segment_key) + if segment_triplet not in seen_segments: + seen_segments.add(segment_triplet) - fallback_durations = ( - {} - if durations_data is not None - else _fallback_statement_durations(segment_dir / f"{source}.jsonl") - ) - embedding_chunks.append(embeddings.astype(np.float32)) - provenance.extend( - { - "day": day, - "stream": stream, - "segment_key": segment_key, - "source": source, - "sentence_id": int(sid), - "duration_s": ( - float(durations_data[idx]) - if durations_data is not None - else fallback_durations.get(int(sid)) - ), - } - for idx, sid in enumerate(statement_ids) - ) + seg_dir = segment_path(day, segment_key, stream, create=False) + if not seg_dir.is_dir(): + skip("missing_segment_dir") + continue - if not embedding_chunks: - _mark_no_cluster(segment_count) - return { - "status": "no_embeddings", - "segments_available": segment_count, - "embeddings_available": 0, - "recommendation": "no_embeddings", - } + jsonl_path = seg_dir / f"{source}.jsonl" + overlap = overlap_cache.setdefault( + jsonl_path, + _read_segment_overlap_fraction(jsonl_path), + ) + if overlap > NOISY_FLYWHEEL_OVERLAP_MAX: + skip("noisy_overlap") + continue - embeddings_matrix = np.vstack(embedding_chunks) - embeddings_matrix, provenance = _subsample_embeddings(embeddings_matrix, provenance) + npz_path = seg_dir / f"{source}.npz" + emb_data = embeddings_cache.setdefault(npz_path, load_embeddings_file(npz_path)) + if emb_data is None: + skip("missing_or_invalid_npz") + continue + embeddings, statement_ids, durations_data = emb_data - if len(embeddings_matrix) < 50: - _mark_no_cluster(segment_count) - return { - "status": "low_data", - "segments_available": segment_count, - "embeddings_available": int(len(embeddings_matrix)), - "recommendation": "low_data", + labels_key = (seg_dir, source) + integer_labels = labels_cache.setdefault( + labels_key, + _load_integer_speaker_labels(seg_dir, source), + ) + if not integer_labels: + skip("missing_integer_labels") + continue + + statement_index = { + int(statement_id): idx for idx, statement_id in enumerate(statement_ids) } + fallback_durations = ( + {} + if durations_data is not None + else fallback_cache.setdefault( + jsonl_path, + _fallback_statement_durations(jsonl_path), + ) + ) - clusterer = HDBSCAN( - min_cluster_size=50, - min_samples=10, - metric="euclidean", - ) - clusterer.fit(embeddings_matrix) - labels = clusterer.labels_ + for sentence_id, speaker_label in sorted(integer_labels.items()): + if len(embedding_rows) >= max_embeddings: + break + if int(speaker_label) != cluster_label: + continue + matched_index = statement_index.get(int(sentence_id)) + if matched_index is None: + skip("sentence_id_absent") + continue + if matched_index >= len(embeddings): + skip("embedding_row_out_of_range") + continue - valid_labels = labels[labels != -1] - if len(valid_labels) == 0: - _mark_no_cluster(segment_count) - return { - "status": "no_clusters", - "segments_available": segment_count, - "embeddings_available": int(len(embeddings_matrix)), - "recommendation": "no_clusters", - } + normalized = normalize_embedding(embeddings[matched_index]) + if normalized is None: + skip("embedding_normalization_failed") + continue - largest_label = int(np.bincount(valid_labels).argmax()) - cluster_indices = np.flatnonzero(labels == largest_label) - if len(cluster_indices) == 0: - _mark_no_cluster(segment_count) - return { - "status": "no_clusters", - "segments_available": segment_count, - "embeddings_available": int(len(embeddings_matrix)), - "recommendation": "no_clusters", - } + duration_s: float | None + if durations_data is not None and matched_index < len(durations_data): + duration_s = float(durations_data[matched_index]) + else: + duration_s = fallback_durations.get(int(sentence_id)) + if duration_s is not None: + durations.append(float(duration_s)) + + embedding_rows.append(np.asarray(normalized, dtype=np.float32)) + provenance.append( + { + "day": day, + "stream": stream, + "segment_key": segment_key, + "source": source, + "sentence_id": int(sentence_id), + "duration_s": duration_s, + } + ) - cluster_embeddings = embeddings_matrix[cluster_indices] - embeddings_count = int(embeddings_matrix.shape[0]) - cluster_durations = [ - float(provenance[int(i)]["duration_s"]) - for i in cluster_indices - if provenance[int(i)].get("duration_s") is not None - ] - low_quality = _apply_owner_quality_gates( - cluster_embeddings, - cluster_durations, - segment_count, - embeddings_count, - source="hdbscan", + if not embedding_rows: + embeddings_matrix = np.empty((0, 0), dtype=np.float32) + else: + embeddings_matrix = np.vstack(embedding_rows).astype(np.float32, copy=False) + + return OwnerCandidateExpansion( + embeddings=embeddings_matrix, + durations=durations, + provenance=provenance, + segments_checked=len(seen_segments), + segments_available=int(candidate.n_segments), + embeddings_available=int(embeddings_matrix.shape[0]), + skipped=skipped, ) - if low_quality is not None: - return low_quality - centroid = normalize_embedding(np.mean(cluster_embeddings, axis=0)) - if centroid is None: - _mark_no_cluster(segment_count) - return { - "status": "no_clusters", - "segments_available": segment_count, - "embeddings_available": embeddings_count, - "recommendation": "no_clusters", - } - cluster_streams = {provenance[int(i)]["stream"] for i in cluster_indices} - streams_represented = len(cluster_streams) - cluster_size = int(cluster_embeddings.shape[0]) - recommendation = "ready" if streams_represented > 1 else "single_stream" - similarities = np.dot(cluster_embeddings, centroid) - sorted_cluster_positions = np.argsort(similarities)[::-1] +def _owner_candidate_samples( + embeddings: np.ndarray, + centroid: np.ndarray, + provenance: list[dict[str, Any]], +) -> list[dict[str, Any]]: + import numpy as np + + similarities = np.dot(embeddings, centroid) + sorted_positions = np.argsort(similarities)[::-1] samples: list[dict[str, Any]] = [] seen_segments: set[tuple[str, str, str]] = set() - for position in sorted_cluster_positions: - record = provenance[int(cluster_indices[position])] + for position in sorted_positions: + record = provenance[int(position)] segment_triplet = (record["day"], record["stream"], record["segment_key"]) if segment_triplet in seen_segments: continue @@ -881,31 +979,112 @@ def detect_owner_candidate() -> dict[str, Any]: ), } ) + if len(samples) == 3: + return samples + + for position in sorted_positions: + record = provenance[int(position)] + sample = { + **record, + "audio_url": _audio_url( + record["day"], + record["stream"], + record["segment_key"], + record["source"], + ), + } + if sample in samples: + continue + samples.append(sample) if len(samples) == 3: break + return samples - if len(samples) < 3: - for position in sorted_cluster_positions: - record = provenance[int(cluster_indices[position])] - sample = { - **record, - "audio_url": _audio_url( - record["day"], - record["stream"], - record["segment_key"], - record["source"], - ), - } - if sample in samples: - continue - samples.append(sample) - if len(samples) == 3: - break + +def detect_owner_candidate() -> dict[str, Any]: + """Detect a likely owner voice centroid from the candidate pool. + + The cheap statement-count prefilter opens no segment files, so its + low-quality payload reports the candidate's recorded n_segments and + n_intervals as segments_available and embeddings_available. + """ + import numpy as np + + confirmed = load_owner_centroid() + if confirmed is not None: + return _confirmed_owner_payload(confirmed) + + candidate_path = _owner_candidate_path() + voiceprint = get_current().get("voiceprint", {}) + if candidate_path.exists() and voiceprint.get("status") == "candidate": + return _candidate_payload_from_awareness(voiceprint) + + principal_id = _principal_id_or_none() + candidate, missing_reason = _select_owner_candidate(principal_id) + if candidate is None: + return _candidate_no_cluster( + missing_reason or "no_eligible_candidate", + segments_checked=0, + segments_available=0, + embeddings_available=0, + ) + + if candidate.n_intervals < OWNER_BOOTSTRAP_MIN_STMTS: + return _bail_low_quality( + LOW_QUALITY_REASON_TOO_FEW_STMTS, + candidate.n_intervals, + OWNER_BOOTSTRAP_MIN_STMTS, + candidate.n_segments, + candidate.n_intervals, + source=OWNER_CANDIDATE_SOURCE, + ) + + expansion = _expand_owner_candidate( + candidate, + max_embeddings=OWNER_CANDIDATE_EXPANSION_MAX_EMBEDDINGS, + ) + if expansion.embeddings_available == 0: + return _candidate_no_cluster( + "candidate_no_usable_embeddings", + segments_checked=expansion.segments_checked, + segments_available=expansion.segments_available, + embeddings_available=0, + ) + + low_quality = _apply_owner_quality_gates( + expansion.embeddings, + expansion.durations, + expansion.segments_checked, + expansion.embeddings_available, + source=OWNER_CANDIDATE_SOURCE, + ) + if low_quality is not None: + return low_quality + + _load_embeddings_file, normalize_embedding, _scan_segment_embeddings = ( + _routes_helpers() + ) + centroid = normalize_embedding(np.mean(expansion.embeddings, axis=0)) + if centroid is None: + return _candidate_no_cluster( + "candidate_centroid_unusable", + segments_checked=expansion.segments_checked, + segments_available=expansion.segments_available, + embeddings_available=expansion.embeddings_available, + ) + + cluster_streams = {record["stream"] for record in expansion.provenance} + streams_represented = len(cluster_streams) + cluster_size = int(expansion.embeddings.shape[0]) + recommendation = "ready" if streams_represented > 1 else "single_stream" + samples = _owner_candidate_samples( + expansion.embeddings, centroid, expansion.provenance + ) version = _iso_now() try: save_npz( - _owner_candidate_path(), + _owner_candidate_path(create=True), { "centroid": centroid.astype(np.float32), "cluster_size": np.array(cluster_size, dtype=np.int32), @@ -938,6 +1117,44 @@ def detect_owner_candidate() -> dict[str, Any]: } +def owner_detection_ready() -> dict[str, Any]: + """Check cheap owner voice candidate state without running detection.""" + if load_owner_centroid() is not None: + return {"ready": False, "reason": "centroid_exists"} + + voiceprint = get_current().get("voiceprint", {}) + rejected_at = voiceprint.get("rejected_at") + if rejected_at: + try: + rejection_time = datetime.fromisoformat(str(rejected_at)) + now = datetime.now(rejection_time.tzinfo) + days_since = (now - rejection_time).days + if days_since < 14: + return { + "ready": False, + "reason": "cooldown", + "days_remaining": 14 - days_since, + } + except (ValueError, TypeError): + pass + + if _owner_candidate_path().exists() and voiceprint.get("status") == "candidate": + if voiceprint.get("recommendation") == "ready": + return { + "ready": True, + "reason": "candidate_found", + "cluster_size": voiceprint.get("cluster_size"), + "streams_represented": voiceprint.get("streams_represented"), + "samples": voiceprint.get("samples", []), + } + return { + "ready": False, + "reason": voiceprint.get("recommendation") or "candidate_not_ready", + } + + return {"ready": False, "reason": "no_candidate"} + + def _load_owner_voiceprint_summary( principal_id: str, ) -> tuple[float | None, list[str]]: diff --git a/solstone/apps/speakers/routes.py b/solstone/apps/speakers/routes.py index 1b1fe1768..cdb16d3b8 100644 --- a/solstone/apps/speakers/routes.py +++ b/solstone/apps/speakers/routes.py @@ -61,6 +61,7 @@ from solstone.apps.speakers.owner import ( load_owner_bootstrap_diagnostics, load_owner_centroid, load_owner_provisional_centroid, + owner_detection_ready, reject_owner_candidate, ) from solstone.apps.speakers.status import get_speakers_status @@ -95,7 +96,7 @@ from solstone.convey.utils import ( safe_day_path, success_response, ) -from solstone.think.awareness import get_current, owner_detection_ready +from solstone.think.awareness import get_current from solstone.think.entities import find_matching_entity from solstone.think.entities.journal import ( ensure_journal_entity_memory, @@ -1464,7 +1465,7 @@ def api_owner_status() -> Any: return jsonify( { "status": "low_quality", - "source": voiceprint.get("source", "hdbscan"), + "source": voiceprint.get("source", "candidate_pool"), "low_quality_reason": voiceprint.get("low_quality_reason", ""), "observed_value": voiceprint.get("observed_value", 0.0), "threshold_value": voiceprint.get("threshold_value", 0.0), diff --git a/solstone/apps/speakers/status.py b/solstone/apps/speakers/status.py index 28381c907..f049b4e68 100644 --- a/solstone/apps/speakers/status.py +++ b/solstone/apps/speakers/status.py @@ -92,7 +92,7 @@ def _owner_section() -> dict[str, Any]: result["streams_represented"] = voiceprint.get("streams_represented") result["recommendation"] = voiceprint.get("recommendation") elif status == "low_quality": - result["source"] = voiceprint.get("source", "hdbscan") + result["source"] = voiceprint.get("source", "candidate_pool") result["low_quality_reason"] = voiceprint.get("low_quality_reason", "") result["observed_value"] = voiceprint.get("observed_value", 0.0) result["threshold_value"] = voiceprint.get("threshold_value", 0.0) diff --git a/solstone/apps/speakers/talent/speakers/SKILL.md b/solstone/apps/speakers/talent/speakers/SKILL.md index d8a5c5332..b3a7c4b06 100644 --- a/solstone/apps/speakers/talent/speakers/SKILL.md +++ b/solstone/apps/speakers/talent/speakers/SKILL.md @@ -75,8 +75,8 @@ Run owner voice candidate detection. Returns the candidate plus sample segments. Behavior notes: -- Only attempt when there are 50+ segments with embeddings across 3+ streams (check `owner-ready` first). -- If fewer segments exist, wait — don't mention speaker ID proactively until there's enough data. +- Only attempt after `owner-ready` reports a usable candidate, or when explicitly refreshing candidate-pool detection. +- Detection expands the existing speaker candidate pool; it does not scan and cluster the full journal. Example: @@ -126,7 +126,7 @@ Behavior notes: sol call speakers owner-ready ``` -Report whether owner voice detection should be surfaced right now (enough embeddings, enough streams, not in cooldown). +Report whether a persisted owner voice candidate should be surfaced right now. This is a cheap state check and does not run detection. Example: @@ -250,7 +250,7 @@ Bootstrap voiceprints from imported-media tracks where the participant roster is Check `speakers owner-ready` (or look at `speakers status owner`). If the owner centroid doesn't exist: -- If readiness passes (50+ segments with embeddings across 3+ streams): good time to try. Run `speakers detect`. +- If readiness reports a candidate: good time to try. Run `speakers detect`. - If not: wait. Don't mention speaker ID proactively until there's enough data. When you have a candidate, present it naturally: "I've been listening to your journal across your different devices and I think I can recognize your voice. Here are a few moments — does this sound right?" Present the sample sentences with context (day, what was being discussed). Don't play audio — show text and context. diff --git a/solstone/apps/speakers/tests/test_candidate_tracker.py b/solstone/apps/speakers/tests/test_candidate_tracker.py index 1a94aa63d..26a3d9ff6 100644 --- a/solstone/apps/speakers/tests/test_candidate_tracker.py +++ b/solstone/apps/speakers/tests/test_candidate_tracker.py @@ -145,6 +145,25 @@ def test_pool_persist_reload_round_trip(speakers_env, tmp_path): ] +def test_load_all_candidates_is_read_only(speakers_env, tmp_path): + env = speakers_env() + seg_dir = _write_labeled_segment( + env, + "20260101", + "090000_300", + {1: np.stack([_unit([0.0, 1.0])] * 3)}, + ) + store = tmp_path / "speaker_candidates.json" + tracker = CandidateTracker(store) + tracker.process_segment("20260101", "090000_300", STREAM, "mic_audio", seg_dir) + before = store.read_text(encoding="utf-8") + + candidates = CandidateTracker(store).load_all_candidates() + + assert [candidate.cand_id for candidate in candidates] == [1] + assert store.read_text(encoding="utf-8") == before + + def test_merge_threshold_updates_existing_candidate(speakers_env, tmp_path): env = speakers_env() store = tmp_path / "speaker_candidates.json" diff --git a/solstone/apps/speakers/tests/test_owner.py b/solstone/apps/speakers/tests/test_owner.py index aa39a2fec..0b7a61f41 100644 --- a/solstone/apps/speakers/tests/test_owner.py +++ b/solstone/apps/speakers/tests/test_owner.py @@ -11,7 +11,7 @@ from pathlib import Path import numpy as np from flask import Flask -from solstone.apps.speakers.encoder_config import OVERLAP_DETECTOR_ID +from solstone.apps.speakers.encoder_config import ENCODER_ID, OVERLAP_DETECTOR_ID from solstone.think.awareness import get_current, update_state @@ -107,15 +107,123 @@ def _normalize_rows(embeddings: np.ndarray) -> np.ndarray: return embeddings / np.where(norms == 0, 1.0, norms) -def _patch_hdbscan(monkeypatch, hdbscan) -> None: - import sklearn.cluster +def _write_labeled_segment( + env, + day: str, + segment_key: str, + clusters: dict[int, np.ndarray], + *, + stream: str = "test", + source: str = "mic_audio", + duration_s: float = 5.0, + overlap_fraction: float = 0.0, +) -> Path: + flat_dir, chronicle_dir = env._segment_dirs(day, segment_key, stream=stream) + embeddings: list[np.ndarray] = [] + statement_ids: list[int] = [] + durations: list[float] = [] + labels: list[int] = [] + sentence_id = 1 + for cluster_label, cluster_embeddings in clusters.items(): + for embedding in cluster_embeddings: + embeddings.append(embedding) + statement_ids.append(sentence_id) + durations.append(duration_s) + labels.append(cluster_label) + sentence_id += 1 + + lines = [ + json.dumps( + { + "raw": f"{source}.flac", + "model": "test", + "overlap_fraction": overlap_fraction, + "overlap_detector": OVERLAP_DETECTOR_ID, + } + ) + ] + for sid, cluster_label in zip(statement_ids, labels): + lines.append( + json.dumps( + { + "start": "09:00:00", + "text": f"sentence {sid}", + "speaker": int(cluster_label), + } + ) + ) - monkeypatch.setattr(sklearn.cluster, "HDBSCAN", hdbscan) + for seg_dir in (flat_dir, chronicle_dir): + (seg_dir / f"{source}.jsonl").write_text( + "\n".join(lines) + "\n", + encoding="utf-8", + ) + np.savez_compressed( + seg_dir / f"{source}.npz", + embeddings=np.stack(embeddings).astype(np.float32), + statement_ids=np.array(statement_ids, dtype=np.int32), + durations_s=np.array(durations, dtype=np.float32), + encoder=np.array(ENCODER_ID), + ) + (seg_dir / f"{source}.flac").write_bytes(b"") + return chronicle_dir - from solstone.apps.speakers import owner as owner_module - if hasattr(owner_module, "HDBSCAN"): - monkeypatch.setattr(owner_module, "HDBSCAN", hdbscan) +def _source_segment( + day: str, + segment_key: str, + *, + stream: str, + source: str = "mic_audio", + cluster_label: int = 1, +) -> dict[str, object]: + return { + "day": day, + "stream": stream, + "segment_key": segment_key, + "source": source, + "cluster_label": cluster_label, + } + + +def _candidate_record( + cand_id: int, + source_segments: list[dict[str, object]], + *, + n_intervals: int, + n_segments: int | None = None, + total_duration_s: float = 300.0, + status: str = "pending", + confirmed_entity: str | None = None, +) -> dict[str, object]: + centroid = np.zeros(256, dtype=np.float32) + centroid[0] = 1.0 + return { + "cand_id": cand_id, + "centroid": centroid.astype(float).tolist(), + "n_segments": n_segments if n_segments is not None else len(source_segments), + "n_intervals": n_intervals, + "total_duration_s": total_duration_s, + "source_segments": source_segments, + "confirmed_entity": confirmed_entity, + "status": status, + } + + +def _write_candidate_pool( + journal: Path, + candidates: list[dict[str, object]], +) -> Path: + path = journal / "awareness" / "speaker_candidates.json" + path.parent.mkdir(parents=True, exist_ok=True) + next_id = ( + max((int(candidate["cand_id"]) for candidate in candidates), default=0) + 1 + ) + path.write_text( + json.dumps({"next_id": next_id, "candidates": candidates}, indent=2) + "\n", + encoding="utf-8", + ) + return path def _save_manual_owner_tags( @@ -174,99 +282,115 @@ def _save_manual_owner_tags( return segment_dir -def test_count_segments_with_embeddings(speakers_env): - from solstone.apps.speakers.owner import count_segments_with_embeddings +def test_load_owner_embedding_inventory_counts_without_materializing( + speakers_env, monkeypatch +): + import solstone.think.journal_io.npz as npz_io + from solstone.apps.speakers import owner as owner_module + from solstone.apps.speakers.owner import load_owner_embedding_inventory + from solstone.apps.speakers.routes import ( + _load_embeddings_file, + _scan_segment_embeddings, + ) + from solstone.think.utils import segment_path env = speakers_env() - env.create_segment("20240101", "090000_300", ["mic_audio"]) - env.create_segment("20240101", "091000_300", ["sys_audio"]) - env.create_segment("20240102", "090000_300", ["audio"]) + _write_labeled_segment( + env, + "20240101", + "090000_300", + {1: _owner_embeddings(4, np.random.default_rng(1))}, + stream="mic", + ) + _write_labeled_segment( + env, + "20240101", + "091000_300", + {1: _owner_embeddings(6, np.random.default_rng(2))}, + stream="sys", + ) - assert count_segments_with_embeddings() == 3 + reference_segments = 0 + reference_embeddings = 0 + for segment in _scan_segment_embeddings("20240101"): + reference_segments += 1 + segment_dir = segment_path("20240101", segment["key"], segment["stream"]) + for source in segment["sources"]: + emb_data = _load_embeddings_file(segment_dir / f"{source}.npz") + assert emb_data is not None + reference_embeddings += int(len(emb_data[0])) + + def fail_materialize(*args, **kwargs): + raise AssertionError("inventory materialized embedding arrays") + + monkeypatch.setattr(owner_module, "_routes_helpers", fail_materialize) + monkeypatch.setattr(owner_module, "load_npz", fail_materialize) + monkeypatch.setattr(npz_io, "load_npz", fail_materialize) + + assert load_owner_embedding_inventory() == { + "segments_available": reference_segments, + "embeddings_available": reference_embeddings, + } -def test_detect_owner_insufficient_segments(speakers_env): +def test_detect_owner_no_candidate_pool_marks_no_cluster(speakers_env): from solstone.apps.speakers.owner import detect_owner_candidate env = speakers_env() - rng = np.random.default_rng(1) - for idx in range(10): - _write_segment( - env.journal, - "20240101", - "mic", - f"{9 + idx:02d}0000_300", - "audio", - _owner_embeddings(1, rng), - ) + _write_labeled_segment( + env, + "20240101", + "090000_300", + {1: _owner_embeddings(40, np.random.default_rng(1))}, + stream="mic", + ) result = detect_owner_candidate() - assert result["status"] == "low_data" - assert result["segments_available"] == 10 - assert result["embeddings_available"] == 10 - assert result["recommendation"] == "low_data" - - -def test_detect_owner_no_cluster(speakers_env): - from solstone.apps.speakers.owner import detect_owner_candidate - env = speakers_env() - for idx in range(50): - embedding = np.zeros((1, 256), dtype=np.float32) - embedding[0, idx] = 1.0 - _write_segment( - env.journal, - "20240101", - "mic", - f"{9 + idx // 12:02d}{(idx % 12) * 5:02d}00_300", - "audio", - embedding, - ) - - result = detect_owner_candidate() - assert result["status"] == "no_clusters" - assert result["segments_available"] == 50 - assert result["recommendation"] == "no_clusters" + assert result["status"] == "no_cluster" + assert result["reason"] == "pool_missing" + assert result["recommendation"] == "no_cluster" assert get_current()["voiceprint"]["status"] == "no_cluster" -def test_detect_owner_basic(speakers_env): +def test_detect_owner_candidate_pool_ready(speakers_env): from solstone.apps.speakers.owner import detect_owner_candidate env = speakers_env() rng = np.random.default_rng(42) - - for idx in range(55): - hour = 9 + (idx // 12) - minute = (idx % 12) * 5 - stream = "mic" if idx % 2 == 0 else "sys" - _write_segment( - env.journal, - "20240101", - stream, - f"{hour:02d}{minute:02d}00_300", - "audio", - _owner_embeddings(2, rng), - ) - - for idx in range(50): - hour = 9 + (idx // 12) - minute = (idx % 12) * 5 - stream = "other" if idx % 2 == 0 else "other_sys" - _write_segment( - env.journal, - "20240102", - stream, - f"{hour:02d}{minute:02d}00_300", - "audio", - _other_cluster_embeddings(2), - ) + _write_labeled_segment( + env, + "20240101", + "090000_300", + {1: _owner_embeddings(20, rng)}, + stream="mic", + ) + _write_labeled_segment( + env, + "20240101", + "091000_300", + {1: _owner_embeddings(20, rng)}, + stream="sys", + ) + _write_candidate_pool( + env.journal, + [ + _candidate_record( + 1, + [ + _source_segment("20240101", "090000_300", stream="mic"), + _source_segment("20240101", "091000_300", stream="sys"), + ], + n_intervals=40, + total_duration_s=200.0, + ) + ], + ) result = detect_owner_candidate() - assert result is not None assert result["status"] == "candidate" - assert result["cluster_size"] >= 50 + assert result["cluster_size"] == 40 assert result["streams_represented"] == 2 assert result["recommendation"] == "ready" assert len(result["samples"]) == 3 @@ -274,180 +398,280 @@ def test_detect_owner_basic(speakers_env): assert get_current()["voiceprint"]["status"] == "candidate" -def test_low_quality_too_few_stmts(speakers_env, monkeypatch): +def test_detect_owner_candidate_selection_skips_rejected_and_non_principal( + speakers_env, +): from solstone.apps.speakers.owner import detect_owner_candidate - def stub_hdbscan(labels: np.ndarray): - class StubHDBSCAN: - def __init__(self, **kwargs): - self.labels_ = np.asarray(labels, dtype=np.int32) - - def fit(self, embeddings: np.ndarray): - assert embeddings.shape[0] == len(self.labels_) - return self - - return StubHDBSCAN - env = speakers_env() - rng = np.random.default_rng(0) - embeddings = _owner_embeddings(60, rng) - _write_segment( - env.journal, + env.create_entity("Self Person", is_principal=True) + _write_labeled_segment( + env, "20240101", - "mic", "090000_300", - "audio", - embeddings, - durations_s=np.full(60, 2.0, dtype=np.float32), + {1: _owner_embeddings(40, np.random.default_rng(3))}, + stream="mic", ) - - labels = np.concatenate( + _write_candidate_pool( + env.journal, [ - np.zeros(29, dtype=np.int32), - np.full(31, -1, dtype=np.int32), - ] + _candidate_record( + 1, + [_source_segment("20240101", "090000_300", stream="missing")], + n_intervals=100, + status="rejected", + ), + _candidate_record( + 2, + [_source_segment("20240101", "090000_300", stream="missing")], + n_intervals=90, + confirmed_entity="someone_else", + ), + _candidate_record( + 3, + [_source_segment("20240101", "090000_300", stream="mic")], + n_intervals=40, + ), + ], ) - _patch_hdbscan(monkeypatch, stub_hdbscan(labels)) result = detect_owner_candidate() - assert result["status"] == "low_quality" - assert result["recommendation"] == "low_quality" - assert result["low_quality_reason"] == "too_few_stmts" - assert get_current()["voiceprint"]["status"] == "low_quality" - assert not _candidate_path(env.journal).exists() + assert result["status"] == "candidate" + assert result["cluster_size"] == 40 -def test_low_quality_median_duration_too_short(speakers_env, monkeypatch): +def test_detect_owner_candidate_prefilter_avoids_npz_load(speakers_env, monkeypatch): + from solstone.apps.speakers import owner as owner_module + from solstone.apps.speakers.encoder_config import OWNER_BOOTSTRAP_MIN_STMTS from solstone.apps.speakers.owner import detect_owner_candidate - def stub_hdbscan(labels: np.ndarray): - class StubHDBSCAN: - def __init__(self, **kwargs): - self.labels_ = np.asarray(labels, dtype=np.int32) - - def fit(self, embeddings: np.ndarray): - assert embeddings.shape[0] == len(self.labels_) - return self - - return StubHDBSCAN - env = speakers_env() - rng = np.random.default_rng(1) - embeddings = _owner_embeddings(60, rng) - _write_segment( + _write_candidate_pool( env.journal, - "20240101", - "mic", - "090000_300", - "audio", - embeddings, - durations_s=np.full(60, 1.0, dtype=np.float32), + [ + _candidate_record( + 1, + [_source_segment("20240101", "090000_300", stream="mic")], + n_intervals=1, + ) + ], ) - _patch_hdbscan(monkeypatch, stub_hdbscan(np.zeros(60, dtype=np.int32))) + def fail_materialize(*args, **kwargs): + raise AssertionError("prefilter opened segment embeddings") + + monkeypatch.setattr( + owner_module, + "_expand_owner_candidate", + lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("expanded")), + ) + monkeypatch.setattr(owner_module, "_routes_helpers", fail_materialize) + monkeypatch.setattr(owner_module, "load_npz", fail_materialize) result = detect_owner_candidate() assert result["status"] == "low_quality" - assert result["recommendation"] == "low_quality" - assert result["low_quality_reason"] == "median_duration_too_short" - assert result["observed_value"] < 1.5 - assert get_current()["voiceprint"]["status"] == "low_quality" - assert not _candidate_path(env.journal).exists() + assert result["source"] == "candidate_pool" + assert result["low_quality_reason"] == "too_few_stmts" + assert result["observed_value"] == 1.0 + assert result["threshold_value"] == float(OWNER_BOOTSTRAP_MIN_STMTS) + assert result["segments_available"] == 1 + assert result["embeddings_available"] == 1 -def test_low_quality_cluster_too_diffuse(speakers_env, monkeypatch): +def test_detect_owner_candidate_round_robin_prevents_stream_starvation( + speakers_env, monkeypatch +): + from solstone.apps.speakers import owner as owner_module from solstone.apps.speakers.owner import detect_owner_candidate - def stub_hdbscan(labels: np.ndarray): - class StubHDBSCAN: - def __init__(self, **kwargs): - self.labels_ = np.asarray(labels, dtype=np.int32) + env = speakers_env() + rng = np.random.default_rng(7) + source_segments: list[dict[str, object]] = [] + for idx in range(4): + segment_key = f"09{idx:02d}00_300" + _write_labeled_segment( + env, + "20240101", + segment_key, + {1: _owner_embeddings(20, rng)}, + stream="a_stream", + ) + source_segments.append( + _source_segment("20240101", segment_key, stream="a_stream") + ) + _write_labeled_segment( + env, + "20240101", + "100000_300", + {1: _owner_embeddings(20, rng)}, + stream="b_stream", + ) + source_segments.append(_source_segment("20240101", "100000_300", stream="b_stream")) + _write_candidate_pool( + env.journal, + [ + _candidate_record( + 1, source_segments, n_intervals=100, total_duration_s=500.0 + ) + ], + ) + monkeypatch.setattr( + owner_module, + "OWNER_CANDIDATE_EXPANSION_MAX_EMBEDDINGS", + 40, + ) - def fit(self, embeddings: np.ndarray): - assert embeddings.shape[0] == len(self.labels_) - return self + result = detect_owner_candidate() - return StubHDBSCAN + assert result["status"] == "candidate" + assert result["cluster_size"] == 40 + assert result["streams_represented"] == 2 + assert result["recommendation"] == "ready" + + +def test_low_quality_too_few_stmts_from_candidate_pool(speakers_env): + from solstone.apps.speakers.owner import detect_owner_candidate env = speakers_env() - rng = np.random.default_rng(0) - template = np.ones((60, 256), dtype=np.float32) - embeddings = template + rng.normal(scale=2.0, size=(60, 256)).astype(np.float32) - _write_segment( - env.journal, + _write_labeled_segment( + env, "20240101", - "mic", "090000_300", - "audio", - embeddings, - durations_s=np.full(60, 1.6, dtype=np.float32), + {1: _owner_embeddings(10, np.random.default_rng(0))}, + stream="mic", + ) + _write_candidate_pool( + env.journal, + [ + _candidate_record( + 1, + [_source_segment("20240101", "090000_300", stream="mic")], + n_intervals=40, + ) + ], ) - - _patch_hdbscan(monkeypatch, stub_hdbscan(np.zeros(60, dtype=np.int32))) result = detect_owner_candidate() assert result["status"] == "low_quality" + assert result["source"] == "candidate_pool" assert result["recommendation"] == "low_quality" - assert result["low_quality_reason"] == "cluster_too_diffuse" - assert result["observed_value"] < 0.30 - assert get_current()["voiceprint"]["status"] == "low_quality" + assert result["low_quality_reason"] == "too_few_stmts" + assert get_current()["voiceprint"]["source"] == "candidate_pool" assert not _candidate_path(env.journal).exists() -def test_detect_owner_candidate_excludes_chaotic_segments(speakers_env, monkeypatch): +def test_detect_owner_candidate_skips_noisy_source_segments(speakers_env): from solstone.apps.speakers.owner import detect_owner_candidate - class StubHDBSCAN: - def __init__(self, **kwargs): - self.labels_ = np.zeros(60, dtype=np.int32) - - def fit(self, embeddings: np.ndarray): - assert embeddings.shape[0] == 60 - return self - env = speakers_env() rng = np.random.default_rng(2) - clean_dir = _write_segment( - env.journal, + _write_labeled_segment( + env, "20240101", - "mic", "090000_300", - "audio", - _owner_embeddings(60, rng), - durations_s=np.full(60, 2.0, dtype=np.float32), + {1: _owner_embeddings(40, rng)}, + stream="mic", + overlap_fraction=0.20, ) - _rewrite_segment_header( - clean_dir, - "audio", - overlap_fraction=0.05, - overlap_detector=OVERLAP_DETECTOR_ID, + _write_labeled_segment( + env, + "20240101", + "091000_300", + {1: _owner_embeddings(40, rng)}, + stream="mic", + overlap_fraction=0.0, ) - - chaotic_dir = _write_segment( + _write_candidate_pool( env.journal, - "20240102", - "mic", - "090000_300", - "audio", - _owner_embeddings(60, rng), - durations_s=np.full(60, 2.0, dtype=np.float32), + [ + _candidate_record( + 1, + [ + _source_segment("20240101", "090000_300", stream="mic"), + _source_segment("20240101", "091000_300", stream="mic"), + ], + n_intervals=80, + total_duration_s=400.0, + ) + ], ) - _rewrite_segment_header( - chaotic_dir, - "audio", - overlap_fraction=0.20, - overlap_detector=OVERLAP_DETECTOR_ID, + + result = detect_owner_candidate() + + assert result["status"] == "candidate" + assert result["cluster_size"] == 40 + assert result["recommendation"] == "single_stream" + + +def test_detect_owner_candidate_reuses_persisted_candidate(speakers_env, monkeypatch): + from solstone.apps.speakers import owner as owner_module + from solstone.apps.speakers.encoder_config import OWNER_THRESHOLD + from solstone.apps.speakers.owner import detect_owner_candidate + + env = speakers_env() + candidate_path = _candidate_path(env.journal) + candidate_path.parent.mkdir(parents=True, exist_ok=True) + version = np.array("2026-03-19T12:00:00Z") + np.savez_compressed( + candidate_path, + centroid=_normalized(np.array([1.0] + [0.0] * 255, dtype=np.float32)), + cluster_size=np.array(40, dtype=np.int32), + threshold=np.array(OWNER_THRESHOLD, dtype=np.float32), + version=version, + ) + update_state( + "voiceprint", + { + "status": "candidate", + "cluster_size": 40, + "streams_represented": 2, + "recommendation": "ready", + "samples": [{"day": "20240101"}], + }, + ) + monkeypatch.setattr( + owner_module, + "_expand_owner_candidate", + lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("recomputed")), ) - _patch_hdbscan(monkeypatch, StubHDBSCAN) + result = detect_owner_candidate() + + assert result == { + "status": "candidate", + "cluster_size": 40, + "streams_represented": 2, + "recommendation": "ready", + "samples": [{"day": "20240101"}], + } + with np.load(candidate_path, allow_pickle=False) as data: + assert str(np.asarray(data["version"]).item()) == str(version.item()) + + +def test_detect_owner_candidate_confirmed_short_circuit(speakers_env): + from solstone.apps.speakers.encoder_config import OWNER_THRESHOLD + from solstone.apps.speakers.owner import detect_owner_candidate + + env = speakers_env() + principal_dir = env.create_entity("Self Person", is_principal=True) + centroid = _normalized(np.array([1.0] + [0.0] * 255, dtype=np.float32)) + np.savez_compressed( + principal_dir / "owner_centroid.npz", + centroid=centroid, + cluster_size=np.array(60, dtype=np.int32), + threshold=np.array(OWNER_THRESHOLD, dtype=np.float32), + last_refreshed_at=np.array("2026-03-15T12:00:00Z"), + ) result = detect_owner_candidate() - assert result["status"] == "candidate" + assert result["status"] == "confirmed" + assert result["recommendation"] == "confirmed" assert result["cluster_size"] == 60 + assert result["samples"] == [] def test_bootstrap_owner_from_manual_tags_confirms(speakers_env): @@ -738,6 +962,184 @@ def test_load_owner_centroid_success(speakers_env): assert loaded.streams == [] +def test_owner_detection_ready_not_ready_when_centroid_exists(speakers_env): + from solstone.apps.speakers.encoder_config import OWNER_THRESHOLD + from solstone.apps.speakers.owner import owner_detection_ready + + env = speakers_env() + principal_dir = env.create_entity("Self Person", is_principal=True) + centroid = _normalized(np.array([1.0] + [0.0] * 255, dtype=np.float32)) + np.savez_compressed( + principal_dir / "owner_centroid.npz", + centroid=centroid, + cluster_size=np.array(60, dtype=np.int32), + threshold=np.array(OWNER_THRESHOLD, dtype=np.float32), + last_refreshed_at=np.array("2026-03-15T12:00:00Z"), + ) + + result = owner_detection_ready() + + assert result["ready"] is False + assert result["reason"] == "centroid_exists" + + +def test_owner_detection_ready_not_ready_during_cooldown(speakers_env, monkeypatch): + from datetime import datetime + + from solstone.apps.speakers import owner as owner_module + from solstone.apps.speakers.owner import owner_detection_ready + + speakers_env() + update_state("voiceprint", {"rejected_at": datetime.now().isoformat()}) + monkeypatch.setattr( + owner_module, + "detect_owner_candidate", + lambda: (_ for _ in ()).throw(AssertionError("called detection")), + ) + + result = owner_detection_ready() + + assert result["ready"] is False + assert result["reason"] == "cooldown" + assert result["days_remaining"] == 14 + + +def test_owner_detection_ready_reads_persisted_candidate(speakers_env, monkeypatch): + from solstone.apps.speakers import owner as owner_module + from solstone.apps.speakers.encoder_config import OWNER_THRESHOLD + from solstone.apps.speakers.owner import owner_detection_ready + + env = speakers_env() + candidate_path = _candidate_path(env.journal) + candidate_path.parent.mkdir(parents=True, exist_ok=True) + np.savez_compressed( + candidate_path, + centroid=_normalized(np.array([1.0] + [0.0] * 255, dtype=np.float32)), + cluster_size=np.array(40, dtype=np.int32), + threshold=np.array(OWNER_THRESHOLD, dtype=np.float32), + version=np.array("2026-03-19T12:00:00Z"), + ) + update_state( + "voiceprint", + { + "status": "candidate", + "cluster_size": 40, + "streams_represented": 2, + "recommendation": "ready", + "samples": [{"day": "20240101"}], + }, + ) + monkeypatch.setattr( + owner_module, + "detect_owner_candidate", + lambda: (_ for _ in ()).throw(AssertionError("called detection")), + ) + + result = owner_detection_ready() + + assert result["ready"] is True + assert result["reason"] == "candidate_found" + assert result["cluster_size"] == 40 + assert result["streams_represented"] == 2 + assert result["samples"] == [{"day": "20240101"}] + + +def test_owner_detection_ready_preserves_single_stream_not_ready( + speakers_env, monkeypatch +): + from solstone.apps.speakers import owner as owner_module + from solstone.apps.speakers.encoder_config import OWNER_THRESHOLD + from solstone.apps.speakers.owner import owner_detection_ready + + env = speakers_env() + candidate_path = _candidate_path(env.journal) + candidate_path.parent.mkdir(parents=True, exist_ok=True) + np.savez_compressed( + candidate_path, + centroid=_normalized(np.array([1.0] + [0.0] * 255, dtype=np.float32)), + cluster_size=np.array(40, dtype=np.int32), + threshold=np.array(OWNER_THRESHOLD, dtype=np.float32), + version=np.array("2026-03-19T12:00:00Z"), + ) + update_state( + "voiceprint", + { + "status": "candidate", + "cluster_size": 40, + "streams_represented": 1, + "recommendation": "single_stream", + "samples": [], + }, + ) + monkeypatch.setattr( + owner_module, + "detect_owner_candidate", + lambda: (_ for _ in ()).throw(AssertionError("called detection")), + ) + + result = owner_detection_ready() + + assert result["ready"] is False + assert result["reason"] == "single_stream" + + +def test_owner_detection_ready_no_candidate_data(speakers_env, monkeypatch): + from solstone.apps.speakers import owner as owner_module + from solstone.apps.speakers.owner import owner_detection_ready + + speakers_env() + monkeypatch.setattr( + owner_module, + "detect_owner_candidate", + lambda: (_ for _ in ()).throw(AssertionError("called detection")), + ) + + result = owner_detection_ready() + + assert result == {"ready": False, "reason": "no_candidate"} + + +def test_owner_detection_ready_cooldown_expired_allows_candidate( + speakers_env, monkeypatch +): + from datetime import datetime, timedelta + + from solstone.apps.speakers import owner as owner_module + from solstone.apps.speakers.encoder_config import OWNER_THRESHOLD + from solstone.apps.speakers.owner import owner_detection_ready + + env = speakers_env() + candidate_path = _candidate_path(env.journal) + candidate_path.parent.mkdir(parents=True, exist_ok=True) + np.savez_compressed( + candidate_path, + centroid=_normalized(np.array([1.0] + [0.0] * 255, dtype=np.float32)), + cluster_size=np.array(40, dtype=np.int32), + threshold=np.array(OWNER_THRESHOLD, dtype=np.float32), + version=np.array("2026-03-19T12:00:00Z"), + ) + update_state( + "voiceprint", + { + "status": "candidate", + "rejected_at": (datetime.now() - timedelta(days=15)).isoformat(), + "cluster_size": 40, + "streams_represented": 2, + "recommendation": "ready", + "samples": [], + }, + ) + monkeypatch.setattr( + owner_module, + "detect_owner_candidate", + lambda: (_ for _ in ()).throw(AssertionError("called detection")), + ) + + result = owner_detection_ready() + + assert result["ready"] is True + + def test_classify_sentences_no_centroid(speakers_env): from solstone.apps.speakers.owner import classify_sentences @@ -908,7 +1310,7 @@ def test_api_owner_status_low_quality(speakers_env): assert response.status_code == 200 assert response.get_json() == { "status": "low_quality", - "source": "hdbscan", + "source": "candidate_pool", "low_quality_reason": "too_few_stmts", "observed_value": 5, "threshold_value": 30, @@ -1037,30 +1439,33 @@ def test_api_owner_detect(speakers_env): env = speakers_env() rng = np.random.default_rng(42) - for idx in range(55): - hour = 9 + (idx // 12) - minute = (idx % 12) * 5 - stream = "mic" if idx % 2 == 0 else "sys" - _write_segment( - env.journal, - "20240101", - stream, - f"{hour:02d}{minute:02d}00_300", - "audio", - _owner_embeddings(2, rng), - ) - for idx in range(50): - hour = 9 + (idx // 12) - minute = (idx % 12) * 5 - stream = "other" if idx % 2 == 0 else "other_sys" - _write_segment( - env.journal, - "20240102", - stream, - f"{hour:02d}{minute:02d}00_300", - "audio", - _other_cluster_embeddings(2), - ) + _write_labeled_segment( + env, + "20240101", + "090000_300", + {1: _owner_embeddings(20, rng)}, + stream="mic", + ) + _write_labeled_segment( + env, + "20240101", + "091000_300", + {1: _owner_embeddings(20, rng)}, + stream="sys", + ) + _write_candidate_pool( + env.journal, + [ + _candidate_record( + 1, + [ + _source_segment("20240101", "090000_300", stream="mic"), + _source_segment("20240101", "091000_300", stream="sys"), + ], + n_intervals=40, + ) + ], + ) app = Flask(__name__) app.register_blueprint(speakers_bp) @@ -1071,9 +1476,112 @@ def test_api_owner_detect(speakers_env): data = response.get_json() assert response.status_code == 200 assert data["status"] == "candidate" - assert data["cluster_size"] >= 50 - assert "streams_represented" in data - assert "recommendation" in data + assert data["cluster_size"] == 40 + assert data["streams_represented"] == 2 + assert data["recommendation"] == "ready" + + +def test_api_owner_detect_no_pool_does_not_loop_needs_detection(speakers_env): + from solstone.apps.speakers.routes import speakers_bp + + env = speakers_env() + _write_labeled_segment( + env, + "20240101", + "090000_300", + {1: _owner_embeddings(40, np.random.default_rng(1))}, + stream="mic", + ) + + app = Flask(__name__) + app.register_blueprint(speakers_bp) + + with app.test_client() as client: + first_status = client.get("/app/speakers/api/owner/status") + detect_response = client.post("/app/speakers/api/owner/detect") + second_status = client.get("/app/speakers/api/owner/status") + + assert first_status.status_code == 200 + assert first_status.get_json()["status"] == "needs_detection" + assert detect_response.status_code == 200 + assert detect_response.get_json()["status"] == "no_cluster" + assert second_status.status_code == 200 + assert second_status.get_json()["status"] == "no_cluster" + + +def test_api_owner_detect_small_pool_does_not_loop_needs_detection(speakers_env): + from solstone.apps.speakers.routes import speakers_bp + + env = speakers_env() + _write_labeled_segment( + env, + "20240101", + "090000_300", + {1: _owner_embeddings(12, np.random.default_rng(1))}, + stream="mic", + ) + _write_candidate_pool( + env.journal, + [ + _candidate_record( + 1, + [_source_segment("20240101", "090000_300", stream="mic")], + n_intervals=12, + n_segments=1, + ) + ], + ) + + app = Flask(__name__) + app.register_blueprint(speakers_bp) + + with app.test_client() as client: + first_status = client.get("/app/speakers/api/owner/status") + detect_response = client.post("/app/speakers/api/owner/detect") + second_status = client.get("/app/speakers/api/owner/status") + + assert first_status.status_code == 200 + assert first_status.get_json()["status"] == "needs_detection" + assert detect_response.status_code == 200 + assert detect_response.get_json()["status"] == "low_quality" + assert detect_response.get_json()["low_quality_reason"] == "too_few_stmts" + assert second_status.status_code == 200 + assert second_status.get_json()["status"] == "low_quality" + + +def test_api_owner_status_does_not_detect_or_materialize_embeddings( + speakers_env, monkeypatch +): + from solstone.apps.speakers import owner as owner_module + from solstone.apps.speakers import routes as speakers_routes + from solstone.apps.speakers.routes import speakers_bp + + env = speakers_env() + _write_labeled_segment( + env, + "20240101", + "090000_300", + {1: _owner_embeddings(40, np.random.default_rng(1))}, + stream="mic", + ) + + def fail_detect(): + raise AssertionError("status called detect_owner_candidate") + + def fail_materialize(*args, **kwargs): + raise AssertionError("status materialized embedding arrays") + + monkeypatch.setattr(speakers_routes, "detect_owner_candidate", fail_detect) + monkeypatch.setattr(owner_module, "load_npz", fail_materialize) + + app = Flask(__name__) + app.register_blueprint(speakers_bp) + + with app.test_client() as client: + response = client.get("/app/speakers/api/owner/status") + + assert response.status_code == 200 + assert response.get_json()["status"] == "needs_detection" def test_confirm_owner_candidate_no_candidate(speakers_env): @@ -1129,8 +1637,10 @@ def test_reject_owner_candidate(speakers_env): def test_reject_owner_candidate_enforces_detection_cooldown(speakers_env, monkeypatch): - from solstone.apps.speakers.owner import reject_owner_candidate - from solstone.think.awareness import owner_detection_ready + from solstone.apps.speakers.owner import ( + owner_detection_ready, + reject_owner_candidate, + ) env = speakers_env() candidate_path = _candidate_path(env.journal) diff --git a/solstone/think/awareness.py b/solstone/think/awareness.py index 20ef8f9ec..6cc0447c7 100644 --- a/solstone/think/awareness.py +++ b/solstone/think/awareness.py @@ -244,53 +244,6 @@ def _recent_chat_exchanges(limit: int = 10000) -> list[dict[str, Any]]: return exchanges[-limit:] -def owner_detection_ready() -> dict[str, Any]: - """Check if owner voice detection should be surfaced to the user. - - Returns a dict with a ``ready`` boolean and contextual fields. - - Checks in order: - 1. Owner centroid already exists → not ready - 2. Recent rejection within 14 days → not ready (cooldown) - 3. Calls ``detect_owner_candidate()`` → ready if positive recommendation - """ - from solstone.apps.speakers.owner import detect_owner_candidate, load_owner_centroid - - if load_owner_centroid() is not None: - return {"ready": False, "reason": "centroid_exists"} - - voiceprint = get_current().get("voiceprint", {}) - rejected_at = voiceprint.get("rejected_at") - if rejected_at: - try: - rejection_time = datetime.fromisoformat(rejected_at) - now = datetime.now(rejection_time.tzinfo) - days_since = (now - rejection_time).days - if days_since < 14: - return { - "ready": False, - "reason": "cooldown", - "days_remaining": 14 - days_since, - } - except (ValueError, TypeError): - pass - - result = detect_owner_candidate() - if result.get("recommendation") == "ready": - return { - "ready": True, - "reason": "candidate_found", - "cluster_size": result.get("cluster_size"), - "streams_represented": result.get("streams_represented"), - "samples": result.get("samples", []), - } - - return { - "ready": False, - "reason": result.get("recommendation", result.get("status", "unknown")), - } - - def record_import( source_type: str, source_display: str | None = None, diff --git a/solstone/think/journal_io/npz.py b/solstone/think/journal_io/npz.py index 3faed9cf3..07c2190c5 100644 --- a/solstone/think/journal_io/npz.py +++ b/solstone/think/journal_io/npz.py @@ -5,6 +5,7 @@ from __future__ import annotations +import zipfile from collections.abc import Callable, Mapping from io import BytesIO from pathlib import Path @@ -29,6 +30,35 @@ def load_npz(path: Path) -> dict[str, np.ndarray] | None: return {key: data[key] for key in data.files} +def load_npz_row_count(path: Path, key: str) -> int | None: + """Return the first dimension for one NPZ array without materializing it.""" + from numpy.lib import format as np_format + + if not path.exists(): + return None + + try: + with zipfile.ZipFile(path) as archive: + with archive.open(f"{key}.npy") as npy_file: + version = np_format.read_magic(npy_file) + if version == (1, 0): + shape, _fortran_order, _dtype = np_format.read_array_header_1_0( + npy_file + ) + elif version == (2, 0): + shape, _fortran_order, _dtype = np_format.read_array_header_2_0( + npy_file + ) + else: + return None + except Exception: + return None + + if not shape: + return None + return int(shape[0]) + + def save_npz( path: Path, arrays: Mapping[str, np.ndarray], diff --git a/tests/test_awareness.py b/tests/test_awareness.py index d1170f828..6dfd5b578 100644 --- a/tests/test_awareness.py +++ b/tests/test_awareness.py @@ -5,7 +5,6 @@ import json import re -import unittest.mock import pytest @@ -156,137 +155,6 @@ class TestJournalState: assert state["journal"]["first_daily_ready_at"] == "20260308T14:00:00" -class TestOwnerDetectionReady: - """Tests for owner_detection_ready().""" - - def test_not_ready_when_centroid_exists(self): - """Returns not ready when owner centroid already exists.""" - from solstone.think.awareness import owner_detection_ready - - with unittest.mock.patch( - "solstone.apps.speakers.owner.load_owner_centroid", - return_value=("centroid", 0.82), - ): - result = owner_detection_ready() - - assert result["ready"] is False - assert result["reason"] == "centroid_exists" - - def test_not_ready_during_cooldown(self): - """Returns not ready when rejection was within 14 days.""" - from datetime import datetime - - from solstone.think.awareness import owner_detection_ready, update_state - - update_state("voiceprint", {"rejected_at": datetime.now().isoformat()}) - - with unittest.mock.patch( - "solstone.apps.speakers.owner.load_owner_centroid", return_value=None - ): - result = owner_detection_ready() - - assert result["ready"] is False - assert result["reason"] == "cooldown" - assert result["days_remaining"] == 14 - - def test_ready_when_candidate_found(self): - """Returns ready when detect_owner_candidate returns positive.""" - from solstone.think.awareness import owner_detection_ready - - mock_detection = { - "status": "candidate", - "recommendation": "ready", - "cluster_size": 88, - "streams_represented": 2, - "samples": [{"day": "20240101"}], - } - - with unittest.mock.patch( - "solstone.apps.speakers.owner.load_owner_centroid", return_value=None - ): - with unittest.mock.patch( - "solstone.apps.speakers.owner.detect_owner_candidate", - return_value=mock_detection, - ): - result = owner_detection_ready() - - assert result["ready"] is True - assert result["reason"] == "candidate_found" - assert result["cluster_size"] == 88 - assert result["streams_represented"] == 2 - - def test_not_ready_low_data(self): - """Returns not ready when detection has insufficient data.""" - from solstone.think.awareness import owner_detection_ready - - mock_detection = { - "status": "low_data", - "recommendation": "low_data", - } - - with unittest.mock.patch( - "solstone.apps.speakers.owner.load_owner_centroid", return_value=None - ): - with unittest.mock.patch( - "solstone.apps.speakers.owner.detect_owner_candidate", - return_value=mock_detection, - ): - result = owner_detection_ready() - - assert result["ready"] is False - assert result["reason"] == "low_data" - - def test_not_ready_single_stream(self): - """Returns not ready when candidate is single_stream (not 'ready').""" - from solstone.think.awareness import owner_detection_ready - - mock_detection = { - "status": "candidate", - "recommendation": "single_stream", - "cluster_size": 60, - } - - with unittest.mock.patch( - "solstone.apps.speakers.owner.load_owner_centroid", return_value=None - ): - with unittest.mock.patch( - "solstone.apps.speakers.owner.detect_owner_candidate", - return_value=mock_detection, - ): - result = owner_detection_ready() - - assert result["ready"] is False - assert result["reason"] == "single_stream" - - def test_cooldown_expires_after_14_days(self): - """Cooldown no longer blocks after 14 days.""" - from datetime import datetime, timedelta - - from solstone.think.awareness import owner_detection_ready, update_state - - old_rejection = (datetime.now() - timedelta(days=15)).isoformat() - update_state("voiceprint", {"rejected_at": old_rejection}) - - mock_detection = { - "status": "candidate", - "recommendation": "ready", - "cluster_size": 100, - "streams_represented": 3, - "samples": [], - } - - with unittest.mock.patch( - "solstone.apps.speakers.owner.load_owner_centroid", return_value=None - ): - with unittest.mock.patch( - "solstone.apps.speakers.owner.detect_owner_candidate", - return_value=mock_detection, - ): - result = owner_detection_ready() - - assert result["ready"] is True - - class TestEnsureIdentityDirectory: """Tests for ensure_identity_directory().""" diff --git a/tests/test_journal_io_npz.py b/tests/test_journal_io_npz.py index db26f8998..c0f77cd61 100644 --- a/tests/test_journal_io_npz.py +++ b/tests/test_journal_io_npz.py @@ -16,7 +16,13 @@ import pytest from solstone.think.journal_io.errors import LockTimeout, MalformedDataError from solstone.think.journal_io.locking import hold_lock as real_hold_lock -from solstone.think.journal_io.npz import load_npz, save_npz, update_npz, write_npz +from solstone.think.journal_io.npz import ( + load_npz, + load_npz_row_count, + save_npz, + update_npz, + write_npz, +) REPO_ROOT = Path(__file__).resolve().parents[1] VOICEPRINT_KEYS = ("embeddings", "metadata") @@ -88,6 +94,27 @@ def test_load_npz_missing_and_old_writer_round_trip(tmp_path) -> None: np.testing.assert_array_equal(loaded["metadata"], old_arrays["metadata"]) +def test_load_npz_row_count_reads_header_only(tmp_path, monkeypatch) -> None: + path = tmp_path / "embeddings.npz" + np.savez_compressed( + path, + embeddings=np.zeros((12, 256), dtype=np.float32), + metadata=np.asarray(["a", "b"]), + scalar=np.array(1), + ) + + def fail_materialize(*args, **kwargs): + raise AssertionError("row count materialized arrays") + + monkeypatch.setattr(np, "load", fail_materialize) + + assert load_npz_row_count(path, "embeddings") == 12 + assert load_npz_row_count(path, "metadata") == 2 + assert load_npz_row_count(path, "scalar") is None + assert load_npz_row_count(path, "missing") is None + assert load_npz_row_count(tmp_path / "missing.npz", "embeddings") is None + + def test_save_npz_owner_centroid_schema_old_reader_compatible(tmp_path) -> None: path = tmp_path / "owner_centroid.npz" arrays = {