Start on live transcription and python sidecar (faster-whisper)
This commit is contained in:
Lucas Winther
+473
-6
@@ -1,10 +1,477 @@
|
||||
/**
|
||||
* Server-side transcription engine — TODO(M5).
|
||||
* Server-side transcription engine (M5, Decisions.md D-013): bridges the
|
||||
* faster-whisper Python sidecar (server-py/sidecar.py) to the streamed
|
||||
* JSON-lines HTTP contract (D-011, wire format in src/lib/engine.ts).
|
||||
*
|
||||
* POST /api/transcribe: auth-guarded, runs the same Transformers.js pipeline
|
||||
* under Bun (ONNX Runtime, CPU), streams progress + partial transcripts as a
|
||||
* chunked JSON-lines response (Decisions.md D-011). Reuses the shared model
|
||||
* catalog in src/lib/models.ts; keep pipelines warm across requests.
|
||||
* Routes (wired in src/index.ts, auth-guarded — the server engine is the one
|
||||
* logged-in-only feature, Architecture.md §5):
|
||||
* GET /api/transcribe/status → 200 {available, reason?} | 401
|
||||
* POST /api/transcribe?model=<catalog id>[&language=..][&task=..]
|
||||
* body: raw float32 mono 16 kHz PCM (application/octet-stream)
|
||||
* → 200 application/x-ndjson (segment* then result | error lines)
|
||||
* | 400 | 401 | 413 | 503 {error}
|
||||
*
|
||||
* The sidecar is spawned lazily on first use and supervised: a crash rejects
|
||||
* the requests it was carrying and the next request respawns it, while a
|
||||
* missing runtime (no python3, faster-whisper not installed) is remembered as
|
||||
* a permanent "unavailable" reason — fixing that requires a deployment
|
||||
* change, so we never busy-respawn. Either way the browser/anonymous path is
|
||||
* untouched: this module holds the only Python dependency in the app.
|
||||
*
|
||||
* Warm-up strategy (WorkBreakdown M5): the sidecar keeps the last-used model
|
||||
* loaded between requests, so only the first request for a model (or a model
|
||||
* switch) pays the load; weights download once into the sidecar's cache.
|
||||
*/
|
||||
|
||||
export {};
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import type { TranscribeStreamLine } from "../lib/engine";
|
||||
import { getFasterWhisperModel } from "../lib/models";
|
||||
import type { TranscriptChunk, WhisperTask } from "../worker/protocol";
|
||||
import { getSessionUser as appGetSessionUser, type SessionUser } from "./auth.ts";
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Sidecar interface (injectable — tests use a fake, see tests/transcribe.test.ts)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export interface SidecarSegment {
|
||||
start: number;
|
||||
end: number;
|
||||
text: string;
|
||||
}
|
||||
|
||||
export interface SidecarTranscribeRequest {
|
||||
/** faster-whisper model name (catalog mapping, src/lib/models.ts). */
|
||||
model: string;
|
||||
/** Path to a file of raw float32 mono 16 kHz PCM. */
|
||||
audioPath: string;
|
||||
language?: string;
|
||||
task?: WhisperTask;
|
||||
}
|
||||
|
||||
export interface SidecarStatus {
|
||||
available: boolean;
|
||||
reason?: string;
|
||||
}
|
||||
|
||||
export interface Sidecar {
|
||||
/** Spawns the process if needed; reports whether inference can run. */
|
||||
status(): Promise<SidecarStatus>;
|
||||
/**
|
||||
* Run one transcription, streaming segments as they decode. Resolves after
|
||||
* the sidecar's "done" (cancelled: true when `signal` fired first);
|
||||
* rejects on sidecar errors or crashes.
|
||||
*/
|
||||
transcribe(
|
||||
request: SidecarTranscribeRequest,
|
||||
onSegment: (segment: SidecarSegment) => void,
|
||||
signal?: AbortSignal,
|
||||
): Promise<{ cancelled: boolean }>;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// The real sidecar: Bun.spawn + JSON-lines over stdio (D-013)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
const SIDECAR_SCRIPT = join(import.meta.dir, "../../server-py/sidecar.py");
|
||||
|
||||
/** How long the spawned process gets to import faster-whisper and report. */
|
||||
const READY_TIMEOUT_MS = 60_000;
|
||||
|
||||
interface SidecarProcess {
|
||||
proc: ReturnType<typeof Bun.spawn>;
|
||||
ready: Promise<void>;
|
||||
}
|
||||
|
||||
interface PendingTranscription {
|
||||
onSegment: (segment: SidecarSegment) => void;
|
||||
resolve: (outcome: { cancelled: boolean }) => void;
|
||||
reject: (error: Error) => void;
|
||||
}
|
||||
|
||||
export function createSidecar(command?: readonly string[]): Sidecar {
|
||||
// "python3" resolved per spawn — SIDECAR_PYTHON overrides for exotic setups.
|
||||
const cmd = command ?? [process.env.SIDECAR_PYTHON || "python3", SIDECAR_SCRIPT];
|
||||
|
||||
let current: SidecarProcess | null = null;
|
||||
/** Deployment-level failure (no python, deps missing): never respawn. */
|
||||
let permanentFailure: string | null = null;
|
||||
const pending = new Map<string, PendingTranscription>();
|
||||
let nextRequestId = 0;
|
||||
|
||||
function failAllPending(error: Error): void {
|
||||
const requests = [...pending.values()];
|
||||
pending.clear();
|
||||
for (const request of requests) request.reject(error);
|
||||
}
|
||||
|
||||
function routeLine(rawLine: string): void {
|
||||
let message: Record<string, unknown>;
|
||||
try {
|
||||
message = JSON.parse(rawLine) as Record<string, unknown>;
|
||||
} catch {
|
||||
return; // stray non-JSON output must not kill the bridge
|
||||
}
|
||||
const request = pending.get(String(message.id));
|
||||
if (!request) return;
|
||||
switch (message.type) {
|
||||
case "segment":
|
||||
if (
|
||||
typeof message.start === "number" &&
|
||||
typeof message.end === "number" &&
|
||||
typeof message.text === "string"
|
||||
) {
|
||||
request.onSegment({
|
||||
start: message.start,
|
||||
end: message.end,
|
||||
text: message.text,
|
||||
});
|
||||
}
|
||||
break;
|
||||
case "done":
|
||||
pending.delete(String(message.id));
|
||||
request.resolve({ cancelled: message.cancelled === true });
|
||||
break;
|
||||
case "error":
|
||||
pending.delete(String(message.id));
|
||||
request.reject(new Error(String(message.message ?? "sidecar error")));
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
function spawn(): SidecarProcess {
|
||||
const proc = Bun.spawn({
|
||||
cmd: [...cmd],
|
||||
stdin: "pipe",
|
||||
stdout: "pipe",
|
||||
stderr: "inherit", // model download progress / tracebacks → server log
|
||||
});
|
||||
|
||||
const ready = new Promise<void>((resolve, reject) => {
|
||||
let settled = false;
|
||||
const timeout = setTimeout(() => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
proc.kill();
|
||||
reject(new Error("the transcription sidecar did not start in time"));
|
||||
}, READY_TIMEOUT_MS);
|
||||
|
||||
void (async () => {
|
||||
// First line decides readiness; everything after is request traffic.
|
||||
const decoder = new TextDecoder();
|
||||
let buffer = "";
|
||||
for await (const chunk of proc.stdout as ReadableStream<Uint8Array>) {
|
||||
buffer += decoder.decode(chunk, { stream: true });
|
||||
let newline: number;
|
||||
while ((newline = buffer.indexOf("\n")) !== -1) {
|
||||
const line = buffer.slice(0, newline);
|
||||
buffer = buffer.slice(newline + 1);
|
||||
if (!settled) {
|
||||
settled = true;
|
||||
clearTimeout(timeout);
|
||||
let handshake: Record<string, unknown> = {};
|
||||
try {
|
||||
handshake = JSON.parse(line) as Record<string, unknown>;
|
||||
} catch {
|
||||
// fall through — treated as a bad handshake below
|
||||
}
|
||||
if (handshake.type === "ready") {
|
||||
resolve();
|
||||
} else {
|
||||
// "fatal": faster-whisper missing — a deployment problem.
|
||||
permanentFailure = String(
|
||||
handshake.message ?? "the transcription sidecar failed to start",
|
||||
);
|
||||
reject(new Error(permanentFailure));
|
||||
}
|
||||
} else {
|
||||
routeLine(line);
|
||||
}
|
||||
}
|
||||
}
|
||||
// stdout closed: the process exited (crash or shutdown).
|
||||
if (!settled) {
|
||||
settled = true;
|
||||
clearTimeout(timeout);
|
||||
reject(new Error("the transcription sidecar exited during startup"));
|
||||
}
|
||||
failAllPending(new Error("the transcription sidecar exited unexpectedly"));
|
||||
if (current?.proc === proc) current = null; // next request respawns
|
||||
})();
|
||||
});
|
||||
// A failed startup must not surface as an unhandled rejection when nobody
|
||||
// is awaiting yet (e.g. crash between requests).
|
||||
ready.catch(() => {});
|
||||
return { proc, ready };
|
||||
}
|
||||
|
||||
async function ensure(): Promise<SidecarProcess> {
|
||||
if (permanentFailure) throw new Error(permanentFailure);
|
||||
if (current) return current;
|
||||
try {
|
||||
current = spawn();
|
||||
} catch (error) {
|
||||
// Bun.spawn throws synchronously when the binary is missing.
|
||||
permanentFailure = `python3 is not available on this server (${
|
||||
error instanceof Error ? error.message : String(error)
|
||||
})`;
|
||||
throw new Error(permanentFailure);
|
||||
}
|
||||
return current;
|
||||
}
|
||||
|
||||
function send(proc: ReturnType<typeof Bun.spawn>, message: unknown): void {
|
||||
const stdin = proc.stdin as { write(data: string): unknown; flush(): unknown };
|
||||
stdin.write(`${JSON.stringify(message)}\n`);
|
||||
stdin.flush();
|
||||
}
|
||||
|
||||
return {
|
||||
async status() {
|
||||
try {
|
||||
const sidecar = await ensure();
|
||||
await sidecar.ready;
|
||||
return { available: true };
|
||||
} catch (error) {
|
||||
return {
|
||||
available: false,
|
||||
reason: error instanceof Error ? error.message : String(error),
|
||||
};
|
||||
}
|
||||
},
|
||||
|
||||
async transcribe(request, onSegment, signal) {
|
||||
const sidecar = await ensure();
|
||||
await sidecar.ready;
|
||||
const id = `req-${++nextRequestId}`;
|
||||
return new Promise<{ cancelled: boolean }>((resolve, reject) => {
|
||||
pending.set(id, { onSegment, resolve, reject });
|
||||
signal?.addEventListener(
|
||||
"abort",
|
||||
() => {
|
||||
// The sidecar checks between segments and still answers "done"
|
||||
// (cancelled: true), which resolves this promise normally.
|
||||
if (pending.has(id) && current?.proc === sidecar.proc) {
|
||||
send(sidecar.proc, { type: "cancel", id });
|
||||
}
|
||||
},
|
||||
{ once: true },
|
||||
);
|
||||
try {
|
||||
send(sidecar.proc, {
|
||||
type: "transcribe",
|
||||
id,
|
||||
model: request.model,
|
||||
audio_path: request.audioPath,
|
||||
language: request.language ?? null,
|
||||
task: request.task ?? "transcribe",
|
||||
});
|
||||
} catch (error) {
|
||||
pending.delete(id);
|
||||
reject(error instanceof Error ? error : new Error(String(error)));
|
||||
}
|
||||
});
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// HTTP handlers
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/** Default cap ≈ 2¼ hours of raw float32 16 kHz mono (64 kB/s). */
|
||||
const DEFAULT_MAX_UPLOAD_BYTES = 512 * 1024 * 1024;
|
||||
|
||||
const LANGUAGE_RE = /^[a-z]{2,3}$/;
|
||||
|
||||
export interface TranscribeHandlers {
|
||||
status(request: Request): Promise<Response>;
|
||||
post(request: Request): Promise<Response>;
|
||||
}
|
||||
|
||||
export interface TranscribeDeps {
|
||||
sidecar: Sidecar;
|
||||
getSessionUser(request: Request): SessionUser | null;
|
||||
maxUploadBytes?: number;
|
||||
/** Directory for uploaded-PCM temp files. Default: the OS temp dir. */
|
||||
tempDir?: string;
|
||||
}
|
||||
|
||||
function json(body: unknown, status: number): Response {
|
||||
return new Response(JSON.stringify(body), {
|
||||
status,
|
||||
headers: { "Content-Type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
interface TranscribeParams {
|
||||
modelId: string;
|
||||
fasterWhisperModel: string;
|
||||
language?: string;
|
||||
task: WhisperTask;
|
||||
}
|
||||
|
||||
/** Parse + validate query params; a string is the 400 error text. */
|
||||
function parseParams(url: URL): TranscribeParams | string {
|
||||
const modelId = url.searchParams.get("model");
|
||||
if (!modelId) return "missing model parameter";
|
||||
const fasterWhisperModel = getFasterWhisperModel(modelId);
|
||||
// Catalog-only (D-006/D-013): never forward arbitrary ids to the sidecar.
|
||||
if (!fasterWhisperModel) return `unknown model: ${modelId}`;
|
||||
|
||||
const rawLanguage = url.searchParams.get("language");
|
||||
let language: string | undefined;
|
||||
if (rawLanguage !== null && rawLanguage !== "" && rawLanguage !== "auto") {
|
||||
if (!LANGUAGE_RE.test(rawLanguage)) return `invalid language: ${rawLanguage}`;
|
||||
language = rawLanguage;
|
||||
}
|
||||
|
||||
const rawTask = url.searchParams.get("task") ?? "transcribe";
|
||||
if (rawTask !== "transcribe" && rawTask !== "translate") {
|
||||
return `invalid task: ${rawTask}`;
|
||||
}
|
||||
|
||||
return { modelId, fasterWhisperModel, language, task: rawTask };
|
||||
}
|
||||
|
||||
/**
|
||||
* Stream the request body to `path`, enforcing the size limit even for
|
||||
* chunked bodies. Returns bytes written, or a Response to answer with.
|
||||
*/
|
||||
async function writeBodyToFile(
|
||||
request: Request,
|
||||
path: string,
|
||||
maxBytes: number,
|
||||
): Promise<number | Response> {
|
||||
const declared = Number(request.headers.get("content-length"));
|
||||
if (Number.isFinite(declared) && declared > maxBytes) {
|
||||
return json({ error: `audio exceeds the ${maxBytes}-byte upload limit` }, 413);
|
||||
}
|
||||
if (!request.body) return json({ error: "missing request body" }, 400);
|
||||
|
||||
const sink = Bun.file(path).writer();
|
||||
let written = 0;
|
||||
try {
|
||||
for await (const chunk of request.body) {
|
||||
written += chunk.byteLength;
|
||||
if (written > maxBytes) {
|
||||
return json({ error: `audio exceeds the ${maxBytes}-byte upload limit` }, 413);
|
||||
}
|
||||
sink.write(chunk);
|
||||
}
|
||||
} finally {
|
||||
await sink.end();
|
||||
}
|
||||
if (written === 0) return json({ error: "missing request body" }, 400);
|
||||
if (written % 4 !== 0) {
|
||||
return json({ error: "body must be raw float32 PCM (length % 4 == 0)" }, 400);
|
||||
}
|
||||
return written;
|
||||
}
|
||||
|
||||
export function createTranscribeHandlers(deps: TranscribeDeps): TranscribeHandlers {
|
||||
const maxUploadBytes = deps.maxUploadBytes ?? DEFAULT_MAX_UPLOAD_BYTES;
|
||||
const tempDir = deps.tempDir ?? tmpdir();
|
||||
|
||||
return {
|
||||
async status(request) {
|
||||
if (!deps.getSessionUser(request)) {
|
||||
return json({ error: "not authenticated" }, 401);
|
||||
}
|
||||
const status = await deps.sidecar.status();
|
||||
return json(status, 200);
|
||||
},
|
||||
|
||||
async post(request) {
|
||||
if (!deps.getSessionUser(request)) {
|
||||
return json({ error: "not authenticated" }, 401);
|
||||
}
|
||||
const params = parseParams(new URL(request.url));
|
||||
if (typeof params === "string") return json({ error: params }, 400);
|
||||
|
||||
// Refuse before consuming a potentially huge body when inference can't
|
||||
// run anyway (degrade gracefully — D-013).
|
||||
const status = await deps.sidecar.status();
|
||||
if (!status.available) {
|
||||
return json({ error: status.reason ?? "server engine unavailable" }, 503);
|
||||
}
|
||||
|
||||
const audioPath = join(tempDir, `whisper-upload-${crypto.randomUUID()}.f32`);
|
||||
const written = await writeBodyToFile(request, audioPath, maxUploadBytes);
|
||||
if (written instanceof Response) {
|
||||
await Bun.file(audioPath).unlink().catch(() => {});
|
||||
return written;
|
||||
}
|
||||
|
||||
// 200 with an ndjson stream (D-011); failures beyond this point travel
|
||||
// in-stream as an "error" line since the status is already committed.
|
||||
const encoder = new TextEncoder();
|
||||
const stream = new ReadableStream<Uint8Array>({
|
||||
async start(controller) {
|
||||
const enqueue = (line: TranscribeStreamLine) =>
|
||||
controller.enqueue(encoder.encode(`${JSON.stringify(line)}\n`));
|
||||
const chunks: TranscriptChunk[] = [];
|
||||
try {
|
||||
await deps.sidecar.transcribe(
|
||||
{
|
||||
model: params.fasterWhisperModel,
|
||||
audioPath,
|
||||
language: params.language,
|
||||
task: params.task,
|
||||
},
|
||||
(segment) => {
|
||||
chunks.push({
|
||||
text: segment.text,
|
||||
timestamp: [segment.start, segment.end],
|
||||
});
|
||||
enqueue({
|
||||
type: "segment",
|
||||
text: segment.text,
|
||||
start: segment.start,
|
||||
end: segment.end,
|
||||
});
|
||||
},
|
||||
// Client gone (abort/refresh) → cancel inference between segments.
|
||||
request.signal,
|
||||
);
|
||||
enqueue({
|
||||
type: "result",
|
||||
text: chunks.map((chunk) => chunk.text).join("").trim(),
|
||||
chunks,
|
||||
});
|
||||
} catch (error) {
|
||||
enqueue({
|
||||
type: "error",
|
||||
message: error instanceof Error ? error.message : String(error),
|
||||
});
|
||||
} finally {
|
||||
await Bun.file(audioPath).unlink().catch(() => {});
|
||||
try {
|
||||
controller.close();
|
||||
} catch {
|
||||
// Already errored/cancelled — the temp file cleanup above is
|
||||
// what actually matters here.
|
||||
}
|
||||
}
|
||||
},
|
||||
});
|
||||
|
||||
return new Response(stream, {
|
||||
headers: {
|
||||
"Content-Type": "application/x-ndjson; charset=utf-8",
|
||||
"Cache-Control": "no-store",
|
||||
},
|
||||
});
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Default instance (real sidecar + app sessions), wired in src/index.ts
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export const transcribe: TranscribeHandlers = createTranscribeHandlers({
|
||||
sidecar: createSidecar(),
|
||||
getSessionUser: appGetSessionUser,
|
||||
maxUploadBytes: Number(process.env.TRANSCRIBE_MAX_UPLOAD_BYTES) || undefined,
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user