Aime/app/diar.js

// Offline acoustic speaker diarization via sherpa-onnx (arm64 CPU, no pytorch).
// Splits a 16 kHz mono wav into speaker segments; AUTO speaker count (cluster
// threshold) because a conference phone ("Raumspinne") can put an unknown number
// of people on the single BT channel. Pinned to the A76 cores.
import { execFile } from 'node:child_process';
import { promisify } from 'node:util';
import fs from 'node:fs';
import * as spk from './spk.js';
const pexec = promisify(execFile);

// Auto-mode over-segmentation fix: sherpa's clustering splits ONE speaker into
// many clusters (e.g. 8-13 on a 2-person call, 61 on a 46-min one), and raising
// the cluster threshold barely helps. So after sherpa, we re-embed each cluster
// with the SAME TitaNet model used for voiceprint identity and MERGE clusters
// whose centroids are cosine-similar — i.e. the same voice that got split. This
// keeps the count dynamic (speakers can join/leave) without a fixed guess.
const MERGE_THRESHOLD = +(process.env.DIAR_MERGE_THRESHOLD || 0.5);
const REP_SEGMENTS = 3;    // longest segments per cluster averaged into its centroid
const MIN_REP_SEC = 0.4;   // skip ultra-short cuts for embedding when a longer one exists
// Stage 2: sub-second stray clusters (overlap crosstalk, noise) survive the
// centroid merge because they're too short to embed reliably. Any speaker with
// less than this much total speech is folded into the temporally-nearest real
// speaker. A genuine join/leave participant speaks well over this.
const MIN_SPEAKER_SEC = +(process.env.DIAR_MIN_SPEAKER_SEC || 3);
// Stage 3: a single >=MIN_SPEAKER_SEC stray cluster can still OPEN an anchor
// when its (short, unreliable) embedding lands below MERGE_THRESHOLD to every
// real speaker (#70: 3 phantom speakers with 2.5-4.8s total talk next to real
// ones with 53-736s). Total talk time over the whole call separates real
// participants from such noise: any FINAL speaker below this is folded into
// the acoustically nearest strong speaker. NOTE: measured on SEGMENT times
// (incl. intra-segment pauses), which run ~20-40% above summed word times —
// 10s let a 9.2s-word-time phantom through on #70, hence 20.
const MIN_TOTAL_SEC = +(process.env.DIAR_MIN_TOTAL_SEC || 20);

// Diarization on the capped A76 cores runs at ~1x wall-clock (a 46-min wav ≈
// 40 min). The old fixed 10-min timeout killed every recording over ~11 min and
// surfaced as an opaque "Command failed" (only sherpa's argv echo in stderr).
// Scale the timeout with the audio length: ~2.5x its duration, floor 20 min.
function diarTimeoutMs(wavPath) {
  let sec = 0;
  try { sec = Math.max(0, (fs.statSync(wavPath).size - 44) / 32000); } catch {} // 16kHz mono s16
  return Math.max(20 * 60 * 1000, Math.ceil(sec * 2.5) * 1000);
}

const DIR = '/opt/sherpa/sherpa-onnx-v1.13.3-linux-aarch64-shared-cpu';
const BIN = `${DIR}/bin/sherpa-onnx-offline-speaker-diarization`;
const SEG = process.env.DIAR_SEG || '/opt/sherpa/sherpa-onnx-pyannote-segmentation-3-0/model.onnx';
const EMB = process.env.DIAR_EMB || '/opt/sherpa/nemo_titanet_small.onnx';
// 0.7 (not 0.5): a lower threshold over-segments — a single (esp. child) voice
// split into several clusters. 0.7 merges more → fewer spurious speakers while
// still auto-detecting the count (needed for a conference phone / "Raumspinne").
const THRESHOLD = process.env.DIAR_THRESHOLD || '0.7';

function env() { return { ...process.env, LD_LIBRARY_PATH: `${DIR}/lib` }; }

