From c7af0c6a7d7c43c7c48f09128cb3ffe1e6def955 Mon Sep 17 00:00:00 2001 From: "[._.]/ Adam Eivy" Date: Fri, 4 Sep 2026 09:02:43 +0000 Subject: [PATCH 1/2] feat: orchestrate continuous-video episode generation across video backends (#6227) Composes the script-to-beats compiler (#6225) and prompt linter (#6226) into an end-to-end pipeline: continuousVideo.js compiles a script + bible into clips, lints every clip prompt before any generation starts, submits clips sequentially to a chosen backend (local/reactor/fal) threading continuation conditioning between clips (a last-frame still, or reactor's native continue_from_clip_id), and stitches the results into one episode. A failed continuation clip re-attempts fresh rather than aborting the whole episode. POST /api/continuous-video submits an episode and rejects a lint failure before any backend call; GET /:jobId/events streams progress over SSE. Also fixes videoPromptLinter.js's private escapeRegExp to import the shared server/lib/textUtils.js implementation, per the repo-wide no-private-escape guard (pre-existing failure surfaced while working in this file). --- server/index.js | 2 + server/lib/apiRouteCatalog.generated.json | 25 +- server/lib/videoPromptLinter.js | 3 +- server/routes/continuousVideoEpisode.js | 122 ++++++++ server/routes/continuousVideoEpisode.test.js | 85 ++++++ server/services/videoGen/continuousVideo.js | 284 ++++++++++++++++++ .../services/videoGen/continuousVideo.test.js | 148 +++++++++ 7 files changed, 663 insertions(+), 6 deletions(-) create mode 100644 server/routes/continuousVideoEpisode.js create mode 100644 server/routes/continuousVideoEpisode.test.js create mode 100644 server/services/videoGen/continuousVideo.js create mode 100644 server/services/videoGen/continuousVideo.test.js diff --git a/server/index.js b/server/index.js index 8d521e1fa7..63bcceb235 100644 --- a/server/index.js +++ b/server/index.js @@ -117,6 +117,7 @@ import characterRoutes from './routes/character.js'; import toolsRoutes from './routes/tools.js'; import imageGenRoutes from './routes/imageGen.js'; import videoGenRoutes from './routes/videoGen.js'; +import continuousVideoEpisodeRoutes from './routes/continuousVideoEpisode.js'; import videoDownloadRoutes from './routes/videoDownload.js'; import videoTimelineRoutes from './routes/videoTimeline.js'; import mediaJobsRoutes from './routes/mediaJobs.js'; @@ -380,6 +381,7 @@ app.use('/api/character', characterRoutes); app.use('/api/tools', toolsRoutes); app.use('/api/image-gen', imageGenRoutes); app.use('/api/video-gen', videoGenRoutes); +app.use('/api/continuous-video', continuousVideoEpisodeRoutes); app.use('/api/devtools/video-download', videoDownloadRoutes); app.use('/api/video-timeline', videoTimelineRoutes); app.use('/api/media-jobs', mediaJobsRoutes); diff --git a/server/lib/apiRouteCatalog.generated.json b/server/lib/apiRouteCatalog.generated.json index bb2845a620..e8fd0e216c 100644 --- a/server/lib/apiRouteCatalog.generated.json +++ b/server/lib/apiRouteCatalog.generated.json @@ -35,6 +35,7 @@ "/api/commands", "/api/conflict-journal", "/api/contacts", + "/api/continuous-video", "/api/cos", "/api/cos/gsd", "/api/creative-commission", @@ -3494,6 +3495,22 @@ "server/routes/contacts.js" ] }, + { + "method": "POST", + "path": "/api/continuous-video", + "mountPath": "/api/continuous-video", + "sources": [ + "server/routes/continuousVideoEpisode.js" + ] + }, + { + "method": "GET", + "path": "/api/continuous-video/:jobId/events", + "mountPath": "/api/continuous-video", + "sources": [ + "server/routes/continuousVideoEpisode.js" + ] + }, { "method": "GET", "path": "/api/cos", @@ -17552,9 +17569,9 @@ } ], "stats": { - "mounts": 147, - "operations": 2174, - "declarations": 2182, - "sourceFiles": 230 + "mounts": 148, + "operations": 2176, + "declarations": 2184, + "sourceFiles": 231 } } diff --git a/server/lib/videoPromptLinter.js b/server/lib/videoPromptLinter.js index fef36e38a6..b180aa95a2 100644 --- a/server/lib/videoPromptLinter.js +++ b/server/lib/videoPromptLinter.js @@ -17,6 +17,7 @@ */ import { resolveBibleDescriptor } from './scriptVideoCompiler.js'; +import { escapeRegExp } from './textUtils.js'; export const MAX_CLIP_PROMPT_LENGTH = 800; const HARD_CUT_PREFIX = 'Hard cut to'; @@ -25,8 +26,6 @@ const BANNED_REFERENTS = ['same', 'still', 'again', 'continues', 'as before']; const BANNED_NEGATIVES = ['no', 'without', 'never']; const BANNED_OVERLAY_TERMS = ['text', 'caption', 'overlay']; -const escapeRegExp = (s) => s.replace(/[.*+?^${}()|[\]\\]/g, '\\$&'); - // \b (not a hand-rolled [^a-z0-9] boundary) so "same_text" isn't flagged as containing // the standalone word "same" — \w already includes '_', matching how prose reads a word. const bannedTermPattern = (term) => ({ term, regex: new RegExp(`\\b${escapeRegExp(term).replace(/\s+/g, '\\s+')}\\b`, 'i') }); diff --git a/server/routes/continuousVideoEpisode.js b/server/routes/continuousVideoEpisode.js new file mode 100644 index 0000000000..3ce85ff2e4 --- /dev/null +++ b/server/routes/continuousVideoEpisode.js @@ -0,0 +1,122 @@ +/** + * Continuous-video episode routes (#6227) — submit a script + bible for + * chained multi-clip generation and track its progress. Mirrors the + * SSE-progress shape `server/routes/videoGen.js` uses for single renders and + * chains, over its own outer-job registry (`continuousVideo.js`) since an + * episode is not a mediaJobQueue entry. + */ + +import { Router } from 'express'; +import { randomUUID } from 'crypto'; +import { z } from 'zod'; +import { asyncHandler, ServerError, failValidation } from '../lib/errorHandler.js'; +import { getSettings } from '../services/settings.js'; +import { lintClips } from '../lib/videoPromptLinter.js'; +import { + generateContinuousVideoEpisode, composeEpisodeClips, attachEpisodeSseClient, CONTINUOUS_VIDEO_BACKENDS, +} from '../services/videoGen/continuousVideo.js'; + +const router = Router(); + +const lineSchema = z.object({ + type: z.enum(['action', 'dialogue']), + speaker: z.string().min(1).max(200).optional(), + voice: z.string().max(200).optional(), + text: z.string().max(4000), +}); + +const sceneSchema = z.object({ + sceneId: z.string().max(200).optional(), + location: z.string().max(200).optional(), + lines: z.array(lineSchema).min(1), +}); + +const bibleEntrySchema = z.object({ descriptor: z.string().min(1).max(2000) }); +const bibleSchema = z.object({ + styleDescriptor: z.string().max(2000).optional(), + cast: z.record(bibleEntrySchema).optional(), + locations: z.record(bibleEntrySchema).optional(), +}); + +// Backend render knobs a caller may steer. `settings`/`pythonPath` are always +// server-resolved below and never accepted here — Zod strips any unknown key +// (including an attempted apiKey/pythonPath/settings override) by default. +const renderOptionsSchema = z.object({ + modelId: z.string().max(64).optional(), + width: z.number().min(64).max(2048).optional(), + height: z.number().min(64).max(2048).optional(), + negativePrompt: z.string().max(8000).optional(), + seed: z.number().optional(), + falModelId: z.string().min(1).max(200).optional(), + falDuration: z.number().min(1).max(60).optional(), + aspectRatio: z.enum(['16:9', '9:16', '1:1']).optional(), + reactorSeconds: z.number().min(1).max(60).optional(), +}); + +const compilerOptionsSchema = z.object({ + maxWords: z.number().int().positive().max(200).optional(), + maxSpeakers: z.number().int().positive().max(10).optional(), + maxChainLength: z.number().int().positive().max(50).optional(), + fps: z.number().positive().max(60).optional(), + frameGrid: z.enum(['uniform', '17n+5']).optional(), +}); + +const submitBodySchema = z.object({ + scenes: z.array(sceneSchema).min(1).max(200), + bible: bibleSchema, + framings: z.array(z.string().max(200).nullable()).max(2000).optional(), + backend: z.enum(CONTINUOUS_VIDEO_BACKENDS).optional(), + renderOptions: renderOptionsSchema.optional(), + compilerOptions: compilerOptionsSchema.optional(), +}); + +router.post('/', asyncHandler(async (req, res) => { + const parsed = submitBodySchema.safeParse(req.body || {}); + if (!parsed.success) failValidation(parsed); + const { + scenes, bible, framings, backend, renderOptions, compilerOptions, + } = parsed.data; + + // Lint BEFORE anything is submitted to a backend — a rule-violating clip + // prompt is rejected here, synchronously, rather than surfacing only later + // over the SSE progress stream. + const clips = composeEpisodeClips({ + scenes, bible, framings, compilerOptions, + }); + const lint = lintClips(clips, { bible }); + if (!lint.pass) { + throw new ServerError('One or more clip prompts failed the continuous-video lint', { + status: 422, code: 'VIDEO_PROMPT_LINT_FAILED', context: { lint }, + }); + } + + const settings = await getSettings(); + const jobId = randomUUID(); + // The orchestrator runs its own multi-clip chain in the background and + // reports progress over `GET /:jobId/events` (attachEpisodeSseClient) — + // matches the queued-then-SSE contract every other video-gen submit uses. + generateContinuousVideoEpisode({ + scenes, + bible, + framings, + backend, + compilerOptions, + jobId, + renderOptions: { + ...renderOptions, + settings, + pythonPath: settings.imageGen?.local?.pythonPath || null, + }, + }).catch((err) => { + console.log(`❌ Continuous video episode [${jobId.slice(0, 8)}] orchestration crashed: ${err.message}`); + }); + + res.json({ jobId, generationId: jobId, status: 'running' }); +})); + +router.get('/:jobId/events', (req, res) => { + const ok = attachEpisodeSseClient(req.params.jobId, res); + if (!ok) throw new ServerError('Job not found or expired', { status: 404 }); +}); + +export default router; diff --git a/server/routes/continuousVideoEpisode.test.js b/server/routes/continuousVideoEpisode.test.js new file mode 100644 index 0000000000..b1e401907b --- /dev/null +++ b/server/routes/continuousVideoEpisode.test.js @@ -0,0 +1,85 @@ +import { + describe, it, expect, vi, beforeEach, +} from 'vitest'; +import express from 'express'; +import { request } from '../lib/testHelper.js'; +import { errorMiddleware } from '../lib/errorHandler.js'; + +vi.mock('../services/settings.js', () => ({ + getSettings: vi.fn(async () => ({ imageGen: { local: { pythonPath: '/usr/bin/python3' } } })), +})); +vi.mock('../services/videoGen/continuousVideo.js', () => ({ + generateContinuousVideoEpisode: vi.fn(async () => ({ ok: true })), + composeEpisodeClips: vi.fn(), + attachEpisodeSseClient: vi.fn(() => false), + CONTINUOUS_VIDEO_BACKENDS: ['local', 'reactor', 'fal'], +})); +vi.mock('../lib/videoPromptLinter.js', () => ({ lintClips: vi.fn() })); + +import * as continuousVideo from '../services/videoGen/continuousVideo.js'; +import { lintClips } from '../lib/videoPromptLinter.js'; +import continuousVideoEpisodeRoutes from './continuousVideoEpisode.js'; + +const scenes = [{ sceneId: 's1', location: 'loc1', lines: [{ type: 'action', text: 'A quiet street.' }] }]; +const bible = { locations: { loc1: { descriptor: 'A quiet, rain-slicked street.' } } }; + +describe('continuousVideoEpisode routes', () => { + let app; + beforeEach(() => { + app = express(); + app.use(express.json()); + app.use('/api/continuous-video', continuousVideoEpisodeRoutes); + app.use(errorMiddleware); + vi.clearAllMocks(); + continuousVideo.composeEpisodeClips.mockReturnValue([{ prompt: 'A quiet street.', cutType: 'fresh' }]); + lintClips.mockReturnValue({ pass: true, results: [] }); + }); + + describe('POST /', () => { + it('rejects a request missing scenes/bible', async () => { + const r = await request(app).post('/api/continuous-video').send({}); + expect(r.status).toBe(400); + expect(continuousVideo.generateContinuousVideoEpisode).not.toHaveBeenCalled(); + }); + + it('rejects a lint failure before starting generation', async () => { + lintClips.mockReturnValue({ pass: false, results: [{ index: 0, pass: false, reasons: ['banned term'] }] }); + const r = await request(app).post('/api/continuous-video').send({ scenes, bible }); + expect(r.status).toBe(422); + expect(r.body.code).toBe('VIDEO_PROMPT_LINT_FAILED'); + expect(continuousVideo.generateContinuousVideoEpisode).not.toHaveBeenCalled(); + }); + + it('starts an episode and returns a running job descriptor', async () => { + const r = await request(app).post('/api/continuous-video').send({ scenes, bible, backend: 'local' }); + expect(r.status).toBe(200); + expect(r.body.status).toBe('running'); + expect(typeof r.body.jobId).toBe('string'); + expect(continuousVideo.generateContinuousVideoEpisode).toHaveBeenCalledTimes(1); + const call = continuousVideo.generateContinuousVideoEpisode.mock.calls[0][0]; + expect(call.jobId).toBe(r.body.jobId); + expect(call.backend).toBe('local'); + // pythonPath/settings are server-resolved, never accepted from the client. + expect(call.renderOptions.pythonPath).toBe('/usr/bin/python3'); + expect(call.renderOptions.settings).toBeTruthy(); + }); + + it('strips a client-supplied pythonPath/settings override from renderOptions', async () => { + const r = await request(app).post('/api/continuous-video').send({ + scenes, bible, renderOptions: { pythonPath: '/evil/python', settings: { hacked: true }, modelId: 'ltx2' }, + }); + expect(r.status).toBe(200); + const call = continuousVideo.generateContinuousVideoEpisode.mock.calls[0][0]; + expect(call.renderOptions.pythonPath).toBe('/usr/bin/python3'); + expect(call.renderOptions.settings).not.toEqual({ hacked: true }); + expect(call.renderOptions.modelId).toBe('ltx2'); + }); + }); + + describe('GET /:jobId/events', () => { + it('404s when the job is not found', async () => { + const r = await request(app).get('/api/continuous-video/missing/events'); + expect(r.status).toBe(404); + }); + }); +}); diff --git a/server/services/videoGen/continuousVideo.js b/server/services/videoGen/continuousVideo.js new file mode 100644 index 0000000000..450f16652e --- /dev/null +++ b/server/services/videoGen/continuousVideo.js @@ -0,0 +1,284 @@ +/** + * Continuous-video episode orchestrator (#6217's end-to-end generation slice). + * Composes the script-to-beats compiler (#6225) and the prompt linter (#6226) + * with sequential submission to a chosen video backend, then stitches the + * completed clips into one episode. + * + * Sequence, per episode: compile the script into clips, lint every clip's + * prompt BEFORE any generation starts (a lint failure never reaches a + * backend), then submit clips one at a time in chain order — a 'continue' + * clip conditions on the clip before it (a last-frame still for local/fal, or + * `continue_from_clip_id` for reactor's native continuation); a 'fresh' clip + * starts unconditioned. When a continuation submission or render fails, the + * chain re-attempts that ONE clip unconditioned (a fresh re-establish) rather + * than aborting the whole episode — a broken mid-chain link should degrade to + * a visible cut, not sink everything rendered before it. + * + * No route/HTTP concerns here — `server/routes/continuousVideoEpisode.js` + * exposes this over SSE progress, mirroring `chainedVideo.js`'s outer-job + * pattern for its own multi-clip chain. + */ + +import { join } from 'path'; +import { randomUUID } from 'crypto'; +import { PATHS } from '../../lib/fileUtils.js'; +import { ServerError } from '../../lib/errorHandler.js'; +import { broadcastSse, closeJobAfterDelay, attachSseClient as attachSse } from '../../lib/sseUtils.js'; +import { compileScriptToClips } from '../../lib/scriptVideoCompiler.js'; +import { lintClips } from '../../lib/videoPromptLinter.js'; +import { videoGenEvents } from './events.js'; +import { extractLastFrame } from './frameExtraction.js'; +import { getHistoryItem } from './history.js'; +import { stitchVideos } from './stitchVideos.js'; + +export const CONTINUOUS_VIDEO_BACKENDS = Object.freeze(['local', 'reactor', 'fal']); + +// Deferred rather than a static top-level import: each backend module (in +// particular the local runtime's model-registry/HF-cache closure) is heavy, +// and most callers of this file only ever exercise ONE of the three — see the +// "widely-reached module" import-scoping rule in server/AGENTS.md. +const BACKEND_MODULES = { + local: () => import('./generateVideo.js'), + reactor: () => import('./reactor.js'), + fal: () => import('./fal.js'), +}; + +// Process-local outer-job registry for episode SSE progress — separate from +// videoJobState (local-lane only) and from reactor.js/fal.js's own per-backend +// job maps, because one episode's chain can run against any single backend. +const episodeJobs = new Map(); + +export const attachEpisodeSseClient = (jobId, res) => attachSse(episodeJobs, jobId, res); + +/** + * Compile a script against a bible and attach what the linter needs: the + * `Hard cut to :` opener on every 'continue' clip that has a framing + * assigned, and the bible references (cast + scene location) each clip's + * prompt must carry verbatim. Deliberately does not invent a framing when the + * caller omits one for a continuing clip — that surfaces as a lint failure + * ("missing hard-cut opener") rather than silently passing an unmarked cut. + * + * @param {Array<{sceneId?: string, location?: string, lines: Array}>} scenes + * @param {object} bible + * @param {string[]} [framings] - camera framing/angle for clip i, parallel to + * the compiled clip array (`compileScriptToClips`'s emission order). + * @param {object} [compilerOptions] - forwarded to compileScriptToClips. + */ +export function composeEpisodeClips({ scenes, bible, framings = [], compilerOptions = {} } = {}) { + const clips = compileScriptToClips({ scenes, bible, ...compilerOptions }); + return clips.map((clip, index) => { + const framing = framings[index] || null; + const locationId = scenes?.[clip.sceneIndex]?.location ?? null; + const references = [ + ...clip.speakers.map((id) => ({ kind: 'cast', id })), + ...(locationId ? [{ kind: 'locations', id: locationId }] : []), + ]; + const prompt = clip.cutType === 'continue' && framing + ? `Hard cut to ${framing}: ${clip.prompt}` + : clip.prompt; + return { + ...clip, framing, references, prompt, + }; + }); +} + +const backendJobParams = ({ backend, clip, conditioning, renderOptions, jobId }) => { + if (backend === 'reactor') { + return { + settings: renderOptions.settings, + prompt: clip.prompt, + negativePrompt: renderOptions.negativePrompt, + continueFromClipId: conditioning.continueFromClipId || undefined, + sourceImagePath: conditioning.sourceImagePath || null, + seconds: renderOptions.reactorSeconds ?? clip.durationSeconds, + seed: renderOptions.seed, + jobId, + }; + } + if (backend === 'fal') { + return { + settings: renderOptions.settings, + modelId: renderOptions.falModelId, + prompt: clip.prompt, + negativePrompt: renderOptions.negativePrompt, + duration: renderOptions.falDuration, + aspectRatio: renderOptions.aspectRatio, + width: renderOptions.width, + height: renderOptions.height, + sourceImagePath: conditioning.sourceImagePath || null, + jobId, + }; + } + return { + pythonPath: renderOptions.pythonPath, + prompt: clip.prompt, + negativePrompt: renderOptions.negativePrompt, + modelId: renderOptions.modelId, + width: renderOptions.width, + height: renderOptions.height, + fps: clip.fps, + numFrames: clip.frames, + sourceImagePath: conditioning.sourceImagePath || null, + mode: conditioning.sourceImagePath ? 'image' : 'text', + hidden: true, + jobId, + }; +}; + +// Wait for the backend's own `completed`/`failed` videoGenEvents pair for one +// inner clip render — the same event contract chainedVideo.js's runChunk() +// consumes, shared by local/reactor/fal generateVideo(). +const awaitClipCompletion = (innerJobId) => new Promise((resolve, reject) => { + const detach = () => { + videoGenEvents.off('completed', onCompleted); + videoGenEvents.off('failed', onFailed); + }; + const onCompleted = (e) => { + if (e.generationId !== innerJobId) return; + detach(); + resolve(e); + }; + const onFailed = (e) => { + if (e.generationId !== innerJobId) return; + detach(); + reject(new Error(e.error || 'clip generation failed')); + }; + videoGenEvents.on('completed', onCompleted); + videoGenEvents.on('failed', onFailed); +}); + +// Build the conditioning the NEXT clip needs from the clip that just +// completed. Reactor conditions natively via continue_from_clip_id (read back +// off the completed clip's history entry); every other backend conditions on +// a still extracted from the completed clip's last frame — the same 'frame' +// hop chainedVideo.js falls back to when no extend pipeline is available. +async function buildConditioning(backend, completedInnerJobId) { + if (backend === 'reactor') { + const entry = await getHistoryItem(completedInnerJobId); + return { continueFromClipId: entry?.clipId || null }; + } + const frame = await extractLastFrame(completedInnerJobId).catch(() => null); + if (!frame?.filename) return { sourceImagePath: null }; + return { sourceImagePath: join(PATHS.images, frame.filename) }; +} + +/** + * Generate one continuous-video episode: compile + lint the script, then + * submit each clip in chain order to `backend`, stitching the completed + * clips into a single episode video. + * + * @param {object} params + * @param {Array} params.scenes + * @param {object} params.bible + * @param {string[]} [params.framings] + * @param {'local'|'reactor'|'fal'} [params.backend] + * @param {object} [params.renderOptions] - backend render knobs (modelId, + * width, height, negativePrompt, seed, pythonPath, falModelId, settings, …) + * — `settings`/`pythonPath` are server-resolved; never accept them from a + * client request. + * @param {object} [params.compilerOptions] - forwarded to compileScriptToClips + * @param {string} [params.jobId] - outer episode job id; minted when absent + */ +export async function generateContinuousVideoEpisode({ + scenes, bible, framings = [], backend = 'local', renderOptions = {}, compilerOptions = {}, jobId, +} = {}) { + if (!CONTINUOUS_VIDEO_BACKENDS.includes(backend)) { + throw new ServerError(`Unknown continuous-video backend: ${backend}`, { status: 400, code: 'VALIDATION_ERROR' }); + } + const outerJobId = jobId || randomUUID(); + const clips = composeEpisodeClips({ scenes, bible, framings, compilerOptions }); + const lint = lintClips(clips, { bible }); + if (!lint.pass) { + return { + ok: false, jobId: outerJobId, stage: 'lint', lint, clips, + }; + } + if (clips.length === 0) { + return { + ok: false, jobId: outerJobId, stage: 'lint', error: 'Script compiled to zero clips', + }; + } + + const { generateVideo: generate } = await BACKEND_MODULES[backend](); + const outerJob = { id: outerJobId, clients: [], status: 'running' }; + episodeJobs.set(outerJobId, outerJob); + + const emitProgress = (index, message) => { + videoGenEvents.emit('progress', { + generationId: outerJobId, + progress: index / clips.length, + message: `Clip ${index + 1}/${clips.length}${message ? ` — ${message}` : ''}`, + }); + broadcastSse(outerJob, { type: 'progress', progress: index / clips.length, message: `Clip ${index + 1}/${clips.length}` }); + }; + + const runOneClip = async (clip, conditioning) => { + const innerJobId = randomUUID(); + const params = backendJobParams({ + backend, clip, conditioning, renderOptions, jobId: innerJobId, + }); + // Listeners are registered BEFORE `generate` is even invoked, so no + // completion event it emits can fire before we're listening for it. + const completion = awaitClipCompletion(innerJobId); + await generate(params); + await completion; + return { innerJobId }; + }; + + const clipIds = []; + let previousClip = null; + for (let i = 0; i < clips.length; i++) { + const clip = clips[i]; + emitProgress(i, clip.cutType === 'continue' ? 'continuing' : 'establishing'); + const wantsContinuation = clip.cutType === 'continue' && previousClip; + const conditioning = wantsContinuation + // eslint-disable-next-line no-await-in-loop + ? await buildConditioning(backend, previousClip) + : {}; + + // eslint-disable-next-line no-await-in-loop + let outcome = await runOneClip(clip, conditioning).catch((err) => ({ error: err.message })); + if (outcome.error && wantsContinuation) { + console.log(`⚠️ Continuous video [${outerJobId.slice(0, 8)}] clip ${i + 1}/${clips.length} continuation failed (${outcome.error}) — re-establishing fresh`); + // eslint-disable-next-line no-await-in-loop + outcome = await runOneClip(clip, {}).catch((err) => ({ error: err.message })); + } + if (outcome.error) { + videoGenEvents.emit('failed', { generationId: outerJobId, error: outcome.error }); + broadcastSse(outerJob, { type: 'error', error: outcome.error }); + closeJobAfterDelay(episodeJobs, outerJobId); + return { + ok: false, jobId: outerJobId, stage: 'generation', failedClipIndex: i, error: outcome.error, clipIds, + }; + } + clipIds.push(outcome.innerJobId); + previousClip = outcome.innerJobId; + } + + const stitched = await stitchVideos(clipIds, { + id: outerJobId, + filenamePrefix: 'episode', + historyKey: 'chainedFrom', + }).catch((err) => ({ error: err.message })); + if (stitched?.error) { + videoGenEvents.emit('failed', { generationId: outerJobId, error: `Stitch failed: ${stitched.error}` }); + broadcastSse(outerJob, { type: 'error', error: `Stitch failed: ${stitched.error}` }); + closeJobAfterDelay(episodeJobs, outerJobId); + return { + ok: false, jobId: outerJobId, stage: 'stitch', error: stitched.error, clipIds, + }; + } + + const result = { + ok: true, + jobId: outerJobId, + filename: stitched.filename, + thumbnail: stitched.thumbnail, + path: `/data/videos/${stitched.filename}`, + clipIds, + }; + videoGenEvents.emit('completed', { generationId: outerJobId, ...result }); + broadcastSse(outerJob, { type: 'complete', result }); + closeJobAfterDelay(episodeJobs, outerJobId); + return result; +} diff --git a/server/services/videoGen/continuousVideo.test.js b/server/services/videoGen/continuousVideo.test.js new file mode 100644 index 0000000000..9c3376250c --- /dev/null +++ b/server/services/videoGen/continuousVideo.test.js @@ -0,0 +1,148 @@ +import { + describe, it, expect, vi, beforeEach, +} from 'vitest'; + +vi.mock('./generateVideo.js', () => ({ generateVideo: vi.fn() })); +vi.mock('./reactor.js', () => ({ generateVideo: vi.fn() })); +vi.mock('./fal.js', () => ({ generateVideo: vi.fn() })); +vi.mock('./frameExtraction.js', () => ({ extractLastFrame: vi.fn(async () => ({ filename: 'frame-still.png' })) })); +vi.mock('./history.js', () => ({ getHistoryItem: vi.fn(async () => ({ clipId: 'reactor-clip-1' })) })); +vi.mock('./stitchVideos.js', () => ({ + stitchVideos: vi.fn(async (videoIds, opts) => ({ + filename: `${opts.id}.mp4`, thumbnail: `${opts.id}-thumb.png`, + })), +})); + +const { generateVideo: localGenerate } = await import('./generateVideo.js'); +const { generateVideo: reactorGenerate } = await import('./reactor.js'); +const { extractLastFrame } = await import('./frameExtraction.js'); +const { getHistoryItem } = await import('./history.js'); +const { stitchVideos } = await import('./stitchVideos.js'); +const { videoGenEvents } = await import('./events.js'); +const { generateContinuousVideoEpisode, composeEpisodeClips } = await import('./continuousVideo.js'); + +// Two-beat scene: a 20-word opening beat (always 'fresh'), then a second +// 20-word beat that overflows BEAT_MAX_WORDS (35) and becomes a 'continue' +// clip — mirrors the shape scriptVideoCompiler.compileScriptToClips emits. +const WORDS_20 = Array(20).fill('word').join(' '); +const scenes = [{ + sceneId: 'scene-1', + location: 'loc1', + lines: [ + { type: 'action', text: WORDS_20 }, + { type: 'action', text: WORDS_20 }, + ], +}]; +const bible = { + styleDescriptor: 'Style: painterly line art.', + locations: { loc1: { descriptor: 'Location: a rain-slicked alley.' } }, + cast: {}, +}; + +// Auto-resolve `completed` once the mocked generateVideo has been invoked for +// this jobId — matches how the real backends emit their terminal event +// asynchronously after returning a sync jobId descriptor. +const succeedOnce = ({ jobId }) => { + queueMicrotask(() => videoGenEvents.emit('completed', { generationId: jobId, filename: `${jobId}.mp4` })); + return Promise.resolve({ jobId }); +}; +const failOnce = ({ jobId }, error = 'render failed') => { + queueMicrotask(() => videoGenEvents.emit('failed', { generationId: jobId, error })); + return Promise.resolve({ jobId }); +}; +const succeedOn = (mockFn) => mockFn.mockImplementation(succeedOnce); +const failOn = (mockFn, error = 'render failed') => mockFn.mockImplementationOnce((args) => failOnce(args, error)); + +beforeEach(() => { + vi.clearAllMocks(); +}); + +describe('composeEpisodeClips', () => { + it('adds the hard-cut opener and bible references to a continuing clip', () => { + const clips = composeEpisodeClips({ scenes, bible, framings: [null, 'medium shot'] }); + expect(clips).toHaveLength(2); + expect(clips[0].cutType).toBe('fresh'); + expect(clips[1].cutType).toBe('continue'); + expect(clips[1].prompt.startsWith('Hard cut to medium shot:')).toBe(true); + expect(clips[1].references).toEqual([{ kind: 'locations', id: 'loc1' }]); + }); + + it('leaves a continuing clip unmarked when no framing is supplied', () => { + const clips = composeEpisodeClips({ scenes, bible }); + expect(clips[1].framing).toBeNull(); + expect(clips[1].prompt.startsWith('Hard cut to')).toBe(false); + }); +}); + +describe('generateContinuousVideoEpisode', () => { + it('short-circuits on a lint failure without generating any clip', async () => { + const result = await generateContinuousVideoEpisode({ scenes, bible, framings: [], backend: 'local' }); + expect(result.ok).toBe(false); + expect(result.stage).toBe('lint'); + expect(result.lint.pass).toBe(false); + expect(localGenerate).not.toHaveBeenCalled(); + }); + + it('generates every clip in chain order and stitches the completed clips', async () => { + succeedOn(localGenerate); + const result = await generateContinuousVideoEpisode({ + scenes, bible, framings: [null, 'medium shot'], backend: 'local', + }); + expect(result.ok).toBe(true); + expect(localGenerate).toHaveBeenCalledTimes(2); + // Second (continuing) clip conditions on the first clip's extracted last frame. + expect(extractLastFrame).toHaveBeenCalledTimes(1); + const secondCallArgs = localGenerate.mock.calls[1][0]; + expect(secondCallArgs.sourceImagePath).toContain('frame-still.png'); + expect(stitchVideos).toHaveBeenCalledWith( + result.clipIds, + expect.objectContaining({ historyKey: 'chainedFrom' }), + ); + expect(result.filename).toBe(`${result.jobId}.mp4`); + }); + + it('re-establishes a failed continuation clip fresh instead of aborting the episode', async () => { + // clip0 (fresh) succeeds; clip1's conditioned attempt fails ONCE; the + // unconditioned retry that follows succeeds. + localGenerate.mockImplementationOnce((args) => succeedOnce(args)); + localGenerate.mockImplementationOnce((args) => failOnce(args, 'continuation render failed')); + localGenerate.mockImplementationOnce((args) => succeedOnce(args)); + const result = await generateContinuousVideoEpisode({ + scenes, bible, framings: [null, 'medium shot'], backend: 'local', + }); + expect(result.ok).toBe(true); + // clip0 (1 call) + clip1 failed attempt (1 call) + clip1 fresh retry (1 call) + expect(localGenerate).toHaveBeenCalledTimes(3); + const retryArgs = localGenerate.mock.calls[2][0]; + expect(retryArgs.sourceImagePath).toBeNull(); + expect(result.clipIds).toHaveLength(2); + }); + + it('aborts the episode when a fresh clip fails outright', async () => { + failOn(localGenerate, 'backend unavailable'); + const result = await generateContinuousVideoEpisode({ + scenes, bible, framings: [null, 'medium shot'], backend: 'local', + }); + expect(result.ok).toBe(false); + expect(result.stage).toBe('generation'); + expect(result.failedClipIndex).toBe(0); + expect(stitchVideos).not.toHaveBeenCalled(); + }); + + it('conditions a reactor continuation clip with continue_from_clip_id instead of a frame still', async () => { + succeedOn(reactorGenerate); + const result = await generateContinuousVideoEpisode({ + scenes, bible, framings: [null, 'medium shot'], backend: 'reactor', + }); + expect(result.ok).toBe(true); + expect(getHistoryItem).toHaveBeenCalledTimes(1); + const secondCallArgs = reactorGenerate.mock.calls[1][0]; + expect(secondCallArgs.continueFromClipId).toBe('reactor-clip-1'); + expect(extractLastFrame).not.toHaveBeenCalled(); + }); + + it('rejects an unknown backend', async () => { + await expect(generateContinuousVideoEpisode({ scenes, bible, backend: 'bogus' })) + .rejects.toThrow(/Unknown continuous-video backend/); + }); +}); From 49df4bd48f3c38d484d9219605fdbfc6d9e695e7 Mon Sep 17 00:00:00 2001 From: "[._.]/ Adam Eivy" Date: Fri, 4 Sep 2026 09:06:38 +0000 Subject: [PATCH 2/2] fix: address opencode review findings on continuous-video orchestrator MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Detach videoGenEvents listeners when a backend's generate() call throws synchronously, instead of only on its own 'completed'/'failed' emit — otherwise a listener pair leaks per synchronously-failing clip. - Reach progress:1.0 once every clip completes (was capped at (N-1)/N) and stop dropping the "continuing"/"establishing" suffix off the SSE progress message. - Fall back fal's clip duration to the compiled beat's durationSeconds, matching reactor's existing fallback. - Route passes its already-compiled+linted clips into the orchestrator instead of having it recompile the script a second time. --- server/routes/continuousVideoEpisode.js | 8 +- server/services/videoGen/continuousVideo.js | 99 +++++++++++++-------- 2 files changed, 67 insertions(+), 40 deletions(-) diff --git a/server/routes/continuousVideoEpisode.js b/server/routes/continuousVideoEpisode.js index 3ce85ff2e4..da4d41ed2e 100644 --- a/server/routes/continuousVideoEpisode.js +++ b/server/routes/continuousVideoEpisode.js @@ -96,12 +96,12 @@ router.post('/', asyncHandler(async (req, res) => { // reports progress over `GET /:jobId/events` (attachEpisodeSseClient) — // matches the queued-then-SSE contract every other video-gen submit uses. generateContinuousVideoEpisode({ - scenes, - bible, - framings, backend, - compilerOptions, jobId, + // Already compiled + linted above — pass the finished clips through + // rather than recompiling the script a second time (bible/scenes are + // only needed to build clips, which is done). + clips, renderOptions: { ...renderOptions, settings, diff --git a/server/services/videoGen/continuousVideo.js b/server/services/videoGen/continuousVideo.js index 450f16652e..19c2b87a6c 100644 --- a/server/services/videoGen/continuousVideo.js +++ b/server/services/videoGen/continuousVideo.js @@ -101,7 +101,7 @@ const backendJobParams = ({ backend, clip, conditioning, renderOptions, jobId }) modelId: renderOptions.falModelId, prompt: clip.prompt, negativePrompt: renderOptions.negativePrompt, - duration: renderOptions.falDuration, + duration: renderOptions.falDuration ?? clip.durationSeconds, aspectRatio: renderOptions.aspectRatio, width: renderOptions.width, height: renderOptions.height, @@ -127,25 +127,33 @@ const backendJobParams = ({ backend, clip, conditioning, renderOptions, jobId }) // Wait for the backend's own `completed`/`failed` videoGenEvents pair for one // inner clip render — the same event contract chainedVideo.js's runChunk() -// consumes, shared by local/reactor/fal generateVideo(). -const awaitClipCompletion = (innerJobId) => new Promise((resolve, reject) => { - const detach = () => { - videoGenEvents.off('completed', onCompleted); - videoGenEvents.off('failed', onFailed); - }; - const onCompleted = (e) => { - if (e.generationId !== innerJobId) return; - detach(); - resolve(e); - }; - const onFailed = (e) => { - if (e.generationId !== innerJobId) return; - detach(); - reject(new Error(e.error || 'clip generation failed')); - }; - videoGenEvents.on('completed', onCompleted); - videoGenEvents.on('failed', onFailed); -}); +// consumes, shared by local/reactor/fal generateVideo(). Returns `detach` +// alongside the promise so a caller whose `generate()` call throws/rejects +// SYNCHRONOUSLY (before it ever gets to emit 'failed') can still unregister +// these listeners itself — otherwise they leak on videoGenEvents forever, +// each holding a promise nothing will ever settle. +function awaitClipCompletion(innerJobId) { + let detach; + const promise = new Promise((resolve, reject) => { + detach = () => { + videoGenEvents.off('completed', onCompleted); + videoGenEvents.off('failed', onFailed); + }; + const onCompleted = (e) => { + if (e.generationId !== innerJobId) return; + detach(); + resolve(e); + }; + const onFailed = (e) => { + if (e.generationId !== innerJobId) return; + detach(); + reject(new Error(e.error || 'clip generation failed')); + }; + videoGenEvents.on('completed', onCompleted); + videoGenEvents.on('failed', onFailed); + }); + return { promise, detach }; +} // Build the conditioning the NEXT clip needs from the clip that just // completed. Reactor conditions natively via continue_from_clip_id (read back @@ -178,20 +186,28 @@ async function buildConditioning(backend, completedInnerJobId) { * client request. * @param {object} [params.compilerOptions] - forwarded to compileScriptToClips * @param {string} [params.jobId] - outer episode job id; minted when absent + * @param {Array} [params.clips] - a caller-precomposed + already-linted clip + * array (`composeEpisodeClips` + a passing `lintClips`) — skips recompiling + * and re-linting the script, for a caller (the route) that already did both + * to fail fast before this async orchestration starts. */ export async function generateContinuousVideoEpisode({ - scenes, bible, framings = [], backend = 'local', renderOptions = {}, compilerOptions = {}, jobId, + scenes, bible, framings = [], backend = 'local', renderOptions = {}, compilerOptions = {}, jobId, clips: precomposedClips, } = {}) { if (!CONTINUOUS_VIDEO_BACKENDS.includes(backend)) { throw new ServerError(`Unknown continuous-video backend: ${backend}`, { status: 400, code: 'VALIDATION_ERROR' }); } const outerJobId = jobId || randomUUID(); - const clips = composeEpisodeClips({ scenes, bible, framings, compilerOptions }); - const lint = lintClips(clips, { bible }); - if (!lint.pass) { - return { - ok: false, jobId: outerJobId, stage: 'lint', lint, clips, - }; + const clips = precomposedClips || composeEpisodeClips({ + scenes, bible, framings, compilerOptions, + }); + if (!precomposedClips) { + const lint = lintClips(clips, { bible }); + if (!lint.pass) { + return { + ok: false, jobId: outerJobId, stage: 'lint', lint, clips, + }; + } } if (clips.length === 0) { return { @@ -203,13 +219,15 @@ export async function generateContinuousVideoEpisode({ const outerJob = { id: outerJobId, clients: [], status: 'running' }; episodeJobs.set(outerJobId, outerJob); + // `progress` is CLIPS COMPLETED / total — index/clips.length while clip + // `index` is still in flight, so the fraction only reaches 1.0 once every + // clip has completed and stitching is what's left (see the pre-stitch call + // below). const emitProgress = (index, message) => { - videoGenEvents.emit('progress', { - generationId: outerJobId, - progress: index / clips.length, - message: `Clip ${index + 1}/${clips.length}${message ? ` — ${message}` : ''}`, - }); - broadcastSse(outerJob, { type: 'progress', progress: index / clips.length, message: `Clip ${index + 1}/${clips.length}` }); + const progress = index / clips.length; + const fullMessage = `Clip ${index + 1}/${clips.length}${message ? ` — ${message}` : ''}`; + videoGenEvents.emit('progress', { generationId: outerJobId, progress, message: fullMessage }); + broadcastSse(outerJob, { type: 'progress', progress, message: fullMessage }); }; const runOneClip = async (clip, conditioning) => { @@ -218,9 +236,15 @@ export async function generateContinuousVideoEpisode({ backend, clip, conditioning, renderOptions, jobId: innerJobId, }); // Listeners are registered BEFORE `generate` is even invoked, so no - // completion event it emits can fire before we're listening for it. - const completion = awaitClipCompletion(innerJobId); - await generate(params); + // completion event it emits can fire before we're listening for it. If + // `generate` itself throws/rejects (a synchronous validation error, never + // reaching its own 'failed' emit), detach here — otherwise these + // listeners would leak on videoGenEvents forever. + const { promise: completion, detach } = awaitClipCompletion(innerJobId); + await generate(params).catch((err) => { + detach(); + throw err; + }); await completion; return { innerJobId }; }; @@ -255,6 +279,9 @@ export async function generateContinuousVideoEpisode({ previousClip = outcome.innerJobId; } + videoGenEvents.emit('progress', { generationId: outerJobId, progress: 1, message: 'Stitching episode' }); + broadcastSse(outerJob, { type: 'progress', progress: 1, message: 'Stitching episode' }); + const stitched = await stitchVideos(clipIds, { id: outerJobId, filenamePrefix: 'episode',