refactor(speaker): pass the tenant to enrolment instead of parsing it out of an id

/enroll/upload and /enroll/batch received no user_id. The service recovered the
tenant with int(speaker_id.split("_")[1]), so the id's `user_{id}_speaker_{hex}`
shape decided which gallery a voiceprint entered. A naming convention is not an
ownership record: it is absent for a client that names ids differently and wrong
for an imported speaker, and nothing checked it against the caller.

Both endpoints now take user_id as a form field, and every caller sends it:
Chronicle's speaker client, the WebUI's Enrollment and Annotation pages and its
api.ts helpers, enroll_speaker.py, laptop_client.py, the diarization-validation
script, and the integration test.

/enroll/append is unchanged and takes no user_id — it requires the speaker to
exist, so the record answers the question and a supplied tenant could only
disagree with it.

extract_user_id_from_speaker_id is now owner_of_speaker: a lookup of
speakers.user_id with no parsing left, 404 when the speaker is unknown. Its
other callers (get/download/delete) all operate on speakers that exist.

Speaker suite green; WebUI typechecks.

Author: AnkushMalaker

backends/advanced/src/advanced_omi_backend/speaker_recognition_client.py

@@ -1525,6 +1525,7 @@ async def enroll_new_speaker(
                 )
                 form_data.add_field("speaker_id", speaker_id)
                 form_data.add_field("speaker_name", speaker_name)
