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

Fix generation progress getting stuck at 1% after Comfy finishes.

Browse Source 
Poll Comfy history and the job snapshot so the browser still gets the video when the live stream is buffered or the websocket misses completion.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
Towsty
2026-08-25 18:23:23 -05:00
co-authored by Cursor
parent ca51fddea3
commit c4e6af9f77
8 changed files with 326 additions and 182 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/api/generate.post.ts
+3 -1
View File
@@ -95,9 +95,11 @@ async function runGeneration(
length: number
}
) {
const done = watchComfyJob(job)
job.status = 'uploading'
const uploaded = await uploadImage(params.image)
emitJob(job, { type: 'status', message: 'Queueing MiniMax H3 job...', progress: 6 })
await waitForComfySocket(job, 4000)
const graph = buildWorkflow({
prompt: params.prompt,
@@ -114,5 +116,5 @@ async function runGeneration(
job.promptId = queued.prompt_id
job.status = 'running'
emitJob(job, { type: 'status', message: 'Job queued on ComfyUI', progress: 8 })
await watchComfyJob(job)
await done
}
server/api/generate/[id].get.ts
+8
View File
@@ -0,0 +1,8 @@
export default defineEventHandler((event) => {
const id = getRouterParam(event, 'id')
const job = id ? getJob(id) : undefined
if (!job) {
throw createError({ statusCode: 404, statusMessage: 'Job not found' })
}
return jobSnapshot(job)
})
server/api/generate/[id]/stream.get.ts
+12 -11
View File
@@ -5,22 +5,18 @@ export default defineEventHandler(async (event) => {
throw createError({ statusCode: 404, statusMessage: 'Job not found' })
}
setResponseHeaders(event, {
'Cache-Control': 'no-cache, no-store, no-transform',
'X-Accel-Buffering': 'no',
Connection: 'keep-alive'
})
const stream = createEventStream(event)
const send = async (payload: unknown) => {
await stream.push(JSON.stringify(payload))
}
await send({
type: 'hello',
status: job.status,
message: job.message,
progress: job.progress,
step: job.step,
maxStep: job.maxStep,
promptId: job.promptId,
elapsedMs: Date.now() - job.startedAt
})
await send(jobSnapshot(job))
for (const past of job.events) {
await send(past)
}
@@ -33,7 +29,12 @@ export default defineEventHandler(async (event) => {
})
})
const ping = setInterval(() => {
void send(jobSnapshot(job))
}, 1000)
stream.onClosed(() => {
clearInterval(ping)
unsubscribe()
})
server/utils/comfy.ts
+52 -17
View File
@@ -68,28 +68,63 @@ export async function fetchHistory(promptId: string) {
return (await res.json()) as Record<string, unknown>
}
export function extractVideo(history: Record<string, unknown> | null, promptId: string) {
if (!history) return null
const entry = (history[promptId] || Object.values(history)[0]) as { outputs?: Record<string, Record<string, unknown>> } | undefined
const outputs = entry?.outputs || {}
for (const node of Object.values(outputs)) {
for (const key of ['videos', 'gifs', 'images', 'files']) {
const items = node[key]
if (!Array.isArray(items)) continue
const match = items.find((item: { filename?: string }) => String(item.filename || '').toLowerCase().endsWith('.mp4'))
|| items[0]
if (match?.filename) {
return {
filename: String(match.filename),
subfolder: String(match.subfolder || 'video'),
type: String(match.type || 'output')
}
}
function isVideoFile(item: { filename?: string; format?: string } | null | undefined) {
if (!item) return false
const name = String(item.filename || '').toLowerCase()
const format = String(item.format || '').toLowerCase()
return name.endsWith('.mp4') || name.endsWith('.webm') || format.includes('video') || format.includes('mp4')
}
function normalizeVideo(item: { filename?: string; subfolder?: string; type?: string }) {
return {
filename: String(item.filename),
subfolder: String(item.subfolder || ''),
type: String(item.type || 'output')
}
}
function findVideo(value: unknown, depth = 0): { filename: string; subfolder: string; type: string } | null {
if (!value || typeof value !== 'object' || depth > 8) return null
if (Array.isArray(value)) {
const files = value.filter((item): item is { filename?: string; format?: string; subfolder?: string; type?: string } => Boolean(item && typeof item === 'object'))
const match = files.find(item => String(item.filename || '').toLowerCase().endsWith('.mp4'))
|| files.find(isVideoFile)
if (match?.filename) return normalizeVideo(match)
for (const item of value) {
const nested = findVideo(item, depth + 1)
if (nested) return nested
}
return null
}
const record = value as { filename?: string; format?: string; subfolder?: string; type?: string }
if (record.filename && isVideoFile(record)) return normalizeVideo(record)
for (const nested of Object.values(value as Record<string, unknown>)) {
const found = findVideo(nested, depth + 1)
if (found) return found
}
return null
}
export function extractVideo(history: Record<string, unknown> | null, promptId: string) {
if (!history) return null
const entry = (history[promptId] || Object.values(history)[0]) as { outputs?: Record<string, unknown> } | undefined
return findVideo(entry?.outputs || {})
}
export function inspectHistory(history: Record<string, unknown> | null, promptId: string) {
const video = extractVideo(history, promptId)
if (!history) return { video: null, completed: false, error: null as string | null }
const entry = (history[promptId] || Object.values(history)[0]) as {
status?: { status_str?: string; completed?: boolean; messages?: unknown[] }
} | undefined
const status = entry?.status
const error = status?.status_str === 'error'
? 'ComfyUI reported an execution error'
: null
const completed = Boolean(video) || status?.completed === true || status?.status_str === 'success'
return { video, completed, error }
}
export async function probeComfy() {
try {
const res = await fetch(`${comfyBase()}/system_stats`, { signal: AbortSignal.timeout(2500) })
server/utils/jobs.ts
+20
View File
@@ -42,6 +42,7 @@ export interface Job {
thumb?: Buffer
}
error?: string
socketReady?: boolean
events: JobEvent[]
listeners: Set<(event: JobEvent) => void>
}
@@ -74,6 +75,25 @@ export function getJob(id: string) {
return jobs.get(id)
}
export function jobSnapshot(job: Job) {
return {
type: 'snapshot' as const,
status: job.status,
message: job.message,
progress: job.progress,
step: job.step,
maxStep: job.maxStep,
promptId: job.promptId,
elapsedMs: Date.now() - job.startedAt,
filename: job.video?.filename,
subfolder: job.video?.subfolder,
mediaType: job.video?.type,
clipId: job.clipId,
hideThumbnail: job.hideThumbnail,
error: job.error
}
}
export function emitJob(job: Job, event: JobEvent) {
if (event.message) job.message = event.message
if (typeof event.progress === 'number') job.progress = event.progress
server/utils/watch.ts
+161 -127
View File
@@ -3,7 +3,7 @@ import type { Job } from '~/server/utils/jobs'
function classifyError(message: string) {
const lower = message.toLowerCase()
if (lower.includes('out of memory') || lower.includes('cuda') && lower.includes('alloc') || lower.includes('vram')) {
if (lower.includes('out of memory') || (lower.includes('cuda') && lower.includes('alloc')) || lower.includes('vram')) {
return 'ComfyUI VRAM allocation failed. Try Turbo (8 steps) or a smaller aspect ratio.'
}
if (lower.includes('econnrefused') || lower.includes('unreachable') || lower.includes('fetch failed')) {
@@ -15,154 +15,188 @@ function classifyError(message: string) {
return message
}
export async function watchComfyJob(job: Job) {
function sleep(ms: number) {
return new Promise(resolve => setTimeout(resolve, ms))
}
export function watchComfyJob(job: Job): Promise<void> {
const ws = new WebSocket(comfyWsUrl(job.clientId))
let settled = false
let pollTimer: ReturnType<typeof setInterval> | null = null
let timeout: ReturnType<typeof setTimeout> | null = null
const finish = async (error?: string) => {
if (settled) return
settled = true
try { ws.close() } catch { /* ignore */ }
if (error) {
job.status = job.status === 'cancelled' ? 'cancelled' : 'error'
job.error = classifyError(error)
emitJob(job, { type: 'error', error: job.error, message: job.error })
return
return new Promise((resolve) => {
const cleanup = () => {
if (timeout) clearTimeout(timeout)
if (pollTimer) clearInterval(pollTimer)
timeout = null
pollTimer = null
try { ws.close() } catch { /* ignore */ }
}
emitJob(job, { type: 'status', message: 'Fetching output...', progress: 96 })
const history = await fetchHistory(job.promptId || '')
const video = extractVideo(history, job.promptId || '')
if (!video) {
job.status = 'error'
job.error = 'Workflow finished but no MP4 was found in ComfyUI history.'
emitJob(job, { type: 'error', error: job.error, message: job.error })
return
}
job.video = video
if (job.library) {
emitJob(job, { type: 'status', message: 'Saving to library...', progress: 98 })
const finish = async (error?: string) => {
if (settled) return
settled = true
cleanup()
try {
const buffer = await downloadComfyVideo(video)
const clip = await saveClip({
folderId: job.library.folderId,
prompt: job.library.prompt,
aspect: job.library.aspect,
width: job.library.width,
height: job.library.height,
steps: job.library.steps,
turbo: job.library.turbo,
seed: job.library.seed,
hideThumbnail: job.library.hideThumbnail,
video: buffer,
thumb: job.library.hideThumbnail ? null : job.library.thumb
if (error) {
job.status = job.status === 'cancelled' ? 'cancelled' : 'error'
job.error = classifyError(error)
emitJob(job, { type: 'error', error: job.error, message: job.error })
return
}
emitJob(job, { type: 'status', message: 'Fetching output...', progress: 96 })
const history = await fetchHistory(job.promptId || '')
const video = extractVideo(history, job.promptId || '')
if (!video) {
job.status = 'error'
job.error = 'Workflow finished but no MP4 was found in ComfyUI history.'
emitJob(job, { type: 'error', error: job.error, message: job.error })
return
}
job.video = video
if (job.library) {
emitJob(job, { type: 'status', message: 'Saving to library...', progress: 98 })
try {
const buffer = await downloadComfyVideo(video)
const clip = await saveClip({
folderId: job.library.folderId,
prompt: job.library.prompt,
aspect: job.library.aspect,
width: job.library.width,
height: job.library.height,
steps: job.library.steps,
turbo: job.library.turbo,
seed: job.library.seed,
hideThumbnail: job.library.hideThumbnail,
video: buffer,
thumb: job.library.hideThumbnail ? null : job.library.thumb
})
job.clipId = clip.id
job.hideThumbnail = clip.hideThumbnail
job.library.thumb = undefined
} catch (saveError) {
const message = saveError instanceof Error ? saveError.message : String(saveError)
job.status = 'error'
job.error = `Video generated but library save failed: ${message}`
emitJob(job, { type: 'error', error: job.error, message: job.error })
return
}
}
job.status = 'complete'
emitJob(job, {
type: 'complete',
message: 'Video ready',
progress: 100,
filename: video.filename,
subfolder: video.subfolder,
mediaType: video.type,
clipId: job.clipId,
hideThumbnail: job.hideThumbnail
})
job.clipId = clip.id
job.hideThumbnail = clip.hideThumbnail
job.library.thumb = undefined
} catch (error) {
const message = error instanceof Error ? error.message : String(error)
job.status = 'error'
job.error = `Video generated but library save failed: ${message}`
emitJob(job, { type: 'error', error: job.error, message: job.error })
return
} finally {
resolve()
}
}
job.status = 'complete'
emitJob(job, {
type: 'complete',
message: 'Video ready',
progress: 100,
filename: video.filename,
subfolder: video.subfolder,
mediaType: video.type,
clipId: job.clipId,
hideThumbnail: job.hideThumbnail
})
}
const timeout = setTimeout(() => {
finish('Timed out waiting for ComfyUI (15 minutes).')
}, 15 * 60 * 1000)
ws.addEventListener('open', () => {
emitJob(job, { type: 'status', message: 'Connected to ComfyUI', progress: 8 })
})
ws.addEventListener('error', () => {
if (!settled) finish('ComfyUI WebSocket connection dropped.')
})
ws.addEventListener('close', () => {
if (!settled) {
// Fall back to history polling in case completion arrived as HTTP only.
setTimeout(async () => {
if (settled) return
const history = await fetchHistory(job.promptId || '')
if (extractVideo(history, job.promptId || '')) {
clearTimeout(timeout)
const pollHistory = async () => {
if (settled || !job.promptId) return
try {
const inspected = inspectHistory(await fetchHistory(job.promptId), job.promptId)
if (inspected.error) {
await finish(inspected.error)
return
}
if (inspected.video || inspected.completed) {
await finish()
}
}, 1500)
} catch {
// Transient ComfyUI history misses are expected while the graph is still running.
}
}
})
ws.addEventListener('message', async (event) => {
let payload: { type?: string; data?: Record<string, unknown> }
try {
payload = JSON.parse(String(event.data))
} catch {
return
}
const data = payload.data || {}
if (job.promptId && data.prompt_id && data.prompt_id !== job.promptId) return
timeout = setTimeout(() => {
void finish('Timed out waiting for ComfyUI (15 minutes).')
}, 15 * 60 * 1000)
if (payload.type === 'progress') {
const value = Number(data.value || 0)
const max = Number(data.max || 1)
const pct = 12 + Math.round((value / Math.max(max, 1)) * 70)
job.status = 'running'
pollTimer = setInterval(() => {
void pollHistory()
}, 2000)
ws.addEventListener('open', () => {
job.socketReady = true
emitJob(job, { type: 'status', message: 'Connected to ComfyUI', progress: Math.max(job.progress, 4) })
})
ws.addEventListener('error', () => {
if (settled) return
emitJob(job, {
type: 'progress',
step: value,
maxStep: max,
progress: pct,
node: data.node ? String(data.node) : null,
message: `Sampling step ${value}/${max} (${Math.round((value / Math.max(max, 1)) * 100)}%)`
type: 'status',
message: job.promptId
? 'Live socket unavailable, polling ComfyUI history...'
: 'Waiting for ComfyUI socket...',
progress: Math.max(job.progress, 4)
})
}
})
if (payload.type === 'executing') {
const node = data.node === null || data.node === undefined ? null : String(data.node)
if (node === null) {
clearTimeout(timeout)
await finish()
ws.addEventListener('message', async (event) => {
let payload: { type?: string; data?: Record<string, unknown> }
try {
payload = JSON.parse(String(event.data))
} catch {
return
}
const label = NODE_LABELS[node] || `Running node ${node}`
const encoding = isEncodingNode(node)
emitJob(job, {
type: 'executing',
node,
progress: encoding ? 90 : Math.max(job.progress, 10),
message: encoding ? 'Encoding video...' : `${label}...`
})
}
const data = payload.data || {}
if (job.promptId && data.prompt_id && data.prompt_id !== job.promptId) return
if (payload.type === 'execution_error') {
clearTimeout(timeout)
const message = String(data.exception_message || data.message || 'ComfyUI node execution failed')
await finish(message)
}
if (payload.type === 'progress') {
const value = Number(data.value || 0)
const max = Number(data.max || 1)
const pct = 12 + Math.round((value / Math.max(max, 1)) * 70)
job.status = 'running'
emitJob(job, {
type: 'progress',
step: value,
maxStep: max,
progress: pct,
node: data.node ? String(data.node) : null,
message: `Sampling step ${value}/${max} (${Math.round((value / Math.max(max, 1)) * 100)}%)`
})
}
if (payload.type === 'execution_interrupted') {
clearTimeout(timeout)
job.status = 'cancelled'
await finish('Job interrupted.')
}
if (payload.type === 'executing') {
const node = data.node === null || data.node === undefined ? null : String(data.node)
if (node === null) {
await finish()
return
}
const label = NODE_LABELS[node] || `Running node ${node}`
const encoding = isEncodingNode(node)
emitJob(job, {
type: 'executing',
node,
progress: encoding ? 90 : Math.max(job.progress, 10),
message: encoding ? 'Encoding video...' : `${label}...`
})
}
if (payload.type === 'execution_error') {
const message = String(data.exception_message || data.message || 'ComfyUI node execution failed')
await finish(message)
}
if (payload.type === 'execution_interrupted') {
job.status = 'cancelled'
await finish('Job interrupted.')
}
})
})
}
return () => {
clearTimeout(timeout)
try { ws.close() } catch { /* ignore */ }
export async function waitForComfySocket(job: Job, ms = 4000) {
const started = Date.now()
while (Date.now() - started < ms) {
if (job.socketReady) return
await sleep(100)
}
}