Aime/app/server.js

import express from 'express';
import path from 'node:path';
import fs from 'node:fs';
import { execFile } from 'node:child_process';
import { promisify } from 'node:util';
import { fileURLToPath } from 'node:url';
import * as bt from './bt.js';
import * as rec from './recorder.js';
import * as db from './db.js';
import * as llm from './llm.js';
import * as diar from './diar.js';
import * as spk from './spk.js';
const pexec = promisify(execFile);

// Glossary → { soundalike(lowercased): correctTerm }. Only entries WITH
// soundalikes get a deterministic swap; term-only entries are left to the LLM hint.
function glossaryMap(glossary) {
  const m = {};
  for (const g of glossary || []) {
    if (!g.term || !g.soundalikes) continue;
    for (const s of String(g.soundalikes).split(',')) {
      const k = s.trim().toLowerCase();
      if (k) m[k] = String(g.term).trim();
    }
  }
  return m;
}
// Whole-word replace (Unicode-aware, case-insensitive match): swaps any token
// equal to a known soundalike for its correct term, leaving punctuation intact.
function applyGlossaryText(text, m) {
  if (!text || !m || !Object.keys(m).length) return text;
  return String(text).replace(/[\p{L}\p{N}]+/gu, (tok) => m[tok.toLowerCase()] || tok);
}

// Build the LLM input from stored words. If acoustic diarization tagged speakers,
// produce a labelled dialogue ("Sprecher 1: …") so the LLM keeps real speakers
// instead of guessing; otherwise the plain transcript.
function buildBasis(rec, nameMap = {}) {
  const words = rec.words || [];
  const hasSpk = words.some(w => w.speaker);
  if (!hasSpk) return { text: rec.transcript || '', labeled: false };
  const turns = diar.toTurns(words);
  const map = {}; let n = 0;
  const label = sp => nameMap[sp] || (map[sp] || (map[sp] = `Sprecher ${++n}`));
  const text = turns.map(t => `${label(t.speaker)}: ${t.text}`).join('\n');
  return { text, labeled: true };
}

const REC_DIR2 = path.join(path.dirname(fileURLToPath(import.meta.url)), 'recordings');
// Match each diarization cluster against enrolled voiceprints → { speaker: name }.
// Cuts the speaker's longest contiguous run from the kept wav, embeds it, cosine-
// matches. Returns only confident matches (others stay "Sprecher N").
async function identifySpeakers(recData) {
  const words = recData.words || [];
  const speakers = [...new Set(words.filter(w => w.speaker).map(w => w.speaker))];
  if (!speakers.length || !recData.wav_file) return {};
  const wavPath = path.join(REC_DIR2, recData.wav_file);
  if (!fs.existsSync(wavPath)) return {};
  const vps = await db.listVoiceprints();
  if (!vps.length) return {};
  const best = {}; let cur = null;
  for (const w of words) {
    if (!cur || cur.speaker !== w.speaker) cur = { speaker: w.speaker, start: w.t_start_ms, end: w.t_end_ms };
    else cur.end = w.t_end_ms;
    const len = cur.end - cur.start;
    if (!best[cur.speaker] || len > best[cur.speaker].len) best[cur.speaker] = { start: cur.start, end: cur.end, len };
  }
  const map = {};
  for (const sp of speakers) {
    const r = best[sp]; if (!r) continue;
    const seg = `/tmp/idseg_${recData.id}_${sp}.wav`;
    try {
      await pexec('ffmpeg', ['-y', '-i', wavPath, '-ss', String(r.start / 1000), '-to', String(r.end / 1000), '-ar', '16000', '-ac', '1', seg]);
      const emb = await spk.embed(seg);
      const m = spk.bestMatch(emb, vps);
      if (m) map[sp] = m.name;
    } catch (e) { console.error('identify', sp, e.message); }
    finally { try { fs.unlinkSync(seg); } catch {} }
  }
  return map;
}

