Merge pull request #3730 from HeyPuter/filipkujundzic/put-1620-video-generation-timeouts-return-500-and-page-on-call

fix: stop video generation timeouts and failures from paging (PUT-1620)
2026-09-10 15:25:49 +00:00 · 2026-09-03 22:54:58 +02:00
parent 799fc4ac3c 8b01077d44
commit 1b777cd760
23 changed files with 1343 additions and 126 deletions
@@ -263,19 +263,24 @@ describe('DriverController — concurrent acquire/release', () => {
    });

    it('does not attach release listeners when the driver declares no concurrent config', async () => {
-        // The optimisation that lets the existing test stubs in
-        // DriverController.test.ts get away without an EventEmitter-shaped
-        // `res`: skip the once() wiring entirely when there's no spec.
+        // With no spec there is nothing to release, so the gate must not
+        // wire `finish`/`close` release listeners. The client-disconnect
+        // hook is separate and always present: exactly one `close` listener.
        const driver = makeSyntheticDriver();
        (driver as { concurrent?: unknown }).concurrent = undefined;
        const controller = buildController(driver);
        const handler = captureCallHandler(controller);

-        // Bare object with no event-emitter surface — exposes the bug
-        // case where the gate would try to call `res.once`.
+        // Minimal `res` that only records which events get listeners.
+        const listened: string[] = [];
        const bareRes = {
            statusCode: 200,
            body: undefined as unknown,
+            writableFinished: false,
+            once(event: string) {
+                listened.push(event);
+                return this;
+            },
            status(code: number) {
                this.statusCode = code;
                return this;
@@ -301,5 +306,6 @@ describe('DriverController — concurrent acquire/release', () => {
            bareRes as unknown as Response,
        );
        expect(bareRes.body).toMatchObject({ success: true, result: 'pong' });
+        expect(listened).toEqual(['close']);
    });
 });
@@ -35,9 +35,10 @@
 /* eslint-disable @typescript-eslint/no-explicit-any */
 import { Readable, Writable } from 'node:stream';
 import type { Request, RequestHandler, Response } from 'express';
+import { APIConnectionTimeoutError } from 'openai';
 import { beforeEach, describe, expect, it, vi } from 'vitest';
 import type { DriverMethodLifecycleEvent } from '../../clients/event/types.js';
-import { runWithContext } from '../../core/context.js';
+import { Context, runWithContext } from '../../core/context.js';
 import { configureRateLimit } from '../../core/http/middleware/rateLimit.js';
 import { DriverController } from './DriverController.js';

@@ -325,6 +326,32 @@ describe('DriverController upstream error translation', () => {
        expect(err).toBe('a bare string');
    });

+    it('maps an SDK connection timeout, which carries no status, to a 504 upstream_timeout', async () => {
+        const raw = new APIConnectionTimeoutError();
+        const { err } = await callWith(throwing(raw));
+        expect(err).toMatchObject({
+            statusCode: 504,
+            legacyCode: 'upstream_timeout',
+            message: 'AI provider timed out',
+            cause: raw,
+        });
+    });
+
+    it('maps a fetch timeout that undici wraps in a `fetch failed` TypeError to a 504 upstream_timeout', async () => {
+        const raw = new TypeError('fetch failed', {
+            cause: Object.assign(new Error('Headers Timeout Error'), {
+                name: 'HeadersTimeoutError',
+                code: 'UND_ERR_HEADERS_TIMEOUT',
+            }),
+        });
+        const { err } = await callWith(throwing(raw));
+        expect(err).toMatchObject({
+            statusCode: 504,
+            legacyCode: 'upstream_timeout',
+            fields: { upstreamCode: 'UND_ERR_HEADERS_TIMEOUT' },
+        });
+    });
+
    it('passes an error with a sub-400 status through untranslated', async () => {
        const raw = { status: 302, message: 'redirected' };
        const { err } = await callWith(throwing(raw));
@@ -435,3 +462,53 @@ describe('DriverController per-method rate limiting', () => {
        expect(alarms).toEqual([]);
    });
 });
+
+// -- Client disconnect --------------------------------------------------
+
+describe('DriverController client disconnect', () => {
+    it('exposes an abort signal in the request context that fires when the client leaves early', async () => {
+        let seen: AbortSignal | undefined;
+        const { handler } = build({
+            run: async () => {
+                seen = Context.get('abortSignal');
+                await new Promise((r) => setImmediate(r));
+                return { ok: true };
+            },
+        });
+        const res = new MockRes();
+        const done = runWithContext({}, () =>
+            handler(
+                makeReq({ interface: 'test-iface', method: 'run' }),
+                res as unknown as Response,
+                () => {},
+            ),
+        );
+        res.destroy();
+        await done;
+
+        expect(seen).toBeInstanceOf(AbortSignal);
+        expect(seen?.aborted).toBe(true);
+    });
+
+    it('does not abort when the response simply finished', async () => {
+        let seen: AbortSignal | undefined;
+        const { handler } = build({
+            run: () => {
+                seen = Context.get('abortSignal');
+                return { ok: true };
+            },
+        });
+        const res = new MockRes();
+        await runWithContext({}, () =>
+            handler(
+                makeReq({ interface: 'test-iface', method: 'run' }),
+                res as unknown as Response,
+                () => {},
+            ),
+        );
+        res.end();
+        await new Promise((r) => setImmediate(r));
+
+        expect(seen?.aborted).toBe(false);
+    });
+});
@@ -31,6 +31,7 @@ import {
 } from '../../core/http/middleware/rateLimit.js';
 import type { PuterRouter } from '../../core/http/PuterRouter.js';
 import type { DriverMeta } from '../../drivers/meta.js';
+import { isUpstreamTimeoutError } from '../../drivers/util/upstreamErrors.js';
 import {
    isDriverStreamResult,
    resolveCallableMethods,
@@ -116,9 +117,20 @@ const translateProviderError = (err: unknown): unknown => {
        message?: string;
        error?: { code?: string; type?: string; message?: string };
        code?: string;
+        cause?: unknown;
    };
    const status = extractUpstreamStatus(e);
-    if (typeof status !== 'number') return err;
+    if (typeof status !== 'number') {
+        if (isUpstreamTimeoutError(e)) {
+            const cause = e.cause as { code?: string } | undefined;
+            return new HttpError(504, 'AI provider timed out', {
+                legacyCode: 'upstream_timeout',
+                fields: { upstreamCode: e.code ?? cause?.code },
+                cause: err,
+            });
+        }
+        return err;
+    }

    const msg = e.error?.message ?? e.message ?? 'Upstream provider error';
    const upstreamCode = e.error?.code ?? e.code;
@@ -301,8 +313,7 @@ export class DriverController extends PuterController {

        if (req.actor) {
            const permService = this.services.permission as unknown as
-                | PermissionService
-                | undefined;
+                PermissionService | undefined;
            if (permService) {
                // Build via PermissionUtil.join so any `:` in a driver or
                // interface name is escaped — raw interpolation would let a
@@ -434,6 +445,15 @@ export class DriverController extends PuterController {
        // a stale value from a prior call.
        Context.set('driverName', requestedDriver);

+        // A caller that hangs up mid-call gets nothing back, so long-running
+        // drivers watch this to stop working (and metering) as soon as it does.
+        // `close` after `finish` is the normal end of a response, not an abort.
+        const abort = new AbortController();
+        res.once('close', () => {
+            if (!res.writableFinished) abort.abort();
+        });
+        Context.set('abortSignal', abort.signal);
+
        // Per-method lifecycle events, scoped to `driver.<iface>.<method>`.
        // Subscribers can listen on `driver.*`, `driver.<iface>.*`, or the
        // exact key. `before` is emitted via `emitAndWait` so a listener may
@@ -61,6 +61,12 @@ export interface KnownContextFields {
     * `/drivers/call` dispatch); drivers read it to pick a provider.
     */
    driverName: string;
+    /**
+     * Aborts when the client disconnects before the response has finished (set
+     * by DriverController). Long-running drivers poll it so work nobody will
+     * receive stops early and is never metered.
+     */
+    abortSignal: AbortSignal;
 }

 // -- Context store ---------------------------------------------------
@@ -419,3 +419,51 @@ describe('ChatCompletionDriver unhealthy-route skipping', () => {
        expect(attempts[0]).toMatchObject({ provider: 'infron' });
    });
 });
+
+// A transport timeout carries no status, so the classifier used to lump a
+// chain of them in with "our bug" and page. It is the provider's pace.
+describe('ChatCompletionDriver timeout classification across the chain', () => {
+    // Shaped like the Stainless SDKs' timeout: no status, only the class.
+    class APIConnectionTimeoutError extends Error {
+        constructor() {
+            super('Request timed out.');
+        }
+    }
+
+    const completeShared = () =>
+        withTestActor(() =>
+            driver.complete({
+                model: 'deepseek-v4-pro',
+                messages: [{ role: 'user', content: 'hi' }],
+            }),
+        ).catch((e: unknown) => e as HttpError);
+
+    it('returns 504 upstream_timeout when every route timed out', async () => {
+        createMock.mockRejectedValue(new APIConnectionTimeoutError());
+
+        const err = await completeShared();
+
+        expect(err).toMatchObject({
+            statusCode: 504,
+            legacyCode: 'upstream_timeout',
+        });
+        const attempts = (
+            err as unknown as { fields: { attempts: { timedOut?: boolean }[] } }
+        ).fields.attempts;
+        expect(attempts.length).toBeGreaterThan(1);
+        expect(attempts.every((a) => a.timedOut === true)).toBe(true);
+    });
+
+    it('counts a timed-out route as an upstream signal, so a mixed chain is not a 500', async () => {
+        createMock
+            .mockRejectedValueOnce(new APIConnectionTimeoutError())
+            .mockRejectedValue(new Error('upstream down'));
+
+        const err = await completeShared();
+
+        expect(err).toMatchObject({
+            statusCode: 400,
+            legacyCode: 'upstream_failed',
+        });
+    });
+});
@@ -1118,6 +1118,36 @@ describe('ChatCompletionDriver.complete OpenAI-shape normalization', () => {
 // ── Fallback / error envelope ───────────────────────────────────────

 describe('ChatCompletionDriver.complete fallback and error envelope', () => {
+    it('returns HTTP 504 upstream_timeout when the only route timed out', async () => {
+        // Shaped like the Stainless SDKs' timeout: no status, only the class.
+        class APIConnectionTimeoutError extends Error {
+            constructor() {
+                super('Request timed out.');
+            }
+        }
+        vi.spyOn(FakeChatProvider.prototype, 'complete').mockRejectedValue(
+            new APIConnectionTimeoutError(),
+        );
+
+        const caught = await withTestActor(() =>
+            driver.complete({
+                model: 'fake',
+                messages: [{ role: 'user', content: 'hi' }],
+            }),
+        ).catch((e: unknown) => e as HttpError);
+
+        expect(caught).toBeInstanceOf(HttpError);
+        expect(caught).toMatchObject({
+            statusCode: 504,
+            legacyCode: 'upstream_timeout',
+            message: 'AI provider timed out',
+        });
+        const attempts = (caught as unknown as { fields: { attempts: { timedOut?: boolean }[] } })
+            .fields.attempts;
+        expect(attempts).toHaveLength(1);
+        expect(attempts[0].timedOut).toBe(true);
+    });
+
    it('returns HTTP 500 with the failure history in `fields.attempts` when all providers fail', async () => {
        vi.spyOn(FakeChatProvider.prototype, 'complete').mockRejectedValue(
            new Error('boom'),
@@ -32,6 +32,7 @@ import { NO_CREDIT_HOLD } from '../../services/metering/types.js';
 import type { DriverStreamResult } from '../meta.js';
 import { PuterDriver } from '../types.js';
 import { AI_CONCURRENT, AI_RATE_LIMIT } from '../util/aiLimits.js';
+import { isUpstreamTimeoutError } from '../util/upstreamErrors.js';
 import { AlibabaProvider } from './providers/alibaba/AlibabaProvider.js';
 import { AzureChatProvider } from './providers/azure/AzureChatProvider.js';
 import { AzureResponsesProvider } from './providers/azure/AzureResponsesProvider.js';
@@ -111,6 +112,8 @@ type ProviderAttempt = {
    status?: number;
    code?: string;
    error: string;
+    /** The attempt died to a transport timeout rather than an answer. */
+    timedOut?: boolean;
 };

 /**
@@ -146,6 +149,7 @@ const toAttempt = (
        status,
        code: e?.error?.code ?? e?.code,
        error: message,
+        ...(isUpstreamTimeoutError(err) ? { timedOut: true } : {}),
    };
 };

@@ -193,6 +197,7 @@ const routeId = (provider: string, modelId: string) => `${provider}:${modelId}`;
 * - All auth failures → 500 `upstream_auth_failed` (paged: our config)
 * - All upstream 5xx → 400 `upstream_provider_unavailable` (no page)
 * - All upstream 4xx (other) → 400 `upstream_bad_request` (no page)
+ * - All timed out → 504 `upstream_timeout` (no page)
 * - Mixed → 400 `upstream_failed` (no page)
 */
 const classifyAttempts = (
@@ -241,10 +246,18 @@ const classifyAttempts = (
        });
    }

+    if (attempts.every((a) => a.timedOut)) {
+        return new HttpError(504, 'AI provider timed out', {
+            legacyCode: 'upstream_timeout',
+            fields,
+        });
+    }
+
    // Mixed failures where at least one attempt is clearly upstream
-    // (had an HTTP status from the SDK) means "AI providers couldn't
-    // satisfy the request" — expose, don't page.
+    // (had an HTTP status from the SDK, or never got one in time) means
+    // "AI providers couldn't satisfy the request" — expose, don't page.
    const isUpstreamSignal = (a: ProviderAttempt) =>
+        a.timedOut === true ||
        a.status !== undefined ||
        isRateLimit(a) ||
        isAuthFailure(a) ||
@@ -176,7 +176,10 @@ describe('TogetherAIProvider construction', () => {
    it('constructs the Together SDK with the configured API key', () => {
        makeProvider();
        expect(togetherCtor).toHaveBeenCalledTimes(1);
-        expect(togetherCtor).toHaveBeenCalledWith({ apiKey: 'test-key' });
+        expect(togetherCtor).toHaveBeenCalledWith({
+            apiKey: 'test-key',
+            timeout: 600_000,
+        });
    });
 });

@@ -42,8 +42,11 @@ export class TogetherAIProvider implements IChatProvider {
    #kvKey = 'togetherai:models';

    constructor(config: { apiKey: string }, meteringService: MeteringService) {
+        // The SDK default is one minute, which long non-streaming
+        // completions exceed; match the ten minutes the other providers get.
        this.#together = new Together({
            apiKey: config.apiKey,
+            timeout: 10 * 60 * 1000,
        });
        this.#meteringService = meteringService;
    }
@@ -29,30 +29,15 @@ import {
    REPLICATE_IMAGE_GENERATION_MODELS,
    type ReplicateImageModel,
 } from './models.js';
+import {
+    CONTENT_FILTER_PATTERN,
+    sanitizeUpstreamMessage,
+} from '../../../util/upstreamErrors.js';

 const DEFAULT_MODEL = 'black-forest-labs/flux-schnell';
 const DEFAULT_RATIO = { w: 1024, h: 1024 };

 const PREDICTION_FAILED_PREFIX = 'Prediction failed:';
-const MAX_UPSTREAM_MESSAGE_LENGTH = 300;
-// Model-side content filters, as worded in failed-prediction errors.
-const CONTENT_FILTER_PATTERN =
-    /\bnsfw\b|flagged as sensitive|sensitive content|content policy|\bE005\b/i;
-
-/**
- * Strips markup and bounds length so an upstream HTML error page never rides
- * through into a response body or an alarm signature.
- */
-const sanitizeUpstreamMessage = (raw: string): string => {
-    const text = raw
-        .replace(/<(style|script)[\s\S]*?<\/\1>/gi, ' ')
-        .replace(/<[^>]*>/g, ' ')
-        .replace(/\s+/g, ' ')
-        .trim();
-    return text.length > MAX_UPSTREAM_MESSAGE_LENGTH
-        ? `${text.slice(0, MAX_UPSTREAM_MESSAGE_LENGTH - 3)}...`
-        : text;
-};

 export class ReplicateImageGenerationProvider implements IImageProvider {
    static readonly #CORE_PARAMS: readonly string[] = [
@@ -44,6 +44,7 @@ import { PuterServer } from '../../../../server.js';
 import { setupTestServer } from '../../../../testUtil.js';
 import { withTestActor } from '../../../integrationTestUtil.js';
 import { BYTEPLUS_VIDEO_GENERATION_MODELS } from './models.js';
+import { VIDEO_POLL_WINDOW_MS } from '../polling.js';
 import { BytePlusVideoProvider } from './BytePlusVideoProvider.js';

 // -- Test harness ----------------------------------------------------
@@ -378,6 +379,42 @@ describe('BytePlusVideoProvider.generate polling and outcomes', () => {
        expect(fetchSpy).toHaveBeenCalledTimes(4); // 1 create + 3 polls
    });

+    it('gives up after the wait window as HttpError 504 upstream_timeout, without metering', async () => {
+        vi.useFakeTimers();
+        try {
+            fetchSpy
+                .mockResolvedValueOnce(
+                    jsonResponse({ id: 'cgt-slow', status: 'queued' }),
+                )
+                .mockImplementation(async () =>
+                    jsonResponse({ id: 'cgt-slow', status: 'running' }),
+                );
+            // The default poll interval keeps the window to ~60 polls.
+            const provider = new BytePlusVideoProvider(
+                { apiKey: 'test-key' },
+                server.services.metering,
+            );
+
+            const rejection = withTestActor(() =>
+                provider.generate({ prompt: 'hi' }),
+            ).catch((e: unknown) => e);
+
+            // Ten-minute wait window, polled every 5s.
+            await vi.advanceTimersByTimeAsync(VIDEO_POLL_WINDOW_MS + 5_000);
+
+            expect(await rejection).toMatchObject({
+                statusCode: 504,
+                legacyCode: 'upstream_timeout',
+                message:
+                    'Timed out waiting for BytePlus video generation to complete',
+                fields: { provider: 'byteplus' },
+            });
+            expect(incrementUsageSpy).not.toHaveBeenCalled();
+        } finally {
+            vi.useRealTimers();
+        }
+    });
+
    it('throws 400 upstream_failed when the task fails, without metering', async () => {
        mockTaskFlow({
            id: 'cgt-test-1',
@@ -386,7 +423,28 @@ describe('BytePlusVideoProvider.generate polling and outcomes', () => {
        });
        await expect(
            withTestActor(() => makeProvider().generate({ prompt: 'hi' })),
-        ).rejects.toMatchObject({ statusCode: 400, message: 'blocked' });
+        ).rejects.toMatchObject({
+            statusCode: 400,
+            code: 'moderation_flagged',
+            message: 'blocked',
+        });
+        expect(incrementUsageSpy).not.toHaveBeenCalled();
+    });
+
+    it('throws 502 upstream_failed when the task fails on the provider side', async () => {
+        mockTaskFlow({
+            id: 'cgt-test-1',
+            status: 'failed',
+            error: { code: 'InternalServiceError', message: 'worker crashed' },
+        });
+        await expect(
+            withTestActor(() => makeProvider().generate({ prompt: 'hi' })),
+        ).rejects.toMatchObject({
+            statusCode: 502,
+            legacyCode: 'upstream_failed',
+            message: 'worker crashed',
+            fields: { provider: 'byteplus', upstreamCode: 'InternalServiceError' },
+        });
        expect(incrementUsageSpy).not.toHaveBeenCalled();
    });

@@ -23,6 +23,7 @@ import type { MeteringService } from '../../../../services/metering/MeteringServ
 import type { IGenerateVideoParams, IVideoModel } from '../../types.js';
 import { capSecondsToRemainingCredits } from '../../creditCap.js';
 import { VideoProvider } from '../VideoProvider.js';
+import { pollUntilSettled, videoJobFailure } from '../polling.js';
 import {
    BYTEPLUS_VIDEO_GENERATION_MODELS,
    BYTEPLUS_VIDEO_SPECS,
@@ -32,7 +33,6 @@ import {
 const DEFAULT_TEST_VIDEO_URL = 'https://assets.puter.site/txt2vid.mp4';
 const DEFAULT_BASE_URL = 'https://ark.ap-southeast.bytepluses.com/api/v3';
 const DEFAULT_POLL_INTERVAL_MS = 5_000;
-const DEFAULT_TIMEOUT_MS = 5 * 60 * 1000;
 const DEFAULT_MODEL = 'dreamina-seedance-2-0-mini-260615';
 // Seedance 2.0 multimodal reference accepts up to 9 reference images.
 const MAX_REFERENCE_IMAGES = 9;
@@ -202,14 +202,11 @@ export class BytePlusVideoProvider extends VideoProvider {
            const errorMessage =
                finalTask.error?.message ??
                `Video generation ${finalTask.status}`;
-            // Ark's `failed` covers both user-input issues (content
-            // moderation) and upstream outages — same ambiguity as the
-            // Together provider, so expose it the same way: a 400 with
-            // `upstream_failed` that the alarm gate skips.
-            throw new HttpError(400, errorMessage, {
-                legacyCode: 'upstream_failed',
-                fields: { provider: 'byteplus' },
-            });
+            throw videoJobFailure(
+                'byteplus',
+                errorMessage,
+                finalTask.error?.code,
+            );
        }

        const videoUrl = finalTask.content?.video_url;
@@ -336,22 +333,18 @@ export class BytePlusVideoProvider extends VideoProvider {
    }

    async #pollUntilComplete(taskId: string): Promise<ArkVideoTask> {
-        const start = Date.now();
-        for (;;) {
-            const task = (await this.#request(
-                'GET',
-                `/contents/generations/tasks/${taskId}`,
-            )) as ArkVideoTask;
-            if (task.status !== 'queued' && task.status !== 'running') {
-                return task;
-            }
-            if (Date.now() - start > DEFAULT_TIMEOUT_MS) {
-                throw new Error(
-                    'Timed out waiting for BytePlus video generation to complete',
-                );
-            }
-            await this.#delay(this.#pollIntervalMs);
-        }
+        return await pollUntilSettled<ArkVideoTask>({
+            provider: 'byteplus',
+            providerLabel: 'BytePlus',
+            intervalMs: this.#pollIntervalMs,
+            fetch: () =>
+                this.#request(
+                    'GET',
+                    `/contents/generations/tasks/${taskId}`,
+                ) as Promise<ArkVideoTask>,
+            isPending: (task) =>
+                task.status === 'queued' || task.status === 'running',
+        });
    }

    async #request(
@@ -384,10 +377,6 @@ export class BytePlusVideoProvider extends VideoProvider {
        return payload;
    }

-    async #delay(ms: number): Promise<void> {
-        return await new Promise((resolve) => setTimeout(resolve, ms));
-    }
-
    #getModel(requestedModel?: string): IVideoModel {
        const wanted = (requestedModel ?? '').trim().toLowerCase();
        const found = BYTEPLUS_VIDEO_GENERATION_MODELS.find(
@@ -45,6 +45,7 @@ import type { MeteringService } from '../../../../services/metering/MeteringServ
 import { PuterServer } from '../../../../server.js';
 import { setupTestServer } from '../../../../testUtil.js';
 import { withTestActor } from '../../../integrationTestUtil.js';
+import { VIDEO_POLL_WINDOW_MS } from '../polling.js';
 import { GeminiVideoProvider } from './GeminiVideoProvider.js';
 import { GEMINI_VIDEO_GENERATION_MODELS } from './models.js';

@@ -422,11 +423,72 @@ describe('GeminiVideoProvider.generate polling', () => {
        }
    });

-    it('throws when the operation finishes with an error', async () => {
+    it('treats a failed poll as a missed poll rather than a failed job', async () => {
+        vi.useFakeTimers();
+        try {
+            const provider = makeProvider();
+            generateVideosMock.mockResolvedValueOnce({ done: false });
+            getVideosOperationMock
+                .mockRejectedValueOnce(
+                    Object.assign(new Error('unavailable'), { status: 503 }),
+                )
+                .mockResolvedValueOnce(completedOperation());
+
+            const promise = withTestActor(() =>
+                provider.generate({
+                    prompt: 'hi',
+                    model: 'veo-3.1-generate-preview',
+                }),
+            );
+            await vi.advanceTimersByTimeAsync(10_000);
+            await vi.advanceTimersByTimeAsync(10_000);
+
+            expect(await promise).toBe('https://gemini/out.mp4');
+            expect(getVideosOperationMock).toHaveBeenCalledTimes(2);
+        } finally {
+            vi.useRealTimers();
+        }
+    });
+
+    it('gives up after the wait window as HttpError 504 upstream_timeout, without metering', async () => {
+        vi.useFakeTimers();
+        try {
+            const provider = makeProvider();
+            generateVideosMock.mockResolvedValueOnce({ done: false });
+            getVideosOperationMock.mockResolvedValue({ done: false });
+
+            const rejection = withTestActor(() =>
+                provider.generate({
+                    prompt: 'hi',
+                    model: 'veo-3.1-generate-preview',
+                }),
+            ).catch((e: unknown) => e);
+
+            // Ten-minute wait window, polled every 10s.
+            await vi.advanceTimersByTimeAsync(VIDEO_POLL_WINDOW_MS + 10_000);
+
+            expect(await rejection).toMatchObject({
+                statusCode: 504,
+                legacyCode: 'upstream_timeout',
+                message:
+                    'Timed out waiting for Gemini video generation to complete',
+                fields: { provider: 'gemini' },
+            });
+            expect(incrementUsageSpy).not.toHaveBeenCalled();
+        } finally {
+            vi.useRealTimers();
+        }
+    });
+
+    it('surfaces an operation that finishes with an error as HttpError 502 upstream_failed', async () => {
        const provider = makeProvider();
        generateVideosMock.mockResolvedValueOnce({
            done: true,
-            error: { message: 'rate limit' },
+            error: {
+                code: 13,
+                message: 'internal server issue',
+                status: 'INTERNAL',
+            },
            response: {},
        });

@@ -437,7 +499,12 @@ describe('GeminiVideoProvider.generate polling', () => {
                    model: 'veo-3.1-generate-preview',
                }),
            ),
-        ).rejects.toThrow(/rate limit/);
+        ).rejects.toMatchObject({
+            statusCode: 502,
+            legacyCode: 'upstream_failed',
+            message: 'internal server issue',
+            fields: { provider: 'gemini', upstreamCode: 'INTERNAL' },
+        });
    });

    it('throws 400 with the filter reason when raiMediaFilteredCount > 0', async () => {
@@ -460,7 +527,10 @@ describe('GeminiVideoProvider.generate polling', () => {
            ),
        ).rejects.toMatchObject({
            statusCode: 400,
+            legacyCode: 'disallowed_value',
+            code: 'moderation_flagged',
            message: expect.stringContaining('unsafe content'),
+            fields: { provider: 'gemini' },
        });
    });

@@ -29,11 +29,11 @@ import type { MeteringService } from '../../../../services/metering/MeteringServ
 import type { IGenerateVideoParams, IVideoModel } from '../../types.js';
 import { capSecondsToRemainingCredits } from '../../creditCap.js';
 import { VideoProvider } from '../VideoProvider.js';
+import { pollUntilSettled, videoJobFailure } from '../polling.js';
 import { GEMINI_VIDEO_GENERATION_MODELS, IGeminiVideoModel } from './models.js';

 const DEFAULT_TEST_VIDEO_URL = 'https://assets.puter.site/txt2vid.mp4';
 const POLL_INTERVAL_MS = 10_000;
-const DEFAULT_TIMEOUT_MS = 10 * 60 * 1000;

 const DIMENSION_MAP: Record<
    string,
@@ -230,7 +230,11 @@ export class GeminiVideoProvider extends VideoProvider {
                throw new HttpError(
                    400,
                    `Video was filtered due to ${reasons}`,
-                    { legacyCode: 'disallowed_value' },
+                    {
+                        legacyCode: 'disallowed_value',
+                        code: 'moderation_flagged',
+                        fields: { provider: 'gemini' },
+                    },
                );
            }
            throw new Error('Gemini response did not include a video');
@@ -271,27 +275,28 @@ export class GeminiVideoProvider extends VideoProvider {
    async #pollUntilComplete(
        operation: GenerateVideosOperation,
    ): Promise<GenerateVideosOperation> {
-        let op = operation;
-        const start = Date.now();
-
-        while (!op.done) {
-            if (Date.now() - start > DEFAULT_TIMEOUT_MS) {
-                throw new Error(
-                    'Timed out waiting for Gemini video generation to complete',
-                );
-            }
-
-            await this.#delay(POLL_INTERVAL_MS);
-            op = await this.#client.operations.getVideosOperation({
-                operation: op,
-            });
-        }
+        const op = await pollUntilSettled<GenerateVideosOperation>({
+            provider: 'gemini',
+            providerLabel: 'Gemini',
+            intervalMs: POLL_INTERVAL_MS,
+            initial: operation,
+            fetch: (previous) =>
+                this.#client.operations.getVideosOperation({
+                    operation: previous ?? operation,
+                }),
+            isPending: (o) => !o.done,
+        });

        if (op.error) {
            const msg =
-                (op.error as Record<string, unknown>).message ??
-                JSON.stringify(op.error);
-            throw new Error(`Gemini video generation failed: ${msg}`);
+                typeof op.error.message === 'string'
+                    ? op.error.message
+                    : JSON.stringify(op.error);
+            const code =
+                typeof op.error.status === 'string'
+                    ? op.error.status
+                    : undefined;
+            throw videoJobFailure('gemini', msg, code);
        }

        return op;
@@ -354,8 +359,4 @@ export class GeminiVideoProvider extends VideoProvider {
        }
        return undefined;
    }
-
-    async #delay(ms: number): Promise<void> {
-        return await new Promise((resolve) => setTimeout(resolve, ms));
-    }
 }
@@ -448,6 +448,42 @@ describe('OpenAIVideoProvider.generate polling', () => {
        }
    });

