Keep remaining multi-shot clips on a Queue page so prompts can be edited between shots, with process-next or overnight run-all.

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-08-27 14:52:03 -05:00
parent d50db8e180
commit 73c4356e08
16 changed files with 1049 additions and 25 deletions
@@ -1,13 +1,23 @@
 import { copyFileSync, existsSync, readFileSync, writeFileSync } from 'node:fs'
 import { join } from 'node:path'
-import type { Job } from '~/server/utils/jobs'
+import { createJob, emitJob, type Job } from '~/server/utils/jobs'
 import { pendingFromJob, remainingAfterCurrentShot, writePendingJob } from '~/server/utils/pending'
 import { emitChainJob, waitForComfySocket, watchComfyJob } from '~/server/utils/watch'
 import { comfyFilenamePrefix, queuePrompt, uploadImage } from '~/server/utils/comfy'
 import { buildWorkflow } from '~/server/utils/workflow'
-import { clipVideoPath, deleteRetryDraft, extendTempDir, stillPath } from '~/server/utils/library'
+import { clipVideoPath, clipTitle, deleteRetryDraft, extendTempDir, getClip, nextClipPartName, removeExtendTemp, stillPath } from '~/server/utils/library'
 import { extractLastFrame, probeHasAudio } from '~/server/utils/ffmpeg'
 import { ensureComfyReady } from '~/server/utils/comfyLifecycle'
+import {
+  finishQueueBurst,
+  getShotQueue,
+  lastCompletedIndex,
+  liveSegmentPrompt,
+  prepareQueueBurst,
+  remainingFromQueue,
+  setQueueJob,
+  updateShotQueue
+} from '~/server/utils/shotQueue'

 export type ChainImage = { filename: string; data: Buffer; type?: string }

@@ -202,11 +212,17 @@ function seedCurrentVideo(job: Job) {
  return { tmpDir, currentPath, part1Path: join(tmpDir, 'part1.mp4'), framePath: join(tmpDir, 'last_frame.png') }
 }

