feat: add sleep to translate-audio call

.github/workflows/integration-tests.yml

@@ -24,7 +24,7 @@ jobs:
       - name: Install dependencies
         run: npm ci
 
-      - name: Run integration tests (excluding long-running audio translation test)
+      - name: Run integration tests
         env:
           MUX_TOKEN_ID: ${{ secrets.MUX_TOKEN_ID }}
           MUX_TOKEN_SECRET: ${{ secrets.MUX_TOKEN_SECRET }}
@@ -49,7 +49,7 @@ jobs:
           MUX_TEST_ASSET_ID_WITHOUT_BURNED_IN_CAPTIONS: ${{ secrets.MUX_TEST_ASSET_ID_WITHOUT_BURNED_IN_CAPTIONS }}
           MUX_TEST_ASSET_ID_AUDIO_ONLY: ${{ secrets.MUX_TEST_ASSET_ID_AUDIO_ONLY }}
           MUX_TEST_ASSET_ID_VIOLENT_AUDIO_ONLY: ${{ secrets.MUX_TEST_ASSET_ID_VIOLENT_AUDIO_ONLY }}
-        run: npm run test:integration -- --exclude tests/integration/translate-audio.test.ts
+        run: npm run test:integration
 
       - name: Simple CLI smoke test - Summarization example
         env:

src/lib/mux-ai-error.ts