// Run LLM analysis for a recording and persist title/summary/todos/participants/
// clean dialogue. Background (NPU ~30-90s) so record/stop stays fast.
// A76 clock cap for the whole run: rkllm burns ~1.7 CPU cores alongside the NPU
// and a long Gemma generation hit 95-97°C uncapped on the fanless board (#70,
// 16-min run) — same shutdown risk transcribeWav/rediarize already guard against.
async function analyzeRecording(id) {
  // Free the resident rkllm (~5.9GB) BEFORE identifySpeakers: the TitaNet
  // spk-embed helper OOM-dies next to it ("spk beendet" per cluster, #71 got
  // 48 raw speakers). llm.analyze() below respawns it (~96s load, noise
  // against a 15-min analysis).
  llm.stopServer();
  await pexec('sudo', ['-n', '/opt/callrec/aime-cpucap.sh', '1608000']).catch(() => {});
  try {
    const rec = await db.getRecording(id);
    if (!rec) return;
    const nameMap = await identifySpeakers(rec).catch(() => ({}));
    // Seed the per-recording speaker names from acoustic identification (fills
    // blanks only; manual edits stay). Manual names win for the LLM labelling.
    await db.mergeSpeakerNames(id, nameMap).catch(() => {});
    const names = { ...nameMap, ...(rec.speaker_names || {}) };
    const glossary = (await db.listGlossary().catch(() => [])).filter(g => g.enabled);
    // Deterministic soundalike replacement: the 3B model is unreliable at obeying
    // a "map Amy->Aime" instruction (it treats "Amy" as a valid name), so entries
    // WITH soundalikes are applied as a hard whole-word swap on the words FIRST —
    // then the LLM sees the correct term and the turn view shows it too. Entries
    // without soundalikes still rely on the (contextual) LLM glossary hint.
    const gmap = glossaryMap(glossary);
    if (Object.keys(gmap).length && Array.isArray(rec.words) && rec.words.length) {
      let changed = false;
      for (const w of rec.words) { const nt = applyGlossaryText(w.word, gmap); if (nt !== w.word) { w.word = nt; changed = true; } }
      if (changed) { await db.replaceWords(id, rec.words); console.log(`glossary: applied soundalike swaps to #${id} words`); }
    }
    const { text, labeled } = buildBasis(rec, names);
    const a = await llm.analyze(text, { labeled, glossary });
    // Participants = the ACOUSTICALLY diarized speakers (named where known),
    // not the LLM's list — the model returned every name MENTIONED in the
    // conversation as a "Teilnehmer" (11 names on a 3-person call, #69).
    // LLM participants stay only as fallback when there is no diarization.
    const spkOrder = [];
    for (const w of (rec.words || [])) if (w.speaker && !spkOrder.includes(w.speaker)) spkOrder.push(w.speaker);
    if (spkOrder.length) a.participants = spkOrder.map((sp, i) => names[sp] || `Sprecher ${i + 1}`);
    // Belt-and-suspenders: also swap in the LLM outputs (cheap, catches any the
    // model reintroduced).
    for (const k of ['title', 'summary', 'transcriptClean']) if (a[k]) a[k] = applyGlossaryText(a[k], gmap);
    a.participants = (a.participants || []).map(p => applyGlossaryText(p, gmap));
    a.todos = (a.todos || []).map(t => applyGlossaryText(t, gmap));
    await db.updateAnalysis(id, { ...a, status: 'done' });
    console.log(`analysis done for #${id}: ${a.title}`);
  } catch (e) {
    console.error(`analysis failed for #${id}:`, e.message);
    await db.updateAnalysis(id, { title: null, summary: null, todos: null, status: 'error' }).catch(() => {});
  } finally {
    await pexec('sudo', ['-n', '/opt/callrec/aime-cpucap.sh', '2256000']).catch(() => {});
  }
}

const __dirname = path.dirname(fileURLToPath(import.meta.url));
const app = express();
app.use(express.json());

const REC_DIR = path.join(__dirname, 'recordings');
app.use('/media', express.static(REC_DIR));
app.use(express.static(path.join(__dirname, 'web/dist')));

const wrap = fn => (req, res) => fn(req, res).catch(e => res.status(500).json({ error: String(e && e.message || e) }));

app.get('/api/health', (req, res) => res.json({ ok: true, ...rec.status() }));

// --- Bluetooth ---
app.post('/api/bt/scan', wrap(async (req, res) => {
  const secs = Math.min(+(req.body && req.body.seconds) || 10, 30);
  res.json(await bt.scan(secs));
}));
app.get('/api/bt/devices', wrap(async (req, res) => res.json(await bt.knownDevices())));
app.post('/api/bt/pair', wrap(async (req, res) => res.json(await bt.pair(req.body.mac))));
app.post('/api/bt/connect', wrap(async (req, res) => res.json(await bt.connect(req.body.mac))));
app.post('/api/bt/repair', wrap(async (req, res) => res.json(await bt.repair(req.body.mac))));
app.post('/api/bt/disconnect', wrap(async (req, res) => res.json(await bt.disconnect(req.body.mac))));

