TowstyandCursor
5df9e2d8ac
Settings and the header menu can cancel invisible queue/pending work and force-stop Beast when Sleep is blocked. Co-authored-by: Cursor <cursoragent@cursor.com>
495 lines
16 KiB
TypeScript
495 lines
16 KiB
TypeScript
import { execFile, spawn } from 'node:child_process'
|
|
import { promisify } from 'node:util'
|
|
import { getComfyHost, setComfyHostOverride, comfyConfigured } from '~/server/utils/comfy'
|
|
|
|
const execFileAsync = promisify(execFile)
|
|
|
|
export type ComfyLifecycleState = 'online' | 'booting' | 'busy' | 'starting' | 'offline'
|
|
|
|
export interface ComfyLifecycleStatus {
|
|
state: ComfyLifecycleState
|
|
message: string
|
|
host: string
|
|
processRunning?: boolean
|
|
queueRunning?: number
|
|
queuePending?: number
|
|
}
|
|
|
|
type StatusFn = (status: ComfyLifecycleStatus) => void
|
|
|
|
let gate: Promise<unknown> = Promise.resolve()
|
|
let lastStartAt = 0
|
|
|
|
function sleep(ms: number) {
|
|
return new Promise(resolve => setTimeout(resolve, ms))
|
|
}
|
|
|
|
function log(event: string, detail: Record<string, unknown> = {}) {
|
|
console.log(JSON.stringify({
|
|
src: 'comfy-lifecycle',
|
|
event,
|
|
at: new Date().toISOString(),
|
|
...detail
|
|
}))
|
|
}
|
|
|
|
function settings() {
|
|
const config = useRuntimeConfig()
|
|
return {
|
|
host: getComfyHost(),
|
|
launchCmd: String(config.comfyLaunchCmd || process.env.COMFY_LAUNCH_CMD || '').trim(),
|
|
controlUrl: String(config.comfyControlUrl || process.env.COMFY_CONTROL_URL || '').replace(/\/$/, ''),
|
|
controlToken: String(config.comfyControlToken || process.env.COMFY_CONTROL_TOKEN || ''),
|
|
healthTimeoutMs: 2500,
|
|
bootPollAttempts: Number(process.env.COMFY_BOOT_POLL_ATTEMPTS || 20),
|
|
bootPollDelayMs: 3000,
|
|
startTimeoutMs: Number(process.env.COMFY_START_TIMEOUT_MS || 300_000),
|
|
busyWaitMs: Number(process.env.COMFY_BUSY_WAIT_MS || 180_000)
|
|
}
|
|
}
|
|
|
|
export async function checkComfyHttp(timeoutMs = 5000, host?: string) {
|
|
let resolved = host || ''
|
|
try {
|
|
if (!resolved) {
|
|
if (!comfyConfigured()) return { ok: false as const, host: '', status: 0, stats: null, timedOut: false }
|
|
resolved = getComfyHost()
|
|
}
|
|
const res = await fetch(`${resolved}/system_stats`, { signal: AbortSignal.timeout(timeoutMs) })
|
|
if (!res.ok) return { ok: false as const, host: resolved, status: res.status, stats: null, timedOut: false }
|
|
const stats = await res.json().catch(() => null)
|
|
return { ok: true as const, host: resolved, status: res.status, stats, timedOut: false }
|
|
} catch (error) {
|
|
const timedOut = error instanceof Error && (error.name === 'TimeoutError' || error.name === 'AbortError' || /timeout/i.test(error.message))
|
|
return { ok: false as const, host: resolved, status: 0, stats: null, timedOut }
|
|
}
|
|
}
|
|
|
|
export async function fetchQueue() {
|
|
if (!comfyConfigured()) return { running: 0, pending: 0, ok: false }
|
|
try {
|
|
const host = getComfyHost()
|
|
const res = await fetch(`${host}/queue`, { signal: AbortSignal.timeout(5000) })
|
|
if (!res.ok) return { running: 0, pending: 0, ok: false }
|
|
const payload = await res.json() as { queue_running?: unknown[]; queue_pending?: unknown[] }
|
|
return {
|
|
running: Array.isArray(payload.queue_running) ? payload.queue_running.length : 0,
|
|
pending: Array.isArray(payload.queue_pending) ? payload.queue_pending.length : 0,
|
|
ok: true
|
|
}
|
|
} catch {
|
|
return { running: 0, pending: 0, ok: false }
|
|
}
|
|
}
|
|
|
|
async function controlRequest(path: string, method = 'GET', timeoutMs = 5000, body?: Record<string, unknown>) {
|
|
const { controlUrl, controlToken } = settings()
|
|
if (!controlUrl) return null
|
|
try {
|
|
const res = await fetch(`${controlUrl}${path}`, {
|
|
method,
|
|
headers: {
|
|
Accept: 'application/json',
|
|
...(body ? { 'Content-Type': 'application/json' } : {}),
|
|
...(controlToken ? { Authorization: `Bearer ${controlToken}` } : {})
|
|
},
|
|
...(body ? { body: JSON.stringify(body) } : {}),
|
|
signal: AbortSignal.timeout(timeoutMs)
|
|
})
|
|
const payload = await res.json().catch(() => null) as {
|
|
ok?: boolean
|
|
http?: boolean
|
|
process?: boolean
|
|
processUp?: boolean
|
|
asleep?: boolean
|
|
lastActivityAt?: string | null
|
|
idleMs?: number
|
|
pid?: number
|
|
started?: boolean
|
|
already?: boolean
|
|
port?: number | null
|
|
proxyPort?: number | null
|
|
stopped?: boolean
|
|
killed?: unknown
|
|
forced?: boolean
|
|
error?: string
|
|
message?: string
|
|
training?: { busy?: boolean; jobId?: string | null; status?: string | null; name?: string | null; message?: string }
|
|
running?: number
|
|
pending?: number
|
|
status?: number
|
|
} | null
|
|
if (!res.ok) {
|
|
return {
|
|
ok: false as const,
|
|
status: res.status,
|
|
error: payload?.error,
|
|
message: payload?.message,
|
|
training: payload?.training,
|
|
running: payload?.running,
|
|
pending: payload?.pending,
|
|
killed: payload?.killed,
|
|
forced: payload?.forced
|
|
}
|
|
}
|
|
return payload || { ok: true }
|
|
} catch {
|
|
return null
|
|
}
|
|
}
|
|
|
|
export async function fetchComfyControlStatus() {
|
|
return await controlRequest('/status')
|
|
}
|
|
|
|
function throwIfTrainingBusy(remote: { training?: { busy?: boolean; message?: string }; error?: string; message?: string } | null) {
|
|
const busy = Boolean(remote?.training?.busy || remote?.error === 'train-busy')
|
|
if (!busy) return
|
|
throw createError({
|
|
statusCode: 409,
|
|
statusMessage: remote?.training?.message || remote?.message || 'AITraining is using the GPU. Stop that job before poking Comfy.'
|
|
})
|
|
}
|
|
|
|
export async function requestComfyWake() {
|
|
const { controlUrl } = settings()
|
|
if (!controlUrl) {
|
|
throw createError({
|
|
statusCode: 503,
|
|
statusMessage: 'COMFY_CONTROL_URL is not set. The studio cannot poke the Beast host agent.'
|
|
})
|
|
}
|
|
const remote = await controlRequest('/start', 'POST', 25_000)
|
|
if (!remote) {
|
|
throw createError({
|
|
statusCode: 503,
|
|
statusMessage: 'Beast host agent is unreachable. Confirm it is running on the desktop.'
|
|
})
|
|
}
|
|
throwIfTrainingBusy(remote)
|
|
if (remote.ok === false) {
|
|
throw createError({
|
|
statusCode: Number(remote.status) || 502,
|
|
statusMessage: String(remote.message || remote.error || 'Beast host agent refused to start Comfy.')
|
|
})
|
|
}
|
|
return {
|
|
ok: true,
|
|
started: Boolean(remote.started),
|
|
already: Boolean(remote.already),
|
|
asleep: false,
|
|
port: remote.port ?? null,
|
|
proxyPort: remote.proxyPort ?? null
|
|
}
|
|
}
|
|
|
|
export async function requestComfySleep() {
|
|
const { controlUrl } = settings()
|
|
if (!controlUrl) {
|
|
throw createError({
|
|
statusCode: 503,
|
|
statusMessage: 'COMFY_CONTROL_URL is not set. The studio cannot poke the Beast host agent.'
|
|
})
|
|
}
|
|
const remote = await controlRequest('/stop', 'POST', 25_000)
|
|
if (!remote) {
|
|
throw createError({
|
|
statusCode: 503,
|
|
statusMessage: 'Beast host agent is unreachable. Confirm it is running on the desktop.'
|
|
})
|
|
}
|
|
if (remote.status === 409 || remote.error === 'queue-busy') {
|
|
const running = Number(remote.running || 0)
|
|
const pending = Number(remote.pending || 0)
|
|
throw createError({
|
|
statusCode: 409,
|
|
statusMessage: `Comfy is still working (${running} running, ${pending} queued). Cancel the job first, then sleep the GPU.`
|
|
})
|
|
}
|
|
if (remote.ok === false) {
|
|
throw createError({
|
|
statusCode: Number(remote.status) || 502,
|
|
statusMessage: 'Beast host agent refused to stop Comfy.'
|
|
})
|
|
}
|
|
return {
|
|
ok: true,
|
|
stopped: true,
|
|
asleep: true,
|
|
killed: remote.killed ?? null
|
|
}
|
|
}
|
|
|
|
/** Interrupt + clear Comfy queue, then kill the process even if a job is wedged. */
|
|
export async function requestComfyForceReset() {
|
|
const { controlUrl } = settings()
|
|
if (!controlUrl) {
|
|
throw createError({
|
|
statusCode: 503,
|
|
statusMessage: 'COMFY_CONTROL_URL is not set. The studio cannot poke the Beast host agent.'
|
|
})
|
|
}
|
|
const { interruptComfy } = await import('~/server/utils/comfy')
|
|
await interruptComfy().catch(() => false)
|
|
const remote = await controlRequest('/stop', 'POST', 30_000, { force: true })
|
|
if (!remote) {
|
|
throw createError({
|
|
statusCode: 503,
|
|
statusMessage: 'Beast host agent is unreachable. Confirm it is running on the desktop.'
|
|
})
|
|
}
|
|
if (remote.ok === false) {
|
|
throw createError({
|
|
statusCode: Number(remote.status) || 502,
|
|
statusMessage: String(remote.message || remote.error || 'Beast host agent refused to force-stop Comfy.')
|
|
})
|
|
}
|
|
return {
|
|
ok: true,
|
|
stopped: true,
|
|
asleep: true,
|
|
forced: true,
|
|
killed: remote.killed ?? null
|
|
}
|
|
}
|
|
|
|
async function findLocalComfyProcess() {
|
|
const platform = process.platform
|
|
try {
|
|
if (platform === 'win32') {
|
|
const { stdout } = await execFileAsync('tasklist', ['/FO', 'CSV', '/NH'], { timeout: 4000, windowsHide: true })
|
|
const hit = stdout.split(/\r?\n/).some(line => /comfy|python/i.test(line) && /comfy/i.test(line))
|
|
if (hit) return true
|
|
const { stdout: verbose } = await execFileAsync('wmic', ['process', 'where', "name='python.exe'", 'get', 'commandline'], { timeout: 4000, windowsHide: true }).catch(() => ({ stdout: '' }))
|
|
return /comfyui|main\.py/i.test(verbose)
|
|
}
|
|
const { stdout } = await execFileAsync('ps', ['-ax', '-o', 'args='], { timeout: 4000 })
|
|
return /comfyui|ComfyUI|main\.py.*8188/i.test(stdout)
|
|
} catch {
|
|
return false
|
|
}
|
|
}
|
|
|
|
async function processRunning() {
|
|
const remote = await controlRequest('/status')
|
|
if (remote?.process) return true
|
|
return findLocalComfyProcess()
|
|
}
|
|
|
|
function spawnLaunchCommand(command: string) {
|
|
log('spawn', { command: command.slice(0, 180) })
|
|
const win = process.platform === 'win32'
|
|
const exe = win
|
|
? command.match(/start\s+""\s+"([^"]+)"/i)?.[1] || command.match(/^"([^"]+\.exe)"\s*$/i)?.[1]
|
|
: ''
|
|
const child = exe
|
|
? spawn(exe, [], { detached: true, stdio: 'ignore', windowsHide: true })
|
|
: spawn(command, {
|
|
shell: true,
|
|
detached: !win,
|
|
stdio: 'ignore',
|
|
windowsHide: true
|
|
})
|
|
child.unref()
|
|
}
|
|
|
|
async function startComfy() {
|
|
const now = Date.now()
|
|
if (now - lastStartAt < 15_000) {
|
|
log('start-skipped', { reason: 'recent-start' })
|
|
return { started: false, reason: 'recent-start' as const }
|
|
}
|
|
lastStartAt = now
|
|
|
|
const remote = await controlRequest('/start', 'POST', 25_000)
|
|
throwIfTrainingBusy(remote)
|
|
if (remote?.ok || remote?.started) {
|
|
log('start-remote', { pid: remote.pid || null })
|
|
return { started: true, reason: 'remote' as const }
|
|
}
|
|
|
|
const { launchCmd } = settings()
|
|
if (launchCmd) {
|
|
spawnLaunchCommand(launchCmd)
|
|
return { started: true, reason: 'launch-cmd' as const }
|
|
}
|
|
|
|
if (process.platform === 'win32') {
|
|
const fallback = process.env.LOCALAPPDATA
|
|
? `"${process.env.LOCALAPPDATA}\\Programs\\Comfy Desktop\\Comfy Desktop.exe"`
|
|
: ''
|
|
if (fallback) {
|
|
spawnLaunchCommand(`start "" ${fallback}`)
|
|
return { started: true, reason: 'comfy-desktop' as const }
|
|
}
|
|
}
|
|
|
|
return { started: false, reason: 'no-launcher' as const }
|
|
}
|
|
|
|
async function pollUntilHealthy(attempts: number, delayMs: number, onStatus: StatusFn, state: ComfyLifecycleState, message: string) {
|
|
const host = getComfyHost()
|
|
for (let i = 1; i <= attempts; i++) {
|
|
onStatus({ state, message: `${message} (${i}/${attempts})`, host, processRunning: true })
|
|
const health = await checkComfyHttp()
|
|
if (health.ok) return true
|
|
if (i < attempts) await sleep(delayMs)
|
|
}
|
|
return false
|
|
}
|
|
|
|
async function waitWhileBusy(onStatus: StatusFn) {
|
|
const { busyWaitMs } = settings()
|
|
const host = getComfyHost()
|
|
const started = Date.now()
|
|
while (Date.now() - started < busyWaitMs) {
|
|
const queue = await fetchQueue()
|
|
if (queue.running === 0) return
|
|
onStatus({
|
|
state: 'busy',
|
|
message: `ComfyUI is busy with another prompt (${queue.running} running, ${queue.pending} queued). Waiting up to 3 minutes, then this job will be saved.`,
|
|
host,
|
|
queueRunning: queue.running,
|
|
queuePending: queue.pending
|
|
})
|
|
log('busy', queue)
|
|
await sleep(2000)
|
|
}
|
|
throw createError({
|
|
statusCode: 409,
|
|
statusMessage: 'ComfyUI is still busy after 3 minutes. Your prompt, still, and settings were saved.',
|
|
data: { code: 'COMFY_BUSY' }
|
|
})
|
|
}
|
|
|
|
function hostWithPort(host: string, port: number) {
|
|
try {
|
|
const url = new URL(host)
|
|
url.port = String(port)
|
|
return url.origin
|
|
} catch {
|
|
return host
|
|
}
|
|
}
|
|
|
|
async function ensureUnlocked(onStatus: StatusFn, skipBusyWait = false) {
|
|
const cfg = settings()
|
|
const host = cfg.host
|
|
const lock = await controlRequest('/status')
|
|
throwIfTrainingBusy(lock)
|
|
onStatus({ state: 'offline', message: 'Checking ComfyUI...', host })
|
|
|
|
const health = await checkComfyHttp(cfg.healthTimeoutMs)
|
|
if (health.ok) {
|
|
log('http-ok', { host })
|
|
if (!skipBusyWait) await waitWhileBusy(onStatus)
|
|
onStatus({ state: 'online', message: 'ComfyUI online', host })
|
|
return
|
|
}
|
|
if (health.timedOut) {
|
|
log('http-busy', { host, timeoutMs: cfg.healthTimeoutMs })
|
|
const retry = await checkComfyHttp(Math.max(cfg.healthTimeoutMs, 12_000))
|
|
if (retry.ok) {
|
|
if (!skipBusyWait) await waitWhileBusy(onStatus)
|
|
onStatus({ state: 'online', message: 'ComfyUI online', host })
|
|
return
|
|
}
|
|
onStatus({
|
|
state: 'busy',
|
|
message: 'ComfyUI is loaded but slow to answer. Not starting another instance.',
|
|
host,
|
|
processRunning: true
|
|
})
|
|
if (!skipBusyWait) await waitWhileBusy(onStatus)
|
|
onStatus({ state: 'online', message: 'ComfyUI online', host, processRunning: true })
|
|
return
|
|
}
|
|
|
|
log('http-fail', { host })
|
|
const remote = await controlRequest('/status')
|
|
if (remote?.http && remote.proxyPort) {
|
|
const proxyUrl = hostWithPort(host, remote.proxyPort)
|
|
onStatus({
|
|
state: 'booting',
|
|
message: `Configured Comfy port is blocked. Trying the desktop LAN proxy on ${remote.proxyPort}...`,
|
|
host: proxyUrl,
|
|
processRunning: true
|
|
})
|
|
const viaProxy = await checkComfyHttp(cfg.healthTimeoutMs, proxyUrl)
|
|
if (viaProxy.ok) {
|
|
setComfyHostOverride(proxyUrl)
|
|
log('http-ok-proxy', { host: proxyUrl, localPort: remote.port || null })
|
|
if (!skipBusyWait) await waitWhileBusy(onStatus)
|
|
onStatus({ state: 'online', message: 'ComfyUI online', host: proxyUrl, processRunning: true })
|
|
return
|
|
}
|
|
log('http-fail-proxy', { host: proxyUrl, localPort: remote.port || null })
|
|
}
|
|
const running = await processRunning()
|
|
if (running) {
|
|
onStatus({
|
|
state: 'booting',
|
|
message: 'ComfyUI process is up but the API is not ready yet. Waiting...',
|
|
host,
|
|
processRunning: true
|
|
})
|
|
const ready = await pollUntilHealthy(cfg.bootPollAttempts, cfg.bootPollDelayMs, onStatus, 'booting', 'Waiting for ComfyUI to finish booting')
|
|
if (ready) {
|
|
if (!skipBusyWait) await waitWhileBusy(onStatus)
|
|
onStatus({ state: 'online', message: 'ComfyUI online', host, processRunning: true })
|
|
return
|
|
}
|
|
throw createError({
|
|
statusCode: 503,
|
|
statusMessage: 'ComfyUI is running but the API never came up. Not launching a second instance.'
|
|
})
|
|
}
|
|
|
|
if (!running) {
|
|
onStatus({
|
|
state: 'starting',
|
|
message: 'ComfyUI is offline. Starting the desktop instance...',
|
|
host,
|
|
processRunning: false
|
|
})
|
|
}
|
|
const started = await startComfy()
|
|
if (!started.started) {
|
|
throw createError({
|
|
statusCode: 503,
|
|
statusMessage: 'ComfyUI is offline and no launch command is configured. Start Comfy Desktop, or set COMFY_LAUNCH_CMD / COMFY_CONTROL_URL.'
|
|
})
|
|
}
|
|
|
|
const deadline = Date.now() + cfg.startTimeoutMs
|
|
let attempt = 0
|
|
while (Date.now() < deadline) {
|
|
attempt += 1
|
|
onStatus({
|
|
state: 'starting',
|
|
message: `Starting ComfyUI... waiting for /system_stats (${attempt})`,
|
|
host
|
|
})
|
|
const next = await checkComfyHttp()
|
|
if (next.ok) {
|
|
log('started', { attempt, reason: started.reason })
|
|
if (!skipBusyWait) await waitWhileBusy(onStatus)
|
|
onStatus({ state: 'online', message: 'ComfyUI online', host, processRunning: true })
|
|
return
|
|
}
|
|
await sleep(2000)
|
|
}
|
|
|
|
throw createError({
|
|
statusCode: 503,
|
|
statusMessage: 'Started ComfyUI but the API never answered. A cold start can take a couple of minutes. Confirm the host agent launched the MiniMax install headlessly.'
|
|
})
|
|
}
|
|
|
|
export function ensureComfyReady(onStatus: StatusFn = () => undefined, options?: { skipBusyWait?: boolean }) {
|
|
const skipBusyWait = options?.skipBusyWait === true
|
|
const run = gate.then(() => ensureUnlocked(onStatus, skipBusyWait))
|
|
gate = run.then(() => undefined, () => undefined)
|
|
return run
|
|
}
|