Start I2V when Beast is idle instead of queuing behind leftover jobs.

Co-authored-by: Cursor <cursoragent@cursor.com>

server/api/edit.post.ts

@@ -1,4 +1,4 @@
-import { addStudioJob, kickStudioQueue, listStudioJobs } from '~/server/utils/studioQueue'
+import { addStudioJob, kickStudioQueue, listStudioJobs, videoJobsBusy } from '~/server/utils/studioQueue'
 import { parsePostedLoraStack, listStudioLoras, persistLoraFields } from '~/server/utils/loras'
 import { assertImageScaleToTotalPixelsNode, comfyConfigured } from '~/server/utils/comfy'
 import { imageDimensions } from '~/server/utils/resolution'
@@ -146,7 +146,11 @@ export default defineEventHandler(async (event) => {
     }
   })
   await kickStudioQueue()
-  const latest = listStudioJobs(ownerKey).find(item => item.id === studio.id)
+  let latest = listStudioJobs(ownerKey).find(item => item.id === studio.id)
+  if (!latest?.liveJobId && !(await videoJobsBusy())) {
+    await kickStudioQueue()
+    latest = listStudioJobs(ownerKey).find(item => item.id === studio.id)
+  }
   const liveJobId = latest?.liveJobId || ''
 
   return {

server/api/generate.post.ts

@@ -1,5 +1,5 @@
 import { existsSync, readFileSync } from 'node:fs'
-import { addStudioJob, kickStudioQueue, listStudioJobs } from '~/server/utils/studioQueue'
+import { addStudioJob, kickStudioQueue, listStudioJobs, videoJobsBusy } from '~/server/utils/studioQueue'
 import { listStudioLoras, parsePostedLoraStack, parseShotLoraStacks, persistLoraFields } from '~/server/utils/loras'
 import type { LoraStackItem } from '~/utils/loras'
 import { clampVideoCfg } from '~/utils/generationPresets'
@@ -259,7 +259,11 @@ export default defineEventHandler(async (event) => {
     }
   })
   await kickStudioQueue()
-  const latest = listStudioJobs(ownerKey).find(item => item.id === studio.id)
+  let latest = listStudioJobs(ownerKey).find(item => item.id === studio.id)
+  if (!latest?.liveJobId && !(await videoJobsBusy())) {
+    await kickStudioQueue()
+    latest = listStudioJobs(ownerKey).find(item => item.id === studio.id)
+  }
   const liveJobId = latest?.liveJobId || ''
   const queueId = latest?.shotQueueId
 

server/api/generate/active.get.ts