// --- v2 pipeline (default) --------------------------------------------------
// The old path let sherpa cluster its own window embeddings and then tried to
// repair the result by merging whole clusters. On a 23-person weekly (#74) that
// produced 44 speakers with the same voice spread over several of them: one
// centroid per (already wrong) cluster from at most 3 cuts is too coarse to
// separate 20+ voices, and a single fixed cosine threshold on RAW embeddings
// can't work at all — untouched TitaNet vectors of one channel share a large
// common component (mean cross-turn cosine ≈ 0.4).
//
// v2 therefore ignores sherpa's labels and only keeps its (pyannote) turn
// BOUNDARIES, then does the identity part itself:
//   1. slice every turn into 3 s windows (1.5 s hop),
//   2. embed each window,
//   3. subtract the per-recording mean and L2-normalise (channel/room
//      compensation — this alone moves cross-turn cosine to ≈ 0 and lifts the
//      same/different separation to d' ≈ 3.5),
//   4. average-linkage agglomerative clustering (all pairs decide a merge, not
//      one greedy first-match link as before) at a FINE threshold,
//   5. fold clusters with almost no talk time into the nearest real speaker,
//   6. keep merging the remaining (strong) clusters up to the FINAL threshold,
//      or until a user-pinned speaker count is reached,
//   7. duration-weighted majority label per turn.
// Live name markers (see resolveMarkers) additionally constrain step 4/6:
// windows around markers with the SAME name must land in one cluster, windows
// around markers with DIFFERENT names must never be merged.
// DIAR_MODE=legacy restores the old behaviour.
const MODE = process.env.DIAR_MODE || 'v2';
const WIN_SEC = +(process.env.DIAR_WIN_SEC || 3.0);
const HOP_SEC = +(process.env.DIAR_HOP_SEC || 1.5);
const MIN_WIN_SEC = +(process.env.DIAR_MIN_WIN_SEC || 0.6);
const MAX_WINDOWS = +(process.env.DIAR_MAX_WINDOWS || 4000);
// Merge thresholds on MEAN-CENTRED cosine (see cluster-lab measurements).
// Stage A over-clusters at a fixed high threshold (0.45 measured stable on #74).
const COS_FINE = +(process.env.DIAR_COS_FINE || 0.45);
// Stage B's threshold is derived PER RECORDING (see mergeThreshold): a fixed one
// cannot serve both a clean 23-person conference and a noisy 4-person call —
// 0.30 gave 21 speakers on #74 (~right) but 9 on #70 (truly 4), while 0.15 fixed
// #70 and collapsed #74 to 14. DIAR_COS_FINAL pins it for debugging.
const COS_FINAL_ENV = process.env.DIAR_COS_FINAL ? +process.env.DIAR_COS_FINAL : null;
const COS_FINAL_MIN = +(process.env.DIAR_COS_FINAL_MIN || 0.05);
const COS_FINAL_MAX = +(process.env.DIAR_COS_FINAL_MAX || 0.35);
// sherpa cluster threshold for the v2 pass: its labels are discarded, we only
// want fine turn boundaries (and min-duration-off=0 so adjacent turns of the
// same label are NOT glued together).
const FINE_THRESHOLD = process.env.DIAR_FINE_THRESHOLD || '0.2';

// Raw sherpa turns. mode 'fine' = boundaries only (v2), else the legacy call
// with its own clustering (threshold or pinned count).
async function sherpaTurns(wavPath, numSpeakers, fine) {
  const clusterArg = fine ? `--clustering.cluster-threshold=${FINE_THRESHOLD}`
    : (numSpeakers > 0 ? `--clustering.num-clusters=${numSpeakers}` : `--clustering.cluster-threshold=${THRESHOLD}`);
  const args = [
    'taskset', '-c', '4-7', BIN,
    `--segmentation.pyannote-model=${SEG}`,
    `--segmentation.num-threads=4`,
    `--embedding.model=${EMB}`,
    `--embedding.num-threads=4`,
    clusterArg,
    ...(fine ? ['--min-duration-on=0.2', '--min-duration-off=0.0'] : []),
    wavPath,
  ];
  const { stdout } = await pexec(args[0], args.slice(1), { env: env(), maxBuffer: 1 << 22, timeout: diarTimeoutMs(wavPath) });
  const segs = [];
  for (const line of stdout.split('\n')) {
    const m = line.match(/([\d.]+)\s*--\s*([\d.]+)\s*(speaker_\d+)/);
    if (m) segs.push({ start: +m[1], end: +m[2], speaker: m[3] });
  }
  segs.sort((a, b) => a.start - b.start);
  return segs;
}

// Returns [{ start, end, speaker }] (seconds, speaker = "speaker_00"…).
// numSpeakers > 0 fixes the count when the user knows it, otherwise the count
// is found automatically (people can join/leave a conference call).
export async function diarize(wavPath, numSpeakers, opts = {}) {
  const n = parseInt(numSpeakers, 10) > 0 ? parseInt(numSpeakers, 10) : 0;
  if (MODE === 'v2') {
    const turns = await sherpaTurns(wavPath, n, true);
    if (turns.length) {
      try { return await reclusterTurns(wavPath, turns, n, opts.markers || []); }
      catch (e) { console.log('diar: v2 re-clustering failed, falling back to legacy:', e.message); }
    }
  }
  const segs = await sherpaTurns(wavPath, n, false);
  // Only auto mode over-segments; a user-pinned count is already trusted.
  if (!n && segs.length) {
    try { return await consolidateSpeakers(wavPath, segs); }
    catch (e) { console.log('diar: consolidation failed, using raw clusters:', e.message); }
  }
  return segs;
}

