towsty/aigen
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Pause the queue after the current clip for real, and carry global locks plus labeled permanence stills through generate and shot prompts.

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Towsty
2026-08-28 07:48:09 -05:00
co-authored by Cursor
parent 010ef6add2
commit efef409537
21 changed files with 1196 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/utils/jobs.ts
+4 -1
View File
@@ -71,10 +71,13 @@ export interface Job {
referenceImageNames?: string[]
referenceStillIds?: Array<string | null>
useIdentityRefs?: boolean
globalLocks?: string
permanenceRefs?: import('~/utils/globalLocks').PermanenceRef[]
shotPermanenceRefs?: import('~/utils/globalLocks').PermanenceRef[][]
duration?: number
sound?: boolean
draftId?: string
extensions?: { prompt: string; duration: number }[]
extensions?: { prompt: string; duration: number; permanenceRefs?: import('~/utils/globalLocks').PermanenceRef[] }[]
chainIndex?: number
chainStep?: number
chainTotal?: number
server/utils/library.ts
+49 -2
View File
@@ -6,6 +6,7 @@ import { randomBytes, scrypt, timingSafeEqual } from 'node:crypto'
import { promisify } from 'node:util'
import type { H3Event } from 'h3'
import { extractFirstFrame, extractLastFrame } from '~/server/utils/ffmpeg'
import { normalizePermanenceRefs, type PermanenceRef } from '~/utils/globalLocks'
const scryptAsync = promisify(scrypt)
@@ -43,6 +44,8 @@ export interface LibraryClip {
stillId?: string
sound?: boolean
hideInput?: boolean
globalLocks?: string
permanenceRefs?: PermanenceRef[]
}
export type StillRole = 'input' | 'output'
@@ -78,6 +81,7 @@ export interface PublicFolder {
export interface QueuedExtension {
prompt: string
duration: number
permanenceRefs?: PermanenceRef[]
}
export interface RetryDraft {
@@ -104,6 +108,9 @@ export interface RetryDraft {
scheduler?: string
extensions?: QueuedExtension[]
workflow?: import('~/utils/videoModels').VideoWorkflowId
globalLocks?: string
permanenceRefs?: PermanenceRef[]
shotPermanenceRefs?: PermanenceRef[][]
}
interface Catalog {
@@ -1120,6 +1127,9 @@ export async function saveRetryDraft(params: {
scheduler?: string
extensions?: QueuedExtension[]
workflow?: import('~/utils/videoModels').VideoWorkflowId
globalLocks?: string
permanenceRefs?: PermanenceRef[]
shotPermanenceRefs?: PermanenceRef[][]
}) {
return mutate(params.ownerKey, (catalog) => {
const existing = params.id ? catalog.drafts.find(item => item.id === params.id) : undefined
@@ -1146,7 +1156,10 @@ export async function saveRetryDraft(params: {
samplerName: params.samplerName,
scheduler: params.scheduler,
extensions: params.extensions?.length ? params.extensions : undefined,
workflow: params.workflow
workflow: params.workflow,
globalLocks: params.globalLocks,
permanenceRefs: params.permanenceRefs,
shotPermanenceRefs: params.shotPermanenceRefs
}
catalog.drafts = [draft, ...catalog.drafts.filter(item => item.id !== draft.id)].slice(0, 20)
return draft
@@ -1298,6 +1311,8 @@ export async function saveClip(params: {
workflow?: import('~/utils/videoModels').VideoWorkflowId
stillId?: string
sound?: boolean
globalLocks?: string
permanenceRefs?: PermanenceRef[]
}) {
const catalog = readCatalog(params.ownerKey)
const folder = catalog.folders.find(item => item.id === params.folderId) || catalog.folders[0]
@@ -1326,7 +1341,9 @@ export async function saveClip(params: {
workflow: params.workflow,
stillId: params.stillId,
sound: params.sound,
hideInput: params.hideInput === true
hideInput: params.hideInput === true,
globalLocks: params.globalLocks,
permanenceRefs: params.permanenceRefs?.length ? normalizePermanenceRefs(params.permanenceRefs) : undefined
}
mkdirSync(clipDir(params.ownerKey, clip.id), { recursive: true })
const videoPath = clipVideoPath(params.ownerKey, clip.id)
@@ -1413,6 +1430,36 @@ export function moveClipFamily(owner: string, id: string, folderId: string) {
})
}
export function patchClipMeta(owner: string, id: string, patch: {
globalLocks?: string
permanenceRefs?: PermanenceRef[]
family?: boolean
}) {
return mutate(owner, (catalog) => {
const clip = catalog.clips.find(item => item.id === id)
if (!clip) throw createError({ statusCode: 404, statusMessage: 'Clip not found' })
const family = patch.family ? clipsInSameFamily(catalog, clip) : [clip]
if (typeof patch.globalLocks === 'string') {
const locks = patch.globalLocks.trim()
for (const item of family) {
if (locks) item.globalLocks = locks
else delete item.globalLocks
}
}
if (patch.permanenceRefs) {
const refs = normalizePermanenceRefs(patch.permanenceRefs)
if (patch.family) {
for (const item of family) {
if (refs.length) item.permanenceRefs = refs
else delete item.permanenceRefs
}
} else if (refs.length) clip.permanenceRefs = refs
else delete clip.permanenceRefs
}
return patch.family ? family : clip
})
}
export function renameClip(owner: string, id: string, name: string) {
const trimmed = name.trim().slice(0, 80)
if (!trimmed) throw createError({ statusCode: 400, statusMessage: 'A name is required' })
server/utils/pending.ts
+16 -4
View File
@@ -1,6 +1,7 @@
import { existsSync, mkdirSync, readdirSync, readFileSync, renameSync, rmSync, writeFileSync } from 'node:fs'
import { join } from 'node:path'
import type { Job } from '~/server/utils/jobs'
import { mergePermanenceRefs } from '~/utils/globalLocks'
export interface PendingJob {
jobId: string
@@ -35,6 +36,9 @@ export interface PendingJob {
referenceImageNames?: string[]
referenceStillIds?: Array<string | null>
useIdentityRefs?: boolean
globalLocks?: string
permanenceRefs?: import('~/utils/globalLocks').PermanenceRef[]
shotPermanenceRefs?: import('~/utils/globalLocks').PermanenceRef[][]
workflow?: import('~/utils/videoModels').VideoWorkflowId
duration?: number
cfg?: number
@@ -42,8 +46,8 @@ export interface PendingJob {
samplerName?: string
scheduler?: string
hideInput?: boolean
extensions?: { prompt: string; duration: number }[]
remainingExtensions?: { prompt: string; duration: number }[]
extensions?: { prompt: string; duration: number; permanenceRefs?: import('~/utils/globalLocks').PermanenceRef[] }[]
remainingExtensions?: { prompt: string; duration: number; permanenceRefs?: import('~/utils/globalLocks').PermanenceRef[] }[]
currentClipId?: string
queueId?: string
queueAutoRun?: boolean
@@ -130,6 +134,9 @@ export function pendingFromJob(job: Job, overrides: Partial<PendingJob> = {}): P
extensions: library.extensions,
remainingExtensions: remaining,
currentClipId: job.clipId,
globalLocks: library.globalLocks,
permanenceRefs: library.permanenceRefs,
shotPermanenceRefs: library.shotPermanenceRefs,
queueId: library.queueId,
queueAutoRun: library.queueAutoRun,
queueBudget: library.queueBudget,
@@ -177,7 +184,10 @@ export function libraryFromPending(pending: PendingJob): NonNullable<Job['librar
extensions: pending.extensions || pending.remainingExtensions,
queueId: pending.queueId,
queueAutoRun: pending.queueAutoRun,
queueBudget: pending.queueBudget
queueBudget: pending.queueBudget,
globalLocks: pending.globalLocks,
permanenceRefs: pending.permanenceRefs,
shotPermanenceRefs: pending.shotPermanenceRefs
}
}
@@ -287,7 +297,9 @@ export async function completePendingIfReady(pending: PendingJob) {
parentClipId: pending.parentClipId,
chainIndex: pending.chainIndex,
stillId: pending.stillId,
sound: pending.sound
sound: pending.sound,
globalLocks: pending.globalLocks,
permanenceRefs: mergePermanenceRefs(pending.permanenceRefs, pending.shotPermanenceRefs?.[pending.chainIndex || 0])
})
await purgeComfyArtifacts({
video,
server/utils/shotQueue.ts
+50 -19
View File
@@ -1,6 +1,6 @@
import { existsSync, mkdirSync, readdirSync, readFileSync, renameSync, writeFileSync } from 'node:fs'
import { join } from 'node:path'
import { getJob, type Job } from '~/server/utils/jobs'
import { getJob, listJobs, type Job } from '~/server/utils/jobs'
export type ShotQueueStatus = 'idle' | 'running' | 'paused' | 'complete' | 'error'
export type ShotSegmentStatus = 'pending' | 'running' | 'complete' | 'error'
@@ -12,6 +12,7 @@ export interface ShotQueueSegment {
status: ShotSegmentStatus
clipId?: string
error?: string
permanenceRefs?: import('~/utils/globalLocks').PermanenceRef[]
}
export interface ShotQueue {
@@ -23,6 +24,7 @@ export interface ShotQueue {
updatedAt: number
status: ShotQueueStatus
autoRun: boolean
stopAfterCurrent?: boolean
currentClipId?: string
currentJobId?: string
stillId?: string
@@ -42,6 +44,8 @@ export interface ShotQueue {
sound?: boolean
useIdentityRefs?: boolean
referenceStillIds?: Array<string | null>
globalLocks?: string
permanenceRefs?: import('~/utils/globalLocks').PermanenceRef[]
segments: ShotQueueSegment[]
lastError?: string
}
@@ -157,8 +161,10 @@ export async function createShotQueue(params: {
sound?: boolean
useIdentityRefs?: boolean
referenceStillIds?: Array<string | null>
initial: { prompt: string; duration: number }
extensions: { prompt: string; duration: number }[]
globalLocks?: string
permanenceRefs?: import('~/utils/globalLocks').PermanenceRef[]
initial: { prompt: string; duration: number; permanenceRefs?: import('~/utils/globalLocks').PermanenceRef[] }
extensions: { prompt: string; duration: number; permanenceRefs?: import('~/utils/globalLocks').PermanenceRef[] }[]
jobId?: string
}): Promise<ShotQueue> {
const now = Date.now()
@@ -171,6 +177,7 @@ export async function createShotQueue(params: {
updatedAt: now,
status: 'running',
autoRun: params.autoRun,
stopAfterCurrent: false,
currentJobId: params.jobId,
stillId: params.stillId,
stillFilename: params.stillFilename,
@@ -189,13 +196,16 @@ export async function createShotQueue(params: {
sound: params.sound,
useIdentityRefs: params.useIdentityRefs,
referenceStillIds: params.referenceStillIds,
globalLocks: params.globalLocks,
permanenceRefs: params.permanenceRefs,
segments: [
{ index: 0, prompt: params.initial.prompt, duration: params.initial.duration, status: 'running' },
{ index: 0, prompt: params.initial.prompt, duration: params.initial.duration, status: 'running', permanenceRefs: params.initial.permanenceRefs },
...params.extensions.map((item, index) => ({
index: index + 1,
prompt: item.prompt,
duration: item.duration,
status: 'pending' as const
status: 'pending' as const,
permanenceRefs: item.permanenceRefs
}))
]
}
@@ -305,6 +315,7 @@ export async function prepareQueueBurst(owner: string, id: string, count: number
delete segment.error
})
queue.autoRun = count === 'all'
queue.stopAfterCurrent = false
queue.status = 'running'
queue.lastError = undefined
queue.updatedAt = Date.now()
@@ -341,23 +352,42 @@ export async function finishQueueBurst(owner: string, id: string, paused: boolea
})
}
export async function pauseShotQueue(owner: string, id: string) {
const jobId = activeJobs.get(id)
if (jobId) {
const job = getJob(jobId)
if (job?.library) job.library.stopAfterCurrent = true
if (job?.library) job.library.queueAutoRun = false
function liveJobForQueue(id: string): Job | undefined {
const mapped = activeJobs.get(id)
if (mapped) {
const job = getJob(mapped)
if (job) return job
}
return listJobs().find(job => (
job.library?.queueId === id
&& (job.status === 'queued' || job.status === 'uploading' || job.status === 'running')
))
}
export async function setShotQueuePause(owner: string, id: string, pause: boolean) {
const live = liveJobForQueue(id)
if (live?.library) {
live.library.stopAfterCurrent = pause
if (pause) live.library.queueAutoRun = false
}
return updateShotQueue(owner, id, (queue) => {
queue.autoRun = false
if (queue.status === 'running') {
queue.lastError = undefined
} else if (queue.status !== 'complete') {
queue.status = 'paused'
if (pause) {
queue.autoRun = false
queue.stopAfterCurrent = true
if (queue.status === 'running') queue.lastError = undefined
else if (queue.status !== 'complete') queue.status = 'paused'
} else {
queue.stopAfterCurrent = false
if (live?.library) live.library.queueAutoRun = queue.autoRun === true
if (queue.status === 'paused') queue.status = 'idle'
}
})
}
export async function pauseShotQueue(owner: string, id: string) {
return setShotQueuePause(owner, id, true)
}
export async function deleteShotQueue(owner: string, id: string) {
if (activeJobs.has(id)) {
throw createError({ statusCode: 409, statusMessage: 'Stop this batch before deleting it' })
@@ -395,10 +425,10 @@ export function listOwnersWithQueues() {
.filter(owner => existsSync(queuesPath(owner)))
}
export function remainingFromQueue(queue: ShotQueue): { prompt: string; duration: number }[] {
export function remainingFromQueue(queue: ShotQueue): { prompt: string; duration: number; permanenceRefs?: import('~/utils/globalLocks').PermanenceRef[] }[] {
return queue.segments
.filter(segment => segment.index > 0)
.map(segment => ({ prompt: segment.prompt, duration: segment.duration }))
.map(segment => ({ prompt: segment.prompt, duration: segment.duration, permanenceRefs: segment.permanenceRefs }))
}
export function lastCompletedIndex(queue: ShotQueue) {
@@ -411,6 +441,7 @@ export function liveSegmentPrompt(queue: ShotQueue, extensionIndex: number) {
const segment = queue.segments.find(item => item.index === extensionIndex + 1)
return {
prompt: segment?.prompt || '',
duration: segment?.duration || 5
duration: segment?.duration || 5,
permanenceRefs: segment?.permanenceRefs
}
}
server/utils/studioQueue.ts
+246 -32
View File
@@ -2,6 +2,7 @@ import { existsSync, mkdirSync, readdirSync, readFileSync, renameSync, writeFile
import { join } from 'node:path'
import { getJob, listJobs, type Job } from '~/server/utils/jobs'
import { parseVideoWorkflow, type VideoWorkflowId } from '~/utils/videoModels'
import type { PermanenceRef } from '~/utils/globalLocks'
export type StudioJobStatus = 'waiting' | 'running' | 'held' | 'complete' | 'error' | 'cancelled'
@@ -29,8 +30,11 @@ export interface StudioJobPayload {
hideInput?: boolean
folderLocked?: boolean
referenceStillIds: Array<string | null>
extensions: { prompt: string; duration: number }[]
extensions: { prompt: string; duration: number; permanenceRefs?: PermanenceRef[] }[]
queueAutoRun: boolean
globalLocks?: string
permanenceRefs?: PermanenceRef[]
shotPermanenceRefs?: PermanenceRef[][]
}
export interface StudioJob {
@@ -48,10 +52,17 @@ export interface StudioJob {
payload: StudioJobPayload
cutIn?: boolean
holdForCutIn?: boolean
pauseAfterCurrent?: boolean
pausedByUser?: boolean
resumeAutoRun?: boolean
lastError?: string
}
type StudioQueueStore = {
paused: boolean
jobs: StudioJob[]
}
const writeChains = new Map<string, Promise<unknown>>()
let dispatchChain: Promise<unknown> = Promise.resolve()
@@ -68,38 +79,68 @@ function ensureOwner(owner: string) {
mkdirSync(join(libraryRoot(), 'users', owner), { recursive: true })
}
function readJobs(owner: string): StudioJob[] {
function readStore(owner: string): StudioQueueStore {
ensureOwner(owner)
const path = queuePath(owner)
if (!existsSync(path)) return []
if (!existsSync(path)) return { paused: false, jobs: [] }
try {
const parsed = JSON.parse(readFileSync(path, 'utf8'))
return Array.isArray(parsed) ? parsed : []
if (Array.isArray(parsed)) return { paused: false, jobs: parsed }
if (parsed && typeof parsed === 'object') {
return {
paused: parsed.paused === true,
jobs: Array.isArray(parsed.jobs) ? parsed.jobs : []
}
}
return { paused: false, jobs: [] }
} catch {
return []
return { paused: false, jobs: [] }
}
}
function writeJobs(owner: string, jobs: StudioJob[]) {
function writeStore(owner: string, store: StudioQueueStore) {
ensureOwner(owner)
const path = queuePath(owner)
const tmp = `${path}.tmp`
writeFileSync(tmp, JSON.stringify(jobs, null, 2))
writeFileSync(tmp, JSON.stringify({
paused: store.paused === true,
jobs: store.jobs
}, null, 2))
renameSync(tmp, path)
}
function readJobs(owner: string): StudioJob[] {
return readStore(owner).jobs
}
function mutate<T>(owner: string, fn: (jobs: StudioJob[]) => T): Promise<T> {
const prev = writeChains.get(owner) || Promise.resolve()
const run = prev.then(() => {
const jobs = readJobs(owner)
const result = fn(jobs)
writeJobs(owner, jobs)
const store = readStore(owner)
const result = fn(store.jobs)
writeStore(owner, store)
return result
})
writeChains.set(owner, run.then(() => undefined, () => undefined))
return run
}
function mutateStore<T>(owner: string, fn: (store: StudioQueueStore) => T): Promise<T> {
const prev = writeChains.get(owner) || Promise.resolve()
const run = prev.then(() => {
const store = readStore(owner)
const result = fn(store)
writeStore(owner, store)
return result
})
writeChains.set(owner, run.then(() => undefined, () => undefined))
return run
}
export function isStudioQueuePaused(owner: string) {
return readStore(owner).paused === true
}
export function listOwnersWithStudioQueues() {
const root = join(libraryRoot(), 'users')
if (!existsSync(root)) return [] as string[]
@@ -133,6 +174,8 @@ export function summarizeStudioJob(job: StudioJob) {
queueAutoRun: job.payload.queueAutoRun === true,
cutIn: job.cutIn === true,
holdForCutIn: job.holdForCutIn === true,
pauseAfterCurrent: job.pauseAfterCurrent === true,
pausedByUser: job.pausedByUser === true,
lastError: job.lastError
}
}
@@ -307,11 +350,14 @@ function pickCutIn(jobs: StudioJob[]) {
}
function pickHeld(jobs: StudioJob[]) {
const held = jobs
.filter(job => job.status === 'held' && job.shotQueueId)
.sort((a, b) => b.updatedAt - a.updatedAt)
const cutInHold = held.find(job => job.holdForCutIn)
return cutInHold || held[0] || null
return jobs
.filter(job => (
job.status === 'held'
&& job.shotQueueId
&& job.holdForCutIn
&& !job.pausedByUser
))
.sort((a, b) => b.updatedAt - a.updatedAt)[0] || null
}
function pickWaiting(jobs: StudioJob[]) {
@@ -347,23 +393,27 @@ function repairStaleJobs(jobs: StudioJob[]) {
async function dispatchStudioQueue() {
if (videoJobsBusy()) return
for (const owner of listOwnersWithStudioQueues()) {
const jobs = await mutate(owner, (list) => {
repairStaleJobs(list)
const next = pruneDone(list)
list.splice(0, list.length, ...next)
return list.map(item => structuredClone(item))
const store = await mutateStore(owner, (current) => {
repairStaleJobs(current.jobs)
const next = pruneDone(current.jobs)
current.jobs.splice(0, current.jobs.length, ...next)
return {
paused: current.paused === true,
jobs: current.jobs.map(item => structuredClone(item))
}
})
const cutIn = pickCutIn(jobs)
const cutIn = pickCutIn(store.jobs)
if (cutIn) {
await startStudioJob(cutIn)
return
}
const held = pickHeld(jobs)
if (store.paused) return
const held = pickHeld(store.jobs)
if (held?.shotQueueId) {
await resumeHeldStudioJob(held)
return
}
const waiting = pickWaiting(jobs)
const waiting = pickWaiting(store.jobs)
if (waiting) {
await startStudioJob(waiting)
return
@@ -445,7 +495,10 @@ export async function startStudioJob(item: StudioJob) {
workflow: parseVideoWorkflow(payload.workflow),
useIdentityRefs: payload.useIdentityRefs,
queueAutoRun: extensions.length ? payload.queueAutoRun : false,
queueBudget: extensions.length && payload.queueAutoRun ? extensions.length : 0
queueBudget: extensions.length && payload.queueAutoRun ? extensions.length : 0,
globalLocks: payload.globalLocks,
permanenceRefs: payload.permanenceRefs,
shotPermanenceRefs: payload.shotPermanenceRefs
}
if (extensions.length) {
@@ -472,7 +525,9 @@ export async function startStudioJob(item: StudioJob) {
sound: payload.sound,
useIdentityRefs: payload.useIdentityRefs,
referenceStillIds: payload.referenceStillIds,
initial: { prompt: payload.prompt, duration: payload.duration },
globalLocks: payload.globalLocks,
permanenceRefs: payload.permanenceRefs,
initial: { prompt: payload.prompt, duration: payload.duration, permanenceRefs: payload.shotPermanenceRefs?.[0] },
extensions,
jobId: job.id
})
@@ -549,22 +604,58 @@ export async function onLiveVideoSettled(job: Job) {
const remaining = job.library
? Math.max(0, (job.library.chainTotal || 1) - (job.library.chainStep || 1))
: 0
const interrupted = job.library?.stopAfterCurrent === true
&& remaining > 0
&& job.status !== 'error'
&& job.status !== 'cancelled'
const failed = job.status === 'error' || job.status === 'cancelled'
const snapshot = readStore(owner)
const rowNow = snapshot.jobs.find(item => item.liveJobId === job.id) || snapshot.jobs.find(item => item.status === 'running')
const userPause = !failed && (
rowNow?.pauseAfterCurrent === true
|| (snapshot.paused && !rowNow?.holdForCutIn)
|| (job.library?.stopAfterCurrent === true && !rowNow?.holdForCutIn)
)
const cutInHold = !failed && remaining > 0 && rowNow?.holdForCutIn === true && !userPause
const interrupted = remaining > 0 && (userPause || cutInHold)
if (interrupted && job.status === 'running') {
job.status = 'complete'
}
const hold = interrupted
if (hold) {
if (userPause && remaining > 0) {
await mutateStore(owner, (store) => {
store.paused = true
const row = store.jobs.find(item => item.liveJobId === job.id) || store.jobs.find(item => item.status === 'running')
if (!row) return
row.status = 'held'
row.liveJobId = undefined
row.pauseAfterCurrent = false
row.pausedByUser = true
row.holdForCutIn = false
row.resumeAutoRun = row.payload.queueAutoRun === true || row.resumeAutoRun === true
row.updatedAt = Date.now()
})
} else if (cutInHold) {
await mutate(owner, (jobs) => {
const row = jobs.find(item => item.liveJobId === job.id) || jobs.find(item => item.status === 'running')
if (!row) return
row.status = 'held'
row.liveJobId = undefined
row.holdForCutIn = true
row.resumeAutoRun = row.payload.queueAutoRun === true || job.library?.queueAutoRun === true
row.resumeAutoRun = row.payload.queueAutoRun === true || job.library?.queueAutoRun === true || row.resumeAutoRun === true
row.updatedAt = Date.now()
})
} else if (userPause) {
await mutateStore(owner, (store) => {
store.paused = true
const row = store.jobs.find(item => item.liveJobId === job.id) || store.jobs.find(item => item.status === 'running')
if (!row) return
row.status = job.status === 'cancelled'
? 'cancelled'
: job.status === 'complete'
? 'complete'
: 'error'
row.liveJobId = undefined
row.cutIn = false
row.holdForCutIn = false
row.pauseAfterCurrent = false
row.pausedByUser = false
row.lastError = job.error
row.updatedAt = Date.now()
})
} else {
@@ -577,3 +668,126 @@ export async function onLiveVideoSettled(job: Job) {
}
kickStudioQueue()
}
function applyLiveJobPause(live: Job | undefined, pause: boolean, restoreAutoRun?: boolean) {
if (!live?.library) return
live.library.stopAfterCurrent = pause
if (pause) live.library.queueAutoRun = false
else if (restoreAutoRun) live.library.queueAutoRun = true
}
export async function toggleStudioQueuePause(owner: string, options: {
resume?: boolean
jobId?: string
queueId?: string
} = {}) {
const { pauseShotQueue, setShotQueuePause, listShotQueues, getShotQueue } = await import('~/server/utils/shotQueue')
const storeNow = readStore(owner)
const running = storeNow.jobs.find(job => job.status === 'running')
const target = options.jobId
? storeNow.jobs.find(job => job.id === options.jobId)
: running || storeNow.jobs.find(job => job.pauseAfterCurrent || job.pausedByUser)
const queueArmed = options.queueId
? listShotQueues(owner).some(queue => queue.id === options.queueId && queue.stopAfterCurrent === true)
: false
const armed = target?.pauseAfterCurrent === true || queueArmed || (storeNow.paused && !running)
const shouldPause = options.resume === true ? false : options.resume === false ? true : !armed
if (!shouldPause) {
const heldPaused = storeNow.jobs.filter(job => (
job.status === 'held'
&& Boolean(job.shotQueueId)
&& job.pausedByUser === true
&& (!options.jobId || job.id === options.jobId)
))
const restored = await mutateStore(owner, (store) => {
store.paused = false
for (const job of store.jobs) {
if (options.jobId && job.id !== options.jobId) continue
job.pauseAfterCurrent = false
job.pausedByUser = false
}
if (!options.jobId) {
for (const job of store.jobs) {
job.pauseAfterCurrent = false
job.pausedByUser = false
}
}
return structuredClone(store)
})
const rows = options.jobId
? restored.jobs.filter(job => job.id === options.jobId)
: restored.jobs
for (const row of rows) {
if (row.liveJobId) applyLiveJobPause(getJob(row.liveJobId), false, row.payload.queueAutoRun === true || row.resumeAutoRun === true)
if (row.shotQueueId) await setShotQueuePause(owner, row.shotQueueId, false).catch(() => null)
}
if (!options.jobId) {
for (const queue of listShotQueues(owner)) {
if (queue.stopAfterCurrent) await setShotQueuePause(owner, queue.id, false).catch(() => null)
}
}
if (heldPaused[0] && !videoJobsBusy()) {
await resumeHeldStudioJob(heldPaused[0]).catch(() => null)
} else {
kickStudioQueue()
}
return { paused: false, willPause: false }
}
const updated = await mutateStore(owner, (store) => {
store.paused = true
const row = options.jobId
? store.jobs.find(job => job.id === options.jobId)
: store.jobs.find(job => job.status === 'running')
if (row) {
if (row.status !== 'running' && row.status !== 'held') {
throw createError({ statusCode: 409, statusMessage: 'Nothing is generating on that job' })
}
row.pauseAfterCurrent = row.status === 'running'
row.resumeAutoRun = row.payload.queueAutoRun === true || row.resumeAutoRun === true
row.updatedAt = Date.now()
}
return {
paused: store.paused,
job: row ? structuredClone(row) : null
}
})
if (updated.job?.liveJobId) {
applyLiveJobPause(getJob(updated.job.liveJobId), true)
}
const queueId = updated.job?.shotQueueId || options.queueId
const queued = queueId ? getShotQueue(owner, queueId) : null
if (updated.job && queued?.autoRun) {
await patchStudioJob(owner, updated.job.id, (job) => {
job.resumeAutoRun = true
}).catch(() => null)
}
if (queueId) await pauseShotQueue(owner, queueId).catch(() => null)
if (!updated.job && !queueId) {
for (const queue of listShotQueues(owner)) {
if (queue.status === 'running') await pauseShotQueue(owner, queue.id).catch(() => null)
}
}
const anyGenerating = storeNow.jobs.some(job => job.status === 'running')
|| listShotQueues(owner).some(queue => queue.status === 'running')
if (!updated.job && !queueId && !anyGenerating) {
throw createError({ statusCode: 409, statusMessage: 'Nothing is generating, so there is nothing to pause after' })
}
return { paused: true, willPause: true, jobId: updated.job?.id, queueId }
}
export async function onStudioQueueBurstStarted(owner: string, queueId: string, liveJobId: string) {
await mutateStore(owner, (store) => {
const row = store.jobs.find(job => job.shotQueueId === queueId)
if (!row) return
store.paused = false
row.status = 'running'
row.liveJobId = liveJobId
row.pauseAfterCurrent = false
row.pausedByUser = false
row.holdForCutIn = false
row.updatedAt = Date.now()
})
}
server/utils/videoChain.ts
+53 -10
View File
@@ -9,7 +9,7 @@ import { isLtxWorkflow, isTextToVideo, parseVideoWorkflow, workflowForExtension,
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 { onLiveVideoSettled } from '~/server/utils/studioQueue'
import { onLiveVideoSettled, onStudioQueueBurstStarted } from '~/server/utils/studioQueue'
import {
finishQueueBurst,
getShotQueue,
@@ -20,6 +20,7 @@ import {
setQueueJob,
updateShotQueue
} from '~/server/utils/shotQueue'
import { composeShotPrompt, type PermanenceRef } from '~/utils/globalLocks'
export type ChainImage = { filename: string; data: Buffer; type?: string }
@@ -42,6 +43,9 @@ type VideoChainParams = {
duration: number
useIdentityRefs: boolean
referenceImages: Array<ChainImage | null>
globalLocks?: string
permanenceRefs?: PermanenceRef[]
shotPermanenceRefs?: PermanenceRef[][]
}
function sleep(ms: number) {
@@ -107,7 +111,10 @@ function paramsFromJob(job: Job): VideoChainParams {
workflow,
duration,
useIdentityRefs: library.useIdentityRefs === true,
referenceImages
referenceImages,
globalLocks: library.globalLocks,
permanenceRefs: library.permanenceRefs,
shotPermanenceRefs: library.shotPermanenceRefs
}
}
@@ -158,8 +165,17 @@ export async function queueMiniMax(
emitChainJob(job, { type: 'status', message: queueing, progress: 6 })
await waitForComfySocket(job, 4000)
const graph = buildWorkflow({
const shotIndex = job.library?.chainIndex || 0
const composedPrompt = composeShotPrompt({
globalLocks: job.library?.globalLocks || params.globalLocks,
prompt: params.prompt,
shotIndex,
familyRefs: job.library?.permanenceRefs || params.permanenceRefs,
shotRefs: job.library?.shotPermanenceRefs?.[shotIndex] || params.shotPermanenceRefs?.[shotIndex]
})
const graph = buildWorkflow({
prompt: composedPrompt,
imageName: uploaded.name,
width: params.width,
height: params.height,
@@ -258,6 +274,9 @@ export async function continueQueuedExtensions(
const liveQueue = job.library.queueId
? getShotQueue(job.library.ownerKey, job.library.queueId)
: null
if (liveQueue?.stopAfterCurrent) job.library.stopAfterCurrent = true
if (liveQueue?.autoRun === false) job.library.queueAutoRun = false
if (job.library.stopAfterCurrent) break
const live = liveQueue ? liveSegmentPrompt(liveQueue, i) : extensions[i]
const ext = {
prompt: live.prompt || extensions[i].prompt,
@@ -267,6 +286,17 @@ export async function continueQueuedExtensions(
throw new Error(`Shot ${i + 2} needs a prompt`)
}
emitChainJob(job, { type: 'status', message: 'Waiting 3s buffer...', progress: 1 })
await sleep(3000)
assertJobActive(job)
const pausedQueue = job.library.queueId
? getShotQueue(job.library.ownerKey, job.library.queueId)
: null
if (job.library.stopAfterCurrent || pausedQueue?.stopAfterCurrent) {
job.library.stopAfterCurrent = true
break
}
const remainingAfter = extensions.length - (i + 1)
const lastOfChain = remainingAfter === 0
const lastOfBurst = !autoRun && shotsLeft === 1
@@ -290,10 +320,6 @@ export async function continueQueuedExtensions(
}).catch(() => null)
}
emitChainJob(job, { type: 'status', message: 'Waiting 3s buffer...', progress: 1 })
await sleep(3000)
assertJobActive(job)
copyFileSync(currentPath, part1Path)
try {
await extractLastFrame(part1Path, framePath)
@@ -306,6 +332,11 @@ export async function continueQueuedExtensions(
const frame = readFileSync(framePath)
job.library.extendPart1Path = part1Path
job.library.prompt = ext.prompt
if (live.permanenceRefs?.length) {
const refs = [...(job.library.shotPermanenceRefs || [])]
refs[i + 1] = live.permanenceRefs
job.library.shotPermanenceRefs = refs
}
const sound = await probeHasAudio(currentPath)
const seed = Math.floor(Math.random() * 2_147_483_647)
@@ -367,11 +398,13 @@ export async function continueQueuedExtensions(
export async function continueQueuedExtensionsIfNeeded(job: Job) {
if (!job.library) return
if (job.status === 'error' || job.status === 'cancelled' || job.status === 'complete') return
if (job.library.stopAfterCurrent) return
const remaining = remainingAfterCurrentShot(job.library)
if (!remaining.length) return
const queue = job.library.queueId
? getShotQueue(job.library.ownerKey, job.library.queueId)
: null
if (queue?.stopAfterCurrent) return
const autoRun = queue ? queue.autoRun === true : job.library.queueAutoRun === true
const budget = job.library.queueBudget || 0
if (queue && !autoRun && budget <= 0) return
@@ -407,12 +440,17 @@ export async function runGeneration(job: Job, params: VideoChainParams) {
persist: extensions.length === 0 || !autoRun
})
if (job.library?.queueId && !autoRun) {
const stopNow = job.library?.stopAfterCurrent === true
|| (job.library?.queueId
? getShotQueue(job.library.ownerKey, job.library.queueId)?.stopAfterCurrent === true
: false)
const keepGoing = Boolean(autoRun && !stopNow && extensions.length && job.library)
if (job.library?.queueId && !keepGoing) {
await finishQueueBurst(job.library.ownerKey, job.library.queueId, true).catch(() => null)
setQueueJob(job.library.queueId, null)
}
if (!extensions.length || !job.library || !autoRun) return
if (!keepGoing) return
await continueQueuedExtensions(job, params, { maxShots: extensions.length, autoRun: true })
} finally {
await onLiveVideoSettled(job)
@@ -467,13 +505,18 @@ export async function startQueueBurst(owner: string, queueId: string, count: num
useIdentityRefs: queue.useIdentityRefs,
queueId: queue.id,
queueAutoRun: count === 'all',
queueBudget: n
queueBudget: n,
globalLocks: queue.globalLocks,
permanenceRefs: queue.permanenceRefs,
shotPermanenceRefs: queue.segments.map(segment => segment.permanenceRefs || [])
}
setQueueJob(queue.id, job.id)
await updateShotQueue(owner, queue.id, (next) => {
next.currentJobId = job.id
next.status = 'running'
next.stopAfterCurrent = false
}).catch(() => null)
await onStudioQueueBurstStarted(owner, queue.id, job.id).catch(() => null)
emitChainJob(job, { type: 'status', message: 'Checking ComfyUI...', progress: 1 })
void continueQueuedExtensions(job, paramsFromJob(job), {
server/utils/watch.ts
+9 -1
View File
@@ -4,6 +4,12 @@ import { getJob, restoreJob } from '~/server/utils/jobs'
import type { PendingJob } from '~/server/utils/pending'
import { isLastChainShot, libraryFromPending, pendingFromJob, remainingAfterCurrentShot, writePendingJob, deletePendingJob } from '~/server/utils/pending'
import { syncQueueFromJob } from '~/server/utils/shotQueue'
import { mergePermanenceRefs, type PermanenceRef } from '~/utils/globalLocks'
function clipRefsFromJob(library: NonNullable<Job['library']>): PermanenceRef[] | undefined {
const refs = mergePermanenceRefs(library.permanenceRefs, library.shotPermanenceRefs?.[library.chainIndex || 0])
return refs.length ? refs : undefined
}
function classifyError(message: string) {
const lower = message.toLowerCase()
@@ -206,7 +212,9 @@ export function watchComfyJob(job: Job, options: { persist?: boolean } = {}): Pr
chainIndex: job.library.chainIndex,
workflow: job.library.workflow,
stillId: job.library.stillId,
sound: job.library.sound
sound: job.library.sound,
globalLocks: job.library.globalLocks,
permanenceRefs: clipRefsFromJob(job.library)
})
job.clipId = clip.id
job.savedPromptId = job.promptId