diff --git a/apps/api/src/router.ts b/apps/api/src/router.ts index 2baf21e..7756f51 100644 --- a/apps/api/src/router.ts +++ b/apps/api/src/router.ts @@ -221,6 +221,7 @@ import { listVoiceCatalog, loadDefaultVoiceCredential, loadVoiceCredential, + mediaTranscriptionSettings, persistVoiceCredential, prepareVoice, toVoiceCredential, @@ -4782,8 +4783,11 @@ export function createRouter(deps: RouterDeps) { voice: { catalog: authed.voice.catalog.handler(async () => listVoiceCatalog()), status: authed.voice.status.handler(async ({ context }) => { - const cred = await findDefaultVoiceCredential(deps.prisma, context.actor); - return toVoiceStatus(cred); + const [cred, server] = await Promise.all([ + findDefaultVoiceCredential(deps.prisma, context.actor), + mediaTranscriptionSettings(deps), + ]); + return toVoiceStatus(cred, { mediaTranscription: Boolean(server) }); }), credentials: authed.voice.credentials.handler(async ({ context }) => { const rows = await deps.prisma.userVoiceCredential.findMany({ @@ -4845,7 +4849,9 @@ export function createRouter(deps: RouterDeps) { { isolationLevel: Prisma.TransactionIsolationLevel.Serializable }, ), ); - return toVoiceStatus(cred); + return toVoiceStatus(cred, { + mediaTranscription: Boolean(await mediaTranscriptionSettings(deps)), + }); }), voices: authed.voice.voices.handler(async ({ context, input }) => { const loaded = await loadDefaultVoiceCredential(deps, context.actor); diff --git a/apps/api/src/voice-transcribe.test.ts b/apps/api/src/voice-transcribe.test.ts new file mode 100644 index 0000000..a023d9e --- /dev/null +++ b/apps/api/src/voice-transcribe.test.ts @@ -0,0 +1,89 @@ +import { createServer, type Server } from "node:http"; +import type { AddressInfo } from "node:net"; +import { NoVoiceConfigured } from "@rakazo/adapters"; +import type { Actor } from "@rakazo/contracts"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { toVoiceStatus, transcribeVoice, type VoiceDeps } from "./voice.js"; + +const actor: Actor = { + userId: "user-1", + spaceId: "space-1", + email: "user@rakazo.test", + isDeploymentOwner: false, +}; + +/** A user without a voice credential, on a deployment whose media settings are `settings`. */ +function deps(settings: object | null, ciphertext: string): VoiceDeps { + return { + prisma: { + spaceVoicePreference: { findFirst: async () => null }, + integrationProviderConfig: { + findUnique: async () => (settings ? { id: "media", ciphertext } : null), + }, + }, + secrets: { load: () => JSON.stringify(settings) }, + } as unknown as VoiceDeps; +} + +describe("dictation through the self-hosted transcription server", () => { + let server: Server; + let baseUrl = ""; + const uploads: string[] = []; + + beforeAll(async () => { + server = createServer((request, response) => { + const chunks: Buffer[] = []; + request.on("data", (chunk: Buffer) => chunks.push(chunk)); + request.on("end", () => { + uploads.push(`${request.url} ${Buffer.concat(chunks).toString("latin1")}`); + response.setHeader("content-type", "application/json"); + response.end(JSON.stringify({ text: " Remind me to call Ada. " })); + }); + }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + baseUrl = `http://127.0.0.1:${(server.address() as AddressInfo).port}/v1`; + }); + + afterAll(async () => { + await new Promise((resolve) => server.close(resolve)); + }); + + it("transcribes when no voice provider can", async () => { + const result = await transcribeVoice( + deps({ comfyui: null, transcription: { baseUrl } }, "media-a"), + actor, + { audio: new Uint8Array([1, 2, 3]), mimeType: "audio/webm;codecs=opus" }, + ); + expect(result).toEqual({ text: "Remind me to call Ada." }); + expect(uploads[0]).toContain("/v1/audio/transcriptions"); + expect(uploads[0]).toContain('filename="audio.webm"'); + }); + + it("keeps asking for a voice key when no server is set up either", async () => { + await expect( + transcribeVoice(deps(null, "none"), actor, { + audio: new Uint8Array([1]), + mimeType: "audio/webm", + }), + ).rejects.toBeInstanceOf(NoVoiceConfigured); + }); + + it("refuses an empty recording before uploading it", async () => { + const before = uploads.length; + await expect( + transcribeVoice(deps({ comfyui: null, transcription: { baseUrl } }, "media-b"), actor, { + audio: new Uint8Array(), + mimeType: "audio/webm", + }), + ).rejects.toThrow("That recording is empty or too large."); + expect(uploads).toHaveLength(before); + }); + + it("reports server dictation in the voice status", () => { + expect(toVoiceStatus(null).transcribe).toBe(false); + expect(toVoiceStatus(null, { mediaTranscription: true })).toMatchObject({ + configured: false, + transcribe: true, + }); + }); +}); diff --git a/apps/api/src/voice.ts b/apps/api/src/voice.ts index 9edeb57..caf2d4d 100644 --- a/apps/api/src/voice.ts +++ b/apps/api/src/voice.ts @@ -1,10 +1,12 @@ import { ORPCError } from "@orpc/server"; import type { AdapterContext } from "@rakazo/adapter-kit"; import { + createOpenAiCompatibleTranscriptionProvider, createVoiceProvider, type EncryptedSecretStore, isVoiceProviderId, listVoiceCatalog, + loadMediaSettings, MAX_SPEAK_CHARS, MAX_TRANSCRIBE_BYTES, NoVoiceConfigured, @@ -51,12 +53,20 @@ export function catalogEntry(provider: string) { return voiceCatalogEntry(provider); } -export function toVoiceStatus(cred: { provider: string; voiceId: string } | null): VoiceStatus { +/** The deployment's self-hosted transcription server, when the owner set one up. */ +export async function mediaTranscriptionSettings(deps: VoiceDeps) { + return (await loadMediaSettings(deps).catch(() => null))?.transcription ?? null; +} + +export function toVoiceStatus( + cred: { provider: string; voiceId: string } | null, + options: { mediaTranscription?: boolean } = {}, +): VoiceStatus { const entry = cred ? catalogEntry(cred.provider) : undefined; return { configured: Boolean(cred), ready: Boolean(cred?.voiceId), - transcribe: Boolean(entry?.transcribe && cred), + transcribe: Boolean(entry?.transcribe && cred) || Boolean(options.mediaTranscription), provider: cred?.provider ?? null, voiceId: cred?.voiceId ?? "", }; @@ -277,9 +287,22 @@ export async function transcribeVoice( input: { audio: Uint8Array; mimeType: string; signal?: AbortSignal }, ) { const loaded = await loadDefaultVoiceCredential(deps, actor); - if (!loaded) throw new NoVoiceConfigured("key"); - const provider = createVoiceProvider(loaded.cred.provider); - if (!provider.transcribe) { + const provider = loaded ? createVoiceProvider(loaded.cred.provider) : null; + if (!loaded || !provider?.transcribe) { + // Without a voice provider that transcribes, the self-hosted media server can. + const server = await mediaTranscriptionSettings(deps); + if (server) { + if (input.audio.byteLength === 0 || input.audio.byteLength > MAX_TRANSCRIBE_BYTES) { + throw new ORPCError("BAD_REQUEST", { message: "That recording is empty or too large." }); + } + const transcript = await createOpenAiCompatibleTranscriptionProvider(server).transcribe({ + audio: input.audio, + mimeType: input.mimeType || "audio/webm", + signal: input.signal, + }); + return { text: transcript.text }; + } + if (!loaded) throw new NoVoiceConfigured("key"); throw new ORPCError("BAD_REQUEST", { message: "This voice provider does not transcribe audio. Use on-device dictation instead.", }); diff --git a/packages/adapters/src/builtin-tools.ts b/packages/adapters/src/builtin-tools.ts index 0ff52fa..d8fc9b3 100644 --- a/packages/adapters/src/builtin-tools.ts +++ b/packages/adapters/src/builtin-tools.ts @@ -5,9 +5,13 @@ import { MODEL_TIERS, SecretAskPurpose, SecretHttpRequest, + TranscribeInputSchema, USER_FACT_CATEGORIES, } from "@rakazo/contracts"; import { z } from "zod"; +import { mediaToolInputSchema } from "./media-tools.js"; + +type MediaToolSchema = Parameters[0]; export const SUBAGENT_MODEL_TIER_DESCRIPTION = "Optional model tier: fast for quick simple work, code for programming, standard otherwise. Uses the model this space maps to the tier, or your own model when it maps none. Ignored when model_provider and model_id are set."; @@ -326,6 +330,12 @@ export const builtinAgentTools: ConnectorTool[] = [ }, }, }, + { + name: "transcribe", + description: + "Transcribe speech in an audio or video artifact into SRT and VTT subtitle files, shown on a media card. Set continue_after to get the transcript back, e.g. to burn the .srt in with edit_video.", + inputSchema: mediaToolInputSchema(z.toJSONSchema(TranscribeInputSchema) as MediaToolSchema), + }, { name: "add_mcp_server", description: diff --git a/packages/adapters/src/executor.ts b/packages/adapters/src/executor.ts index 5cb9181..9d2c8bd 100644 --- a/packages/adapters/src/executor.ts +++ b/packages/adapters/src/executor.ts @@ -357,6 +357,7 @@ import { skillReadFromTool, skillUpdateFromTool, } from "./skill-tools.js"; +import { ensureStudioWhenMediaConfigured } from "./studio-specialist.js"; import { continueRunClaimFence, DESKTOP_HELD_FOR_TAKEOVER_MESSAGE, @@ -3620,6 +3621,8 @@ export function createRunExecutor(deps: ExecutorDeps) { }); } const runtimeHistory = [...historicalContext, ...history]; + // Studio joins the directory once a media server is set up. + if (exposedToolNames.has("delegate_task")) await ensureStudioWhenMediaConfigured(deps, run); // The main agent sees every specialist it can delegate to, across the user's spaces. // Other runs see the teammates they can address with message_bot; runs without it // (every group run) list their reachable members in instead. diff --git a/packages/adapters/src/media-operations.ts b/packages/adapters/src/media-operations.ts index 98dbf7e..3905c49 100644 --- a/packages/adapters/src/media-operations.ts +++ b/packages/adapters/src/media-operations.ts @@ -1,4 +1,5 @@ import type { MediaSettings } from "@rakazo/contracts"; +import { createTranscriptionOperations } from "./media/transcription.js"; import type { MediaOperationRegistration } from "./media-runtime.js"; export interface MediaOperationDeps { @@ -10,6 +11,6 @@ export interface MediaOperationDeps { * Every media operation this deployment can run. The API and the worker build their runtime * from this one list, so both processes run the same operations. */ -export function createMediaOperations(_deps: MediaOperationDeps): MediaOperationRegistration[] { - return []; +export function createMediaOperations(deps: MediaOperationDeps): MediaOperationRegistration[] { + return [...createTranscriptionOperations({ loadSettings: deps.loadSettings })]; } diff --git a/packages/adapters/src/media/transcription.test.ts b/packages/adapters/src/media/transcription.test.ts index bf6c010..34c39af 100644 --- a/packages/adapters/src/media/transcription.test.ts +++ b/packages/adapters/src/media/transcription.test.ts @@ -1,10 +1,24 @@ -import { mkdtemp, rm, writeFile } from "node:fs/promises"; +import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; -import { afterEach, describe, expect, it } from "vitest"; -import { EmulatorTranscriptionProvider } from "../transcription-emulator.js"; import { + type MediaJobContext, + MediaJobDeferredError, + type MediaJobInputFile, + type MediaJobOutputFile, +} from "@rakazo/adapter-kit"; +import type { MediaOutput, MediaSettings } from "@rakazo/contracts"; +import { afterEach, describe, expect, it } from "vitest"; +import { createOpenAiCompatibleTranscriptionProvider } from "../openai-compatible-transcription.js"; +import { + EmulatorTranscriptionProvider, + OpenAiCompatibleTranscriptionServerEmulator, +} from "../transcription-emulator.js"; +import { + createTranscriptionOperations, + planAudioChunks, TRANSCRIPT_SUMMARY_MAX_CHARS, + type TranscriptionAudioTools, transcribeChunks, transcriptionChunkSeconds, transcriptSummary, @@ -100,3 +114,215 @@ describe("transcriptSummary", () => { expect(body.endsWith("word …")).toBe(true); }); }); + +const TRANSCRIPTION: MediaSettings = { + comfyui: null, + transcription: { baseUrl: "http://127.0.0.1:8080/v1", model: undefined, apiKey: undefined }, +}; + +async function jobContext(source: Omit, input: unknown) { + const tempDir = await mkdtemp(join(tmpdir(), "transcribe-job-")); + dirs.push(tempDir); + const sourcePath = join(tempDir, `input${source.name.slice(source.name.lastIndexOf("."))}`); + await writeFile(sourcePath, new Uint8Array([82, 73, 70, 70, 1, 2, 3])); + const outputs: Array = []; + const progress: number[] = []; + const context: MediaJobContext = { + jobId: "job-1", + spaceId: "space-1", + userId: "user-1", + botId: "bot-1", + input, + tempDir, + signal: new AbortController().signal, + progress: async (fraction) => { + progress.push(fraction); + }, + fetchArtifact: async (artifactId) => { + if (artifactId !== source.artifactId) throw new Error("unknown artifact"); + return { ...source, path: sourcePath }; + }, + addOutput: async (file) => { + outputs.push({ ...file, body: await readFile(file.path, "utf8") }); + return { artifactId: `out-${outputs.length}` } as MediaOutput; + }, + }; + return { context, outputs, progress }; +} + +const audioSource = { + artifactId: "audio-1", + name: "Interview take 2.m4a", + mimeType: "audio/mp4", + kind: "audio" as const, + size: 7, +}; + +function operation( + settings: MediaSettings = TRANSCRIPTION, + options: { flavor?: "openai" | "text"; audio?: TranscriptionAudioTools; maxBytes?: number } = {}, +) { + const server = new OpenAiCompatibleTranscriptionServerEmulator({ flavor: options.flavor }); + const [registration] = createTranscriptionOperations({ + loadSettings: async () => settings, + audio: options.audio, + createProvider: (transcription) => { + const provider = createOpenAiCompatibleTranscriptionProvider({ + ...transcription, + fetch: server.fetch, + }); + if (!options.maxBytes) return provider; + const maxBytes = options.maxBytes; + return { + transcribe: provider.transcribe, + describe: () => { + const descriptor = provider.describe(); + return { ...descriptor, capabilities: { ...descriptor.capabilities, maxBytes } }; + }, + }; + }, + }); + return { registration: registration!, server }; +} + +describe("transcribe operation", () => { + it("is available only with a transcription server and reads the source artifact", async () => { + expect(await operation().registration.available({ spaceId: "space-1" })).toBe(true); + expect( + await operation({ comfyui: null, transcription: null }).registration.available({ + spaceId: "space-1", + }), + ).toBe(false); + const { registration } = operation(); + const input = registration.inputSchema.parse({ artifactId: "audio-1", language: "de" }); + expect(registration.artifactIds?.(input)).toEqual(["audio-1"]); + expect(registration.title(input)).toBe("Transcript"); + expect( + registration.inputSchema.safeParse({ artifactId: "audio-1", language: "German!" }).success, + ).toBe(false); + }); + + it("turns an audio artifact into SRT and VTT files next to it", async () => { + const { registration, server } = operation(); + const job = await jobContext(audioSource, { artifactId: "audio-1", language: "de" }); + const result = await registration.handler.run(job.context); + expect(server.requests).toEqual([ + expect.objectContaining({ + language: "de", + fileName: "Interview_take_2.m4a", + fileType: "audio/mp4", + }), + ]); + expect( + job.outputs.map(({ name, mimeType, parentArtifactId }) => [name, mimeType, parentArtifactId]), + ).toEqual([ + ["Interview take 2.srt", "application/x-subrip", "audio-1"], + ["Interview take 2.vtt", "text/vtt", "audio-1"], + ]); + expect(job.outputs[0]?.body).toContain("00:00:00,000 --> 00:00:02,400\nWelcome to the studio."); + expect(job.outputs[1]?.body.startsWith("WEBVTT\n")).toBe(true); + expect(job.progress.at(-1)).toBeCloseTo(0.95); + expect(result.summary).toBe( + "Transcript of Interview take 2.m4a: Welcome to the studio. Today we cut a vertical clip.\nSubtitles are attached as SRT and VTT.", + ); + }); + + it("writes a text file when the server returns no timings", async () => { + const { registration } = operation(TRANSCRIPTION, { flavor: "text" }); + const job = await jobContext(audioSource, { artifactId: "audio-1" }); + const result = await registration.handler.run(job.context); + expect(job.outputs.map((output) => [output.name, output.mimeType])).toEqual([ + ["Interview take 2.txt", "text/plain"], + ]); + expect(job.outputs[0]?.body).toBe("Welcome to the studio. Today we cut a vertical clip.\n"); + expect(result.summary).toContain("attached as text"); + }); + + it("refuses what it cannot transcribe without ffmpeg", async () => { + const video = { + ...audioSource, + artifactId: "video-1", + name: "clip.mp4", + kind: "video" as const, + }; + const { registration } = operation(); + await expect( + registration.handler.run((await jobContext(video, { artifactId: "video-1" })).context), + ).rejects.toThrow("Transcribing video needs ffmpeg on the server."); + const image = { + ...audioSource, + artifactId: "image-1", + name: "still.png", + kind: "image" as const, + }; + await expect( + registration.handler.run((await jobContext(image, { artifactId: "image-1" })).context), + ).rejects.toThrow("Only audio and video can be transcribed."); + const long = operation(TRANSCRIPTION, { maxBytes: 4 }); + await expect( + long.registration.handler.run( + (await jobContext(audioSource, { artifactId: "audio-1" })).context, + ), + ).rejects.toThrow("too long to transcribe without ffmpeg"); + }); + + it("converts with ffmpeg and transcribes long media in chunks on one timeline", async () => { + const extracted: Array<{ startMs: number; endMs: number; mimeType: string }> = []; + const audio: TranscriptionAudioTools = { + available: async () => true, + durationMs: async () => 150_000, + extract: async (_input, output, options) => { + extracted.push({ + startMs: options.startMs, + endMs: options.endMs, + mimeType: options.mimeType, + }); + await writeFile(output, new Uint8Array([1, 2, 3])); + }, + }; + // 2 MiB per upload: 62 seconds of 16 kHz mono WAV. + const { registration, server } = operation(TRANSCRIPTION, { audio, maxBytes: 2 * 1024 * 1024 }); + const video = { + ...audioSource, + artifactId: "video-1", + name: "talk.mp4", + mimeType: "video/mp4", + kind: "video" as const, + }; + const job = await jobContext(video, { artifactId: "video-1" }); + await registration.handler.run(job.context); + expect(extracted).toEqual([ + { startMs: 0, endMs: 62_000, mimeType: "video/mp4" }, + { startMs: 62_000, endMs: 124_000, mimeType: "video/mp4" }, + { startMs: 124_000, endMs: 150_000, mimeType: "video/mp4" }, + ]); + expect(server.requests.map((request) => [request.fileName, request.fileType])).toEqual([ + ["talk.wav", "audio/wav"], + ["talk.wav", "audio/wav"], + ["talk.wav", "audio/wav"], + ]); + expect(job.outputs[0]?.body).toContain("00:02:04,000 --> 00:02:06,400"); + }); + + it("waits for a transcription server that is down instead of failing", async () => { + const { registration, server } = operation(); + server.failNextStatus = 503; + const job = await jobContext(audioSource, { artifactId: "audio-1" }); + await expect(registration.handler.run(job.context)).rejects.toBeInstanceOf( + MediaJobDeferredError, + ); + expect(job.outputs).toEqual([]); + }); +}); + +describe("planAudioChunks", () => { + it("covers the whole duration in upload-sized ranges", () => { + expect(planAudioChunks(0, 25 * 1024 * 1024)).toEqual([]); + expect(planAudioChunks(30_000, 25 * 1024 * 1024)).toEqual([{ startMs: 0, endMs: 30_000 }]); + expect(planAudioChunks(1_500_000, 25 * 1024 * 1024)).toEqual([ + { startMs: 0, endMs: 720_000 }, + { startMs: 720_000, endMs: 1_440_000 }, + { startMs: 1_440_000, endMs: 1_500_000 }, + ]); + }); +}); diff --git a/packages/adapters/src/media/transcription.ts b/packages/adapters/src/media/transcription.ts index 0b081e9..41f9c70 100644 --- a/packages/adapters/src/media/transcription.ts +++ b/packages/adapters/src/media/transcription.ts @@ -1,5 +1,23 @@ -import { readFile } from "node:fs/promises"; -import type { Transcript, TranscriptionProvider, TranscriptSegment } from "@rakazo/adapter-kit"; +import { readFile, writeFile } from "node:fs/promises"; +import path from "node:path"; +import { + MediaJobDeferredError, + type Transcript, + type TranscriptionProvider, + type TranscriptSegment, +} from "@rakazo/adapter-kit"; +import { + type MediaSettings, + type TranscribeInput, + TranscribeInputSchema, + type TranscriptionSettings, +} from "@rakazo/contracts"; +import { toSrt, toVtt } from "@rakazo/core"; +import { MediaJobRejectedError, type MediaOperationRegistration } from "../media-runtime.js"; +import { + createOpenAiCompatibleTranscriptionProvider, + TranscriptionServerUnavailableError, +} from "../openai-compatible-transcription.js"; /** Bytes per second of the 16 kHz mono 16-bit WAV that audio is converted to before upload. */ export const TRANSCRIPTION_WAV_BYTES_PER_SECOND = 16_000 * 2; @@ -77,3 +95,163 @@ export function transcriptSummary(name: string, transcript: Transcript): string : "The server returned no timings, so the transcript is attached as text."; return `Transcript of ${name}: ${cut}\n${timing}`; } + +/** Retry delay while the transcription server is down or busy. */ +const UNAVAILABLE_RETRY_MS = 15_000; + +/** ffmpeg steps the operation uses when the server has ffmpeg. */ +export interface TranscriptionAudioTools { + available(): Promise; + durationMs(input: string, options: { signal: AbortSignal; mimeType: string }): Promise; + /** Writes `[startMs, endMs)` of the first audio track as 16 kHz mono WAV. */ + extract( + input: string, + output: string, + options: { signal: AbortSignal; mimeType: string; startMs: number; endMs: number }, + ): Promise; +} + +/** Time ranges of converted audio that each fit one upload. */ +export function planAudioChunks( + durationMs: number, + maxBytes: number, +): Array<{ startMs: number; endMs: number }> { + const chunkMs = transcriptionChunkSeconds(maxBytes) * 1000; + const ranges: Array<{ startMs: number; endMs: number }> = []; + for (let startMs = 0; startMs < durationMs; startMs += chunkMs) { + ranges.push({ startMs, endMs: Math.min(durationMs, startMs + chunkMs) }); + } + return ranges; +} + +export interface TranscriptionOperationDeps { + loadSettings(): Promise; + /** Without ffmpeg, audio is sent as stored and video cannot be transcribed. */ + audio?: TranscriptionAudioTools; + /** The client for the configured server; tests pass one backed by an emulator. */ + createProvider?(settings: TranscriptionSettings): TranscriptionProvider; +} + +/** + * `transcribe`: speech in an audio or video artifact becomes SRT and VTT subtitle files (or a + * text file when the server returns no timings), and the bot gets the transcript back. + */ +export function createTranscriptionOperations( + deps: TranscriptionOperationDeps, +): MediaOperationRegistration[] { + const createProvider = deps.createProvider ?? createOpenAiCompatibleTranscriptionProvider; + return [ + { + operation: "transcribe", + inputSchema: TranscribeInputSchema, + title: () => "Transcript", + available: async () => Boolean((await deps.loadSettings()).transcription), + artifactIds: (input) => [input.artifactId], + handler: { + operation: "transcribe", + async run(context) { + const input = TranscribeInputSchema.parse(context.input); + const settings = (await deps.loadSettings()).transcription; + if (!settings) throw new MediaJobRejectedError("Transcription is not set up."); + const provider = createProvider(settings); + const source = await context.fetchArtifact(input.artifactId); + if (source.kind !== "audio" && source.kind !== "video") { + throw new MediaJobRejectedError("Only audio and video can be transcribed."); + } + const audio = await prepareAudio(deps.audio, provider, source, context); + await context.progress(0.05); + let transcript: Transcript; + try { + transcript = await transcribeChunks(provider, audio.chunks, { + mimeType: audio.mimeType, + fileName: audio.fileName, + ...(input.language ? { language: input.language } : {}), + signal: context.signal, + onProgress: (fraction) => context.progress(0.05 + fraction * 0.9), + }); + } catch (error) { + // Nothing is stored yet, so the job can wait for the server to come back. + if (error instanceof TranscriptionServerUnavailableError) { + throw new MediaJobDeferredError(error.message, UNAVAILABLE_RETRY_MS); + } + throw error; + } + if (!transcript.text.trim() && transcript.segments.length === 0) { + throw new MediaJobRejectedError(`No speech found in ${source.name}.`); + } + const base = path.parse(source.name).name || "transcript"; + const files = + transcript.segments.length > 0 + ? [ + { + name: `${base}.srt`, + mimeType: "application/x-subrip", + body: toSrt(transcript.segments), + }, + { name: `${base}.vtt`, mimeType: "text/vtt", body: toVtt(transcript.segments) }, + ] + : [ + { + name: `${base}.txt`, + mimeType: "text/plain", + body: `${transcript.text.trim()}\n`, + }, + ]; + for (const [index, file] of files.entries()) { + const filePath = path.join( + context.tempDir, + `transcript-${index}${path.extname(file.name)}`, + ); + await writeFile(filePath, file.body, { mode: 0o600 }); + await context.addOutput({ + path: filePath, + name: file.name, + mimeType: file.mimeType, + parentArtifactId: source.artifactId, + }); + } + return { summary: transcriptSummary(source.name, transcript) }; + }, + }, + }, + ]; +} + +async function prepareAudio( + tools: TranscriptionAudioTools | undefined, + provider: TranscriptionProvider, + source: { path: string; name: string; mimeType: string; kind: string; size: number }, + context: { tempDir: string; signal: AbortSignal }, +): Promise<{ chunks: AudioChunk[]; mimeType: string; fileName: string }> { + const maxBytes = provider.describe().capabilities.maxBytes; + if (!tools || !(await tools.available())) { + if (source.kind === "video") { + throw new MediaJobRejectedError("Transcribing video needs ffmpeg on the server."); + } + if (source.size > maxBytes) { + throw new MediaJobRejectedError( + "This recording is too long to transcribe without ffmpeg on the server.", + ); + } + return { + chunks: [{ path: source.path, offsetMs: 0 }], + mimeType: source.mimeType, + fileName: source.name, + }; + } + // Whisper servers expect 16 kHz mono; converting also keeps long files within upload limits. + const options = { signal: context.signal, mimeType: source.mimeType }; + const durationMs = await tools.durationMs(source.path, options); + const chunks: AudioChunk[] = []; + for (const [index, range] of planAudioChunks(durationMs, maxBytes).entries()) { + const chunkPath = path.join(context.tempDir, `speech-${index}.wav`); + await tools.extract(source.path, chunkPath, { ...options, ...range }); + chunks.push({ path: chunkPath, offsetMs: range.startMs }); + } + if (chunks.length === 0) throw new MediaJobRejectedError(`No speech found in ${source.name}.`); + return { + chunks, + mimeType: "audio/wav", + fileName: `${path.parse(source.name).name || "speech"}.wav`, + }; +} diff --git a/packages/adapters/src/openai-compatible-transcription.ts b/packages/adapters/src/openai-compatible-transcription.ts index 1b71c27..866ce1a 100644 --- a/packages/adapters/src/openai-compatible-transcription.ts +++ b/packages/adapters/src/openai-compatible-transcription.ts @@ -23,6 +23,14 @@ export const OPENAI_COMPATIBLE_TRANSCRIPTION_DEFAULT_MODEL = "whisper-1"; const TRANSCRIPTION_TIMEOUT_MS = 15 * 60_000; const MAX_TRANSCRIPTION_RESPONSE_BYTES = 8 * 1024 * 1024; +/** The server is down, restarting or busy; trying again later may succeed. */ +export class TranscriptionServerUnavailableError extends Error { + constructor(message: string) { + super(message); + this.name = "TranscriptionServerUnavailableError"; + } +} + export interface OpenAiCompatibleTranscriptionConfig { baseUrl: string; model?: string; @@ -94,7 +102,11 @@ export function createOpenAiCompatibleTranscriptionProvider( ) { return { rejectedFormat: true }; } - throw new Error(httpErrorMessage(response.status, message)); + const failure = httpErrorMessage(response.status, message); + if (response.status === 429 || response.status === 502 || response.status === 503) { + throw new TranscriptionServerUnavailableError(failure); + } + throw new Error(failure); } return parseTranscriptionResponse(text, response.headers.get("content-type")); } catch (error) { @@ -102,7 +114,7 @@ export function createOpenAiCompatibleTranscriptionProvider( if (timeout.aborted) throw new Error("Transcription server timed out"); // undici reports refused or reset connections as a bare "fetch failed". if (error instanceof TypeError && error.message === "fetch failed") { - throw new Error("Transcription server is unreachable"); + throw new TranscriptionServerUnavailableError("Transcription server is unreachable"); } throw error; } diff --git a/packages/adapters/src/transcription-conformance.test.ts b/packages/adapters/src/transcription-conformance.test.ts index 9957653..dc9a643 100644 --- a/packages/adapters/src/transcription-conformance.test.ts +++ b/packages/adapters/src/transcription-conformance.test.ts @@ -7,6 +7,7 @@ import { OPENAI_COMPATIBLE_TRANSCRIPTION_DEFAULT_MODEL, OPENAI_COMPATIBLE_TRANSCRIPTION_MAX_BYTES, parseTranscriptionResponse, + TranscriptionServerUnavailableError, } from "./openai-compatible-transcription.js"; import { EMULATED_TRANSCRIPT, @@ -209,7 +210,11 @@ describe("OpenAI-compatible transcription provider", () => { }, }); await expect(unreachable.transcribe({ audio: AUDIO, mimeType: "audio/wav" })).rejects.toThrow( - "Transcription server is unreachable", + TranscriptionServerUnavailableError, + ); + server.failNextStatus = 503; + await expect(provider.transcribe({ audio: AUDIO, mimeType: "audio/wav" })).rejects.toThrow( + TranscriptionServerUnavailableError, ); });