@@ -66,11 +66,36 @@ export class MuxAiError extends Error {
  * trigger the much stronger field-level scrubber on the preceding
  * model output.
  */
+/**
+ * Extract a human-readable detail from an unknown thrown value.
+ *
+ * An error thrown inside a Workflow DevKit step crosses a VM/serialization
+ * boundary before it reaches a workflow-level catch, arriving as a plain
+ * object that fails `instanceof Error` but still carries a `message` string.
+ * Duck-typing `message` avoids collapsing the real cause (e.g. "fetch failed")
+ * to "Unknown error".
+ */
+function extractErrorDetail(error: unknown): string {
+  if (error instanceof Error && error.message) {
+    return error.message;
+  }
+  if (typeof error === "string" && error.length > 0) {
+    return error;
+  }
+  if (typeof error === "object" && error !== null) {
+    const message = (error as { message?: unknown }).message;
+    if (typeof message === "string" && message.length > 0) {
+      return message;
+    }
+  }
+  return "Unknown error";
+}
+
 export function wrapError(error: unknown, message: string): never {
   if (error instanceof MuxAiError) {
     throw error;
   }
-  const rawDetail = error instanceof Error ? error.message : "Unknown error";
+  const rawDetail = extractErrorDetail(error);
   const leakReason = detectLeakReason(rawDetail);
   const shouldSuppress = leakReason === "canary" || leakReason === "prompt_tag";
   const detail = shouldSuppress ?

src/workflows/translate-audio.ts

@@ -1,3 +1,5 @@
+import { sleep } from "workflow";
+
 import env from "../env.ts";
 import { getApiKeyFromEnv } from "../lib/client-factory.ts";
 import { getLanguageCodePair, toISO639_1, toISO639_3 } from "../lib/language-codes.ts";
@@ -84,11 +86,6 @@ export interface AudioTranslationOptions extends MuxAIOptions {
 const STATIC_RENDITION_POLL_INTERVAL_MS = 5000;
 const STATIC_RENDITION_MAX_ATTEMPTS = 36; // ~3 minutes
 
-async function sleep(ms: number): Promise<void> {
-  "use step";
-  await new Promise(resolve => setTimeout(resolve, ms));
-}
-
 function getReadyAudioStaticRendition(asset: any) {
   const files = asset.static_renditions?.files as any[] | undefined;
   if (!files || files.length === 0) {
@@ -144,6 +141,18 @@ async function requestStaticRenditionCreation(
   }
 }
 
+async function retrieveAsset(
+  assetId: string,
+  credentials?: WorkflowCredentialsInput,
+): Promise<any> {
+  "use step";
+  const muxClient = await resolveMuxClient(credentials);
+  const mux = await muxClient.createClient();
+  return mux.video.assets.retrieve(assetId);
+}
+
+// Orchestration-level (not a step): the poll loop must call the durable `sleep`,
+// which suspends the workflow between retries without holding the function open.
 async function waitForAudioStaticRendition({
   assetId,
   initialAsset,
@@ -153,9 +162,6 @@ async function waitForAudioStaticRendition({
   initialAsset: any;
   credentials?: WorkflowCredentialsInput;
 }): Promise<any> {
-  "use step";
-  const muxClient = await resolveMuxClient(credentials);
-  const mux = await muxClient.createClient();
   let currentAsset = initialAsset;
 
   if (hasReadyAudioStaticRendition(currentAsset)) {
@@ -174,7 +180,7 @@ async function waitForAudioStaticRendition({
 
   for (let attempt = 1; attempt <= STATIC_RENDITION_MAX_ATTEMPTS; attempt++) {
     await sleep(STATIC_RENDITION_POLL_INTERVAL_MS);
-    currentAsset = await mux.video.assets.retrieve(assetId);
+    currentAsset = await retrieveAsset(assetId, credentials);
 
     if (hasReadyAudioStaticRendition(currentAsset)) {
       return currentAsset;
@@ -198,26 +204,15 @@ async function waitForAudioStaticRendition({
   );
 }
 
-async function fetchAudioFromMux(audioUrl: string): Promise<ArrayBuffer> {
-  "use step";
-
-  const audioResponse = await fetch(audioUrl);
-  if (!audioResponse.ok) {
-    throw new Error(`Failed to fetch audio file: ${audioResponse.statusText}`);
-  }
-
-  return audioResponse.arrayBuffer();
-}
-
 async function createElevenLabsDubbingJob({
-  audioBuffer,
+  sourceUrl,
   assetId,
   elevenLabsLangCode,
   elevenLabsSourceLangCode,
   numSpeakers,
   credentials,
 }: {
-  audioBuffer: ArrayBuffer;
+  sourceUrl: string;
   assetId: string;
   elevenLabsLangCode: string;
   elevenLabsSourceLangCode?: string;
@@ -227,10 +222,10 @@ async function createElevenLabsDubbingJob({
   "use step";
   const elevenLabsApiKey = await getApiKeyFromEnv("elevenlabs", credentials);
 
-  const audioBlob = new Blob([audioBuffer], { type: "audio/mp4" });
-
+  // Hand ElevenLabs the Mux audio URL directly so it fetches the source itself,
+  // instead of downloading the bytes onto the workflow runner and re-uploading them.
   const formData = new FormData();
-  formData.append("file", audioBlob);
+  formData.append("source_url", sourceUrl);
   formData.append("target_lang", elevenLabsLangCode);
   if (elevenLabsSourceLangCode) {
     formData.append("source_lang", elevenLabsSourceLangCode);
@@ -460,16 +455,6 @@ export async function translateAudio(
     audioUrl = await signUrl(audioUrl, playbackId, "video", undefined, credentials);
   }
 
-  // Fetch audio from Mux
-  console.warn("🎙️ Fetching audio from Mux...");
-
-  let audioBuffer: ArrayBuffer;
-  try {
-    audioBuffer = await fetchAudioFromMux(audioUrl);
-  } catch (error) {
-    wrapError(error, "Failed to fetch audio from Mux");
-  }
-
   // Create dubbing job in ElevenLabs
   console.warn("🎙️ Creating dubbing job in ElevenLabs...");
 
@@ -484,7 +469,7 @@ export async function translateAudio(
   let dubbingId: string;
   try {
     dubbingId = await createElevenLabsDubbingJob({
-      audioBuffer,
+      sourceUrl: audioUrl,
       assetId,
       elevenLabsLangCode,
       elevenLabsSourceLangCode,
@@ -496,15 +481,15 @@ export async function translateAudio(
     wrapError(error, "Failed to create ElevenLabs dubbing job");
   }
 
-  // Poll for completion
+  // Poll for completion. ElevenLabs added intermediate dubbing states
   console.warn("⏳ Waiting for dubbing to complete...");
 
-  let dubbingStatus: string = "dubbing";
+  let dubbingStatus = "dubbing";
   let pollAttempts = 0;
   const maxPollAttempts = 180; // 30 minutes at 10s intervals
   let targetLanguages: string[] = [];
 
-  while (dubbingStatus === "dubbing" && pollAttempts < maxPollAttempts) {
+  while (dubbingStatus !== "dubbed" && pollAttempts < maxPollAttempts) {
     await sleep(10000); // Wait 10 seconds
     pollAttempts++;
 
@@ -515,13 +500,13 @@ export async function translateAudio(
       });
       dubbingStatus = statusResult.status;
       targetLanguages = statusResult.targetLanguages;
-
-      if (dubbingStatus === "failed") {
-        throw new Error("ElevenLabs dubbing job failed");
-      }
     } catch (error) {
       wrapError(error, "Failed to check dubbing status");
     }
+
+    if (dubbingStatus === "failed") {
+      throw new MuxAiError("ElevenLabs reported that the dubbing job failed.", { type: "processing_error" });
+    }
   }
 
   if (dubbingStatus !== "dubbed") {