+                form_data.add_field("user_id", user_id)
 
                 async with session.post(
                     f"{self.service_url}/enroll/upload",

extras/ml-experiments/experiments/pyannote_diarization_validation/enroll_and_identify.py

@@ -0,0 +1,128 @@
+"""Validate the PRODUCTION diarization path: enroll speakers, then /diarize-and-identify, and
+score DER on the IDENTIFIED labels (identified_as) — which is what the backend actually stores.
+
+This tests whether per-segment centroid matching reconciles the cross-chunk label inconsistency
+(the user's point). Enrollment segments are EXCLUDED from scoring via a held-out UEM, so it's not
+train-on-test. Throwaway user_id=9999; speakers deleted by the caller afterward.
+"""
+
+import io
+import json
+import sys
+from pathlib import Path
+
+import numpy as np
+import requests
+import soundfile as sf
+
+SVC = "http://localhost:8085"
+SR = 16000
+USER = 9999
+N_ENROLL = 3  # segments per speaker used for enrollment
+MIN_ENROLL_S = 2.5  # only use segments at least this long for enrollment
+
+
+def parse_rttm(p: Path):
+    by_spk = {}
+    for ln in p.read_text().splitlines():
+        f = ln.split()
+        if f and f[0] == "SPEAKER":
+            start, dur, spk = float(f[3]), float(f[4]), f[7]
+            by_spk.setdefault(spk, []).append((start, start + dur))
+    return by_spk
+
+
+def main():
+    meeting = sys.argv[1]
+    wav = Path(f"data/ami_sdm_slice/audio/{meeting}.wav")
+    ref = Path(f"data/ami_sdm_slice/ref_rttm/{meeting}.rttm")
+    outdir = Path("experiments/pyannote_diarization_validation/enrolled")
+    outdir.mkdir(parents=True, exist_ok=True)
+    uemdir = outdir / "uem"
+    uemdir.mkdir(exist_ok=True)
+    hypdir = outdir / f"{meeting}_results"
+    hypdir.mkdir(exist_ok=True)
+
+    audio, sr = sf.read(wav, dtype="float32")
+    assert sr == SR
+    by_spk = parse_rttm(ref)
+
+    enroll_ranges = []  # (start,end) excluded from scoring
+    for spk, segs in by_spk.items():
+        longest = sorted(segs, key=lambda s: s[1] - s[0], reverse=True)
+        picked = [s for s in longest if s[1] - s[0] >= MIN_ENROLL_S][:N_ENROLL]
+        if not picked:
+            picked = longest[:N_ENROLL]
+        clip = np.concatenate([audio[int(s * SR) : int(e * SR)] for s, e in picked])
+        enroll_ranges += picked
+        buf = io.BytesIO()
+        sf.write(buf, clip, SR, format="WAV", subtype="PCM_16")
+        buf.seek(0)
+        sid = f"user_{USER}_{meeting}_{spk}"
+        r = requests.post(
+            f"{SVC}/enroll/upload",
+            files={"file": ("enroll.wav", buf, "audio/wav")},
+            data={"speaker_id": sid, "speaker_name": spk, "user_id": str(USER)},
+            timeout=120,
+        )
+        print(
+            f"enroll {spk}: {r.status_code} ({sum(e-s for s,e in picked):.1f}s)",
+            flush=True,
+        )
+
+    # held-out UEM = whole meeting minus enrolled ranges
+    total = len(audio) / SR
+    ranges = sorted(enroll_ranges)
+    held, cur = [], 0.0
+    for s, e in ranges:
+        if s > cur:
+            held.append((cur, s))
+        cur = max(cur, e)
+    if cur < total:
+        held.append((cur, total))
+    (uemdir / f"{meeting}.uem").write_text(
+        "\n".join(f"{meeting} 1 {s:.3f} {e:.3f}" for s, e in held) + "\n"
+    )
+
+    # diarize + identify with enrolled speakers
+    with open(wav, "rb") as f:
+        r = requests.post(
+            f"{SVC}/diarize-and-identify",
+            files={"file": (wav.name, f, "audio/wav")},
+            data={
+                "user_id": str(USER),
+                "min_duration": 0.5,
+                "collar": 2.0,
+                "min_duration_off": 1.5,
+            },
+            timeout=1800,
+        )
+    segs = r.json().get("segments", [])
+    n_id = sum(1 for s in segs if s.get("identified_as"))
+    print(
+        f"{meeting}: {len(segs)} segs, {n_id} identified, "
+        f"{len({s['speaker'] for s in segs})} diar-labels",
+        flush=True,
+    )
+
+    # hyp speaker = identified_as (production label) else fall back to diar label
+    hyp = [
+        {
+            "start": s["start"],
+            "end": s["end"],
+            "speaker": s.get("identified_as") or f"diar_{s.get('speaker')}",
+        }
+        for s in segs
+    ]
+    json.dump(
+        {"segments": hyp, "provider": "enrolled_identified"},
+        open(hypdir / f"{meeting}.pyannote.json", "w"),
+    )
+    print(
+        f"wrote {hypdir}/{meeting}.pyannote.json + held-out uem ({len(held)} regions)",
+        flush=True,
+    )
+
+
+if __name__ == "__main__":
+    main()

extras/speaker-recognition/laptop_client.py

@@ -82,13 +82,19 @@ async def health_check(self):
         logger.info(f"Health check response: {health_response}")
         return health_response
 
-    async def enroll_speaker(self, speaker_id: str, speaker_name: str, audio_path: str):
+    async def enroll_speaker(
+        self, speaker_id: str, speaker_name: str, user_id: str, audio_path: str
+    ):
         """Enroll a speaker from audio file by uploading it."""
         if not self.session:
             raise RuntimeError("Client not initialized")
 
         # Prepare query parameters
-        params = {"speaker_id": speaker_id, "speaker_name": speaker_name}
+        params = {
+            "speaker_id": speaker_id,
+            "speaker_name": speaker_name,
+            "user_id": user_id,
+        }
 
         # Prepare file upload
         with open(audio_path, "rb") as f:
@@ -275,7 +281,7 @@ async def cmd_enroll(args):
 
             # Enroll speaker
             result = await client.enroll_speaker(
-                args.speaker_id, args.speaker_name, audio_path
+                args.speaker_id, args.speaker_name, args.user_id, audio_path
             )
             logger.info(f"Server response: {result}")
 
@@ -524,6 +530,9 @@ def main():
     # Enroll command
     enroll_parser = subparsers.add_parser("enroll", help="Enroll a new speaker")
     enroll_parser.add_argument("--speaker-id", required=True, help="Unique speaker ID")
+    enroll_parser.add_argument(
+        "--user-id", required=True, help="Chronicle user id owning the speaker"
+    )
     enroll_parser.add_argument(
         "--speaker-name", required=True, help="Speaker display name"
     )

extras/speaker-recognition/scripts/enroll_speaker.py

@@ -70,6 +70,7 @@ def enroll_single_file(
     file_path: str,
     speaker_id: str,
     speaker_name: str,
+    user_id: str,
     start: Optional[float] = None,
     end: Optional[float] = None,
     service_url: str = None,
@@ -84,7 +85,11 @@ def enroll_single_file(
     try:
         with open(file_path, "rb") as f:
             files = {"file": (os.path.basename(file_path), f, "audio/wav")}
-            data = {"speaker_id": speaker_id, "speaker_name": speaker_name}
+            data = {
+                "speaker_id": speaker_id,
+                "speaker_name": speaker_name,
+                "user_id": user_id,
+            }
 
             if start is not None:
                 data["start"] = start
@@ -118,7 +123,7 @@ def enroll_single_file(
 
 
 def enroll_multiple_files(
-    file_paths: List[str], speaker_id: str, speaker_name: str
+    file_paths: List[str], speaker_id: str, speaker_name: str, user_id: str
 ) -> bool:
     """Enroll speaker from multiple audio files for better accuracy."""
     valid_files = [f for f in file_paths if os.path.exists(f)]
@@ -138,7 +143,11 @@ def enroll_multiple_files(
                     ("files", (os.path.basename(file_path), content, "audio/wav"))
                 )
 
-        data = {"speaker_id": speaker_id, "speaker_name": speaker_name}
+        data = {
+            "speaker_id": speaker_id,
+            "speaker_name": speaker_name,
+            "user_id": user_id,
+        }
 
         response = requests.post(
             f"{SPEAKER_SERVICE_URL}/enroll/batch", files=files, data=data
@@ -161,7 +170,9 @@ def enroll_multiple_files(
         return False
 
 
-def enroll_from_directory(directory: str, speaker_id: str, speaker_name: str) -> bool:
+def enroll_from_directory(
+    directory: str, speaker_id: str, speaker_name: str, user_id: str
+) -> bool:
     """Enroll speaker from all audio files in a directory."""
     audio_extensions = {".wav", ".flac", ".mp3", ".m4a", ".ogg"}
     dir_path = Path(directory)
@@ -179,7 +190,7 @@ def enroll_from_directory(directory: str, speaker_id: str, speaker_name: str) ->
         return False
 
     logger.info(f"Found {len(audio_files)} audio files in {directory}")
-    return enroll_multiple_files(audio_files, speaker_id, speaker_name)
+    return enroll_multiple_files(audio_files, speaker_id, speaker_name, user_id)
 
 
 def download_youtube_audio(
@@ -316,6 +327,9 @@ def main():
 
     # Speaker info arguments
     parser.add_argument("--id", help="Speaker ID (required for enrollment)")
+    parser.add_argument(
+        "--user-id", help="Chronicle user id owning the speaker (required to enroll)"
+    )
     parser.add_argument("--name", help="Speaker display name (required for enrollment)")
 
     # Optional arguments
@@ -347,25 +361,29 @@ def main():
 
     else:
         # Enrollment actions require ID and name
-        if not args.id or not args.name:
-            logger.error("❌ --id and --name are required for enrollment")
+        if not args.id or not args.name or not args.user_id:
+            logger.error("❌ --id, --name and --user-id are required for enrollment")
             return 1
 
         if args.file:
             success = enroll_single_file(
-                args.file, args.id, args.name, args.start, args.end
+                args.file, args.id, args.name, args.user_id, args.start, args.end
             )
 
         elif args.files:
-            success = enroll_multiple_files(args.files, args.id, args.name)
+            success = enroll_multiple_files(
+                args.files, args.id, args.name, args.user_id
+            )
 
         elif args.dir:
-            success = enroll_from_directory(args.dir, args.id, args.name)
+            success = enroll_from_directory(args.dir, args.id, args.name, args.user_id)
 
         elif args.youtube:
             audio_path = download_youtube_audio(args.youtube, args.start, args.end)
             if audio_path:
-                success = enroll_single_file(audio_path, args.id, args.name)
+                success = enroll_single_file(
+                    audio_path, args.id, args.name, args.user_id
+                )
                 os.unlink(audio_path)  # Clean up temporary file
             else:
                 success = False

extras/speaker-recognition/src/simple_speaker_recognition/api/core/__init__.py

@@ -1,8 +1,8 @@
 """Core utilities and shared components."""
 
 from .utils import (
-    extract_user_id_from_speaker_id,
     get_data_directory,
+    owner_of_speaker,
     safe_format_confidence,
     secure_temp_file,
     validate_confidence,
@@ -12,6 +12,6 @@
     "get_data_directory",
     "safe_format_confidence",
     "secure_temp_file",
-    "extract_user_id_from_speaker_id",
+    "owner_of_speaker",
     "validate_confidence",
 ]

extras/speaker-recognition/src/simple_speaker_recognition/api/core/utils.py

@@ -105,23 +105,19 @@ def secure_temp_file(suffix: str = ".wav") -> tempfile._TemporaryFileWrapper:
     return tempfile.NamedTemporaryFile(delete=False, suffix=suffix)
 
 
-def extract_user_id_from_speaker_id(speaker_id: str) -> str:
-    """Return the tenant for ``speaker_id``: its recorded owner, else the id's prefix.
-
-    The stored ``speakers.user_id`` is the ownership record and wins whenever the
-    speaker exists — it stays right for a speaker that was imported, renamed, or
-    created by a client that formats ids differently, none of which the id convention
-    survives.
-
-    The convention ``user_{user_id}_speaker_{...}`` is only consulted when no such
-    speaker exists yet, because enrolment *creates* one and the id is currently how
-    the caller states the tenant. That is a weak contract and the reason the audit
-    says a tenant should not be inferred from a SpeakerId; removing it needs an
-    explicit ``user_id`` on /enroll/upload, /enroll/batch and /enroll/append and on
-    each of their callers (Chronicle's client, the speaker WebUI, and two scripts).
-
-    Returns a string either way. It used to return ``int(parts[1])``, which stopped
-    parsing the moment a tenant became a Chronicle ObjectId.
+def owner_of_speaker(speaker_id: str) -> str:
+    """Return the tenant that owns ``speaker_id``, from the speaker's own record.
+
+    ``speakers.user_id`` is the ownership record, so it is the only thing consulted.
+    This used to parse the tenant out of the id — ``int(speaker_id.split("_")[1])`` —
+    which the audit flags as inferring a tenant from a SpeakerId: the
+    ``user_{user_id}_speaker_{...}`` shape is a naming convention, not a fact, and it
+    is wrong for a speaker that was imported or created by a client that names ids
+    differently. It also stopped parsing entirely once a tenant became a Chronicle
+    ObjectId rather than an integer.
+
+    Only for speakers that already exist. Enrolment creates one, so it takes an
+    explicit ``user_id`` instead of leaving the service to guess.
     """
 
     # Imported here rather than at module scope: database.queries imports the API
@@ -132,22 +128,12 @@ def extract_user_id_from_speaker_id(speaker_id: str) -> str:
     db = get_db_session()
     try:
         speaker = db.query(Speaker).filter(Speaker.id == speaker_id).first()
-        if speaker is not None:
-            return str(speaker.user_id)
+        if speaker is None:
+            raise HTTPException(404, f"Speaker not found: {speaker_id}")
+        return str(speaker.user_id)
     finally:
         db.close()
 
-    if not speaker_id.startswith("user_"):
-        raise HTTPException(
-            400,
-            "Unknown speaker and unrecognised id: expected "
-            f"'user_{{user_id}}_...', got: {speaker_id}",
-        )
-    parts = speaker_id.split("_")
-    if len(parts) < 3 or not parts[1]:
-        raise HTTPException(400, f"Cannot determine the owning user from: {speaker_id}")
-    return parts[1]
-
 
 def validate_confidence(confidence: Any, context: str = "") -> float:
     """Validate and sanitize confidence values from speaker identification.