// --- wav slicing (16 kHz mono s16, no ffmpeg per window) --------------------
function loadWav(p) {
  const b = fs.readFileSync(p);
  if (b.toString('ascii', 0, 4) !== 'RIFF' || b.toString('ascii', 8, 12) !== 'WAVE') throw new Error('kein RIFF/WAVE: ' + p);
  let off = 12, fmt = null, data = null;
  while (off + 8 <= b.length) {
    const id = b.toString('ascii', off, off + 4), size = b.readUInt32LE(off + 4);
    if (id === 'fmt ') fmt = { ch: b.readUInt16LE(off + 10), rate: b.readUInt32LE(off + 12), bits: b.readUInt16LE(off + 22) };
    if (id === 'data') { data = { start: off + 8, size: Math.min(size, b.length - off - 8) }; break; }
    off += 8 + size + (size & 1);
  }
  if (!fmt || !data) throw new Error('wav-Chunks fehlen');
  if (fmt.bits !== 16) throw new Error('wav ist nicht s16 (' + fmt.bits + ')');
  return { buf: b, fmt, data };
}
function writeSlice(w, startSec, endSec, outPath) {
  const bpf = 2 * w.fmt.ch;
  const s = Math.max(0, Math.floor(startSec * w.fmt.rate) * bpf);
  const e = Math.min(w.data.size, Math.ceil(endSec * w.fmt.rate) * bpf);
  const n = Math.max(0, e - s);
  const h = Buffer.alloc(44);
  h.write('RIFF', 0); h.writeUInt32LE(36 + n, 4); h.write('WAVE', 8);
  h.write('fmt ', 12); h.writeUInt32LE(16, 16); h.writeUInt16LE(1, 20);
  h.writeUInt16LE(w.fmt.ch, 22); h.writeUInt32LE(w.fmt.rate, 24);
  h.writeUInt32LE(w.fmt.rate * bpf, 28); h.writeUInt16LE(bpf, 32); h.writeUInt16LE(16, 34);
  h.write('data', 36); h.writeUInt32LE(n, 40);
  fs.writeFileSync(outPath, Buffer.concat([h, w.buf.subarray(w.data.start + s, w.data.start + e)]));
  return n / (w.fmt.rate * bpf);
}

// --- vector helpers ---------------------------------------------------------
function l2(v) { let s = 0; for (const x of v) s += x * x; s = Math.sqrt(s) + 1e-9; return v.map(x => x / s); }
function meanVec(vs) {
  const d = vs[0].length, m = new Array(d).fill(0);
  for (const v of vs) for (let i = 0; i < d; i++) m[i] += v[i] / vs.length;
  return m;
}
function dot(a, b) { let s = 0; for (let i = 0; i < a.length; i++) s += a[i] * b[i]; return s; }

// Per-recording merge threshold. The pairwise window-similarity histogram of one
// recording is bimodal: a big "different speaker" mode around 0 and a "same
// speaker" mode further right. How far right depends entirely on the audio — on
// a clean conference recording the same-speaker mode sits at ~0.30, on a noisy
// BT call at ~0.21 with much more spread. A 2-Gaussian mixture (EM) is fitted to
// a deterministic sample of the pairs and the point where the two weighted
// densities cross becomes the merge threshold: measured 0.20 for #74 (→ 20
// speakers for 23 participants) and 0.07 for #70 (→ exactly the 4 known ones),
// where no single fixed value could do both.
function mergeThreshold(X) {
  const n = X.length;
  const vals = [];
  for (let k = 1; k < 400000 && vals.length < 60000; k++) {
    const i = (k * 7919) % n, j = (k * 104729 + 13) % n;
    if (i !== j) vals.push(dot(X[i], X[j]));
  }
  if (vals.length < 50) return (COS_FINAL_MIN + COS_FINAL_MAX) / 2;
  const N = (x, m, s) => Math.exp(-((x - m) ** 2) / (2 * s * s)) / (s * Math.sqrt(2 * Math.PI));
  let m1 = -0.1, m2 = 0.5, s1 = 0.15, s2 = 0.15, w1 = 0.8;
  for (let it = 0; it < 100; it++) {
    let sw1 = 0, sw2 = 0, sm1 = 0, sm2 = 0, sv1 = 0, sv2 = 0;
    for (const x of vals) {
      const a = w1 * N(x, m1, s1), b = (1 - w1) * N(x, m2, s2);
      const r = a / (a + b + 1e-300);
      sw1 += r; sw2 += 1 - r; sm1 += r * x; sm2 += (1 - r) * x;
    }
    const nm1 = sm1 / sw1, nm2 = sm2 / sw2;
    for (const x of vals) {
      const a = w1 * N(x, m1, s1), b = (1 - w1) * N(x, m2, s2);
      const r = a / (a + b + 1e-300);
      sv1 += r * (x - nm1) ** 2; sv2 += (1 - r) * (x - nm2) ** 2;
    }
    m1 = nm1; m2 = nm2;
    s1 = Math.max(0.02, Math.sqrt(sv1 / sw1));
    s2 = Math.max(0.02, Math.sqrt(sv2 / sw2));
    w1 = sw1 / vals.length;
  }
  if (!(m2 > m1)) return (COS_FINAL_MIN + COS_FINAL_MAX) / 2;
  let lo = m1, hi = m2, t = (lo + hi) / 2;
  for (let i = 0; i < 60; i++) {
    t = (lo + hi) / 2;
    if (w1 * N(t, m1, s1) - (1 - w1) * N(t, m2, s2) > 0) lo = t; else hi = t;
  }
  return Math.min(COS_FINAL_MAX, Math.max(COS_FINAL_MIN, t));
}

