Keysat
1038 lines
44 KiB
JavaScript
1038 lines
44 KiB
JavaScript
// Operator-hardware fallback backend. Forwards transcribe requests to
|
|
// a Parakeet endpoint (or any Whisper-API-compatible server — same wire
|
|
// format) and analyze requests to a Gemma endpoint (or any
|
|
// OpenAI-compatible chat-completions server).
|
|
//
|
|
// Used when a Pro/Max user has exceeded their monthly Gemini cap.
|
|
// Returns the same shape gemini.js produces so route handlers don't
|
|
// need a backend-specific branch downstream:
|
|
// transcribeAudio → { text, segments, duration_seconds }
|
|
// analyzeText → { text }
|
|
//
|
|
// Long-audio handling: Parakeet wrappers on operator GPUs OOM on long
|
|
// audio (one shot of a 108-min file returned "CUDA error: unknown
|
|
// error" on a 24GB card). transcribeAudio chunks audio > 5 minutes
|
|
// into 5-minute slices, transcribes each sequentially, and stitches
|
|
// the segments with continuous timestamps. Single-pass for short
|
|
// audio, no perf hit.
|
|
|
|
import fs from "fs/promises";
|
|
import os from "os";
|
|
import path from "path";
|
|
import {
|
|
getAudioDurationSeconds,
|
|
splitAudioFile,
|
|
} from "../audio-meta.js";
|
|
import {
|
|
sortAndDedupeTranscript,
|
|
mergeShortEntries,
|
|
} from "./gemini.js";
|
|
import { lanFetch } from "../lan-fetch.js";
|
|
import {
|
|
clusterSpeakers,
|
|
assignSpeakersToSegments,
|
|
} from "../speaker-clustering.js";
|
|
|
|
const ANALYZE_MAX_TOKENS = 16000;
|
|
// Gemma served locally tends to live on the host's LAN, not the public
|
|
// internet, so generous timeouts. Same scale as Recap's defaults.
|
|
const DEFAULT_TIMEOUT_MS = 900_000;
|
|
|
|
// Default Parakeet model name — used only when the route handler
|
|
// doesn't supply one (e.g. a unit test instantiating the backend
|
|
// directly). In production the value comes from Spark Control
|
|
// discovery via resolveHardwareConfig().
|
|
//
|
|
// NOTE: there is NO default for the analyze (Gemma) model. An empty
|
|
// gemmaModel intentionally triggers autodiscovery via /v1/models in
|
|
// analyzeText() below. Substituting a hardcoded default here (the
|
|
// old "gemma3:27b") defeats autodiscovery and causes "model does
|
|
// not exist" errors when the operator's hardware hosts a different
|
|
// model name.
|
|
const DEFAULT_TRANSCRIBE_MODEL = "parakeet-tdt-0.6b-v3";
|
|
|
|
// Normalize an OpenAI-API-compatible base URL: strip trailing slash
|
|
// AND strip a trailing `/v1` segment if the operator pasted one,
|
|
// because we always append `/v1/...` below. Without this, a base URL
|
|
// of `http://192.168.1.87:8000/v1` would produce
|
|
// `http://192.168.1.87:8000/v1/v1/audio/transcriptions` → 404.
|
|
function normalizeApiBase(url) {
|
|
let s = (url || "").trim().replace(/\/$/, "");
|
|
s = s.replace(/\/v1$/, "");
|
|
return s;
|
|
}
|
|
|
|
// 503 retry policy — both Parakeet and Sortformer/TitaNet endpoints
|
|
// on Spark Control return 503 + a "retry in ~60s" body when the GPU
|
|
// container CUDA-wedges and auto-restarts. The right thing to do is
|
|
// wait for the documented retry window (or the Retry-After header if
|
|
// supplied) and try again. Honor Retry-After when present, default
|
|
// to 60s, clamp to [5, 120], and add ±5s jitter so parallel chunks
|
|
// don't all wake up and pile on at the exact same moment (thundering
|
|
// herd → instant re-wedge).
|
|
const SPARK_503_MAX_ATTEMPTS = 4;
|
|
function parse503WaitSec(res) {
|
|
const ra = res?.headers?.get?.("retry-after");
|
|
const raNum = parseInt(ra || "60", 10);
|
|
const base = Math.min(120, Math.max(5, Number.isFinite(raNum) ? raNum : 60));
|
|
const jitter = Math.round((Math.random() * 10 - 5) * 10) / 10; // ±5s
|
|
return Math.max(5, Math.round((base + jitter) * 10) / 10);
|
|
}
|
|
function sleepMs(ms) {
|
|
return new Promise((r) => setTimeout(r, ms));
|
|
}
|
|
|
|
export function createHardwareBackend({
|
|
parakeetBaseURL = "",
|
|
gemmaBaseURL = "",
|
|
// Spark Control base URL (no path) — derived by the caller from
|
|
// relay_spark_control_url with the /api/endpoints suffix stripped.
|
|
// Used only for diarize-chunk (POST {spark}/api/audio/diarize-chunk).
|
|
// Transcribe and analyze go to parakeet/gemma respectively, which
|
|
// discovery reports as separate hosts on the operator's LAN.
|
|
// Empty = diarization can't run (the route handler should not have
|
|
// enabled it).
|
|
sparkControlBaseURL = "",
|
|
parakeetModel = DEFAULT_TRANSCRIBE_MODEL,
|
|
gemmaModel = "",
|
|
timeoutMs = DEFAULT_TIMEOUT_MS,
|
|
// Chunking knobs — caller MUST source from relay-config.json. The
|
|
// defaults here are for unit-test ergonomics only. Production
|
|
// callers always pass these from getConfigSnapshot() so the
|
|
// operator's Settings-tab edits flow through.
|
|
//
|
|
// Parallel chunks: per the spark-control LLM developer, Parakeet can
|
|
// handle ~4 concurrent transcribe POSTs on a typical single-GPU
|
|
// operator setup. Earlier versions hard-serialized to 1 in flight
|
|
// out of caution; we now bound by the operator's chosen concurrency.
|
|
txChunkSeconds = 5 * 60,
|
|
// Overlap (seconds) between consecutive chunks. Defaults to 0
|
|
// (matches pre-v0.2.77 behavior); production callers pass the
|
|
// operator's configured `relay_hardware_tx_chunk_overlap_seconds`.
|
|
// When > 0, chunks at the boundary share audio and the stitcher
|
|
// drops words/segments in chunk N+1 whose global timestamps fall
|
|
// within the prior chunk's tail. See splitAudioFile() for the
|
|
// chunking math.
|
|
txChunkOverlapSeconds = 0,
|
|
txConcurrency = 4,
|
|
// Diarization: when true, every chunk gets TWO parallel HTTP calls
|
|
// — the existing /v1/audio/transcriptions to Parakeet AND a sibling
|
|
// /api/audio/diarize-chunk to Sortformer+TitaNet on the same Spark
|
|
// Control host. Diarization output (segments + per-chunk voice
|
|
// fingerprints) is collected per chunk and returned alongside the
|
|
// transcript on the backend response so the relay's caller can run
|
|
// cross-chunk speaker clustering. In Phase 1C this just gathers the
|
|
// data; the clustering + transcript-merge step lands in Phase 1D.
|
|
// Default false (matches the current transcribe-only behavior).
|
|
diarizationEnabled = false,
|
|
// Cosine-similarity threshold for cross-chunk speaker clustering,
|
|
// as an integer percentage (slider system is int-only). 70 means
|
|
// "merge two chunks' speaker fingerprints into the same global
|
|
// speaker if their cosine similarity >= 0.70" — NeMo's recommended
|
|
// default for TitaNet embeddings. Range 50-95 enforced inside the
|
|
// clustering module; values outside the range are clamped.
|
|
clusterThresholdPct = 70,
|
|
// Post-cluster suppression knobs (Phase 2). Operator-tunable via
|
|
// Settings → Operator hardware. Used only when diarization is on.
|
|
// Defaults match speaker-clustering.js's hardcoded fallbacks.
|
|
anchorMinSpeakingSec = 30,
|
|
smallClusterMaxSpeakingSec = 15,
|
|
uncertainMarginPct = 10,
|
|
// Optional per-chunk completion callback for the pipelined-analyze
|
|
// path. Fires AFTER both transcribe + diarize have resolved for
|
|
// each chunk (Promise.allSettled), in the order chunks finish (not
|
|
// necessarily monotonic by chunkIndex when concurrency > 1).
|
|
// Receives:
|
|
// {
|
|
// chunkIndex, // 0-based
|
|
// totalChunks, // emitted on the first call so the caller
|
|
// // knows the count up front
|
|
// startSeconds, // global audio offset where this chunk begins
|
|
// durationSeconds, // chunk audio length
|
|
// overlapBoundarySec, // RELATIVE to chunk start; segments with
|
|
// // global start < (startSeconds + overlap-
|
|
// // BoundarySec) belong to the prior chunk
|
|
// segments, // raw chunk segments, globally-timestamped
|
|
// // (start/end already shifted by startSeconds)
|
|
// diarOk, // true | false | null (null = not run)
|
|
// }
|
|
// Caller uses this to feed a chunk buffer that the pipelined-analyze
|
|
// workers read from. When this callback is null (the default),
|
|
// transcribe behaves as today — caller gets the final stitched
|
|
// result only at the end.
|
|
onChunkComplete = null,
|
|
// Output cap passed to the operator's vLLM/Ollama chat-completion
|
|
// endpoint as `max_tokens`. Default 16000 was the historical
|
|
// hardcoded value. Operators with smaller models that produce
|
|
// shorter JSON sections may want to drop this; operators on
|
|
// larger models with reasoning preambles may want to bump it.
|
|
// Configurable via Settings → `relay_hardware_an_max_tokens`.
|
|
anMaxTokens = ANALYZE_MAX_TOKENS,
|
|
} = {}) {
|
|
const parakeet = normalizeApiBase(parakeetBaseURL);
|
|
const gemma = normalizeApiBase(gemmaBaseURL);
|
|
// Spark Control URL — used ONLY by postDiarizeChunk. Strip any
|
|
// trailing /api/endpoints (the discovery path) and any trailing
|
|
// slash so the diarize URL composes cleanly as `${sparkBase}/api/
|
|
// audio/diarize-chunk`.
|
|
const sparkBase = (sparkControlBaseURL || "")
|
|
.trim()
|
|
.replace(/\/$/, "")
|
|
.replace(/\/api\/endpoints$/, "");
|
|
const transcribeModel = parakeetModel || DEFAULT_TRANSCRIBE_MODEL;
|
|
// Intentionally NOT coalesced to a default — empty string means
|
|
// "autodiscover from /v1/models at request time". See the comment
|
|
// at DEFAULT_TRANSCRIBE_MODEL above for why.
|
|
const analyzeModel = gemmaModel || "";
|
|
|
|
// Single Parakeet POST. The chunking layer in transcribeAudio calls
|
|
// this once per chunk; for short audio it's called once with the
|
|
// whole buffer. Returns { segments, duration_seconds } that the
|
|
// caller stitches.
|
|
//
|
|
// Tries the OpenAI standard `/v1/audio/transcriptions` first, falls
|
|
// back to bare `/audio/transcriptions` on 404. Retries without
|
|
// verbose_json + timestamp_granularities[] if the wrapper rejects
|
|
// the rich shape.
|
|
async function postOneChunk({ audioBuffer, mimeType, offsetSeconds }) {
|
|
const buildForm = (richMode) => {
|
|
const form = new FormData();
|
|
const blob = new Blob([audioBuffer], { type: mimeType });
|
|
form.append("file", blob, "audio.bin");
|
|
form.append("model", transcribeModel);
|
|
if (richMode) {
|
|
form.append("response_format", "verbose_json");
|
|
form.append("timestamp_granularities[]", "segment");
|
|
}
|
|
return form;
|
|
};
|
|
const pathCandidates = [
|
|
"/v1/audio/transcriptions",
|
|
"/audio/transcriptions",
|
|
];
|
|
|
|
// Resolve which path the wrapper exposes (404 from one → try the
|
|
// other). Done before the 503 loop so we don't waste retry
|
|
// budget on a path that doesn't exist.
|
|
let resolvedUrl = null;
|
|
let res = null;
|
|
for (const p of pathCandidates) {
|
|
const url = `${parakeet}${p}`;
|
|
try {
|
|
res = await lanFetch(url, {
|
|
method: "POST",
|
|
body: buildForm(true),
|
|
redirect: "follow",
|
|
signal: AbortSignal.timeout(timeoutMs),
|
|
});
|
|
} catch (err) {
|
|
const cause =
|
|
err?.cause?.message || err?.cause?.code || err?.cause || "";
|
|
const detail = cause ? `${err.message} (cause: ${cause})` : err?.message || String(err);
|
|
const e = new Error(`Parakeet transcribe network error at ${url}: ${detail}`);
|
|
e.status = 502;
|
|
throw e;
|
|
}
|
|
if (res.status !== 404) {
|
|
resolvedUrl = url;
|
|
break;
|
|
}
|
|
console.warn(
|
|
`[hardware] 404 at ${url} — trying next path candidate`
|
|
);
|
|
}
|
|
if (!resolvedUrl) {
|
|
// All path candidates 404'd — return that 404 below.
|
|
resolvedUrl = `${parakeet}${pathCandidates[pathCandidates.length - 1]}`;
|
|
}
|
|
|
|
// ── 503 retry loop ──
|
|
// Spark Control returns 503 + "retry in ~60s" body when the
|
|
// Parakeet container is CUDA-wedged and auto-restarting. We wait
|
|
// (honoring Retry-After) and retry up to SPARK_503_MAX_ATTEMPTS
|
|
// total. Without this loop, the very first chunks in a multi-
|
|
// chunk job all hit a single wedge simultaneously and each chunk
|
|
// permanently fails after one round-trip.
|
|
let attempt = 1;
|
|
while (res.status === 503 && attempt < SPARK_503_MAX_ATTEMPTS) {
|
|
const waitSec = parse503WaitSec(res);
|
|
console.warn(
|
|
`[hardware/tx] 503 at ${resolvedUrl} (attempt ${attempt}/${SPARK_503_MAX_ATTEMPTS}) — Spark Control reports transient CUDA wedge, waiting ${waitSec}s then retrying`
|
|
);
|
|
await sleepMs(waitSec * 1000);
|
|
try {
|
|
res = await lanFetch(resolvedUrl, {
|
|
method: "POST",
|
|
body: buildForm(true),
|
|
redirect: "follow",
|
|
signal: AbortSignal.timeout(timeoutMs),
|
|
});
|
|
} catch (err) {
|
|
const cause =
|
|
err?.cause?.message || err?.cause?.code || err?.cause || "";
|
|
const detail = cause ? `${err.message} (cause: ${cause})` : err?.message || String(err);
|
|
const e = new Error(`Parakeet transcribe network error at ${resolvedUrl} (retry ${attempt}): ${detail}`);
|
|
e.status = 502;
|
|
throw e;
|
|
}
|
|
attempt += 1;
|
|
}
|
|
|
|
// ── Rich-mode fallback ──
|
|
// If the wrapper rejected verbose_json + timestamp_granularities
|
|
// (some non-OpenAI Parakeet servers do — this is a PARAMETER
|
|
// issue, not a server-availability issue), retry once without
|
|
// those fields. Distinct from the 503 retry above — different
|
|
// failure mode, different fix. We skip this on 503 because the
|
|
// 503 loop has already exhausted retries.
|
|
if (
|
|
!res.ok &&
|
|
res.status !== 503 &&
|
|
res.status >= 400 &&
|
|
res.status < 600 &&
|
|
res.status !== 404
|
|
) {
|
|
const richBody = await safeBody(res);
|
|
console.warn(
|
|
`[hardware] rich Parakeet request to ${resolvedUrl} returned ${res.status}: ${richBody.slice(0, 200)} — retrying without verbose_json`
|
|
);
|
|
try {
|
|
res = await lanFetch(resolvedUrl, {
|
|
method: "POST",
|
|
body: buildForm(false),
|
|
redirect: "follow",
|
|
signal: AbortSignal.timeout(timeoutMs),
|
|
});
|
|
} catch (err) {
|
|
const e = new Error(
|
|
`Parakeet transcribe network error (bare fallback) at ${resolvedUrl}: ${err?.message || err}`
|
|
);
|
|
e.status = 502;
|
|
throw e;
|
|
}
|
|
}
|
|
if (!res.ok) {
|
|
const body = await safeBody(res);
|
|
const hint =
|
|
res.status === 404
|
|
? ` (tried ${pathCandidates.join(" and ")} on base ${parakeet} — wrapper may expose the endpoint at a different path; check the Parakeet URL or container logs)`
|
|
: res.status === 503
|
|
? ` (Spark Control reported transient CUDA wedge ${SPARK_503_MAX_ATTEMPTS} times in a row — operator container may need manual restart)`
|
|
: "";
|
|
const e = new Error(
|
|
`Parakeet transcribe ${res.status} at ${resolvedUrl}: ${body.slice(0, 300)}${hint}`
|
|
);
|
|
e.status = res.status;
|
|
throw e;
|
|
}
|
|
const data = await res.json();
|
|
const segments = Array.isArray(data.segments) ? data.segments : [];
|
|
const shifted = segments.map((s) => ({
|
|
start: (s.start || 0) + offsetSeconds,
|
|
end: (s.end || 0) + offsetSeconds,
|
|
text: (s.text || "").trim(),
|
|
}));
|
|
return {
|
|
segments: shifted,
|
|
duration_seconds: data.duration || 0,
|
|
};
|
|
}
|
|
|
|
// Parallel diarization call for the same audio chunk. Hits
|
|
// POST {spark_control}/api/audio/diarize-chunk (Sortformer for
|
|
// segmentation, TitaNet for per-speaker voice fingerprints). The
|
|
// returned `speaker` labels are LOCAL to this chunk — Speaker_0
|
|
// here is not necessarily the same person as Speaker_0 in chunk
|
|
// N+1; the relay's cross-chunk clustering step (Phase 1D)
|
|
// reconciles them via cosine similarity on the fingerprints.
|
|
//
|
|
// Retry policy from the SC dev:
|
|
// 503 + Retry-After: Parakeet/Sortformer CUDA wedge — container
|
|
// is auto-restarting in the background. Wait + retry once.
|
|
// 502: container unreachable. Fail.
|
|
// 500: real error. Fail.
|
|
//
|
|
// Segment timestamps are shifted by offsetSeconds so they
|
|
// already sit on the global audio timeline when returned —
|
|
// matches postOneChunk's behavior so the merge step in Phase 1D
|
|
// can align word timestamps and segment timestamps directly
|
|
// without re-shifting.
|
|
async function postDiarizeChunk({ audioBuffer, mimeType, offsetSeconds }) {
|
|
if (!sparkBase) {
|
|
const e = new Error(
|
|
"Diarization called but sparkControlBaseURL is empty — caller should not have enabled diarization without Service Discovery configured"
|
|
);
|
|
e.status = 500;
|
|
throw e;
|
|
}
|
|
const buildForm = () => {
|
|
const form = new FormData();
|
|
const blob = new Blob([audioBuffer], { type: mimeType });
|
|
form.append("file", blob, "audio.bin");
|
|
return form;
|
|
};
|
|
// Diarize-chunk lives on Spark Control (the same host that serves
|
|
// /api/endpoints) — NOT on the Parakeet wrapper. Discovery's
|
|
// parakeet.base_url usually points to a separate STT-only service
|
|
// that doesn't have this endpoint. Use lanFetch so Spark Control's
|
|
// StartOS Local Intermediate CA cert doesn't fail TLS validation.
|
|
const url = `${sparkBase}/api/audio/diarize-chunk`;
|
|
let res = null;
|
|
// 503 retry loop — same policy as the transcribe path. Up to
|
|
// SPARK_503_MAX_ATTEMPTS, honoring Retry-After with ±5s jitter
|
|
// so parallel chunks don't synchronize their retries and pile
|
|
// back onto the freshly-restarted container at the same instant.
|
|
for (let attempt = 0; attempt < SPARK_503_MAX_ATTEMPTS; attempt++) {
|
|
try {
|
|
res = await lanFetch(url, {
|
|
method: "POST",
|
|
body: buildForm(),
|
|
redirect: "follow",
|
|
signal: AbortSignal.timeout(timeoutMs),
|
|
});
|
|
} catch (err) {
|
|
// undici wraps the underlying socket error in err.cause —
|
|
// surfacing it makes "fetch failed" diagnosable (ECONNREFUSED
|
|
// vs ETIMEDOUT vs TLS handshake vs DNS).
|
|
const cause =
|
|
err?.cause?.message || err?.cause?.code || err?.cause || "";
|
|
const detail = cause ? `${err.message} (cause: ${cause})` : err?.message || String(err);
|
|
const e = new Error(`Diarize network error at ${url}: ${detail}`);
|
|
e.status = 502;
|
|
throw e;
|
|
}
|
|
if (res.status === 503 && attempt < SPARK_503_MAX_ATTEMPTS - 1) {
|
|
const waitSec = parse503WaitSec(res);
|
|
console.warn(
|
|
`[hardware/diarize] 503 at ${url} (attempt ${attempt + 1}/${SPARK_503_MAX_ATTEMPTS}) — Spark Control reports transient CUDA wedge, waiting ${waitSec}s then retrying`
|
|
);
|
|
await sleepMs(waitSec * 1000);
|
|
continue;
|
|
}
|
|
break;
|
|
}
|
|
if (!res.ok) {
|
|
const body = await safeBody(res);
|
|
const hint =
|
|
res.status === 503
|
|
? ` (Spark Control reported transient CUDA wedge ${SPARK_503_MAX_ATTEMPTS} times in a row — operator container may need manual restart)`
|
|
: "";
|
|
const e = new Error(
|
|
`Diarize ${res.status} at ${url}: ${body.slice(0, 300)}${hint}`
|
|
);
|
|
e.status = res.status;
|
|
throw e;
|
|
}
|
|
const data = await res.json();
|
|
const rawSegments = Array.isArray(data.segments) ? data.segments : [];
|
|
const shiftedSegments = rawSegments.map((s) => ({
|
|
// Use the same `start` / `end` naming as postOneChunk to keep
|
|
// the merge code uniform across segment shapes.
|
|
start: (s.start_s || 0) + offsetSeconds,
|
|
end: (s.end_s || 0) + offsetSeconds,
|
|
speaker_local: s.speaker || "Speaker_unknown",
|
|
confidence:
|
|
typeof s.confidence === "number" && Number.isFinite(s.confidence)
|
|
? s.confidence
|
|
: null,
|
|
}));
|
|
return {
|
|
duration_s: typeof data.duration === "number" ? data.duration : 0,
|
|
segments: shiftedSegments,
|
|
speakers_local: Array.isArray(data.speakers_detected)
|
|
? data.speakers_detected
|
|
: Array.isArray(data.speakers_local)
|
|
? data.speakers_local
|
|
: [],
|
|
fingerprints:
|
|
data.fingerprints && typeof data.fingerprints === "object"
|
|
? data.fingerprints
|
|
: {},
|
|
models: data.models || null,
|
|
};
|
|
}
|
|
|
|
return {
|
|
hasTranscribe: !!parakeet,
|
|
hasAnalyze: !!gemma,
|
|
|
|
// Public API. Receives a full audio buffer; decides whether to
|
|
// send it to Parakeet in one shot or to chunk-and-stitch.
|
|
async transcribeAudio({
|
|
audio,
|
|
mimeType = "application/octet-stream",
|
|
offsetSeconds = 0,
|
|
}) {
|
|
if (!parakeet) {
|
|
const err = new Error(
|
|
"operator-hardware transcribe is not configured — Spark Control discovery isn't reporting a ready parakeet endpoint"
|
|
);
|
|
err.status = 503;
|
|
throw err;
|
|
}
|
|
|
|
// Write to temp file so ffprobe + ffmpeg can read it (no
|
|
// reliable stdin pathway for either tool on every codec).
|
|
// Cleanup happens unconditionally in `finally`.
|
|
const tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), "hw-tx-"));
|
|
const inputPath = path.join(tmpDir, "input." + extFromMime(mimeType));
|
|
await fs.writeFile(inputPath, audio);
|
|
|
|
try {
|
|
const duration = await getAudioDurationSeconds(inputPath);
|
|
// Build the chunk list. For audio that fits in a single chunk
|
|
// (or where ffprobe couldn't read the duration), we synthesize
|
|
// a 1-entry descriptor pointing at the ORIGINAL file — no
|
|
// ffmpeg cut needed. This routes the short-audio case through
|
|
// the SAME worker loop as multi-chunk audio so:
|
|
// - diarization still fires (when enabled)
|
|
// - onChunkComplete still fires (otherwise pipelined-analyze
|
|
// workers wait forever on a callback that never comes — the
|
|
// bug that hung Internal Meetings + short YouTube videos)
|
|
// - cross-chunk clustering still runs (N=1 cluster, but
|
|
// speakers are still labeled on the segments)
|
|
// - bracketed-text stitching runs (the single-shot path used
|
|
// to return text="" because postOneChunk doesn't emit text)
|
|
// For longer audio, splitAudioFile cuts on disk as before.
|
|
let chunks;
|
|
if (!duration || duration <= txChunkSeconds) {
|
|
chunks = [{
|
|
filePath: inputPath,
|
|
startSeconds: 0,
|
|
durationSeconds: duration || 0,
|
|
overlapBoundarySec: 0,
|
|
index: 0,
|
|
}];
|
|
} else {
|
|
chunks = await splitAudioFile({
|
|
inputPath,
|
|
outputDir: tmpDir,
|
|
chunkSeconds: txChunkSeconds,
|
|
overlapSeconds: txChunkOverlapSeconds,
|
|
});
|
|
}
|
|
const effectiveConcurrency = Math.min(chunks.length, Math.max(1, txConcurrency));
|
|
const durLabel = duration ? `${duration.toFixed(0)}s` : "unknown-length";
|
|
if (chunks.length === 1) {
|
|
console.log(
|
|
`[hardware] ${durLabel} audio fits in a single chunk — sending in one shot (no ffmpeg split)`
|
|
);
|
|
} else {
|
|
console.log(
|
|
`[hardware] chunking ${durLabel} audio into ${chunks.length} pieces of up to ${txChunkSeconds}s (overlap=${txChunkOverlapSeconds}s), ${effectiveConcurrency} in parallel`
|
|
);
|
|
}
|
|
if (diarizationEnabled) {
|
|
console.log(
|
|
`[hardware] diarization ENABLED — each chunk fires 2 parallel calls (transcribe + diarize) against Spark Control`
|
|
);
|
|
}
|
|
const chunkResults = new Array(chunks.length);
|
|
// Per-chunk wall-time tracking. Summed by callers into the
|
|
// audit row's transcribe_ms_sum so the operator can see total
|
|
// backend compute across all Parakeet POSTs (distinct from
|
|
// the outer wall-time which captures the parallel-fan-out).
|
|
const chunkDurationsMs = new Array(chunks.length).fill(null);
|
|
// Per-chunk diarization data. Populated when
|
|
// diarizationEnabled is true. Each entry mirrors the chunk's
|
|
// index in `chunks` so the clustering step (Phase 1D) can
|
|
// correlate per-chunk speaker labels back to chunk positions.
|
|
// null = diarization didn't run OR failed for this chunk.
|
|
const chunkDiarization = new Array(chunks.length).fill(null);
|
|
let nextIdx = 0;
|
|
const worker = async () => {
|
|
while (true) {
|
|
const i = nextIdx++;
|
|
if (i >= chunks.length) return;
|
|
const chunk = chunks[i];
|
|
const chunkStart = Date.now();
|
|
try {
|
|
console.log(
|
|
`[hardware] chunk ${i + 1}/${chunks.length} (start=${chunk.startSeconds}s, len=${chunk.durationSeconds.toFixed(0)}s)`
|
|
);
|
|
const buf = await fs.readFile(chunk.filePath);
|
|
const chunkOffset = offsetSeconds + chunk.startSeconds;
|
|
// Diarization-enabled = parallel transcribe + diarize.
|
|
// Diarization-disabled = transcribe only.
|
|
// We use Promise.allSettled so a diarize failure doesn't
|
|
// take down the transcript for the chunk — the transcript
|
|
// is the critical output, diarization is additive metadata.
|
|
const calls = [
|
|
postOneChunk({
|
|
audioBuffer: buf,
|
|
mimeType,
|
|
offsetSeconds: chunkOffset,
|
|
}),
|
|
];
|
|
if (diarizationEnabled) {
|
|
calls.push(
|
|
postDiarizeChunk({
|
|
audioBuffer: buf,
|
|
mimeType,
|
|
offsetSeconds: chunkOffset,
|
|
})
|
|
);
|
|
}
|
|
const settled = await Promise.allSettled(calls);
|
|
const txSettled = settled[0];
|
|
if (txSettled.status === "rejected") {
|
|
throw txSettled.reason;
|
|
}
|
|
chunkDurationsMs[i] = Date.now() - chunkStart;
|
|
chunkResults[i] = {
|
|
ok: true,
|
|
...txSettled.value,
|
|
startSeconds: chunk.startSeconds,
|
|
};
|
|
let diarOk = null; // null = not run, true/false = ran
|
|
if (diarizationEnabled) {
|
|
const diarSettled = settled[1];
|
|
if (diarSettled.status === "fulfilled") {
|
|
diarOk = true;
|
|
chunkDiarization[i] = {
|
|
ok: true,
|
|
chunkIndex: i,
|
|
startSeconds: chunk.startSeconds,
|
|
// Global timestamp BEFORE which segments in this
|
|
// chunk are duplicates of the prior chunk's tail
|
|
// (overlap zone). Used by clusterSpeakers to
|
|
// dedup diar segments — without this the same
|
|
// 30s overlap zone gets counted twice toward
|
|
// speaker totals on every chunk boundary.
|
|
chunkOverlapBoundarySec:
|
|
offsetSeconds + (chunk.overlapBoundarySec || 0),
|
|
...diarSettled.value,
|
|
};
|
|
} else {
|
|
diarOk = false;
|
|
console.warn(
|
|
`[hardware] diarize chunk ${i + 1}/${chunks.length} failed (transcript intact): ${diarSettled.reason?.message || diarSettled.reason}`
|
|
);
|
|
chunkDiarization[i] = {
|
|
ok: false,
|
|
chunkIndex: i,
|
|
startSeconds: chunk.startSeconds,
|
|
error: diarSettled.reason,
|
|
};
|
|
}
|
|
}
|
|
// Per-chunk completion log so the operator can see
|
|
// parallelism in the timeline. Earlier we logged only
|
|
// when each chunk STARTED, which made concurrent
|
|
// execution invisible (start lines look sequential
|
|
// because each pair of chunks finishes ~simultaneously,
|
|
// and only then does the next pair's first start log
|
|
// appear). Logging completion makes the in-flight
|
|
// overlap obvious.
|
|
const dur = ((chunkDurationsMs[i] || 0) / 1000).toFixed(1);
|
|
const diarTag =
|
|
diarOk === true ? " · diarize ok" :
|
|
diarOk === false ? " · diarize FAIL" : "";
|
|
console.log(
|
|
`[hardware] chunk ${i + 1}/${chunks.length} done in ${dur}s${diarTag}`
|
|
);
|
|
// Fire pipelined-analyze hook (no-op when not in
|
|
// pipelined mode). Hands the caller this chunk's
|
|
// segments + position so it can advance its time-
|
|
// covered cursor and unblock any window analyses
|
|
// that were waiting on this range.
|
|
if (onChunkComplete) {
|
|
try {
|
|
onChunkComplete({
|
|
chunkIndex: i,
|
|
totalChunks: chunks.length,
|
|
startSeconds: chunk.startSeconds,
|
|
durationSeconds: chunk.durationSeconds,
|
|
overlapBoundarySec: chunk.overlapBoundarySec || 0,
|
|
segments: chunkResults[i].segments || [],
|
|
diarOk,
|
|
});
|
|
} catch (cbErr) {
|
|
// Don't kill transcribe if the caller's callback
|
|
// throws — pipelined mode is best-effort, the
|
|
// final analyze fallback still works.
|
|
console.warn(
|
|
`[hardware] onChunkComplete callback failed for chunk ${i + 1}: ${cbErr?.message || cbErr}`
|
|
);
|
|
}
|
|
}
|
|
} catch (err) {
|
|
chunkDurationsMs[i] = Date.now() - chunkStart;
|
|
console.warn(
|
|
`[hardware] chunk ${i + 1}/${chunks.length} failed: ${err?.message || err}`
|
|
);
|
|
chunkResults[i] = { ok: false, error: err, startSeconds: chunk.startSeconds };
|
|
}
|
|
}
|
|
};
|
|
const workers = Array.from({ length: effectiveConcurrency }, worker);
|
|
await Promise.all(workers);
|
|
|
|
const failedChunks = chunkResults.filter((r) => !r || !r.ok);
|
|
const succeededChunks = chunkResults.filter((r) => r && r.ok);
|
|
if (succeededChunks.length === 0) {
|
|
const first = failedChunks[0]?.error;
|
|
const e = new Error(
|
|
`Parakeet chunked transcribe: all ${chunks.length} chunks failed. First error: ${first?.message || "unknown"}`
|
|
);
|
|
e.status = first?.status || 502;
|
|
throw e;
|
|
}
|
|
// Merge in chunk order so timestamps appear chronologically.
|
|
// Failed chunks leave gaps (their segments are simply absent);
|
|
// the stitched transcript may have time-gaps the analyzer can
|
|
// tolerate. Better than aborting the whole job for one
|
|
// transient chunk failure.
|
|
//
|
|
// Overlap dedup: when consecutive chunks share audio at their
|
|
// boundary (txChunkOverlapSeconds > 0), chunk N+1's first
|
|
// `overlap` seconds will repeat content from chunk N's tail.
|
|
// We drop segments from chunk N+1 whose global start time
|
|
// falls before the chunk's overlap boundary. Chunk 0 has no
|
|
// prior chunk so its boundary equals its own start time and
|
|
// the filter is a no-op.
|
|
const allSegments = [];
|
|
let totalDurationSec = 0;
|
|
let dedupedSegmentsCount = 0;
|
|
for (let i = 0; i < chunkResults.length; i++) {
|
|
const r = chunkResults[i];
|
|
if (!r || !r.ok) continue;
|
|
const globalOverlapBoundary =
|
|
offsetSeconds + (chunks[i]?.overlapBoundarySec || 0);
|
|
for (const seg of r.segments) {
|
|
if (seg.start >= globalOverlapBoundary) {
|
|
allSegments.push(seg);
|
|
} else {
|
|
dedupedSegmentsCount += 1;
|
|
}
|
|
}
|
|
totalDurationSec = Math.max(
|
|
totalDurationSec,
|
|
(r.startSeconds || 0) + (r.duration_seconds || 0)
|
|
);
|
|
}
|
|
if (dedupedSegmentsCount > 0) {
|
|
console.log(
|
|
`[hardware] dropped ${dedupedSegmentsCount} duplicate segment(s) in chunk overlap regions`
|
|
);
|
|
}
|
|
const lines = allSegments.length
|
|
? allSegments.map((s) => `[${formatMmSs(s.start)}] ${s.text}`)
|
|
: [`[0:00] `];
|
|
// Apply the same sort + merge passes Gemini's backend uses
|
|
// so the user-facing transcript is monotonically ordered and
|
|
// grouped into readable thought-sized entries instead of one
|
|
// entry per Parakeet segment (which can be 1-3 seconds each).
|
|
const stitchedText = mergeShortEntries(sortAndDedupeTranscript(lines.join("\n")));
|
|
// Phase 1C: when diarization was enabled, surface the
|
|
// per-chunk diarization data so the caller can run the
|
|
// cross-chunk clustering step in Phase 1D. Null/absent when
|
|
// diarization was off (current default), preserving the
|
|
// existing response shape for callers that don't consume it.
|
|
// Each entry: { ok, chunkIndex, startSeconds, duration_s,
|
|
// segments, speakers_local, fingerprints, models } on
|
|
// success; { ok: false, chunkIndex, startSeconds, error } on
|
|
// failure (chunk's transcript is still in `segments`/`text`).
|
|
const diarizationOut = diarizationEnabled
|
|
? chunkDiarization
|
|
: null;
|
|
|
|
// ── Phase 1D: cross-chunk speaker clustering ──
|
|
// When diarization ran, cluster the per-chunk fingerprints
|
|
// into global speaker IDs (Speaker_A, Speaker_B, ...) and
|
|
// stamp them onto the merged transcript segments. The raw
|
|
// per-chunk diarization data stays in `diarizationOut` for
|
|
// debugging; `speakers` is the operator-facing summary; each
|
|
// entry in `allSegments` gains `.speaker` + `.speaker_confidence`
|
|
// in place.
|
|
let speakersOut = null;
|
|
if (diarizationEnabled) {
|
|
const okCount = chunkDiarization.filter((d) => d && d.ok).length;
|
|
const totalFps = chunkDiarization.reduce(
|
|
(n, d) =>
|
|
n + (d && d.ok ? Object.keys(d.fingerprints || {}).length : 0),
|
|
0
|
|
);
|
|
if (totalFps > 0) {
|
|
const t0 = Date.now();
|
|
const { globalMap, uncertaintyMap, speakers, clusterCount, thresholdSimilarity } =
|
|
clusterSpeakers(chunkDiarization, clusterThresholdPct, {
|
|
anchorMinSpeakingSec,
|
|
smallClusterMaxSpeakingSec,
|
|
uncertainMarginPct,
|
|
});
|
|
assignSpeakersToSegments(allSegments, chunkDiarization, globalMap, uncertaintyMap);
|
|
const clusterMs = Date.now() - t0;
|
|
speakersOut = speakers;
|
|
console.log(
|
|
`[hardware] diarization: ${okCount}/${chunks.length} chunks succeeded, ${totalFps} fingerprints → ${clusterCount} distinct speaker(s) at ${(thresholdSimilarity * 100).toFixed(0)}% cosine-sim threshold (clustering took ${clusterMs}ms)`
|
|
);
|
|
} else {
|
|
console.log(
|
|
`[hardware] diarization: ${okCount}/${chunks.length} chunks succeeded, 0 fingerprints — clustering skipped`
|
|
);
|
|
}
|
|
}
|
|
return {
|
|
text: stitchedText,
|
|
segments: allSegments,
|
|
duration_seconds: totalDurationSec || duration,
|
|
usage: null,
|
|
model: transcribeModel,
|
|
chunk_count: chunks.length,
|
|
chunks_failed: failedChunks.length,
|
|
chunk_durations_ms: chunkDurationsMs,
|
|
diarization: diarizationOut,
|
|
// Map of globalId → summary stats. Null when diarization
|
|
// was off OR ran but produced zero fingerprints. See
|
|
// server/speaker-clustering.js for the shape.
|
|
speakers: speakersOut,
|
|
};
|
|
} finally {
|
|
try {
|
|
await fs.rm(tmpDir, { recursive: true, force: true });
|
|
} catch {}
|
|
}
|
|
},
|
|
|
|
// (postOneChunk + helpers live below the returned object so they
|
|
// can close over `parakeet` / `transcribeModel` / `timeoutMs`.)
|
|
|
|
// POST <gemma>/v1/chat/completions with the OpenAI shape. Ollama's
|
|
// server, vLLM, llama.cpp's HTTP server, and most other OSS LLM
|
|
// runners support this wire format — so we don't lock the relay
|
|
// to one specific Gemma deployment.
|
|
//
|
|
// Model-name autodiscovery: if `gemmaModel` is empty OR the
|
|
// configured model returns "model does not exist", we query
|
|
// /v1/models and use the first listed model id. Spark Control's
|
|
// discovery normally supplies the model name explicitly so this
|
|
// path runs only when discovery's vllm entry is missing the
|
|
// `model` field. Result is cached for the process lifetime so we
|
|
// don't pay the /v1/models round-trip on every analyze call.
|
|
async analyzeText({ prompt }) {
|
|
if (!gemma) {
|
|
const err = new Error(
|
|
"operator-hardware analyze is not configured — Spark Control discovery isn't reporting a ready vllm endpoint"
|
|
);
|
|
err.status = 503;
|
|
throw err;
|
|
}
|
|
|
|
// Resolve which model to send. Configured value wins; empty
|
|
// string triggers autodiscovery.
|
|
let modelToUse = (analyzeModel || "").trim();
|
|
if (!modelToUse) {
|
|
modelToUse = await discoverFirstModel(gemma, timeoutMs);
|
|
if (!modelToUse) {
|
|
const e = new Error(
|
|
`No analyze model configured and no models discoverable at ${gemma}/v1/models — set Gemma Model Name or check the Gemma endpoint`
|
|
);
|
|
e.status = 503;
|
|
throw e;
|
|
}
|
|
}
|
|
|
|
const result = await callChatCompletions({
|
|
base: gemma,
|
|
model: modelToUse,
|
|
prompt,
|
|
maxTokens: anMaxTokens,
|
|
timeoutMs,
|
|
});
|
|
if (result.ok) {
|
|
return {
|
|
text: result.text,
|
|
usage: null,
|
|
model: modelToUse,
|
|
};
|
|
}
|
|
|
|
// Detect "model does not exist" (vLLM / Ollama wording) and
|
|
// retry once with autodiscovered first model — operator may
|
|
// have swapped models on their hardware without updating the
|
|
// relay's config field, very common ergonomic miss.
|
|
const looksLikeModelMissing =
|
|
result.status === 404 &&
|
|
/model.*(does not exist|not found|unknown)/i.test(result.body || "");
|
|
if (looksLikeModelMissing) {
|
|
const discovered = await discoverFirstModel(gemma, timeoutMs);
|
|
if (discovered && discovered !== modelToUse) {
|
|
console.warn(
|
|
`[hardware] configured model "${modelToUse}" not found at ${gemma}; auto-using discovered "${discovered}"`
|
|
);
|
|
const retry = await callChatCompletions({
|
|
base: gemma,
|
|
model: discovered,
|
|
prompt,
|
|
maxTokens: anMaxTokens,
|
|
timeoutMs,
|
|
});
|
|
if (retry.ok) {
|
|
return {
|
|
text: retry.text,
|
|
usage: null,
|
|
model: discovered,
|
|
};
|
|
}
|
|
// Discovered model also failed — fall through to surface
|
|
// the FIRST failure (the configured-model one) so the
|
|
// operator sees what they configured + what we tried.
|
|
}
|
|
const e = new Error(
|
|
`Gemma analyze: configured model "${modelToUse}" does not exist on ${gemma}. ` +
|
|
`Update "Gemma Model Name" in Set Gemma URL — or leave it blank to auto-pick from /v1/models. ` +
|
|
`Response: ${(result.body || "").slice(0, 200)}`
|
|
);
|
|
e.status = 404;
|
|
throw e;
|
|
}
|
|
|
|
const hint =
|
|
result.status === 404
|
|
? ` (tried both /v1/chat/completions and /chat/completions on ${gemma} — check the Gemma URL)`
|
|
: "";
|
|
const e = new Error(
|
|
`Gemma analyze ${result.status} at ${result.lastUrl}: ${(result.body || "").slice(0, 300)}${hint}`
|
|
);
|
|
e.status = result.status;
|
|
throw e;
|
|
},
|
|
};
|
|
}
|
|
|
|
// Shared helper that tries both /v1/chat/completions and
|
|
// /chat/completions, returns the unified result shape.
|
|
async function callChatCompletions({ base, model, prompt, maxTokens, timeoutMs }) {
|
|
const pathCandidates = ["/v1/chat/completions", "/chat/completions"];
|
|
let res = null;
|
|
let body = null;
|
|
let lastUrl = null;
|
|
for (const p of pathCandidates) {
|
|
const url = `${base}${p}`;
|
|
lastUrl = url;
|
|
try {
|
|
// lanFetch — analyze goes through Spark Control over HTTPS
|
|
// with the StartOS Local CA cert. See lan-fetch.js.
|
|
res = await lanFetch(url, {
|
|
method: "POST",
|
|
headers: { "Content-Type": "application/json" },
|
|
redirect: "follow",
|
|
body: JSON.stringify({
|
|
model,
|
|
max_tokens: maxTokens,
|
|
messages: [{ role: "user", content: prompt }],
|
|
stream: false,
|
|
// Force JSON-mode output. Recap's analyze prompt already asks
|
|
// for a JSON shape; this constrains the model so it can't
|
|
// wrap the answer in "Here are the sections I found:" prose,
|
|
// which both costs tokens and trips our parser.
|
|
response_format: { type: "json_object" },
|
|
// Qwen3.6 ships with a reasoning/thinking mode that's great
|
|
// for math but pure latency-noise for structured extraction.
|
|
// The chat_template_kwargs field is honored by vLLM's
|
|
// tokenizer when applying the chat template; non-Qwen
|
|
// backends ignore it.
|
|
chat_template_kwargs: { enable_thinking: false },
|
|
}),
|
|
signal: AbortSignal.timeout(timeoutMs),
|
|
});
|
|
} catch (err) {
|
|
const cause =
|
|
err?.cause?.message || err?.cause?.code || err?.cause || "";
|
|
const detail = cause ? `${err.message} (cause: ${cause})` : err?.message || String(err);
|
|
const e = new Error(
|
|
`Gemma analyze network error at ${url}: ${detail}`
|
|
);
|
|
e.status = 502;
|
|
throw e;
|
|
}
|
|
body = await safeBody(res);
|
|
// Only fall through to the next path candidate when the 404 is a
|
|
// PATH-shaped 404 (FastAPI's default `{"detail":"Not Found"}` or
|
|
// an empty body). A 404 with a content body — e.g. vLLM's
|
|
// `{"error":{"message":"The model X does not exist..."}}` —
|
|
// means the endpoint exists, the request was rejected for a
|
|
// different reason. Falling through would mask the real error
|
|
// behind a path-not-found from the alternate path.
|
|
const looksLikePath404 =
|
|
res.status === 404 &&
|
|
(!body ||
|
|
body.length < 60 ||
|
|
/^\s*\{\s*"detail"\s*:\s*"Not Found"\s*\}\s*$/.test(body));
|
|
if (!looksLikePath404) break;
|
|
console.warn(`[hardware] path-404 at ${url} — trying next path candidate`);
|
|
}
|
|
if (!res.ok) {
|
|
return { ok: false, status: res.status, body, lastUrl };
|
|
}
|
|
let data;
|
|
try {
|
|
data = JSON.parse(body);
|
|
} catch {
|
|
return { ok: false, status: 502, body: "non-JSON response", lastUrl };
|
|
}
|
|
const text = data?.choices?.[0]?.message?.content || "";
|
|
return { ok: true, text, lastUrl };
|
|
}
|
|
|
|
// Cache of {base → modelId} so we don't pay the /v1/models round-trip
|
|
// on every call. Per-process; reset on container restart.
|
|
const _discoveredModelCache = new Map();
|
|
async function discoverFirstModel(base, timeoutMs) {
|
|
if (_discoveredModelCache.has(base)) {
|
|
return _discoveredModelCache.get(base);
|
|
}
|
|
try {
|
|
const res = await lanFetch(`${base}/v1/models`, {
|
|
signal: AbortSignal.timeout(timeoutMs),
|
|
});
|
|
if (!res.ok) {
|
|
console.warn(`[hardware] /v1/models returned ${res.status} at ${base}`);
|
|
return null;
|
|
}
|
|
const data = await res.json();
|
|
const first = Array.isArray(data?.data) ? data.data[0] : null;
|
|
const id = first?.id || null;
|
|
if (id) _discoveredModelCache.set(base, id);
|
|
return id;
|
|
} catch (err) {
|
|
console.warn(`[hardware] /v1/models discovery failed at ${base}: ${err?.message || err}`);
|
|
return null;
|
|
}
|
|
}
|
|
|
|
function formatMmSs(seconds) {
|
|
const s = Math.max(0, Math.floor(seconds));
|
|
const h = Math.floor(s / 3600);
|
|
const m = Math.floor((s % 3600) / 60);
|
|
const sec = s % 60;
|
|
if (h > 0)
|
|
return `${h}:${String(m).padStart(2, "0")}:${String(sec).padStart(2, "0")}`;
|
|
return `${m}:${String(sec).padStart(2, "0")}`;
|
|
}
|
|
|
|
async function safeBody(res) {
|
|
try {
|
|
return await res.text();
|
|
} catch {
|
|
return "";
|
|
}
|
|
}
|
|
|
|
// Map common audio MIME types to a sensible file extension so ffprobe
|
|
// + ffmpeg pick the right demuxer. Defaults to `mp3` which is the
|
|
// fallback Recap normalizes to before sending here.
|
|
function extFromMime(mime) {
|
|
if (!mime) return "mp3";
|
|
const m = mime.toLowerCase();
|
|
if (m.includes("mp4") || m.includes("m4a")) return "m4a";
|
|
if (m.includes("ogg")) return "ogg";
|
|
if (m.includes("opus")) return "opus";
|
|
if (m.includes("wav")) return "wav";
|
|
if (m.includes("aac")) return "aac";
|
|
if (m.includes("webm")) return "webm";
|
|
if (m.includes("flac")) return "flac";
|
|
return "mp3";
|
|
}
|