-export async function continueQueuedExtensions(job: Job, params: VideoChainParams) {
+export async function continueQueuedExtensions(
+  job: Job,
+  params: VideoChainParams,
+  options: { maxShots?: number; autoRun?: boolean } = {}
+) {
  const extensions = params.extensions || []
  if (!extensions.length || !job.library) return
  if (job.library.chainContinuing) return
  job.library.chainContinuing = true
+  const autoRun = options.autoRun === true || job.library.queueAutoRun === true
+  let shotsLeft = options.maxShots ?? (autoRun ? extensions.length : (job.library.queueBudget || 0))

  const ready = (status: { state: string; message: string; queueRunning?: number; queuePending?: number }) => {
    emitChainJob(job, {
@@ -220,18 +236,49 @@ export async function continueQueuedExtensions(job: Job, params: VideoChainParam
  }

  try {
+    if (shotsLeft <= 0) return
    const { currentPath, part1Path, framePath } = seedCurrentVideo(job)
    const startFrom = job.library.chainIndex || 0

    for (let i = startFrom; i < extensions.length; i++) {
+      if (shotsLeft <= 0) break
      assertJobActive(job)
-      const ext = extensions[i]
-      const isLast = i === extensions.length - 1
+      if (job.library.stopAfterCurrent && i > startFrom) break
+
+      const liveQueue = job.library.queueId
+        ? getShotQueue(job.library.ownerKey, job.library.queueId)
+        : null
+      const live = liveQueue ? liveSegmentPrompt(liveQueue, i) : extensions[i]
+      const ext = {
+        prompt: live.prompt || extensions[i].prompt,
+        duration: live.duration || extensions[i].duration
+      }
+      if (!ext.prompt.trim()) {
+        throw new Error(`Shot ${i + 2} needs a prompt`)
+      }
+
+      const remainingAfter = extensions.length - (i + 1)
+      const lastOfChain = remainingAfter === 0
+      const lastOfBurst = !autoRun && shotsLeft === 1
+      const persist = lastOfChain || lastOfBurst || job.library.stopAfterCurrent === true
+
      job.library.chainIndex = i + 1
      job.library.chainStep = i + 2
      job.library.chainLabel = `Extension ${i + 1}`
      job.library.extendPart1Path = undefined
      job.library.thumb = undefined
+      job.library.duration = ext.duration
+      if (liveQueue) {
+        await updateShotQueue(job.library.ownerKey, liveQueue.id, (queue) => {
+          const segment = queue.segments.find(item => item.index === i + 1)
+          if (segment) {
+            segment.status = 'running'
+            delete segment.error
+          }
+          queue.status = 'running'
+          queue.currentJobId = job.id
+        }).catch(() => null)
+      }

      emitChainJob(job, { type: 'status', message: 'Waiting 3s buffer...', progress: 1 })
      await sleep(3000)
@@ -269,14 +316,17 @@ export async function continueQueuedExtensions(job: Job, params: VideoChainParam
        fps: params.fps,
        samplerName: params.samplerName,
        scheduler: params.scheduler,
-        persist: isLast,
+        persist,
        workflow: params.workflow,
        duration: ext.duration,
        useIdentityRefs: params.useIdentityRefs,
        referenceImages: params.useIdentityRefs ? params.referenceImages : []
      })

-      if (!isLast) {
+      shotsLeft -= 1
+      if (job.library) job.library.queueBudget = shotsLeft
+
+      if (!lastOfChain && !persist) {
        if (!job.segmentBuffer?.length) {
          throw new Error('Extension finished without a stitched video')
        }
@@ -284,9 +334,22 @@ export async function continueQueuedExtensions(job: Job, params: VideoChainParam
        job.segmentBuffer = undefined
        job.library.extendPart1Path = undefined
      }
+
+      if (persist && remainingAfter > 0) break
+      if (job.library.stopAfterCurrent) break
    }
  } finally {
    if (job.library) job.library.chainContinuing = false
+    if (job.library?.queueId) {
+      const paused = job.status !== 'error' && job.status !== 'cancelled'
+      await finishQueueBurst(
+        job.library.ownerKey,
+        job.library.queueId,
+        paused && !job.library.queueAutoRun,
+        job.status === 'error' || job.status === 'cancelled' ? (job.error || 'Stopped') : undefined
+      ).catch(() => null)
+      setQueueJob(job.library.queueId, null)
+    }
  }
 }

@@ -295,7 +358,20 @@ export async function continueQueuedExtensionsIfNeeded(job: Job) {
  if (job.status === 'error' || job.status === 'cancelled' || job.status === 'complete') return
  const remaining = remainingAfterCurrentShot(job.library)
  if (!remaining.length) return
-  await continueQueuedExtensions(job, paramsFromJob(job))
+  const queue = job.library.queueId
+    ? getShotQueue(job.library.ownerKey, job.library.queueId)
+    : null
+  const autoRun = queue ? queue.autoRun === true : job.library.queueAutoRun === true
+  const budget = job.library.queueBudget || 0
+  if (queue && !autoRun && budget <= 0) return
+  if (!queue && !autoRun && budget <= 0) {
+    // Legacy chains without a queue keep the old auto-continue behavior.
+    await continueQueuedExtensions(job, paramsFromJob(job), { autoRun: true })
+    return
+  }
+  const maxShots = autoRun ? remaining.length : budget
+  if (maxShots <= 0) return
+  await continueQueuedExtensions(job, paramsFromJob(job), { maxShots, autoRun })
 }

 export async function runGeneration(job: Job, params: VideoChainParams) {
@@ -313,11 +389,95 @@ export async function runGeneration(job: Job, params: VideoChainParams) {
  await ensureComfyReady(ready)

  const { extensions, ...base } = params
+  const autoRun = job.library?.queueAutoRun === true
  await queueMiniMax(job, {
    ...base,
-    persist: extensions.length === 0
+    persist: extensions.length === 0 || !autoRun
  })

-  if (!extensions.length || !job.library) return
-  await continueQueuedExtensions(job, params)
+  if (job.library?.queueId && !autoRun) {
+    await finishQueueBurst(job.library.ownerKey, job.library.queueId, true).catch(() => null)
+    setQueueJob(job.library.queueId, null)
+  }
+
+  if (!extensions.length || !job.library || !autoRun) return
+  await continueQueuedExtensions(job, params, { maxShots: extensions.length, autoRun: true })
+}
+
+export async function startQueueBurst(owner: string, queueId: string, count: number | 'all') {
+  const { queue, count: n } = await prepareQueueBurst(owner, queueId, count)
+  const lastIndex = lastCompletedIndex(queue)
+  const clipId = queue.currentClipId
+  if (!clipId) {
+    throw createError({ statusCode: 409, statusMessage: 'The previous clip is missing, so the next shot cannot start' })
+  }
+  const source = getClip(owner, clipId)
+  const destFolderLocked = false
+  const job = createJob()
+  job.kind = 'video'
+  job.maxStep = queue.steps
+  job.hideThumbnail = queue.hideThumbnail
+  job.clipId = clipId
+  job.library = {
+    ownerKey: owner,
+    folderId: queue.folderId,
+    hideThumbnail: queue.hideThumbnail,
+    hideInput: queue.hideInput,
+    folderLocked: destFolderLocked,
+    name: nextClipPartName(clipTitle(source)),
+    prompt: queue.segments[lastIndex]?.prompt || source.prompt,
+    aspect: queue.aspect,
+    width: queue.width,
+    height: queue.height,
+    steps: queue.steps,
+    turbo: queue.turbo,
+    seed: Math.floor(Math.random() * 2_147_483_647),
+    cfg: queue.cfg,
+    fps: queue.fps,
+    samplerName: queue.samplerName,
+    scheduler: queue.scheduler,
+    duration: queue.segments[lastIndex]?.duration || source.duration,
+    stillId: queue.stillId,
+    stillFilename: queue.stillFilename,
+    referenceStillIds: queue.referenceStillIds,
+    sound: queue.sound,
+    extensions: remainingFromQueue(queue),
+    chainIndex: Math.max(0, lastIndex),
+    chainStep: lastIndex + 1,
+    chainTotal: queue.segments.length,
+    chainLabel: lastIndex <= 0 ? 'Initial' : `Extension ${lastIndex}`,
+    familyId: queue.familyId,
+    parentClipId: clipId,
+    workflow: queue.workflow,
+    useIdentityRefs: queue.useIdentityRefs,
+    queueId: queue.id,
+    queueAutoRun: count === 'all',
+    queueBudget: n
+  }
+  setQueueJob(queue.id, job.id)
+  await updateShotQueue(owner, queue.id, (next) => {
+    next.currentJobId = job.id
+    next.status = 'running'
+  }).catch(() => null)
+
+  emitChainJob(job, { type: 'status', message: 'Checking ComfyUI...', progress: 1 })
+  void continueQueuedExtensions(job, paramsFromJob(job), {
+    maxShots: n,
+    autoRun: count === 'all'
+  }).catch(async (error) => {
+    removeExtendTemp(job.library?.extendTmpDir)
+    const message = error instanceof Error ? error.message : String(error)
+    if (job.status === 'error' || job.status === 'cancelled') {
+      await finishQueueBurst(owner, queue.id, false, message).catch(() => null)
+      setQueueJob(queue.id, null)
+      return
+    }
+    job.status = 'error'
+    job.error = message
+    emitJob(job, { type: 'error', error: message, message })
+    await finishQueueBurst(owner, queue.id, false, message).catch(() => null)
+    setQueueJob(queue.id, null)
+  })
+
+  return { jobId: job.id, queueId: queue.id, count: n, chainTotal: queue.segments.length, chainStep: lastIndex + 1 }
 }