// Average-linkage agglomerative clustering over unit vectors X.
// Stops when the best average-linkage cosine drops below `threshold`, or when
// `kStop` clusters remain (user-pinned speaker count). Best-match caching keeps
// this ~O(n^2) for the few-thousand windows of a long recording.
//   initGroups: continue from an existing clustering (arrays of X indices)
//   names:      per-init-group name tag; two clusters that both carry a (then
//               necessarily different) name are never merged — the live markers'
//               cannot-link constraint.
// Returns the surviving groups (arrays of X indices) and their name tags.
function ahcAverage(X, threshold, kStop = 0, initGroups = null, names = null) {
  const groups = (initGroups || X.map((_, i) => [i])).map(g => [...g]);
  const tag = groups.map((_, i) => (names && names[i]) || null);
  const m = groups.length;
  const A = new Float32Array(m * m);
  for (let i = 0; i < m; i++) {
    A[i * m + i] = 1;
    for (let j = i + 1; j < m; j++) {
      let s = 0;
      for (const a of groups[i]) for (const b of groups[j]) s += dot(X[a], X[b]);
      s /= groups[i].length * groups[j].length;
      A[i * m + j] = s; A[j * m + i] = s;
    }
  }
  const cnt = new Int32Array(m); groups.forEach((g, i) => cnt[i] = g.length);
  const alive = new Uint8Array(m).fill(1);
  const bestJ = new Int32Array(m).fill(-1);
  const bestS = new Float32Array(m).fill(-Infinity);
  const blocked = (i, j) => tag[i] && tag[j] && tag[i] !== tag[j];
  const rescan = i => {
    let bj = -1, bs = -Infinity;
    for (let j = 0; j < m; j++) {
      if (j === i || !alive[j] || blocked(i, j)) continue;
      const s = A[i * m + j]; if (s > bs) { bs = s; bj = j; }
    }
    bestJ[i] = bj; bestS[i] = bs;
  };
  for (let i = 0; i < m; i++) rescan(i);
  let k = m;
  while (k > 1) {
    let a = -1, bs = -Infinity;
    for (let i = 0; i < m; i++) if (alive[i] && bestJ[i] >= 0 && bestS[i] > bs) { bs = bestS[i]; a = i; }
    if (a < 0) break;
    const b = bestJ[a];
    if (kStop > 0 ? k <= kStop : bs < threshold) break;
    const ca = cnt[a], cb = cnt[b];
    for (let c = 0; c < m; c++) {
      if (!alive[c] || c === a || c === b) continue;
      const v = (A[a * m + c] * ca + A[b * m + c] * cb) / (ca + cb);
      A[a * m + c] = v; A[c * m + a] = v;
    }
    cnt[a] = ca + cb; alive[b] = 0; k--;
    groups[a] = groups[a].concat(groups[b]); groups[b] = [];
    const gained = !tag[a] && !!tag[b];
    if (gained) tag[a] = tag[b];
    rescan(a);
    // A cluster that just gained a name blocks pairs that were allowed before,
    // so every cached best-match has to be re-checked; otherwise only those
    // pointing at the two merged clusters.
    for (let i = 0; i < m; i++)
      if (alive[i] && (gained || bestJ[i] === b || bestJ[i] === a)) rescan(i);
  }
  const out = [], outTags = [];
  for (let i = 0; i < m; i++) if (alive[i]) { out.push(groups[i]); outTags.push(tag[i]); }
  return { groups: out, tags: outTags };
}

// v2 entry: embed the windows, then cluster them.
async function reclusterTurns(wavPath, turns, numSpeakers = 0, markers = []) {
  const ok = await embedWindows(wavPath, turns);
  return clusterWindows(turns, ok, numSpeakers, markers);
}