@@ -1,5 +1,13 @@
-function livePayload(job: ReturnType<typeof getJob> & {}) {
-  if (!job) return null
+import { fetchLiveQueue } from '~/server/utils/comfy'
+import { jobSnapshot, listJobs, type Job } from '~/server/utils/jobs'
+import { libraryOwnerKey } from '~/server/utils/library'
+import { listPendingJobs, resolvePendingJob } from '~/server/utils/pending'
+import { jobIsLocallySubmitting } from '~/server/utils/studioQueue'
+import { ensurePendingWatch } from '~/server/utils/watch'
+
+const SUBMIT_WINDOW_MS = 45 * 1000
+
+function livePayload(job: Job) {
   return {
     jobId: job.id,
     kind: job.kind || 'video',
@@ -9,10 +17,20 @@ function livePayload(job: ReturnType<typeof getJob> & {}) {
   }
 }
 
+function jobLooksLive(job: Job, comfyBusy: boolean) {
+  if (job.library?.stopAfterCurrent) return false
+  if (jobIsLocallySubmitting(job)) return true
+  if (job.status === 'running' && job.promptId) return comfyBusy
+  if (job.status === 'queued' && Date.now() - job.startedAt < SUBMIT_WINDOW_MS) return true
+  return false
+}
+
 export default defineEventHandler(async (event) => {
   const owner = libraryOwnerKey(event)
+  const queue = await fetchLiveQueue()
+  const comfyBusy = queue ? queue.running > 0 || queue.pending > 0 : true
   const live = listJobs()
-    .filter(job => job.library?.ownerKey === owner && (job.status === 'queued' || job.status === 'uploading' || job.status === 'running'))
+    .filter(job => job.library?.ownerKey === owner && jobLooksLive(job, comfyBusy))
     .sort((a, b) => b.startedAt - a.startedAt)
   const videoLive = live.find(job => job.kind !== 'edit')
   const editLive = live.find(job => job.kind === 'edit')
@@ -22,11 +40,15 @@ export default defineEventHandler(async (event) => {
 
   if (!video) {
     const pending = listPendingJobs()
-      .filter(job => job.ownerKey === owner)
+      .filter(job => (
+        job.ownerKey === owner
+        && !job.stopAfterCurrent
+        && Date.now() - job.startedAt < SUBMIT_WINDOW_MS * 4
+      ))
       .sort((a, b) => b.startedAt - a.startedAt)[0]
-    if (pending) {
+    if (pending && comfyBusy) {
       const resolved = await resolvePendingJob(pending).catch(() => null)
-      if (resolved) {
+      if (resolved && resolved.status !== 'complete' && resolved.status !== 'error') {
         video = {
           jobId: pending.jobId,
           kind: 'video' as const,
@@ -34,7 +56,7 @@ export default defineEventHandler(async (event) => {
           folderLocked: pending.folderLocked === true,
           ...resolved
         }
-      } else {
+      } else if (!resolved) {
         const job = ensurePendingWatch(pending)
         video = {
           jobId: pending.jobId,

server/utils/studioQueue.ts

@@ -3,6 +3,7 @@ import { join } from 'node:path'
 import { getJob, listJobs, emitJob, type Job } from '~/server/utils/jobs'
 import { listPendingJobs, patchPendingJob, readPendingJob, deletePendingJob } from '~/server/utils/pending'
 import { getShotQueue } from '~/server/utils/shotQueue'
+import { fetchLiveQueue } from '~/server/utils/comfy'
 import { isLtxWorkflow, LTX_DISABLED_MESSAGE, ltxWorkflowEnabled, parseVideoWorkflow, type VideoWorkflowId } from '~/utils/videoModels'
 import { persistLoraFields } from '~/server/utils/loras'
 import { resolveLoraStack } from '~/utils/loras'
@@ -221,11 +222,19 @@ export function summarizeStudioJob(job: StudioJob) {
 }
 
 const QUEUED_GRACE_MS = 4 * 60 * 1000
+const SUBMIT_WINDOW_MS = 45 * 1000
 const PENDING_ORPHAN_MS = 2 * 60 * 1000
 
-function liveJobIsActive(job: Job) {
-  if (job.status === 'uploading' || job.status === 'running') return true
-  if (job.status === 'queued') return Date.now() - job.startedAt < QUEUED_GRACE_MS
+function jobAgeMs(job: { startedAt: number }) {
+  return Date.now() - job.startedAt
+}
+
+/** Uploading, or created just now and not yet on Comfy. Paused leftovers do not count. */
+export function jobIsLocallySubmitting(job: Job) {
+  if (job.library?.stopAfterCurrent) return false
+  if (job.status === 'uploading') return true
+  if (job.status === 'queued' && !job.promptId && jobAgeMs(job) < SUBMIT_WINDOW_MS) return true
+  if (job.status === 'running' && !job.promptId && jobAgeMs(job) < SUBMIT_WINDOW_MS) return true
   return false
 }
 
@@ -240,13 +249,20 @@ function sweepStaleLiveJobs() {
   }
 }
 
-export function videoJobsBusy() {
+export async function videoJobsBusy() {
   sweepStaleLiveJobs()
-  if (listJobs().some(liveJobIsActive)) return true
-  return listPendingJobs().some((pending) => {
-    if (!pending.promptId) return false
+  if (listJobs().some(jobIsLocallySubmitting)) return true
+  const queue = await fetchLiveQueue()
+  if (queue) return queue.running > 0 || queue.pending > 0
+  return listJobs().some((job) => {
+    if (job.library?.stopAfterCurrent) return false
+    if (job.status === 'uploading' || job.status === 'running') return true
+    if (job.status === 'queued') return jobAgeMs(job) < SUBMIT_WINDOW_MS
+    return false
+  }) || listPendingJobs().some((pending) => {
+    if (!pending.promptId || pending.stopAfterCurrent) return false
     const live = getJob(pending.jobId)
-    if (live) return liveJobIsActive(live)
+    if (live) return jobIsLocallySubmitting(live) || live.status === 'running' || live.status === 'uploading'
     return Date.now() - pending.startedAt < PENDING_ORPHAN_MS
   })
 }
@@ -571,26 +587,22 @@ async function dispatchStudioQueue() {
         jobs: current.jobs.map(item => structuredClone(item))
       }
     })
-    if (videoJobsBusy()) return
+    if (await videoJobsBusy()) return
     const cutIn = pickCutIn(store.jobs)
     if (cutIn) {
       await startStudioJob(cutIn)
       return
     }
     const waiting = pickWaiting(store.jobs)
-    const held = pickHeld(store.jobs)
-    if (waiting && (!held || waiting.createdAt >= held.updatedAt)) {
-      await startStudioJob(waiting)
-      return
-    }
-    if (held?.shotQueueId) {
-      const resumed = await resumeHeldStudioJob(held)
-      if (resumed) return
-    }
     if (waiting) {
       await startStudioJob(waiting)
       return
     }
+    const held = pickHeld(store.jobs)
+    if (held?.shotQueueId) {
+      const resumed = await resumeHeldStudioJob(held)
+      if (resumed) return
+    }
   }
 }
 
@@ -1050,10 +1062,11 @@ export async function toggleStudioQueuePause(owner: string, options: {
       const rowHasQueue = rows.some(job => job.shotQueueId === options.queueId)
       if (!rowHasQueue) await setShotQueuePause(owner, options.queueId, false).catch(() => null)
     }
-    if (heldPaused[0] && !videoJobsBusy()) {
+    const busy = await videoJobsBusy()
+    if (heldPaused[0] && !busy) {
       await resumeHeldStudioJob(heldPaused[0]).catch(() => null)
     } else {
-      if (heldPaused.length && videoJobsBusy()) {
+      if (heldPaused.length && busy) {
         await mutate(owner, (jobs) => {
           for (const job of jobs) {
             if (!heldPaused.some(item => item.id === job.id)) continue