// --- Audio ---
app.get('/api/audio/sources', wrap(async (req, res) => res.json(await rec.listSources())));
app.post('/api/audio/headset-mic', wrap(async (req, res) => res.json(await rec.enableHeadsetMic((req.body && req.body.codec) || 'auto'))));

// --- Global NPU pipeline serializer ---------------------------------------
// Transcription (whisper-turbo ENCODER on the NPU) and LLM analysis (rkllm on
// the NPU) cannot run simultaneously: while the whisper encoder holds the NPU,
// rkllm_init fails with "failed to allocate handle, ret:-1, errno:14, Bad
// address", which surfaces as "rkllm READY timeout" and a failed/empty
// analysis. This bit when two recordings finished within seconds of each other
// (both fired transcribe+analyze fire-and-forget with no coordination). Every
// heavy per-recording job (transcribe / analyze / re-diarize) now runs through
// this single FIFO queue, so exactly one touches the NPU at a time; a second
// recording simply waits its turn instead of colliding.
let npuChain = Promise.resolve();
let npuDepth = 0;
function withNpuLock(label, fn) {
  npuDepth++;
  const waiting = npuDepth - 1;
  const run = npuChain.then(async () => {
    if (waiting > 0) console.log(`npu-queue: ${label} start (${waiting} ahead done)`);
    try { return await fn(); }
    finally { npuDepth--; }
  });
  npuChain = run.catch(() => {});   // keep the chain alive across a job's failure
  return run;
}

// --- Recording ---
app.post('/api/record/start', wrap(async (req, res) => res.json(await rec.startRecording(req.body || {}))));
// Transcribe (thermal-guarded whisper + diarize) then analyze, in the background.
// The wav + DB row already exist, so a crash here loses nothing — the recording
// stays with status 'transcribing' and can be recovered via /retranscribe.
async function transcribeAndAnalyze(id, wavFile, numSpeakers) {
  await db.markAnalysisStart(id).catch(() => {});
  try {
    const { transcript, words } = await rec.transcribeWav(wavFile, numSpeakers);
    await db.setTranscript(id, transcript, words);
    console.log(`transcription done for #${id}: ${words.length} segments`);
  } catch (e) {
    console.error(`transcription failed for #${id}:`, e.message);
    await db.setStatus(id, 'error').catch(() => {});
    // also mark analysis 'error' so the UI stops polling and shows a retry
    await db.updateAnalysis(id, { title: null, summary: null, todos: null, status: 'error' }).catch(() => {});
    return; // no transcript -> skip LLM analysis
  }
  analyzeRecording(id);
}

app.post('/api/record/stop', wrap(async (req, res) => {
  // Persist wav + mp3 + DB row FIRST (crash-safe), respond immediately, then
  // transcribe/analyze in the background. The frontend already polls while
  // analysis_status === 'pending', so the transcript + analysis fill in live.
  const r = await rec.stopAndSave();
  const id = await db.addRecording({ ...r, transcript: '', status: 'transcribing' });
  res.json({ id, title: r.title, mp3_file: r.mp3_file, duration_ms: r.duration_ms,
             transcript: '', status: 'transcribing', analysis_status: 'pending' });
  withNpuLock(`transcribe #${id}`, () => transcribeAndAnalyze(id, r.wav_file, r.numSpeakers));
}));

// --- Recordings ---
app.get('/api/recordings', wrap(async (req, res) => res.json(await db.listRecordings())));
app.get('/api/recordings/:id', wrap(async (req, res) => {
  const r = await db.getRecording(+req.params.id);
  r ? res.json(r) : res.status(404).json({ error: 'not found' });
}));
app.delete('/api/recordings/:id', wrap(async (req, res) => { await db.deleteRecording(+req.params.id); res.json({ ok: true }); }));