// Map each live marker to the turn the named person was speaking in: the turn
// covering the marker, else the last one ending shortly before it (the name is
// typed a few seconds AFTER the voice was heard), else the next one starting
// right after. Returns { name -> [turnIndex…] }.
const ANCHOR_BACK_SEC = +(process.env.MARK_PRE_MS || 25000) / 1000;
const ANCHOR_FWD_SEC = +(process.env.MARK_POST_MS || 8000) / 1000;
const ANCHOR_MIN_SEC = 1.0;   // too short to identify a voice from
function markerAnchors(turns, markers) {
  const byName = new Map();
  const usable = turns.map((t, i) => ({ i, ...t })).filter(t => t.end - t.start >= ANCHOR_MIN_SEC);
  for (const mk of markers || []) {
    const t = (mk.t_ms || 0) / 1000;
    let pick = -1, bd = Infinity;
    for (const c of usable) {
      const back = t - c.end;       // >0: turn ended before the marker
      const fwd = c.start - t;      // >0: turn starts after the marker
      let d = Infinity;
      if (back <= 0 && fwd <= 0) d = 0;                                   // covers it
      else if (back > 0 && back <= ANCHOR_BACK_SEC) d = back;             // typing lag
      else if (fwd > 0 && fwd <= ANCHOR_FWD_SEC) d = fwd * (ANCHOR_BACK_SEC / Math.max(1, ANCHOR_FWD_SEC));
      if (d < bd) { bd = d; pick = c.i; }
    }
    if (pick < 0) continue;
    const key = String(mk.name).trim().toLowerCase();
    (byName.get(key) || byName.set(key, []).get(key)).push(pick);
  }
  return byName;
}

// Slice + embed every window of every turn (the only part that touches audio).
export async function embedWindows(wavPath, turns) {
  const wav = loadWav(wavPath);
  // Adaptive hop: keep the window count (and the O(n^2) clustering) bounded on
  // very long recordings.
  const winCount = (hop) => turns.reduce((a, t) => {
    const d = t.end - t.start;
    return a + (d < MIN_WIN_SEC ? 0 : d <= WIN_SEC ? 1 : Math.max(1, Math.ceil((d - MIN_WIN_SEC) / hop)));
  }, 0);
  let hop = HOP_SEC;
  while (winCount(hop) > MAX_WINDOWS && hop < 60) hop *= 1.5;

  const wins = [];
  turns.forEach((t, ti) => {
    const d = t.end - t.start;
    if (d < MIN_WIN_SEC) return;
    if (d <= WIN_SEC) { wins.push({ ti, start: t.start, end: t.end }); return; }
    for (let s = t.start; s + MIN_WIN_SEC <= t.end; s += hop) wins.push({ ti, start: s, end: Math.min(s + WIN_SEC, t.end) });
  });
  if (!wins.length) throw new Error('keine Fenster für Sprecher-Embeddings');

  const tmp = `/tmp/diarwin_${process.pid}.wav`;
  const t0 = Date.now();
  for (const w of wins) {
    writeSlice(wav, w.start, w.end, tmp);
    try { w.vec = await spk.embed(tmp); } catch (e) { w.err = e.message; }
  }
  try { fs.unlinkSync(tmp); } catch {}
  const ok = wins.filter(w => Array.isArray(w.vec) && w.vec.length);
  if (ok.length < 2) throw new Error('Embedding fehlgeschlagen (' + (wins[0] && wins[0].err || 'unbekannt') + ')');
  console.log(`diar: ${ok.length}/${wins.length} Fenster-Embeddings (hop ${hop.toFixed(1)}s) in ${((Date.now() - t0) / 1000).toFixed(0)}s`);
  return ok;
}

