diff --git a/AGENTS.md b/AGENTS.md index db967b50f..bc93f343a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -191,6 +191,7 @@ Each domain has exactly **one** write-owning module (or one tightly-scoped famil | Config (`config/journal.json`) | `solstone/think/journal_config.py` | | Chat config (`config/chat.json`) | `solstone/apps/chat/config.py` | | Vertex credentials (`.config/vertex-credentials.json`) | `solstone/apps/settings/vertex_credentials.py` | +| Speaker labels (`chronicle/**/talents/speaker_labels.json`) | `solstone/apps/speakers/attribution.py` | | Chronicle day content (`chronicle/YYYYMMDD/**`) | The capturing module (observer, importer) per its declared outputs | | Index (SQLite, `indexer/*`) | `solstone/think/indexer/*` | diff --git a/scripts/check_journal_io_access.py b/scripts/check_journal_io_access.py index 44eae8fd4..f14592551 100644 --- a/scripts/check_journal_io_access.py +++ b/scripts/check_journal_io_access.py @@ -81,6 +81,7 @@ OWNER_FILES: frozenset[str] = frozenset( { "solstone/apps/entities/call.py", "solstone/apps/chat/config.py", + "solstone/apps/speakers/attribution.py", "solstone/apps/speakers/owner.py", "solstone/apps/speakers/routes.py", "solstone/apps/settings/vertex_credentials.py", diff --git a/solstone/apps/speakers/attribution.py b/solstone/apps/speakers/attribution.py index fd06fbab9..39c31363c 100644 --- a/solstone/apps/speakers/attribution.py +++ b/solstone/apps/speakers/attribution.py @@ -23,6 +23,7 @@ import json import logging import re from collections import defaultdict +from collections.abc import Callable from pathlib import Path from typing import Any @@ -41,6 +42,12 @@ from solstone.think.entities.journal import ( get_journal_principal, load_all_journal_entities, ) +from solstone.think.journal_io import ( + MalformedPolicy, + hold_lock, + read_json, + write_json, +) from solstone.think.utils import day_path, now_ms, segment_path logger = logging.getLogger(__name__) @@ -450,12 +457,45 @@ def attribute_segment( # --------------------------------------------------------------------------- +def update_speaker_labels( + seg_dir: Path, + transform: Callable[[dict | None], dict | None], +) -> None: + """Apply a locked read-modify-write transform to speaker_labels.json.""" + path = seg_dir / "talents" / "speaker_labels.json" + with hold_lock(path): + current = read_json( + path, + on_error=MalformedPolicy.WARN_AND_SKIP, + default=None, + ) + new = transform(current) + if new is None: + return + write_json(path, new) + + +def _label_sentence_id(label: dict) -> int | None: + sid = label.get("sentence_id") + if sid is None: + return None + try: + return int(sid) + except (TypeError, ValueError): + return None + + +def _is_user_label(label: dict) -> bool: + method = label.get("method") + return isinstance(method, str) and method.startswith("user_") + + def save_speaker_labels( seg_dir: Path, labels: list[dict], metadata: dict[str, Any], ) -> Path: - """Write speaker_labels.json to the segment's agents/ directory. + """Write speaker_labels.json to the segment's talents/ directory. Preserves user corrections: if speaker_corrections.json exists, any sentence that was corrected by the user keeps the corrected attribution @@ -503,20 +543,102 @@ def save_speaker_labels( ) out_path = agents_dir / "speaker_labels.json" - data = { - "labels": labels, - "owner_centroid_last_refreshed_at": metadata.get( - "owner_centroid_last_refreshed_at" - ), - "voiceprint_versions": metadata.get("voiceprint_versions", {}), - } - with open(out_path, "w", encoding="utf-8") as f: - json.dump(data, f, indent=2) + + def transform(current: dict | None) -> dict: + user_by_sid: dict[int, dict] = {} + if isinstance(current, dict): + current_labels = current.get("labels", []) + if isinstance(current_labels, list): + for current_label in current_labels: + if not isinstance(current_label, dict): + continue + sid = _label_sentence_id(current_label) + if sid is not None and _is_user_label(current_label): + user_by_sid[sid] = current_label + + merged_labels: list[dict] = [] + fresh_sids: set[int] = set() + for label in labels: + sid = _label_sentence_id(label) + if sid is None: + merged_labels.append(label) + continue + fresh_sids.add(sid) + merged_labels.append(user_by_sid.get(sid, label)) + + user_only = [ + label for sid, label in sorted(user_by_sid.items()) if sid not in fresh_sids + ] + merged_labels.extend(user_only) + + return { + "labels": merged_labels, + "owner_centroid_last_refreshed_at": metadata.get( + "owner_centroid_last_refreshed_at" + ), + "voiceprint_versions": metadata.get("voiceprint_versions", {}), + } + + update_speaker_labels(seg_dir, transform) logger.info("Wrote %d labels to %s", len(labels), out_path) return out_path +def save_speaker_labels_stub(seg_dir: Path, reason: str) -> None: + """Write a locked attribution stub for segments that cannot be labeled.""" + update_speaker_labels( + seg_dir, + lambda _current: {"labels": [], "skipped": True, "reason": reason}, + ) + + +def apply_label_patches( + seg_dir: Path, + patches: dict[int, dict], + *, + allow_insert: bool, +) -> None: + """Apply locked per-sentence speaker label patches.""" + + def transform(current: dict | None) -> dict: + if isinstance(current, dict): + base = dict(current) + labels_value = current.get("labels", []) + labels = list(labels_value) if isinstance(labels_value, list) else [] + else: + base = {} + labels = [] + + labels_by_sid: dict[int, dict] = {} + for label in labels: + if not isinstance(label, dict): + continue + sid = _label_sentence_id(label) + if sid is not None: + labels_by_sid[sid] = label + + for sid, fields in patches.items(): + sid_int = int(sid) + existing = labels_by_sid.get(sid_int) + if existing is not None: + existing.update(fields) + continue + if not allow_insert: + raise ValueError(f"speaker label sentence_id {sid_int} not found") + label = {"sentence_id": sid_int, **fields} + labels.append(label) + labels_by_sid[sid_int] = label + + if allow_insert: + labels = sorted(labels, key=lambda label: int(label["sentence_id"])) + + base["labels"] = labels + return base + + update_speaker_labels(seg_dir, transform) + + # --------------------------------------------------------------------------- # Voiceprint accumulation # --------------------------------------------------------------------------- diff --git a/solstone/apps/speakers/discovery.py b/solstone/apps/speakers/discovery.py index cf8635d9f..6e49e1889 100644 --- a/solstone/apps/speakers/discovery.py +++ b/solstone/apps/speakers/discovery.py @@ -34,7 +34,6 @@ def _routes_helpers(): _load_embeddings_file, _load_speaker_labels, _normalize_embedding, - _save_speaker_labels, _scan_segment_embeddings, ) @@ -42,7 +41,6 @@ def _routes_helpers(): _load_embeddings_file, _load_speaker_labels, _normalize_embedding, - _save_speaker_labels, _scan_segment_embeddings, _append_speaker_correction, _check_owner_contamination, @@ -128,7 +126,6 @@ def discover_unknown_speakers() -> dict[str, Any]: load_embeddings_file, load_speaker_labels, normalize_embedding, - _, scan_segment_embeddings, _, _, @@ -340,6 +337,7 @@ def identify_cluster( cluster_id: int, name: str, entity_id: str | None = None ) -> dict[str, Any]: """Identify a discovered unknown speaker cluster.""" + from solstone.apps.speakers.attribution import apply_label_patches from solstone.think.entities import ( load_existing_voiceprint_keys, save_voiceprints_batch, @@ -349,7 +347,6 @@ def identify_cluster( load_embeddings_file, load_speaker_labels, normalize_embedding, - save_speaker_labels, _scan, append_speaker_correction, check_owner_contamination, @@ -489,6 +486,7 @@ def identify_cluster( } updated = False + patches: dict[int, dict[str, Any]] = {} for sentence_id in sorted(set(sentence_ids)): original = labels_by_sid.get(sentence_id, {}) new_label = { @@ -500,7 +498,11 @@ def identify_cluster( if original != new_label: updated = True sentences_attributed += 1 - labels_by_sid[sentence_id] = new_label + patches[sentence_id] = { + "speaker": entity_id, + "confidence": "high", + "method": "user_identified", + } correction_key = (sentence_id, entity_id) if correction_key in existing_correction_keys: @@ -519,11 +521,7 @@ def identify_cluster( existing_correction_keys.add(correction_key) if updated: - labels_data["labels"] = sorted( - labels_by_sid.values(), - key=lambda label: int(label["sentence_id"]), - ) - save_speaker_labels(seg_dir, labels_data) + apply_label_patches(seg_dir, patches, allow_insert=True) segments_updated += 1 _write_resolved_cluster(cluster_id, entity_id, entity_name) diff --git a/solstone/apps/speakers/routes.py b/solstone/apps/speakers/routes.py index 5c5eaf26c..34a5b17b7 100644 --- a/solstone/apps/speakers/routes.py +++ b/solstone/apps/speakers/routes.py @@ -26,6 +26,7 @@ from flask import ( send_file, ) +from solstone.apps.speakers.attribution import apply_label_patches from solstone.apps.speakers.copy import ( SPK_OVERVIEW_KNOWN_VOICES_SORTS, speaker_copy_payload, @@ -66,6 +67,7 @@ from solstone.convey.reasons import ( MISSING_REQUEST_BODY, MISSING_REQUIRED_FIELD, SPEAKER_ATTRIBUTION_STATE_INVALID, + SPEAKER_LABELS_BUSY, SPEAKER_NOT_FOUND, SPEAKER_OWNER_VOICE_TOO_CLOSE, SPEAKER_REVIEW_UNAVAILABLE, @@ -326,17 +328,6 @@ def _load_speaker_labels(segment_dir: Path) -> dict | None: return None -def _save_speaker_labels(segment_dir: Path, labels_data: dict) -> None: - """Atomically write speaker_labels.json to a segment's talents/ directory.""" - talents_dir = segment_dir / "talents" - talents_dir.mkdir(parents=True, exist_ok=True) - out_path = talents_dir / "speaker_labels.json" - tmp_path = out_path.with_suffix(".tmp") - with open(tmp_path, "w", encoding="utf-8") as f: - json.dump(labels_data, f, indent=2) - tmp_path.rename(out_path) - - def _load_speaker_corrections(segment_dir: Path) -> list[dict]: """Load speaker_corrections.json from a segment's talents/ directory. @@ -406,6 +397,14 @@ def _voiceprint_busy_response(exc: LockTimeout) -> Any: ) +def _labels_busy_response(exc: LockTimeout) -> Any: + logger.warning("speaker labels busy for %s", exc.path) + return error_response( + SPEAKER_LABELS_BUSY, + detail="speaker labels are busy; try again", + ) + + def _owner_bootstrap_status_fields() -> dict[str, Any]: """Return shared owner bootstrap diagnostics for status surfaces.""" diagnostics = load_owner_bootstrap_diagnostics(_principal_id_or_none()) @@ -1018,14 +1017,12 @@ def api_confirm_attribution() -> Any: ) label = None - label_idx = None - for i, item in enumerate(labels_data.get("labels", [])): + for item in labels_data.get("labels", []): if item.get("sentence_id") == sentence_id: label = item - label_idx = i break - if label is None or label_idx is None: + if label is None: return error_response( SPEAKER_SENTENCE_MISSING, detail="Sentence not found in labels", @@ -1069,9 +1066,14 @@ def api_confirm_attribution() -> Any: return _voiceprint_busy_response(exc) old_method = label.get("method") - labels_data["labels"][label_idx]["confidence"] = "high" - labels_data["labels"][label_idx]["method"] = "user_confirmed" - _save_speaker_labels(segment_dir, labels_data) + try: + apply_label_patches( + segment_dir, + {sentence_id: {"confidence": "high", "method": "user_confirmed"}}, + allow_insert=False, + ) + except LockTimeout as exc: + return _labels_busy_response(exc) _append_speaker_correction( segment_dir, @@ -1152,14 +1154,12 @@ def api_correct_attribution() -> Any: ) label = None - label_idx = None - for i, item in enumerate(labels_data.get("labels", [])): + for item in labels_data.get("labels", []): if item.get("sentence_id") == sentence_id: label = item - label_idx = i break - if label is None or label_idx is None: + if label is None: return error_response( SPEAKER_SENTENCE_MISSING, detail="Sentence not found in labels", @@ -1211,10 +1211,20 @@ def api_correct_attribution() -> Any: except LockTimeout as exc: return _voiceprint_busy_response(exc) - labels_data["labels"][label_idx]["speaker"] = new_speaker - labels_data["labels"][label_idx]["confidence"] = "high" - labels_data["labels"][label_idx]["method"] = "user_corrected" - _save_speaker_labels(segment_dir, labels_data) + try: + apply_label_patches( + segment_dir, + { + sentence_id: { + "speaker": new_speaker, + "confidence": "high", + "method": "user_corrected", + } + }, + allow_insert=False, + ) + except LockTimeout as exc: + return _labels_busy_response(exc) _append_speaker_correction( segment_dir, @@ -1301,14 +1311,12 @@ def api_assign_attribution() -> Any: ) label = None - label_idx = None - for i, item in enumerate(labels_data.get("labels", [])): + for item in labels_data.get("labels", []): if item.get("sentence_id") == sentence_id: label = item - label_idx = i break - if label is None or label_idx is None: + if label is None: return error_response( SPEAKER_SENTENCE_MISSING, detail="Sentence not found in labels", @@ -1344,10 +1352,20 @@ def api_assign_attribution() -> Any: except LockTimeout as exc: return _voiceprint_busy_response(exc) - labels_data["labels"][label_idx]["speaker"] = speaker - labels_data["labels"][label_idx]["confidence"] = "high" - labels_data["labels"][label_idx]["method"] = "user_assigned" - _save_speaker_labels(segment_dir, labels_data) + try: + apply_label_patches( + segment_dir, + { + sentence_id: { + "speaker": speaker, + "confidence": "high", + "method": "user_assigned", + } + }, + allow_insert=False, + ) + except LockTimeout as exc: + return _labels_busy_response(exc) _append_speaker_correction( segment_dir, @@ -1564,7 +1582,13 @@ def api_discovery_identify() -> Any: detail="cluster_id must be an integer", ) - result = identify_cluster(cluster_id, name) + try: + result = identify_cluster(cluster_id, name) + except LockTimeout as exc: + if exc.path.name == "speaker_labels.json": + return _labels_busy_response(exc) + return _voiceprint_busy_response(exc) + if "error" in result: resolved = load_resolved_cluster(cluster_id) if resolved and resolved.get("label", "").strip().lower() == name.lower(): diff --git a/solstone/apps/speakers/tests/test_attribution.py b/solstone/apps/speakers/tests/test_attribution.py index a86622e10..1bb34aba6 100644 --- a/solstone/apps/speakers/tests/test_attribution.py +++ b/solstone/apps/speakers/tests/test_attribution.py @@ -292,7 +292,7 @@ def test_unmatched_sentences_get_null(speakers_env): # --------------------------------------------------------------------------- -def test_save_speaker_labels(tmp_path): +def test_public_speaker_labels_save(tmp_path): from solstone.apps.speakers.attribution import save_speaker_labels labels = [ @@ -323,6 +323,100 @@ def test_save_speaker_labels(tmp_path): assert data["voiceprint_versions"]["alice"] == 10 +def test_public_labels_save_preserves_current_user_labels(tmp_path): + from solstone.apps.speakers.attribution import save_speaker_labels + + existing = { + "labels": [ + { + "sentence_id": 1, + "speaker": "old_pipeline", + "confidence": "high", + "method": "acoustic", + }, + { + "sentence_id": 2, + "speaker": "user_choice", + "confidence": "high", + "method": "user_corrected", + }, + { + "sentence_id": 3, + "speaker": "user_only", + "confidence": "high", + "method": "user_assigned", + }, + ], + "owner_centroid_last_refreshed_at": None, + "voiceprint_versions": {}, + } + talents_dir = tmp_path / "talents" + talents_dir.mkdir() + labels_path = talents_dir / "speaker_labels.json" + labels_path.write_text(json.dumps(existing), encoding="utf-8") + + labels = [ + { + "sentence_id": 1, + "speaker": "fresh_pipeline", + "confidence": "high", + "method": "acoustic", + }, + { + "sentence_id": 2, + "speaker": "fresh_pipeline", + "confidence": "high", + "method": "acoustic", + }, + ] + + save_speaker_labels(tmp_path, labels, {}) + + data = json.loads(labels_path.read_text(encoding="utf-8")) + by_sid = {label["sentence_id"]: label for label in data["labels"]} + assert by_sid[1]["speaker"] == "fresh_pipeline" + assert by_sid[2]["speaker"] == "user_choice" + assert by_sid[2]["method"] == "user_corrected" + assert by_sid[3]["speaker"] == "user_only" + assert by_sid[3]["method"] == "user_assigned" + + +def test_public_labels_save_keeps_corrections_overlay(tmp_path): + from solstone.apps.speakers.attribution import save_speaker_labels + + talents_dir = tmp_path / "talents" + talents_dir.mkdir() + (talents_dir / "speaker_corrections.json").write_text( + json.dumps( + { + "corrections": [ + { + "sentence_id": 1, + "original_speaker": "alice", + "corrected_speaker": "bob", + } + ] + } + ), + encoding="utf-8", + ) + labels = [ + { + "sentence_id": 1, + "speaker": "alice", + "confidence": "high", + "method": "acoustic", + } + ] + + save_speaker_labels(tmp_path, labels, {}) + + data = json.loads((talents_dir / "speaker_labels.json").read_text()) + assert data["labels"][0]["speaker"] == "bob" + assert data["labels"][0]["method"] == "user_corrected" + assert labels[0]["speaker"] == "bob" + + # --------------------------------------------------------------------------- # Voiceprint accumulation # --------------------------------------------------------------------------- diff --git a/solstone/apps/speakers/tests/test_routes.py b/solstone/apps/speakers/tests/test_routes.py index 78c9aae6d..f6a4d9098 100644 --- a/solstone/apps/speakers/tests/test_routes.py +++ b/solstone/apps/speakers/tests/test_routes.py @@ -992,6 +992,56 @@ def test_api_confirm_attribution(speakers_env): assert metadata["stream"] == "test" +def test_api_confirm_attribution_labels_busy(speakers_env, monkeypatch): + """Confirm returns speaker_labels_busy when the labels lock times out.""" + from pathlib import Path + + from flask import Flask + + from solstone.apps.speakers import attribution + from solstone.apps.speakers.routes import speakers_bp + from solstone.think.journal_io.errors import LockTimeout + + env = speakers_env() + env.create_segment("20240101", "143022_300", ["mic_audio"]) + env.create_entity("Alice Test") + env.create_speaker_labels( + "20240101", + "143022_300", + [ + { + "sentence_id": 1, + "speaker": "alice_test", + "confidence": "medium", + "method": "acoustic", + }, + ], + ) + + def busy_hold_lock(path: Path): + raise LockTimeout(path=path, timeout=0.0) + + monkeypatch.setattr(attribution, "hold_lock", busy_hold_lock) + + app = Flask(__name__) + app.register_blueprint(speakers_bp) + + with app.test_client() as client: + resp = client.post( + "/app/speakers/api/confirm-attribution", + json={ + "day": "20240101", + "stream": "test", + "segment_key": "143022_300", + "source": "mic_audio", + "sentence_id": 1, + }, + ) + + assert resp.status_code == 503 + assert resp.get_json()["reason_code"] == "speaker_labels_busy" + + def test_api_confirm_idempotent(speakers_env): """Confirming an already-confirmed attribution is a no-op success.""" from flask import Flask diff --git a/solstone/convey/reasons.py b/solstone/convey/reasons.py index e349f8a7d..377761119 100644 --- a/solstone/convey/reasons.py +++ b/solstone/convey/reasons.py @@ -296,6 +296,11 @@ SPEAKER_VOICEPRINT_BUSY = Reason( "I couldn't update that voice right now because it was busy. Try again in a moment.", 503, ) +SPEAKER_LABELS_BUSY = Reason( + "speaker_labels_busy", + "I couldn't update those speaker attributions right now because they were busy. Try again in a moment.", + 503, +) # reprocess REPROCESS_PAST_ONLY = Reason( diff --git a/solstone/talent/entities.py b/solstone/talent/entities.py index b7d6a1347..075499aa6 100644 --- a/solstone/talent/entities.py +++ b/solstone/talent/entities.py @@ -113,7 +113,7 @@ def post_process(result: str, context: dict) -> str | None: len(unique_entities), ) - # Write entities.jsonl alongside the agent output in the agents/ directory + # Write entities.jsonl alongside the agent output in the talents/ directory output_path_value = context.get("output_path") if not output_path_value: logging.error("entities hook: missing output_path in context") diff --git a/solstone/talent/speaker_attribution.py b/solstone/talent/speaker_attribution.py index 8f2c56e59..a1b63f291 100644 --- a/solstone/talent/speaker_attribution.py +++ b/solstone/talent/speaker_attribution.py @@ -25,6 +25,7 @@ def pre_process(context: dict) -> dict | None: accumulate_voiceprints, attribute_segment, save_speaker_labels, + save_speaker_labels_stub, ) from solstone.think.utils import segment_path @@ -42,15 +43,8 @@ def pre_process(context: dict) -> dict | None: logger.info("Attribution skipped: %s", result["error"]) reason = result["error"] if any(seg_dir.glob("*.npz")): - agents_dir = seg_dir / "talents" - agents_dir.mkdir(parents=True, exist_ok=True) - out_path = agents_dir / "speaker_labels.json" - with open(out_path, "w", encoding="utf-8") as fh: - json.dump( - {"labels": [], "skipped": True, "reason": reason}, - fh, - indent=2, - ) + save_speaker_labels_stub(seg_dir, reason) + out_path = seg_dir / "talents" / "speaker_labels.json" logger.info("Wrote attribution stub: %s (%s)", out_path, reason) return {"skip_reason": reason} @@ -58,15 +52,8 @@ def pre_process(context: dict) -> dict | None: if not labels: reason = "no_embeddings" if any(seg_dir.glob("*.npz")): - agents_dir = seg_dir / "talents" - agents_dir.mkdir(parents=True, exist_ok=True) - out_path = agents_dir / "speaker_labels.json" - with open(out_path, "w", encoding="utf-8") as fh: - json.dump( - {"labels": [], "skipped": True, "reason": reason}, - fh, - indent=2, - ) + save_speaker_labels_stub(seg_dir, reason) + out_path = seg_dir / "talents" / "speaker_labels.json" logger.info("Wrote attribution stub: %s (%s)", out_path, reason) return {"skip_reason": reason} diff --git a/solstone/think/cluster.py b/solstone/think/cluster.py index ac08ae8e6..9a473fe62 100644 --- a/solstone/think/cluster.py +++ b/solstone/think/cluster.py @@ -254,7 +254,7 @@ def _process_segment( file=sys.stderr, ) - # Process agent output summaries from agents/**/*.md files (with optional filtering) + # Process agent output summaries from talents/**/*.md files (with optional filtering) if agents: # Convert bool to filter: True -> None (all), False handled by outer if agent_filter = ( diff --git a/solstone/think/retention.py b/solstone/think/retention.py index fb418fb24..bcee273d7 100644 --- a/solstone/think/retention.py +++ b/solstone/think/retention.py @@ -74,10 +74,10 @@ def is_segment_complete(segment_path: Path) -> bool: """Check if a segment has finished all processing. Completion checks (ALL must pass): - 1. No _active.jsonl files in agents/ + 1. No _active.jsonl files in talents/ 2. audio.jsonl exists if any audio raw media was captured 3. screen.jsonl exists if any video raw media was captured - 4. agents/speaker_labels.json exists if embeddings (.npz) are present + 4. talents/speaker_labels.json exists if embeddings (.npz) are present """ agents_dir = segment_path / "talents" diff --git a/tests/test_speaker_labels_locking.py b/tests/test_speaker_labels_locking.py new file mode 100644 index 000000000..68cde0169 --- /dev/null +++ b/tests/test_speaker_labels_locking.py @@ -0,0 +1,184 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +from __future__ import annotations + +import json +import multiprocessing +import os +import time +import traceback +from pathlib import Path +from queue import Empty +from typing import Any + +DAY = "20240101" +STREAM = "test" +SEGMENT = "143022_300" + + +def _pipeline_worker( + journal_path: str, + seg_dir_path: str, + barrier: Any, + errors: Any, + delay: float, +) -> None: + os.environ["SOLSTONE_JOURNAL"] = journal_path + try: + barrier.wait(timeout=5) + if delay: + time.sleep(delay) + + from solstone.apps.speakers.attribution import save_speaker_labels + + save_speaker_labels( + Path(seg_dir_path), + [ + { + "sentence_id": 1, + "speaker": "X", + "confidence": "high", + "method": "acoustic", + }, + { + "sentence_id": 2, + "speaker": "pipeline_overwrite", + "confidence": "high", + "method": "acoustic", + }, + ], + { + "owner_centroid_last_refreshed_at": "2026-03-15T12:00:00Z", + "voiceprint_versions": {"X": 1}, + }, + ) + except BaseException: + errors.put(traceback.format_exc()) + raise + + +def _patch_worker( + journal_path: str, + seg_dir_path: str, + barrier: Any, + errors: Any, + delay: float, +) -> None: + os.environ["SOLSTONE_JOURNAL"] = journal_path + try: + barrier.wait(timeout=5) + if delay: + time.sleep(delay) + + from solstone.apps.speakers.attribution import apply_label_patches + + apply_label_patches( + Path(seg_dir_path), + { + 2: { + "speaker": "Y", + "confidence": "high", + "method": "user_corrected", + } + }, + allow_insert=False, + ) + except BaseException: + errors.put(traceback.format_exc()) + raise + + +def _seed_labels(journal: Path) -> Path: + seg_dir = journal / "chronicle" / DAY / STREAM / SEGMENT + talents_dir = seg_dir / "talents" + talents_dir.mkdir(parents=True) + (talents_dir / "speaker_labels.json").write_text( + json.dumps( + { + "labels": [ + { + "sentence_id": 1, + "speaker": "old_one", + "confidence": "high", + "method": "acoustic", + }, + { + "sentence_id": 2, + "speaker": "old_two", + "confidence": "high", + "method": "acoustic", + }, + ], + "owner_centroid_last_refreshed_at": None, + "voiceprint_versions": {}, + } + ), + encoding="utf-8", + ) + return seg_dir + + +def _drain_errors(errors: Any) -> list[str]: + found = [] + while True: + try: + found.append(errors.get_nowait()) + except Empty: + return found + + +def _run_case( + tmp_path: Path, + name: str, + pipeline_delay: float, + patch_delay: float, +) -> None: + ctx = multiprocessing.get_context("spawn") + journal = tmp_path / name + seg_dir = _seed_labels(journal) + barrier = ctx.Barrier(2) + errors = ctx.Queue() + + processes = [ + ctx.Process( + target=_pipeline_worker, + args=(str(journal), str(seg_dir), barrier, errors, pipeline_delay), + ), + ctx.Process( + target=_patch_worker, + args=(str(journal), str(seg_dir), barrier, errors, patch_delay), + ), + ] + + for process in processes: + process.start() + for process in processes: + process.join(timeout=10) + for process in processes: + if process.is_alive(): + process.terminate() + process.join(timeout=2) + + error_text = "\n".join(_drain_errors(errors)) + assert all(not process.is_alive() for process in processes), error_text + assert all(process.exitcode == 0 for process in processes), error_text + + labels_path = seg_dir / "talents" / "speaker_labels.json" + data = json.loads(labels_path.read_text(encoding="utf-8")) + by_sid = {int(label["sentence_id"]): label for label in data["labels"]} + assert by_sid[1]["speaker"] == "X" + assert by_sid[2]["speaker"] == "Y" + assert by_sid[2]["method"] == "user_corrected" + + +def test_speaker_labels_locked_merge_survives_pipeline_then_user( + tmp_path: Path, +) -> None: + _run_case(tmp_path, "pipeline_first", pipeline_delay=0.0, patch_delay=0.2) + + +def test_speaker_labels_locked_merge_survives_user_then_pipeline( + tmp_path: Path, +) -> None: + _run_case(tmp_path, "user_first", pipeline_delay=0.2, patch_delay=0.0)