// --- Voiceprints / enrollment ---
app.get('/api/voiceprints', wrap(async (req, res) => {
  const vps = await db.listVoiceprints();
  res.json(vps.map(v => ({ id: v.id, name: v.name, dim: v.dim, created_at: v.created_at })));
}));
app.delete('/api/voiceprints/:id', wrap(async (req, res) => { await db.deleteVoiceprint(+req.params.id); res.json({ ok: true }); }));
app.post('/api/enroll/start', wrap(async (req, res) => res.json(await rec.startRecording({ source: req.body && req.body.source }))));
app.post('/api/enroll/stop', wrap(async (req, res) => {
  const name = (req.body && req.body.name || '').trim();
  if (!name) return res.status(400).json({ error: 'Name fehlt' });
  const wav = await rec.stopRaw();
  try {
    const emb = await spk.embed(wav);
    const id = await db.addVoiceprint(name, emb);
    res.json({ id, name, dim: emb.length });
  } finally { try { fs.unlinkSync(wav); } catch {} }
}));

// --- Glossary (known proper nouns / terms; fed into LLM correction) ---
app.get('/api/glossary', wrap(async (req, res) => res.json(await db.listGlossary())));
app.post('/api/glossary', wrap(async (req, res) => {
  const term = (req.body && req.body.term || '').trim();
  const soundalikes = (req.body && req.body.soundalikes || '').trim();
  if (!term) return res.status(400).json({ error: 'Begriff fehlt' });
  res.json(await db.addGlossary(term, soundalikes));
}));
app.delete('/api/glossary/:id', wrap(async (req, res) => { await db.deleteGlossary(+req.params.id); res.json({ ok: true }); }));

// --- LLM ---
app.get('/api/llm/status', (req, res) => res.json(llm.status()));
app.get('/api/spk/status', (req, res) => res.json(spk.status()));
// Re-diarize a recording's words with an (optional) fixed speaker count, then
// persist the new speaker labels. Text/timestamps are untouched. Needs the wav.
async function rediarize(id, numSpeakers) {
  const r = await db.getRecording(id);
  if (!r || !r.wav_file) throw new Error('wav für Re-Diarisierung fehlt');
  const wav = path.join(REC_DIR, r.wav_file);
  if (!fs.existsSync(wav)) throw new Error('wav-Datei nicht gefunden: ' + r.wav_file);
  const words = r.words || [];
  // Free the resident rkllm first — diar's consolidation embeds every cluster
  // via TitaNet, which OOM-dies next to a resident 5.9GB LLM (see
  // analyzeRecording; on #71 that silently yielded 48 raw speakers).
  llm.stopServer();
  // Diarization runs on the A76 cores; the board is fanless, so cap the clock
  // for the whole run (transcribeWav does this for the record path — rediarize
  // is the other entry point and must do it too, or a long re-diar overheats).
  await pexec('sudo', ['-n', '/opt/callrec/aime-cpucap.sh', '1608000']).catch(() => {});
  try {
    const segs = await diar.diarize(wav, numSpeakers);
    diar.assignSpeakers(words, segs);
    await db.replaceWords(id, words);
    console.log(`re-diarized #${id} (numSpeakers=${numSpeakers || 'auto'}) → ${new Set(segs.map(s => s.speaker)).size} speakers`);
  } finally {
    await pexec('sudo', ['-n', '/opt/callrec/aime-cpucap.sh', '2256000']).catch(() => {});
  }
}

// Re-run analysis (e.g. the model wasn't downloaded yet at record time). Optional
// body { numSpeakers } re-diarizes first (fixes over-segmentation on a known count).
app.post('/api/recordings/:id/reanalyze', wrap(async (req, res) => {
  const id = +req.params.id;
  const r = await db.getRecording(id);
  if (!r) return res.status(404).json({ error: 'not found' });
  const numSpeakers = +(req.body && req.body.numSpeakers) || 0;
  await db.updateAnalysis(id, { title: null, summary: null, todos: null, status: 'pending' });
  await db.markAnalysisStart(id).catch(() => {});
  res.json({ ok: true, status: 'pending', reDiarize: true });
  // Always re-diarize: numSpeakers>0 pins the count, 0 = auto (now consolidated,
  // so auto no longer over-segments). This is the "fix my speakers" path.
  withNpuLock(`reanalyze #${id}`, async () => {
    try { await rediarize(id, numSpeakers); }
    catch (e) { console.error(`re-diarize failed for #${id}:`, e.message); }
    await analyzeRecording(id);
  });
}));

