From 382bd37a3ea4fc3639ddf9040dc0430569c19218 Mon Sep 17 00:00:00 2001 From: Towsty Date: Fri, 28 Aug 2026 17:45:32 -0500 Subject: [PATCH] Start I2V when Beast is idle instead of queuing behind leftover jobs. Co-authored-by: Cursor --- pages/index.vue | 19 +++++++++-- server/api/edit.post.ts | 8 +++-- server/api/generate.post.ts | 8 +++-- server/api/generate/active.get.ts | 36 +++++++++++++++++---- server/utils/studioQueue.ts | 53 +++++++++++++++++++------------ 5 files changed, 91 insertions(+), 33 deletions(-) diff --git a/pages/index.vue b/pages/index.vue index 34afb31..80f5635 100644 --- a/pages/index.vue +++ b/pages/index.vue @@ -2008,6 +2008,7 @@ const studioJobs = ref>([]) const outputStudioJobId = ref('') +const videoAwaitingBeast = ref(false) const pickerOpen = ref(false) const pickerKind = ref<'image' | 'permanence'>('image') const pickerSlot = ref<'main' | 'editRef' | number | 'perm' | `perm:${number}` | `clip:${string}`>('main') @@ -2133,11 +2134,15 @@ const liveStatusMessage = computed(() => { return idleInputStatus('edit') } if (videoBusy.value || videoUrl.value || lockedSave.value) return statusMessage.value - if (outputQueuedJob.value) return statusMessage.value || 'Queued. Waiting for Beast to be free.' + if (outputQueuedJob.value || videoAwaitingBeast.value) { + return statusMessage.value || 'Queued. Waiting for Beast to be free.' + } return idleInputStatus('video') }) const liveStatusBusy = computed(() => ( - outputProgressKind.value === 'edit' ? editStatusBusy.value : statusBusy.value + outputProgressKind.value === 'edit' + ? editStatusBusy.value + : (statusBusy.value || videoAwaitingBeast.value) )) const liveProgressPct = computed(() => ( outputProgressKind.value === 'edit' @@ -3648,6 +3653,14 @@ async function refreshStudioQueue() { }>('/api/studio-queue').catch(() => ({ jobs: [] })) studioJobs.value = data.jobs || [] const running = studioJobs.value.find(job => job.status === 'running' && job.liveJobId) + const queuedMine = Boolean( + outputStudioJobId.value + && studioJobs.value.some(job => job.id === outputStudioJobId.value && (job.status === 'waiting' || job.status === 'held')) + ) + if (queuedMine && running && running.id !== outputStudioJobId.value) { + return + } + if (running?.liveJobId) videoAwaitingBeast.value = false if (running?.kind === 'edit') { if (running.liveJobId && running.liveJobId !== editJobId.value && !editBusy.value) { outputStudioJobId.value = running.id @@ -4977,6 +4990,7 @@ async function generate() { if (started.queued) { outputStudioJobId.value = started.studioJobId || '' outputFocus.value = 'video' + videoAwaitingBeast.value = true statusMessage.value = 'Queued. Waiting for Beast to be free.' const n = 1 + queued.length toast(n > 1 @@ -4985,6 +4999,7 @@ async function generate() { await refreshStudioQueue() return } + videoAwaitingBeast.value = false clearVideoForm() statusBusy.value = false videoBusy.value = true diff --git a/server/api/edit.post.ts b/server/api/edit.post.ts index c7e1a1a..a1584bf 100644 --- a/server/api/edit.post.ts +++ b/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 { diff --git a/server/api/generate.post.ts b/server/api/generate.post.ts index 8791570..3a86979 100644 --- a/server/api/generate.post.ts +++ b/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 diff --git a/server/api/generate/active.get.ts b/server/api/generate/active.get.ts index aeb1038..5184132 100644 --- a/server/api/generate/active.get.ts +++ b/server/api/generate/active.get.ts @@ -1,5 +1,13 @@ -function livePayload(job: ReturnType & {}) { - 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 & {}) { } } +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, diff --git a/server/utils/studioQueue.ts b/server/utils/studioQueue.ts index e230bac..0eb0554 100644 --- a/server/utils/studioQueue.ts +++ b/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