// The heart of v2: window embeddings → clusters → per-turn speaker labels.
// Split from the audio/embedding part so the clustering can be replayed offline
// on cached embeddings (lab/lab-eval.mjs) — the tuning ran against THIS code.
export function clusterWindows(turns, ok, numSpeakers = 0, markers = []) {
  // Channel compensation: L2 → subtract the recording mean → L2 again.
  let X = ok.map(w => l2(w.vec));
  const mu = meanVec(X);
  X = X.map(v => l2(v.map((x, i) => x - mu[i])));

  // Windows of a turn, and the turn label from its windows (duration-weighted).
  const winsOfTurn = new Map();
  ok.forEach((w, i) => { (winsOfTurn.get(w.ti) || winsOfTurn.set(w.ti, []).get(w.ti)).push(i); });
  const labelTurns = (groups) => {
    const of = new Int32Array(ok.length).fill(-1);
    groups.forEach((g, gi) => g.forEach(i => of[i] = gi));
    const lab = new Array(turns.length).fill(-1);
    for (const [ti, idxs] of winsOfTurn) {
      const m = new Map();
      for (const i of idxs) m.set(of[i], (m.get(of[i]) || 0) + (ok[i].end - ok[i].start));
      let bl = -1, bd = -1;
      for (const [l, d] of m) if (d > bd) { bd = d; bl = l; }
      lab[ti] = bl;
    }
    const talk = new Map();
    turns.forEach((t, ti) => { if (lab[ti] >= 0) talk.set(lab[ti], (talk.get(lab[ti]) || 0) + (t.end - t.start)); });
    return { lab, talk };
  };
  const centroidOf = g => l2(meanVec(g.map(i => X[i])));

  const cosFinal = COS_FINAL_ENV !== null ? COS_FINAL_ENV : mergeThreshold(X);

  // Stage A: deliberate over-clustering — every real speaker ends up in one or
  // more clusters, but (almost) no cluster mixes two speakers.
  let { groups } = ahcAverage(X, COS_FINE, 0);
  const fineCount = groups.length;

  // Fold clusters with almost no talk time (overlap, crosstalk, door noise) into
  // the acoustically nearest real speaker; they must not survive as "speakers"
  // and must not be merge partners in stage B. The floor scales with the length
  // of the recording — a flat 20 s would swallow a real second speaker on a
  // 2-minute call, while 3 s would let noise through on a 33-minute conference.
  const speechSec = turns.reduce((a, t) => a + (t.end - t.start), 0);
  const minTotal = Math.min(MIN_TOTAL_SEC, Math.max(3, speechSec * 0.02));
  const { talk } = labelTurns(groups);
  const strongIdx = groups.map((_, i) => i).filter(i => (talk.get(i) || 0) >= minTotal);
  let folded = 0;
  if (strongIdx.length && strongIdx.length < groups.length) {
    const cents = new Map(strongIdx.map(i => [i, centroidOf(groups[i])]));
    const keep = new Map(strongIdx.map(i => [i, [...groups[i]]]));
    groups.forEach((g, i) => {
      if (keep.has(i)) return;
      const c = centroidOf(g);
      let best = strongIdx[0], bs = -Infinity;
      for (const s of strongIdx) { const v = dot(c, cents.get(s)); if (v > bs) { bs = v; best = s; } }
      keep.get(best).push(...g);
      folded++;
    });
    groups = [...keep.values()];
  }

  // Live markers → must-link (same name in one group) + cannot-link (different
  // names never merge). Applied to the folded stage-A clusters: the cluster a
  // marker's turn landed in carries that name.
  const anchors = markerAnchors(turns, markers);
  let tags = null;
  if (anchors.size) {
    const { lab } = labelTurns(groups);
    const nameOfGroup = new Map();     // groupIdx -> name (first/most-voted wins)
    const votes = new Map();
    for (const [name, tis] of anchors) for (const ti of tis) {
      const g = lab[ti];
      if (g < 0) continue;
      const key = g + '\0' + name;
      votes.set(key, (votes.get(key) || 0) + 1);
    }
    const bestPerGroup = new Map();
    for (const [key, v] of votes) {
      const [g, name] = key.split('\0');
      const cur = bestPerGroup.get(g);
      if (!cur || v > cur.v) bestPerGroup.set(g, { name, v });
    }
    for (const [g, { name }] of bestPerGroup) nameOfGroup.set(+g, name);
    // must-link: all groups tagged with the same name become ONE group
    const byName = new Map();
    for (const [g, name] of nameOfGroup) (byName.get(name) || byName.set(name, []).get(name)).push(g);
    const used = new Set();
    const merged = [], mtags = [];
    for (const [name, gs] of byName) {
      merged.push(gs.flatMap(g => groups[g]));
      mtags.push(name);
      gs.forEach(g => used.add(g));
    }
    groups.forEach((g, i) => { if (!used.has(i)) { merged.push(g); mtags.push(null); } });
    groups = merged; tags = mtags;
    console.log(`diar v2: ${anchors.size} Marker-Name(n) als Zwangsbedingung (${[...byName.values()].reduce((a, g) => a + g.length, 0)} Cluster verankert)`);
  }

  // Stage B: merge the remaining strong clusters up to the final threshold — or
  // down to the user-pinned speaker count. Marker constraints are respected.
  const stageB = ahcAverage(X, cosFinal, numSpeakers, groups, tags);
  groups = stageB.groups;

  const { lab: turnLab } = labelTurns(groups);
  // Turns too short to embed inherit the temporally nearest labelled turn.
  for (let i = 0; i < turns.length; i++) {
    if (turnLab[i] >= 0) continue;
    let best = -1, bd = Infinity;
    for (let j = 0; j < turns.length; j++) {
      if (turnLab[j] < 0) continue;
      const d = Math.min(Math.abs(turns[j].end - turns[i].start), Math.abs(turns[i].end - turns[j].start));
      if (d < bd) { bd = d; best = turnLab[j]; }
    }
    turnLab[i] = best >= 0 ? best : 0;
  }

  // Renumber by first appearance and glue adjacent turns of the same speaker
  // (gap <= 0.5 s — what sherpa's min-duration-off would have done).
  const rename = new Map(); let nSp = 0;
  const out = [];
  turns.forEach((t, ti) => {
    const l = turnLab[ti];
    if (!rename.has(l)) rename.set(l, `speaker_${String(nSp++).padStart(2, '0')}`);
    const sp = rename.get(l);
    const last = out[out.length - 1];
    if (last && last.speaker === sp && t.start - last.end <= 0.5) last.end = Math.max(last.end, t.end);
    else out.push({ start: t.start, end: t.end, speaker: sp });
  });
  console.log(`diar v2: ${turns.length} Turns, ${ok.length} Fenster → ${fineCount} Cluster (fein ${COS_FINE})` +
    ` → ${folded} schwache <${minTotal.toFixed(0)}s eingefaltet → ${nSp} Sprecher` +
    (numSpeakers ? ` [fix ${numSpeakers}]` : ` (final ${cosFinal.toFixed(3)}${COS_FINAL_ENV === null ? ' automatisch' : ''})`));
  return out;
}