+    it('gives up after the wait window as HttpError 504 upstream_timeout, without metering', async () => {
+        vi.useFakeTimers();
+        try {
+            const provider = makeProvider();
+            videosCreateMock.mockResolvedValueOnce({
+                id: 'job-slow',
+                status: 'queued',
+                size: '720x1280',
+                seconds: '4',
+            });
+            videosRetrieveMock.mockResolvedValue({
+                id: 'job-slow',
+                status: 'in_progress',
+            });
+
+            const rejection = withTestActor(() =>
+                provider.generate({ prompt: 'hi', model: 'sora-2' }),
+            ).catch((e: unknown) => e);
+
+            // Five-minute wait window, polled every 5s.
+            await vi.advanceTimersByTimeAsync(5 * 60 * 1000 + 5_000);
+
+            expect(await rejection).toMatchObject({
+                statusCode: 504,
+                legacyCode: 'upstream_timeout',
+                message:
+                    'Timed out waiting for Sora video generation to complete',
+                fields: { provider: 'openai' },
+            });
+            expect(videosDownloadContentMock).not.toHaveBeenCalled();
+            expect(incrementUsageSpy).not.toHaveBeenCalled();
+        } finally {
+            vi.useRealTimers();
+        }
+    });
+
    it('surfaces failed jobs as HttpError 400 upstream_failed (not a 500 page)', async () => {
        const provider = makeProvider();
        videosCreateMock.mockResolvedValueOnce({
@@ -25,6 +25,7 @@ import type { MeteringService } from '../../../../services/metering/MeteringServ
 import type { IGenerateVideoParams, IVideoModel } from '../../types.js';
 import { capSecondsToRemainingCredits } from '../../creditCap.js';
 import { VideoProvider } from '../VideoProvider.js';
+import { pollTimeoutError } from '../polling.js';
 import { OPENAI_VIDEO_MODELS, OPENAI_VIDEO_ALLOWED_SECONDS } from './models.js';
 import { Readable } from 'stream';

@@ -216,9 +217,7 @@ export class OpenAIVideoProvider extends VideoProvider {

        while (job.status === 'queued' || job.status === 'in_progress') {
            if (Date.now() - start > DEFAULT_TIMEOUT_MS) {
-                throw new Error(
-                    'Timed out waiting for Sora video generation to complete',
-                );
+                throw pollTimeoutError('openai', 'Sora');
            }

            await this.#delay(POLL_INTERVAL_MS);
@@ -0,0 +1,297 @@
+/*
+ * Copyright (C) 2024-present Puter Technologies Inc.
+ *
+ * This file is part of Puter.
+ *
+ * Puter is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published
+ * by the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+import { afterEach, describe, expect, it, vi } from 'vitest';
+import { Context, runWithContext } from '../../../core/context.js';
+import { HttpError } from '../../../core/http/HttpError.js';
+import {
+    VIDEO_POLL_WINDOW_MS,
+    abortableDelay,
+    clientAbortedError,
+    pollTimeoutError,
+    pollUntilSettled,
+    videoJobFailure,
+} from './polling.js';
+
+afterEach(() => {
+    vi.useRealTimers();
+});
+
+describe('pollTimeoutError', () => {
+    it('builds a 504 upstream_timeout that names the provider and keeps the cause', () => {
+        const cause = new Error('last poll failed');
+        const err = pollTimeoutError('together', 'Together AI', cause);
+        expect(err).toBeInstanceOf(HttpError);
+        expect(err).toMatchObject({
+            statusCode: 504,
+            legacyCode: 'upstream_timeout',
+            message:
+                'Timed out waiting for Together AI video generation to complete',
+            fields: { provider: 'together' },
+            cause,
+        });
+    });
+});
+
+describe('clientAbortedError', () => {
+    it('builds a 400 client_aborted that names the provider', () => {
+        expect(clientAbortedError('gemini')).toMatchObject({
+            statusCode: 400,
+            legacyCode: 'client_aborted',
+            fields: { provider: 'gemini' },
+        });
+    });
+});
+
+describe('videoJobFailure', () => {
+    it('maps content-filter wording to 400 moderation_flagged', () => {
+        expect(
+            videoJobFailure('together', 'content policy violation'),
+        ).toMatchObject({
+            statusCode: 400,
+            legacyCode: 'bad_request',
+            code: 'moderation_flagged',
+            message: 'content policy violation',
+            fields: { provider: 'together' },
+        });
+    });
+
+    it('reads the filter from the code when the message does not say', () => {
+        expect(
+            videoJobFailure(
+                'byteplus',
+                'blocked',
+                'OutputVideoSensitiveContentDetected',
+            ),
+        ).toMatchObject({
+            statusCode: 400,
+            code: 'moderation_flagged',
+            fields: {
+                provider: 'byteplus',
+                upstreamCode: 'OutputVideoSensitiveContentDetected',
+            },
+        });
+    });
+
+    it('maps a rejected parameter to 400 upstream_bad_request', () => {
+        expect(
+            videoJobFailure(
+                'together',
+                'fps is not supported by this model',
+                'unsupportedParameter',
+            ),
+        ).toMatchObject({
+            statusCode: 400,
+            legacyCode: 'upstream_bad_request',
+        });
+    });
+
+    it('reads Together-style camelCase input codes as rejections', () => {
+        for (const [code, message] of [
+            ['invalidDuration', "Invalid value for 'seconds' parameter."],
+            [
+                'missingFrameImagesForImageToVideoModel',
+                'Missing required parameter for Kling 2.1 Standard image-to-video model',
+            ],
+        ]) {
+            expect(videoJobFailure('together', message, code)).toMatchObject({
+                statusCode: 400,
+                legacyCode: 'upstream_bad_request',
+                fields: { upstreamCode: code },
+            });
+        }
+    });
+
+    it('maps anything else to 502 upstream_failed and strips markup', () => {
+        const err = videoJobFailure(
+            'gemini',
+            '<html><body><h1>Bad gateway</h1></body></html>',
+        );
+        expect(err).toMatchObject({
+            statusCode: 502,
+            legacyCode: 'upstream_failed',
+            message: 'Bad gateway',
+        });
+    });
+
+    it('falls back to a generic message when the provider sent none', () => {
+        expect(videoJobFailure('gemini', '').message).toBe(
+            'Video generation failed',
+        );
+    });
+});
+
+describe('abortableDelay', () => {
+    it('resolves early when the signal aborts', async () => {
+        vi.useFakeTimers();
+        const abort = new AbortController();
+        const delay = abortableDelay(60_000, abort.signal);
+        abort.abort();
+        await expect(delay).resolves.toBeUndefined();
+    });
+
+    it('resolves after the delay otherwise', async () => {
+        vi.useFakeTimers();
+        let done = false;
+        void abortableDelay(1_000).then(() => {
+            done = true;
+        });
+        await vi.advanceTimersByTimeAsync(999);
+        expect(done).toBe(false);
+        await vi.advanceTimersByTimeAsync(1);
+        expect(done).toBe(true);
+    });
+});
+
+describe('pollUntilSettled', () => {
+    type Job = { status: 'pending' | 'done' };
+    const pending: Job = { status: 'pending' };
+    const done: Job = { status: 'done' };
+
+    const poll = (
+        fetch: (previous: Job | undefined) => Promise<Job>,
+        extra: { initial?: Job; windowMs?: number } = {},
+    ) =>
+        runWithContext({}, () =>
+            pollUntilSettled<Job>({
+                provider: 'together',
+                providerLabel: 'Together AI',
+                intervalMs: 5_000,
+                fetch,
+                isPending: (job) => job.status === 'pending',
+                ...extra,
+            }),
+        );
+
+    it('fetches immediately, then on the interval, until the job settles', async () => {
+        vi.useFakeTimers();
+        const fetch = vi
+            .fn<(p: Job | undefined) => Promise<Job>>()
+            .mockResolvedValueOnce(pending)
+            .mockResolvedValueOnce(pending)
+            .mockResolvedValueOnce(done);
+
+        const result = poll(fetch);
+        await vi.advanceTimersByTimeAsync(0);
+        expect(fetch).toHaveBeenCalledTimes(1);
+        await vi.advanceTimersByTimeAsync(5_000);
+        await vi.advanceTimersByTimeAsync(5_000);
+
+        expect(await result).toBe(done);
+        expect(fetch).toHaveBeenCalledTimes(3);
+    });
+
+    it('waits one interval before re-fetching a pending initial state', async () => {
+        vi.useFakeTimers();
+        const fetch = vi
+            .fn<(p: Job | undefined) => Promise<Job>>()
+            .mockResolvedValueOnce(done);
+
+        const result = poll(fetch, { initial: pending });
+        await vi.advanceTimersByTimeAsync(0);
+        expect(fetch).not.toHaveBeenCalled();
+        await vi.advanceTimersByTimeAsync(5_000);
+
+        expect(await result).toBe(done);
+        expect(fetch).toHaveBeenCalledWith(pending);
+    });
+
+    it('treats a transient poll failure as a missed poll', async () => {
+        vi.useFakeTimers();
+        const fetch = vi
+            .fn<(p: Job | undefined) => Promise<Job>>()
+            .mockRejectedValueOnce(
+                Object.assign(new Error('bad gateway'), { status: 502 }),
+            )
+            .mockResolvedValueOnce(done);
+
+        const result = poll(fetch);
+        await vi.advanceTimersByTimeAsync(5_000);
+
+        expect(await result).toBe(done);
+        expect(fetch).toHaveBeenCalledTimes(2);
+    });
+
+    it('rethrows a poll failure that is the provider\'s verdict', async () => {
+        const notFound = Object.assign(new Error('no such job'), {
+            status: 404,
+        });
+        await expect(poll(() => Promise.reject(notFound))).rejects.toBe(
+            notFound,
+        );
+    });
+
+    it('gives up after the window with a 504 carrying the last poll error', async () => {
+        vi.useFakeTimers();
+        const flaky = Object.assign(new Error('still down'), { status: 503 });
+        const fetch = vi
+            .fn<(p: Job | undefined) => Promise<Job>>()
+            .mockRejectedValue(flaky);
+
+        const rejection = poll(fetch).catch((e: unknown) => e);
+        await vi.advanceTimersByTimeAsync(VIDEO_POLL_WINDOW_MS + 5_000);
+
+        expect(await rejection).toMatchObject({
+            statusCode: 504,
+            legacyCode: 'upstream_timeout',
+            cause: flaky,
+        });
+    });
+
+    it('honours a shorter window when one is given', async () => {
+        vi.useFakeTimers();
+        const rejection = poll(() => Promise.resolve(pending), {
+            windowMs: 20_000,
+        }).catch((e: unknown) => e);
+        await vi.advanceTimersByTimeAsync(25_000);
+
+        expect(await rejection).toMatchObject({ statusCode: 504 });
+    });
+
+    it('stops as soon as the request context signals a client abort', async () => {
+        vi.useFakeTimers();
+        const abort = new AbortController();
+        const fetch = vi
+            .fn<(p: Job | undefined) => Promise<Job>>()
+            .mockResolvedValue(pending);
+
+        const rejection = runWithContext({}, () => {
+            Context.set('abortSignal', abort.signal);
+            return pollUntilSettled<Job>({
+                provider: 'gemini',
+                providerLabel: 'Gemini',
+                intervalMs: 10_000,
+                fetch,
+                isPending: (job) => job.status === 'pending',
+            });
+        }).catch((e: unknown) => e);
+
+        await vi.advanceTimersByTimeAsync(10_000);
+        abort.abort();
+        await vi.advanceTimersByTimeAsync(0);
+
+        expect(await rejection).toMatchObject({
+            statusCode: 400,
+            legacyCode: 'client_aborted',
+            fields: { provider: 'gemini' },
+        });
+        // The abort cut the second wait short; no third poll went out.
+        expect(fetch).toHaveBeenCalledTimes(2);
+    });
+});
@@ -0,0 +1,163 @@
+/*
+ * Copyright (C) 2024-present Puter Technologies Inc.
+ *
+ * This file is part of Puter.
+ *
+ * Puter is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published
+ * by the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+import { Context } from '../../../core/context.js';
+import { HttpError } from '../../../core/http/HttpError.js';
+import {
+    CONTENT_FILTER_PATTERN,
+    isTransientUpstreamError,
+    sanitizeUpstreamMessage,
+} from '../../util/upstreamErrors.js';
+
+/** How long a provider's job may run before we stop waiting for it. */
+export const VIDEO_POLL_WINDOW_MS = 10 * 60 * 1000;
+
+// Provider wording for input the model does not accept, including Together's
+// camelCase codes (`invalidDuration`, `missingFrameImagesForImageToVideoModel`).
+const INVALID_INPUT_PATTERN =
+    /\b(unsupported|invalid|missing)\w*|not supported|required parameter/i;
+
+/**
+ * The upstream job outlived the wait window. That is the provider's pace, not a
+ * fault of ours, so it leaves as a 504 tagged `upstream_timeout`, which the
+ * alarm gate skips.
+ */
+export const pollTimeoutError = (
+    provider: string,
+    providerLabel: string,
+    cause?: unknown,
+): HttpError =>
+    new HttpError(
+        504,
+        `Timed out waiting for ${providerLabel} video generation to complete`,
+        {
+            legacyCode: 'upstream_timeout',
+            fields: { provider },
+            ...(cause !== undefined ? { cause } : {}),
+        },
+    );
+
+/** The caller hung up mid-job; nobody is left to bill or answer. */
+export const clientAbortedError = (provider: string): HttpError =>
+    new HttpError(400, 'Client disconnected before video generation finished', {
+        legacyCode: 'client_aborted',
+        fields: { provider },
+    });
+
+/**
+ * Classifies a job the provider reports as failed. A content-filter refusal is
+ * the caller's to act on; a rejected parameter is theirs to fix; anything else
+ * failed on the provider's side.
+ */
+export const videoJobFailure = (
+    provider: string,
+    message: string,
+    code?: string,
+): HttpError => {
+    const detail =
+        sanitizeUpstreamMessage(message) || 'Video generation failed';
+    const haystack = `${code ?? ''} ${detail}`;
+    const fields = { provider, upstreamCode: code };
+    if (CONTENT_FILTER_PATTERN.test(haystack)) {
+        return new HttpError(400, detail, {
+            legacyCode: 'bad_request',
+            code: 'moderation_flagged',
+            fields,
+        });
+    }
+    if (INVALID_INPUT_PATTERN.test(haystack)) {
+        return new HttpError(400, detail, {
+            legacyCode: 'upstream_bad_request',
+            fields,
+        });
+    }
+    return new HttpError(502, detail, {
+        legacyCode: 'upstream_failed',
+        fields,
+    });
+};
+
+/** Resolves after `ms`, or as soon as `signal` aborts. */
+export const abortableDelay = (
+    ms: number,
+    signal?: AbortSignal,
+): Promise<void> =>
+    new Promise((resolve) => {
+        if (signal?.aborted) return resolve();
+        const done = () => {
+            clearTimeout(timer);
+            signal?.removeEventListener('abort', done);
+            resolve();
+        };
+        const timer = setTimeout(done, ms);
+        signal?.addEventListener('abort', done, { once: true });
+    });
+
+export interface PollOptions<T> {
+    /** Goes into `fields.provider` on every error raised here. */
+    provider: string;
+    /** Human-readable name for the timeout message. */
+    providerLabel: string;
+    intervalMs: number;
+    windowMs?: number;
+    /** The state the create call already returned, if it returned one. */
+    initial?: T;
+    /** Fetches the current job state; receives the last state seen. */
+    fetch: (previous: T | undefined) => Promise<T>;
+    isPending: (job: T) => boolean;
+}
+
+/**
+ * Polls a provider job until it settles. A transient poll failure is a missed
+ * poll, not a failed job; a client disconnect ends the wait before anything is
+ * metered; the window closes with a 504 that carries the last poll error.
+ */
+export const pollUntilSettled = async <T>(opts: PollOptions<T>): Promise<T> => {
+    const windowMs = opts.windowMs ?? VIDEO_POLL_WINDOW_MS;
+    const signal = Context.get('abortSignal');
+    const start = Date.now();
+    let job = opts.initial;
+    let lastError: unknown;
+    let firstFetch = job === undefined;
+
+    for (;;) {
+        if (job !== undefined && !opts.isPending(job)) return job;
+        if (signal?.aborted) throw clientAbortedError(opts.provider);
+        if (Date.now() - start > windowMs) {
+            throw pollTimeoutError(
+                opts.provider,
+                opts.providerLabel,
+                lastError,
+            );
+        }
+        if (!firstFetch) {
+            await abortableDelay(opts.intervalMs, signal);
+            // The delay ends early on abort; do not spend a poll on it.
+            if (signal?.aborted) throw clientAbortedError(opts.provider);
+        }
+        firstFetch = false;
+        try {
+            job = await opts.fetch(job);
+            lastError = undefined;
+        } catch (err) {
+            if (!isTransientUpstreamError(err)) throw err;
+            lastError = err;
+        }
+    }
+};
@@ -40,10 +40,12 @@ import {
    type MockInstance,
 } from 'vitest';

+import { Context } from '../../../../core/context.js';
 import type { MeteringService } from '../../../../services/metering/MeteringService.js';
 import { PuterServer } from '../../../../server.js';
 import { setupTestServer } from '../../../../testUtil.js';
 import { withTestActor } from '../../../integrationTestUtil.js';
+import { VIDEO_POLL_WINDOW_MS } from '../polling.js';
 import { TogetherVideoProvider } from './TogetherVideoProvider.js';
 import { TOGETHER_VIDEO_GENERATION_MODELS } from './models.js';

@@ -118,7 +120,10 @@ describe('TogetherVideoProvider construction', () => {
    it('constructs the Together SDK with the configured api key', () => {
        makeProvider();
        expect(togetherCtor).toHaveBeenCalledTimes(1);
-        expect(togetherCtor).toHaveBeenCalledWith({ apiKey: 'test-key' });
+        expect(togetherCtor).toHaveBeenCalledWith({
+            apiKey: 'test-key',
+            timeout: 60_000,
+        });
    });

    it('throws when no apiKey is supplied', () => {
@@ -409,7 +414,96 @@ describe('TogetherVideoProvider.generate polling', () => {
        }
    });

-    it('surfaces failed jobs as HttpError 400 upstream_failed (not 500) so they do not page', async () => {
+    it('gives up after the wait window as HttpError 504 upstream_timeout, without metering', async () => {
+        vi.useFakeTimers();
+        try {
+            const provider = makeProvider();
+            videosCreateMock.mockResolvedValueOnce({ id: 'job-slow' });
+            videosRetrieveMock.mockResolvedValue({
+                id: 'job-slow',
+                status: 'in_progress',
+            });
+
+            const rejection = withTestActor(() =>
+                provider.generate({ prompt: 'hi' }),
+            ).catch((e: unknown) => e);
+
+            // Ten-minute wait window, polled every 5s.
+            await vi.advanceTimersByTimeAsync(VIDEO_POLL_WINDOW_MS + 5_000);
+
+            expect(await rejection).toMatchObject({
+                statusCode: 504,
+                legacyCode: 'upstream_timeout',
+                message:
+                    'Timed out waiting for Together AI video generation to complete',
+                fields: { provider: 'together' },
+            });
+            expect(incrementUsageSpy).not.toHaveBeenCalled();
+        } finally {
+            vi.useRealTimers();
+        }
+    });
+
+    it('treats a failed poll as a missed poll rather than a failed job', async () => {
+        vi.useFakeTimers();
+        try {
+            const provider = makeProvider();
+            videosCreateMock.mockResolvedValueOnce({ id: 'job-flaky' });
+            videosRetrieveMock
+                .mockRejectedValueOnce(
+                    Object.assign(new Error('bad gateway'), { status: 502 }),
+                )
+                .mockResolvedValueOnce({
+                    id: 'job-flaky',
+                    status: 'completed',
+                    outputs: { video_url: 'https://together/flaky.mp4' },
+                });
+
+            const promise = withTestActor(() =>
+                provider.generate({ prompt: 'hi' }),
+            );
+            await vi.advanceTimersByTimeAsync(5_000);
+
+            expect(await promise).toBe('https://together/flaky.mp4');
+            expect(videosRetrieveMock).toHaveBeenCalledTimes(2);
+        } finally {
+            vi.useRealTimers();
+        }
+    });
+
+    it('stops polling and skips metering when the client disconnects', async () => {
+        vi.useFakeTimers();
+        try {
+            const provider = makeProvider();
+            videosCreateMock.mockResolvedValueOnce({ id: 'job-gone' });
+            videosRetrieveMock.mockResolvedValue({
+                id: 'job-gone',
+                status: 'in_progress',
+            });
+            const abort = new AbortController();
+
+            const rejection = withTestActor(() => {
+                Context.set('abortSignal', abort.signal);
+                return provider.generate({ prompt: 'hi' });
+            }).catch((e: unknown) => e);
+
+            await vi.advanceTimersByTimeAsync(5_000);
+            abort.abort();
+            await vi.advanceTimersByTimeAsync(0);
+
+            expect(await rejection).toMatchObject({
+                statusCode: 400,
+                legacyCode: 'client_aborted',
+                fields: { provider: 'together' },
+            });
+            expect(videosRetrieveMock).toHaveBeenCalledTimes(2);
+            expect(incrementUsageSpy).not.toHaveBeenCalled();
+        } finally {
+            vi.useRealTimers();
+        }
+    });
+
+    it('surfaces a content-policy refusal as 400 moderation_flagged', async () => {
        const provider = makeProvider();
        videosCreateMock.mockResolvedValueOnce({ id: 'job-3' });
        videosRetrieveMock.mockResolvedValueOnce({
@@ -422,9 +516,54 @@ describe('TogetherVideoProvider.generate polling', () => {
            withTestActor(() => provider.generate({ prompt: 'hi' })),
        ).rejects.toMatchObject({
            statusCode: 400,
-            legacyCode: 'upstream_failed',
+            legacyCode: 'bad_request',
+            code: 'moderation_flagged',
            message: 'content policy violation',
+            fields: { provider: 'together' },
        });
+        expect(incrementUsageSpy).not.toHaveBeenCalled();
+    });
+
+    it('surfaces a rejected parameter as 400 upstream_bad_request', async () => {
+        const provider = makeProvider();
+        videosCreateMock.mockResolvedValueOnce({ id: 'job-4' });
+        videosRetrieveMock.mockResolvedValueOnce({
+            id: 'job-4',
+            status: 'failed',
+            error: {
+                code: 'unsupportedParameter',
+                message: 'fps is not supported by this model',
+            },
+        });
+
+        await expect(
+            withTestActor(() => provider.generate({ prompt: 'hi' })),
+        ).rejects.toMatchObject({
+            statusCode: 400,
+            legacyCode: 'upstream_bad_request',
+            message: 'fps is not supported by this model',
+            fields: { provider: 'together', upstreamCode: 'unsupportedParameter' },
+        });
+    });
+
+    it('surfaces any other failed job as 502 upstream_failed, without metering', async () => {
+        const provider = makeProvider();
+        videosCreateMock.mockResolvedValueOnce({ id: 'job-5' });
+        videosRetrieveMock.mockResolvedValueOnce({
+            id: 'job-5',
+            status: 'failed',
+            error: { message: 'worker crashed' },
+        });
+
+        await expect(
+            withTestActor(() => provider.generate({ prompt: 'hi' })),
+        ).rejects.toMatchObject({
+            statusCode: 502,
+            legacyCode: 'upstream_failed',
+            message: 'worker crashed',
+            fields: { provider: 'together' },
+        });
+        expect(incrementUsageSpy).not.toHaveBeenCalled();
    });

    it('throws when a finished job has no video_url', async () => {
@@ -23,11 +23,12 @@ import { HttpError } from '../../../../core/http/HttpError.js';
 import type { MeteringService } from '../../../../services/metering/MeteringService.js';
 import type { IGenerateVideoParams, IVideoModel } from '../../types.js';
 import { VideoProvider } from '../VideoProvider.js';
+import { pollUntilSettled, videoJobFailure } from '../polling.js';
 import { TOGETHER_VIDEO_GENERATION_MODELS } from './models.js';

 const DEFAULT_TEST_VIDEO_URL = 'https://assets.puter.site/txt2vid.mp4';
 const POLL_INTERVAL_MS = 5_000;
-const DEFAULT_TIMEOUT_MS = 5 * 60 * 1000;
+const REQUEST_TIMEOUT_MS = 60 * 1000;
 const DEFAULT_MODEL = 'minimax/video-01-director';
 const DEFAULT_DURATION_SECONDS = 6;

@@ -40,7 +41,12 @@ export class TogetherVideoProvider extends VideoProvider {
        if (!config.apiKey) {
            throw new Error('Together AI video generation requires an API key');
        }
-        this.#client = new Together({ apiKey: config.apiKey });
+        // Bounds each create/retrieve call; a slow poll is retried by the
+        // loop rather than failing the job.
+        this.#client = new Together({
+            apiKey: config.apiKey,
+            timeout: REQUEST_TIMEOUT_MS,
+        });
        this.#meteringService = meteringService;
    }

@@ -190,24 +196,22 @@ export class TogetherVideoProvider extends VideoProvider {

        if (finalJob.status === 'failed') {
            const errorMessage =
-                finalJob?.error.message ??
+                finalJob?.error?.message ??
                finalJob?.info?.errors?.[0]?.message ??
                finalJob?.info?.errors?.message ??
                finalJob?.info?.errors ??
                'Video generation failed';
-            // Together returns `failed` for both user-input issues
-            // (content policy / unsupported params) and their own
-            // outages — we can't reliably tell from the payload, so
-            // expose as 4xx with `upstream_failed`. The alarm gate
-            // skips `upstream_*` legacy codes so this no longer pages.
-            throw new HttpError(400, errorMessage, {
-                legacyCode: 'upstream_failed',
-                fields: { provider: 'together' },
-            });
+            throw videoJobFailure(
+                'together',
+                typeof errorMessage === 'string'
+                    ? errorMessage
+                    : JSON.stringify(errorMessage),
+                finalJob?.error?.code,
+            );
        }

        if (finalJob.status === 'cancelled') {
-            throw new Error('Video generation was cancelled');
+            throw videoJobFailure('together', 'Video generation was cancelled');
        }

        const usageKey = `together-video:${model}`;
@@ -228,25 +232,14 @@ export class TogetherVideoProvider extends VideoProvider {

    async #pollUntilComplete(jobId: string): Promise<any> {
        // any here because sdk types are wrong https://docs.together.ai/docs/videos-overview -> "Job Status Reference"
-        let job = await (this.#client as any).videos.retrieve(jobId);
-        const start = Date.now();
-
-        while (job.status === 'queued' || job.status === 'in_progress') {
-            if (Date.now() - start > DEFAULT_TIMEOUT_MS) {
-                throw new Error(
-                    'Timed out waiting for Together AI video generation to complete',
-                );
-            }
-
-            await this.#delay(POLL_INTERVAL_MS);
-            job = await (this.#client as any).videos.retrieve(jobId);
-        }
-
-        return job;
-    }
-
-    async #delay(ms: number): Promise<void> {
-        return await new Promise((resolve) => setTimeout(resolve, ms));
+        return await pollUntilSettled<any>({
+            provider: 'together',
+            providerLabel: 'Together AI',
+            intervalMs: POLL_INTERVAL_MS,
+            fetch: () => (this.#client as any).videos.retrieve(jobId),
+            isPending: (job) =>
+                job.status === 'queued' || job.status === 'in_progress',
+        });
    }

    async #getModel(requestedModel?: string): Promise<IVideoModel | undefined> {
@@ -0,0 +1,129 @@
+/*
+ * Copyright (C) 2024-present Puter Technologies Inc.
+ *
+ * This file is part of Puter.
+ *
+ * Puter is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published
+ * by the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+import { APIConnectionError, APIConnectionTimeoutError } from 'openai';
+import { describe, expect, it } from 'vitest';
+import { HttpError } from '../../core/http/HttpError.js';
+import {
+    CONTENT_FILTER_PATTERN,
+    isTransientUpstreamError,
+    isUpstreamTimeoutError,
+    sanitizeUpstreamMessage,
+} from './upstreamErrors.js';
+
+const withStatus = (status: number) =>
+    Object.assign(new Error(`status ${status}`), { status });
+
+describe('isUpstreamTimeoutError', () => {
+    it('recognises the Stainless SDK timeout class', () => {
+        expect(isUpstreamTimeoutError(new APIConnectionTimeoutError())).toBe(
+            true,
+        );
+    });
+
+    it('recognises an undici timeout wrapped in a fetch failed TypeError', () => {
+        const err = new TypeError('fetch failed', {
+            cause: Object.assign(new Error('Headers Timeout Error'), {
+                name: 'HeadersTimeoutError',
+                code: 'UND_ERR_HEADERS_TIMEOUT',
+            }),
+        });
+        expect(isUpstreamTimeoutError(err)).toBe(true);
+    });
+
+    it('recognises an AbortSignal.timeout rejection by name', () => {
+        expect(isUpstreamTimeoutError({ name: 'TimeoutError' })).toBe(true);
+    });
+
+    it('ignores ordinary errors, non-objects, and status-bearing failures', () => {
+        expect(isUpstreamTimeoutError(new Error('boom'))).toBe(false);
+        expect(isUpstreamTimeoutError('timed out')).toBe(false);
+        expect(isUpstreamTimeoutError(withStatus(504))).toBe(false);
+    });
+});
+
+describe('isTransientUpstreamError', () => {
+    it('treats timeouts and dropped connections as transient', () => {
+        expect(isTransientUpstreamError(new APIConnectionTimeoutError())).toBe(
+            true,
+        );
+        expect(
+            isTransientUpstreamError(new APIConnectionError({ message: 'x' })),
+        ).toBe(true);
+        expect(
+            isTransientUpstreamError(
+                Object.assign(new Error('reset'), { code: 'ECONNRESET' }),
+            ),
+        ).toBe(true);
+    });
+
+    it('treats 5xx, 408 and 429 statuses as transient, however they are carried', () => {
+        expect(isTransientUpstreamError(withStatus(503))).toBe(true);
+        expect(isTransientUpstreamError(withStatus(429))).toBe(true);
+        expect(isTransientUpstreamError(withStatus(408))).toBe(true);
+        expect(isTransientUpstreamError(new HttpError(502, 'x'))).toBe(true);
+        expect(
+            isTransientUpstreamError({ response: { status: 500 } }),
+        ).toBe(true);
+    });
+
+    it('does not treat other 4xx responses or plain errors as transient', () => {
+        expect(isTransientUpstreamError(withStatus(404))).toBe(false);
+        expect(isTransientUpstreamError(withStatus(400))).toBe(false);
+        expect(isTransientUpstreamError(new HttpError(400, 'x'))).toBe(false);
+        expect(isTransientUpstreamError(new Error('boom'))).toBe(false);
+    });
+});
+
+describe('CONTENT_FILTER_PATTERN', () => {
+    it('matches the wording providers use for refusals', () => {
+        for (const text of [
+            'Error generating image: NSFW content detected.',
+            'OutputVideoSensitiveContentDetected',
+            'content policy violation',
+            'blocked by our safety filters',
+            'The input or output was flagged as sensitive. (E005)',
+        ]) {
+            expect(text).toMatch(CONTENT_FILTER_PATTERN);
+        }
+    });
+
+    it('does not match ordinary failures', () => {
+        expect('q_descale must have shape (batch_size, num_heads_k)').not.toMatch(
+            CONTENT_FILTER_PATTERN,
+        );
+        expect('internal server issue').not.toMatch(CONTENT_FILTER_PATTERN);
+    });
+});
+
+describe('sanitizeUpstreamMessage', () => {
+    it('strips markup and collapses whitespace', () => {
+        expect(
+            sanitizeUpstreamMessage(
+                '<html><style>body{color:red}</style><h1>Bad</h1>\n<p>gateway</p></html>',
+            ),
+        ).toBe('Bad gateway');
+    });
+
+    it('bounds the length', () => {
+        const out = sanitizeUpstreamMessage('x'.repeat(1000));
+        expect(out.length).toBe(300);
+        expect(out.endsWith('...')).toBe(true);
+    });
+});
@@ -0,0 +1,129 @@
+/*
+ * Copyright (C) 2024-present Puter Technologies Inc.
+ *
+ * This file is part of Puter.
+ *
+ * Puter is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published
+ * by the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+
+// -- Transport failures --
+
+// Transport timeouts carry no HTTP status. The Stainless SDKs (OpenAI,
+// Together, Gemini) throw `APIConnectionTimeoutError`; undici wraps its own
+// in a `fetch failed` TypeError whose cause carries the code.
+const TIMEOUT_ERROR_NAMES = new Set([
+    'APIConnectionTimeoutError',
+    'TimeoutError',
+    'ConnectTimeoutError',
+    'HeadersTimeoutError',
+    'BodyTimeoutError',
+]);
+const TIMEOUT_ERROR_CODES = new Set([
+    'ETIMEDOUT',
+    'UND_ERR_CONNECT_TIMEOUT',
+    'UND_ERR_HEADERS_TIMEOUT',
+    'UND_ERR_BODY_TIMEOUT',
+]);
+const CONNECTION_ERROR_NAMES = new Set(['APIConnectionError', 'SocketError']);
+const CONNECTION_ERROR_CODES = new Set([
+    'ECONNRESET',
+    'ECONNREFUSED',
+    'EPIPE',
+    'EAI_AGAIN',
+    'UND_ERR_SOCKET',
+]);
+const TRANSIENT_STATUSES = new Set([408, 429]);
+
+interface ErrorShape {
+    name?: unknown;
+    code?: unknown;
+    cause?: unknown;
+    status?: unknown;
+    statusCode?: unknown;
+    response?: { status?: unknown };
+    constructor?: { name?: string };
+}
+
+const asShape = (e: unknown): ErrorShape | undefined =>
+    e && typeof e === 'object' ? (e as ErrorShape) : undefined;
+
+const matches = (
+    e: ErrorShape,
+    names: Set<string>,
+    codes: Set<string>,
+): boolean =>
+    (typeof e.name === 'string' && names.has(e.name)) ||
+    names.has(e.constructor?.name ?? '') ||
+    (typeof e.code === 'string' && codes.has(e.code));
+
+const upstreamStatus = (e: ErrorShape): number | undefined => {
+    const s = e.status ?? e.statusCode ?? e.response?.status;
+    return typeof s === 'number' ? s : undefined;
+};
+
+/** A request to an upstream provider ran out of time before it answered. */
+export const isUpstreamTimeoutError = (err: unknown): boolean => {
+    const e = asShape(err);
+    if (!e) return false;
+    if (matches(e, TIMEOUT_ERROR_NAMES, TIMEOUT_ERROR_CODES)) return true;
+    const cause = asShape(e.cause);
+    return (
+        cause !== undefined &&
+        matches(cause, TIMEOUT_ERROR_NAMES, TIMEOUT_ERROR_CODES)
+    );
+};
+
+/**
+ * A failure worth retrying on the next poll: a timeout, a dropped connection,
+ * or a status the provider itself treats as temporary (408, 429, 5xx). Any
+ * other 4xx is the provider's verdict on the request and is not transient.
+ */
+export const isTransientUpstreamError = (err: unknown): boolean => {
+    if (isUpstreamTimeoutError(err)) return true;
+    const e = asShape(err);
+    if (!e) return false;
+    const status = upstreamStatus(e);
+    if (status !== undefined) {
+        return status >= 500 || TRANSIENT_STATUSES.has(status);
+    }
+    if (matches(e, CONNECTION_ERROR_NAMES, CONNECTION_ERROR_CODES)) return true;
+    const cause = asShape(e.cause);
+    return (
+        cause !== undefined &&
+        matches(cause, CONNECTION_ERROR_NAMES, CONNECTION_ERROR_CODES)
+    );
+};
+
+// -- Upstream messages --
+
+const MAX_UPSTREAM_MESSAGE_LENGTH = 300;
+
+/** Model-side content filters, as worded in provider failure payloads. */
+export const CONTENT_FILTER_PATTERN =
+    /\bnsfw\b|sensitive|content[\s_-]?policy|moderation|safety|\bunsafe\b|\bflagged\b|prohibited|\bE005\b/i;
+
+/**
+ * Strips markup and bounds length so an upstream HTML error page never rides
+ * through into a response body or an alarm signature.
+ */
+export const sanitizeUpstreamMessage = (raw: string): string => {
+    const text = raw
+        .replace(/<(style|script)[\s\S]*?<\/\1>/gi, ' ')
+        .replace(/<[^>]*>/g, ' ')
+        .replace(/\s+/g, ' ')
+        .trim();
+    return text.length > MAX_UPSTREAM_MESSAGE_LENGTH
+        ? `${text.slice(0, MAX_UPSTREAM_MESSAGE_LENGTH - 3)}...`
+        : text;
+};
@@ -106,6 +106,29 @@ A `Promise` that resolves to an `HTMLVideoElement`. The element is preloaded, ha

 > **Note:** Video generation can take several minutes to complete. The returned promise resolves only when the video is ready, so keep your UI responsive (for example, by showing a spinner) while you wait. Each successful generation consumes the user’s AI credits in accordance with the model, duration, and resolution you request.

+## Errors
+
+A rejection carries the error body exactly as the backend sent it. Every error has `message` and `code`; the other fields appear when they apply.
+
+| Field | Meaning |
+| --- | --- |
+| `message` | Human-readable reason. `error` carries the same text for older clients. |
+| `code` | Stable error code; see the table below. |
+| `errorCode` | A more specific code alongside a general `code`. Today the only value is `moderation_flagged`. |
+| `provider` | Which upstream handled the request: `gemini` (Veo), `together`, `byteplus` or `openai`. Present on errors raised while a job was running. |
+| `upstreamCode` | The provider's own error code, when it gave one. |
+| `upstreamStatus` | The HTTP status the provider returned, when it rejected the request before a job started. |
+
+| Code | Meaning |
+| --- | --- |
+| `upstream_timeout` | The provider did not finish the clip within the time Puter waits for it, or stopped answering. Puter waits ten minutes for Veo, Together and BytePlus models and five minutes for Sora models. Arrives as HTTP 504. The request itself was fine; retry it, ideally with a shorter clip or a faster model. |
+| `errorCode: moderation_flagged` | The provider's content filter refused the prompt or removed the generated video. Arrives as HTTP 400, with `code: bad_request` from Together and BytePlus and `code: disallowed_value` from Veo. Change the prompt rather than retrying it as-is. Sora does not report refusals distinctly; they arrive as `upstream_failed`. |
+| `upstream_bad_request` | The provider rejected the request itself, for example a duration the model does not support. Arrives as HTTP 400; `message` and `upstreamCode` carry the provider's reason. |
+| `upstream_failed` | The provider accepted the request but generation failed on their side. From Veo, Together and BytePlus it arrives as HTTP 502 and is safe to retry. From Sora it arrives as HTTP 400 and may also be a content-policy refusal, so read the `message` before retrying the same prompt. |
+| `insufficient_funds` | Your balance cannot cover the estimated cost of the clip. Arrives as HTTP 402. |
+
+Other `upstream_*` codes mean the provider rejected the request or was unavailable before a job started; `message` carries the provider's reason.
+
 ## Examples

 <strong class="example-title">Generate a sample clip (test mode)</strong>