// LLM-only re-analysis: re-run title/summary/todos on the EXISTING words +
// speaker labels, WITHOUT re-diarizing (fast — skips the ~40min sherpa pass when
// the speakers are already correct and only the LLM/prompts changed).
app.post('/api/recordings/:id/reanalyze-llm', wrap(async (req, res) => {
  const id = +req.params.id;
  const r = await db.getRecording(id);
  if (!r) return res.status(404).json({ error: 'not found' });
  await db.updateAnalysis(id, { title: null, summary: null, todos: null, status: 'pending' });
  await db.markAnalysisStart(id).catch(() => {});
  res.json({ ok: true, status: 'pending', reDiarize: false });
  withNpuLock(`reanalyze-llm #${id}`, async () => { await analyzeRecording(id); });
}));

// Re-run transcription from the saved wav (recovery: a crash/thermal-shutdown
// during whisper leaves the row at status 'transcribing' with an empty
// transcript, or 'error'; the wav is on disk). Optional { numSpeakers }.
app.post('/api/recordings/:id/retranscribe', wrap(async (req, res) => {
  const id = +req.params.id;
  const r = await db.getRecording(id);
  if (!r) return res.status(404).json({ error: 'not found' });
  if (!r.wav_file) return res.status(400).json({ error: 'kein wav für Re-Transkription' });
  const numSpeakers = +(req.body && req.body.numSpeakers) || 0;
  await db.setStatus(id, 'transcribing');
  await db.updateAnalysis(id, { title: null, summary: null, todos: null, status: 'pending' });
  res.json({ ok: true, status: 'transcribing' });
  withNpuLock(`retranscribe #${id}`, () => transcribeAndAnalyze(id, r.wav_file, numSpeakers));
}));

// Longest contiguous run of one diarization cluster (best clean sample of a voice).
function longestRun(words, speaker) {
  let best = null, cur = null;
  for (const w of words || []) {
    if (w.speaker !== speaker) { cur = null; continue; }
    if (!cur) cur = { start: w.t_start_ms, end: w.t_end_ms };
    else cur.end = w.t_end_ms;
    if (!best || (cur.end - cur.start) > (best.end - best.start)) best = { start: cur.start, end: cur.end };
  }
  return best;
}

// Name a diarized speaker in a recording and (default) LEARN their voice: fold
// the speaker's longest clean segment into the named voiceprint centroid, so
// future recordings auto-identify them. Empty name clears the label.
app.post('/api/recordings/:id/speaker', wrap(async (req, res) => {
  const id = +req.params.id;
  const speaker = String((req.body && req.body.speaker) || '').trim();
  const name = String((req.body && req.body.name) || '').trim();
  const enroll = !(req.body && req.body.enroll === false);
  if (!speaker) return res.status(400).json({ error: 'speaker fehlt' });
  const r = await db.getRecording(id);
  if (!r) return res.status(404).json({ error: 'not found' });
  await db.setSpeakerName(id, speaker, name);

  let enrolled = false, learnMsg = null;
  if (enroll && name && r.wav_file) {
    const seg = longestRun(r.words, speaker);
    const wavPath = path.join(REC_DIR, r.wav_file);
    if (!seg || (seg.end - seg.start) < 1500) {
      learnMsg = 'Segment zu kurz zum Lernen (<1,5s)';
    } else if (!fs.existsSync(wavPath)) {
      learnMsg = 'wav fehlt – nicht gelernt';
    } else {
      const cut = `/tmp/enr_${id}_${speaker}.wav`;
      try {
        await pexec('ffmpeg', ['-y', '-i', wavPath, '-ss', String(seg.start / 1000),
          '-to', String(seg.end / 1000), '-ar', '16000', '-ac', '1', cut]);
        const emb = await spk.embed(cut);
        await db.upsertVoiceprint(name, emb);
        enrolled = true;
      } catch (e) { console.error('enroll-from-recording failed:', e.message); learnMsg = 'Lernen fehlgeschlagen'; }
      finally { try { fs.unlinkSync(cut); } catch {} }
    }
  }
  const r2 = await db.getRecording(id);
  res.json({ ok: true, speaker_names: r2.speaker_names || {}, enrolled, learnMsg });
}));

app.get(/^(?!\/api|\/media).*/, (req, res) => res.sendFile(path.join(__dirname, 'web/dist/index.html')));

const PORT = process.env.PORT || 3000;
db.ensureSchema()
  .catch(e => console.error('ensureSchema:', e.message))
  .finally(() => app.listen(PORT, '0.0.0.0', () => console.log(`callrec listening on :${PORT}`)));