// Average voiceprint of a cluster: embed its longest few segments and mean them
// (short cuts give noisy embeddings, so prefer the longest and average out noise).
async function clusterCentroid(wavPath, clusterSegs) {
  const sorted = [...clusterSegs].sort((a, b) => (b.end - b.start) - (a.end - a.start));
  const pick = sorted.filter(s => s.end - s.start >= MIN_REP_SEC).slice(0, REP_SEGMENTS);
  const use = pick.length ? pick : sorted.slice(0, 1);
  const vecs = [];
  for (const s of use) {
    const seg = `/tmp/diarcons_${process.pid}_${Math.round(s.start * 1000)}.wav`;
    try {
      await pexec('ffmpeg', ['-y', '-i', wavPath, '-ss', String(s.start),
        '-to', String(Math.min(s.end, s.start + 8)), '-ar', '16000', '-ac', '1', seg]);
      const v = await spk.embed(seg);
      if (Array.isArray(v) && v.length) vecs.push(v);
    } catch { /* skip this cut */ }
    finally { try { fs.unlinkSync(seg); } catch {} }
  }
  if (!vecs.length) return null;
  const dim = vecs[0].length, c = new Array(dim).fill(0);
  for (const v of vecs) for (let i = 0; i < dim; i++) c[i] += v[i];
  for (let i = 0; i < dim; i++) c[i] /= vecs.length;
  return c;
}

// Merge sherpa's over-segmented clusters (anchor-based, see fn body),
// then renumber the surviving speakers by first appearance. Segments whose
// centroid couldn't be embedded keep their own label (safe: never lose a turn).
export async function consolidateSpeakers(wavPath, segs) {
  const labels = [...new Set(segs.map(s => s.speaker))];
  if (labels.length <= 1) return segs;

  // Centroid + total speech duration per raw sherpa cluster.
  const cents = {}, dur = {};
  for (const lab of labels) {
    const cs = segs.filter(s => s.speaker === lab);
    cents[lab] = await clusterCentroid(wavPath, cs);
    dur[lab] = cs.reduce((a, s) => a + (s.end - s.start), 0);
  }

  // ANCHOR-based merge (replaces the old pairwise union-find, which chained
  // transitively — A~B, B~C folded A and C together — so one borderline link
  // near the 0.5 cosine could cascade and the count flipped run-to-run under
  // load/thermal-cap FP noise: the SAME 2-3 person call came out 2, 3, or 7).
  // Only long clusters (reliable embeddings) become anchors and, processed
  // LONGEST-FIRST, either join an existing speaker or open a new one — a stable,
  // order-deterministic decision. Every short/stray cluster is then assigned to
  // its NEAREST anchor (never opens a speaker), so noise can't fragment.
  const anchors = labels
    .filter(l => cents[l] && dur[l] >= MIN_SPEAKER_SEC)
    .sort((a, b) => (dur[b] - dur[a]) || (a < b ? -1 : 1));

  const speakerOf = {};   // rawLabel -> canonical speaker rep (a rawLabel)
  const reps = [];        // accepted speaker representatives
  for (const a of anchors) {
    let joined = null;
    for (const r of reps) if (spk.cosine(cents[a], cents[r]) >= MERGE_THRESHOLD) { joined = r; break; }
    if (joined) speakerOf[a] = joined; else { speakerOf[a] = a; reps.push(a); }
  }

  const nearestRep = (lab) => {
    if (!reps.length) return lab;                 // no anchors -> keep own label
    if (!cents[lab]) return reps[0];              // unembeddable -> dominant speaker
    let best = reps[0], bs = -Infinity;
    for (const r of reps) { const c = spk.cosine(cents[lab], cents[r]); if (c > bs) { bs = c; best = r; } }
    return best;
  };

  const merged = segs.map(s => ({ ...s, speaker: speakerOf[s.speaker] || nearestRep(s.speaker) }));

  // Stage 3: fold weak FINAL speakers (< MIN_TOTAL_SEC total talk over the
  // whole call) into the acoustically nearest strong one — see MIN_TOTAL_SEC.
  const total = {};
  for (const s of merged) total[s.speaker] = (total[s.speaker] || 0) + (s.end - s.start);
  const strong = reps.filter(r => (total[r] || 0) >= MIN_TOTAL_SEC);
  if (strong.length && strong.length < reps.length) {
    const fold = {};
    for (const r of Object.keys(total)) {
      if (strong.includes(r)) continue;
      let best = strong[0], bs = -Infinity;
      for (const t of strong) {
        const c = (cents[r] && cents[t]) ? spk.cosine(cents[r], cents[t]) : -1;
        if (c > bs) { bs = c; best = t; }
      }
      fold[r] = best;
    }
    for (const s of merged) if (fold[s.speaker]) s.speaker = fold[s.speaker];
    console.log(`diar: folded ${Object.keys(fold).length} weak speaker(s) <${MIN_TOTAL_SEC}s total talk`);
  }

  // Renumber surviving speakers by first appearance.
  const rootMap = {}; let n = 0;
  for (const s of merged) if (!(s.speaker in rootMap)) rootMap[s.speaker] = `speaker_${String(n++).padStart(2, '0')}`;
  if (n !== labels.length) console.log(`diar: consolidated ${labels.length} -> ${n} speakers (anchors ${reps.length}, merge>=${MERGE_THRESHOLD}, floor ${MIN_SPEAKER_SEC}s)`);
  return merged.map(s => ({ ...s, speaker: rootMap[s.speaker] }));
}

// Assign each word (t_start_ms/t_end_ms) the speaker whose segment overlaps it
// most. Words with no overlap inherit the previous word's speaker.
export function assignSpeakers(words, segs) {
  if (!segs.length) return words;
  let last = segs[0].speaker;
  for (const w of words) {
    const ws = w.t_start_ms / 1000, we = w.t_end_ms / 1000;
    let best = null, bestOv = 0;
    for (const s of segs) {
      const ov = Math.min(we, s.end) - Math.max(ws, s.start);
      if (ov > bestOv) { bestOv = ov; best = s.speaker; }
    }
    w.speaker = best || last;
    last = w.speaker;
  }
  return words;
}

// --- live name markers ------------------------------------------------------
// Contiguous runs of one speaker over the word list (ms).
export function speakerRuns(words) {
  const runs = [];
  for (const w of words || []) {
    if (!w.speaker) continue;
    const last = runs[runs.length - 1];
    if (last && last.speaker === w.speaker) last.end = w.t_end_ms;
    else runs.push({ speaker: w.speaker, start: w.t_start_ms, end: w.t_end_ms });
  }
  return runs;
}

// Resolve names the user typed DURING the recording to diarization clusters.
// A marker is stamped when the name is submitted, i.e. typically a few seconds
// AFTER that person started talking (hear voice → type → Enter), occasionally
// just before they start. So we score every speaker run in an asymmetric window
// around the marker by overlap × proximity, and prefer the run that covers the
// marker itself. Returns [{ ...marker, speaker, conf }] (conf: high | low | null).
const MARK_PRE = +(process.env.MARK_PRE_MS || 25000);   // look back (typing lag)
const MARK_POST = +(process.env.MARK_POST_MS || 8000);  // look ahead (typed early)
const MARK_DECAY = +(process.env.MARK_DECAY_MS || 10000);
export function resolveMarkers(words, markers) {
  const runs = speakerRuns(words);
  return (markers || []).map(m => {
    const t = m.t_ms || 0, lo = t - MARK_PRE, hi = t + MARK_POST;
    const score = {}, covering = {};
    for (const r of runs) {
      const ov = Math.min(hi, r.end) - Math.max(lo, r.start);
      if (ov <= 0) continue;
      const dist = r.start > t ? r.start - t : (r.end < t ? t - r.end : 0);
      score[r.speaker] = (score[r.speaker] || 0) + ov * Math.exp(-dist / MARK_DECAY);
      if (r.start <= t && r.end >= t) covering[r.speaker] = Math.max(covering[r.speaker] || 0, r.end - r.start);
    }
    const ranked = Object.entries(score).sort((a, b) => b[1] - a[1]);
    if (!ranked.length) return { ...m, speaker: null, conf: null };
    const [sp, s1] = ranked[0];
    const s2 = ranked[1] ? ranked[1][1] : 0;
    const conf = (covering[sp] >= 1500 && s1 >= 1.5 * s2) ? 'high'
      : (s1 >= 2 * s2 || ranked.length === 1) ? 'mid' : 'low';
    return { ...m, speaker: sp, conf };
  });
}

// Group consecutive same-speaker words into turns [{ speaker, text }].
export function toTurns(words) {
  const turns = [];
  for (const w of words) {
    const sp = w.speaker || 'speaker_00';
    if (!turns.length || turns[turns.length - 1].speaker !== sp) turns.push({ speaker: sp, text: w.word });
    else turns[turns.length - 1].text += ' ' + w.word;
  }
  return turns;
}