From e02f031378f7678cb402eeb239b1278926a684ee Mon Sep 17 00:00:00 2001 From: Codex Date: Thu, 12 Nov 2026 11:24:49 +0800 Subject: [PATCH] Show ComfyUI task progress and measured sampling time estimates --- README.md | 8 + huobao-source/backend/src/routes/hd.ts | 4 +- .../src/services/adapters/minimax-video.ts | 3 +- .../backend/src/services/adapters/types.ts | 1 + .../backend/src/services/generation.ts | 6 + .../backend/src/services/task-progress.ts | 29 +++ .../frontend/app/components/HdVideoStudio.vue | 1 + .../frontend/app/components/TaskProgress.vue | 64 +++++++ .../frontend/app/views/drama/episode.vue | 3 + .../browser-test/verify_progress_live.cjs | 18 ++ .../browser-test/verify_progress_ui.cjs | 39 ++++ integration-work/comfy_progress.py | 169 +++++++++++++++++ integration-work/general-normalizer-panels.py | 175 +++++++++++++++++- integration-work/install_comfy_progress.py | 40 ++++ integration-work/test_comfy_progress.py | 46 +++++ .../test_gpu_replacement_references.py | 16 +- integration-work/test_task_progress.cjs | 11 ++ integration-work/vdn-normalizer-panels.py | 175 +++++++++++++++++- integration-work/verify_progress_ui.py | 22 +++ model-bridge/bridge.py | 1 + model-bridge/build_upgrade.py | 2 +- model-bridge/test_bridge.py | 3 +- 22 files changed, 826 insertions(+), 10 deletions(-) create mode 100644 huobao-source/backend/src/services/task-progress.ts create mode 100644 huobao-source/frontend/app/components/TaskProgress.vue create mode 100644 integration-work/browser-test/verify_progress_live.cjs create mode 100644 integration-work/browser-test/verify_progress_ui.cjs create mode 100644 integration-work/comfy_progress.py create mode 100644 integration-work/install_comfy_progress.py create mode 100644 integration-work/test_comfy_progress.py create mode 100644 integration-work/test_task_progress.cjs create mode 100644 integration-work/verify_progress_ui.py diff --git a/README.md b/README.md index 43fb6da..4c76b5a 100644 --- a/README.md +++ b/README.md @@ -26,6 +26,14 @@ 前镜头版本或起止设定改变后，已生成的下游视频显示核对提示，旧版保留，不自动重跑。依赖链按顺序提交，独立镜头仍可多卡并行。VDN-H3 高清制作仍使用其参考图工作流，不支持固定首帧接续。 +## 视频进度与预计耗时 + +视频预览下方、任务列表和高清制作页会显示排队、模型加载、采样、解码与编码阶段，以及已提交时长。采样步数来自对应 ComfyUI 任务的实时消息；积累足够的步速记录后显示当前采样阶段的剩余时间区间，后续解码、编码和保存另计。 + +整段视频耗时只参考同显卡、同工作流及规格的成功记录，不包含排队与下载；样本不足时不报估算。采样完成不代表成片完成，连接中断或数据过期会隐藏旧估算。进度读取不会取消、重新提交或改换显卡。 + +服务器的进度扩展源文件为 `integration-work/comfy_progress.py`，由 `install_comfy_progress.py` 备份后嵌入现有适配器；无需重启 ComfyUI。图片、视频及私有配置仍不进入 Git。 + ## 目录 | 目录或文件 | 用途 | diff --git a/huobao-source/backend/src/routes/hd.ts b/huobao-source/backend/src/routes/hd.ts index 0393c40..9f3703a 100644 --- a/huobao-source/backend/src/routes/hd.ts +++ b/huobao-source/backend/src/routes/hd.ts @@ -16,7 +16,7 @@ import { taskRouting } from '../services/worker-routing.js' const app = new Hono() const MODEL = 'MiniMax-H3-VDN-H3' -type HdParams = { variant?: string; originalVideoUrl?: string; resolution?: string; gpu?: string } +type HdParams = { variant?: string; originalVideoUrl?: string; resolution?: string; gpu?: string; progress?: import('../services/task-progress').TaskProgress } function params(raw: string | null): HdParams { try { return JSON.parse(raw || '{}') } catch { return {} } } @@ -51,7 +51,7 @@ app.get('/episodes/:id', async c => { return { id: sb.id, title: sb.title, order: sb.storyboardNumber, current: sb.videoUrl, original, busy: tasks.some(t => t.status === 'processing'), busyTaskId: [...tasks].reverse().find(t => t.status === 'processing')?.id, - activeTask: [...tasks].reverse().filter(t => params(t.params).variant === 'hd').map(t => ({ id: t.id, status: t.status, model: t.model, configId: t.configId, gpu: params(t.params).gpu, routing: taskRouting(t) }))[0], + activeTask: [...tasks].reverse().filter(t => params(t.params).variant === 'hd').map(t => ({ id: t.id, status: t.status, stage: t.stage, createdAt: t.createdAt, progress: params(t.params).progress, model: t.model, configId: t.configId, gpu: params(t.params).gpu, routing: taskRouting(t) }))[0], versions: tasks.filter(t => params(t.params).variant === 'hd').map(t => ({ id: t.id, status: t.status, resolution: params(t.params).resolution, gpu: params(t.params).gpu, routing: taskRouting(t), url: t.localPath, error: t.errorMsg ? safeTaskError(t.errorMsg) : null, createdAt: t.createdAt })) } })), diff --git a/huobao-source/backend/src/services/adapters/minimax-video.ts b/huobao-source/backend/src/services/adapters/minimax-video.ts index ab4aa39..b31148d 100644 --- a/huobao-source/backend/src/services/adapters/minimax-video.ts +++ b/huobao-source/backend/src/services/adapters/minimax-video.ts @@ -23,6 +23,7 @@ import type { } from './types' import { joinProviderUrl } from './url' import { parseWorkerRouting } from '../worker-routing' +import { parseTaskProgress } from '../task-progress' /** 仅支持 MiniMax-H3 系列（前缀匹配，兼容未来 H3.x 变体） */ const H3_MODEL_PREFIX = 'minimax-h3' @@ -152,7 +153,7 @@ export class MiniMaxVideoAdapter implements VideoProviderAdapter { return { status: 'failed', error: `${code}${msg}` } } const workerStage = ['queued', 'running', 'checking', 'missing'].includes(task.worker_state) ? task.worker_state : undefined - return { status: 'processing', stage: task.delivery?.status === 'transferring' ? 'transferring' : workerStage, routing: parseWorkerRouting(task.routing) } + return { status: 'processing', stage: task.delivery?.status === 'transferring' ? 'transferring' : workerStage, routing: parseWorkerRouting(task.routing), progress: parseTaskProgress(task.progress) } } extractVideoUrl(result: any): string | null { diff --git a/huobao-source/backend/src/services/adapters/types.ts b/huobao-source/backend/src/services/adapters/types.ts index f9e1ff8..c51ed0b 100644 --- a/huobao-source/backend/src/services/adapters/types.ts +++ b/huobao-source/backend/src/services/adapters/types.ts @@ -138,6 +138,7 @@ export interface VideoGenResponse { } export interface VideoPollResponse { + progress?: import('../task-progress').TaskProgress | null routing?: import('../worker-routing').WorkerRouting | null status: 'pending' | 'processing' | 'completed' | 'failed' stage?: 'transferring' | 'queued' | 'running' | 'checking' | 'missing' diff --git a/huobao-source/backend/src/services/generation.ts b/huobao-source/backend/src/services/generation.ts index 7aa2411..6adde76 100644 --- a/huobao-source/backend/src/services/generation.ts +++ b/huobao-source/backend/src/services/generation.ts @@ -568,6 +568,12 @@ async function pollTask(record: SysTaskRecord, config: AIConfig, taskId: string) .where(and(eq(schema.sysTask.id, record.id), eq(schema.sysTask.status, 'processing'))) } if (type === 'video' && pollResp.routing) await saveRouting(record.id, pollResp.routing) + if (type === 'video' && local && pollResp.status === 'processing') { + const latest = await db.select().from(schema.sysTask).where(eq(schema.sysTask.id, record.id)).get() + if (latest?.status === 'processing') await db.update(schema.sysTask).set({ + params: JSON.stringify({ ...parseTaskParams(latest.params), progress: pollResp.progress || null }), + }).where(and(eq(schema.sysTask.id, record.id), eq(schema.sysTask.status, 'processing'))) + } if (type === 'video' && pollResp.stage === 'transferring') { await db.update(schema.sysTask).set({ stage: 'transferring', updatedAt: now() }).where(eq(schema.sysTask.id, record.id)) } diff --git a/huobao-source/backend/src/services/task-progress.ts b/huobao-source/backend/src/services/task-progress.ts new file mode 100644 index 0000000..4ce651f --- /dev/null +++ b/huobao-source/backend/src/services/task-progress.ts @@ -0,0 +1,29 @@ +/** Allowlisted, finite telemetry only; model/worker payloads are untrusted. */ +export interface TaskProgress { + phase: string + observedAt: number + step?: number + totalSteps?: number + stepUpdatedAt?: number + samplingMinSeconds?: number + samplingMaxSeconds?: number + historySamples?: number + historyMinSeconds?: number + historyMaxSeconds?: number + queuePosition?: number + queueAhead?: number +} +export function parseTaskProgress(raw: unknown): TaskProgress | null { + if (!raw || typeof raw !== 'object') return null + const value = raw as Record + const phases = ['queued', 'running', 'sampling', 'decoding', 'encoding', 'loading', 'conditioning', 'postprocessing'] + if (!phases.includes(String(value.phase)) || typeof value.observedAt !== 'number' || !Number.isFinite(value.observedAt)) return null + const result: TaskProgress = { phase: String(value.phase), observedAt: value.observedAt } + const keys = ['step', 'totalSteps', 'stepUpdatedAt', 'samplingMinSeconds', 'samplingMaxSeconds', + 'historySamples', 'historyMinSeconds', 'historyMaxSeconds', 'queuePosition', 'queueAhead'] as const + for (const key of keys) { + const n = value[key] + if (typeof n === 'number' && Number.isFinite(n) && n >= 0 && n <= (key === 'stepUpdatedAt' ? 1e14 : 1e7)) result[key] = n + } + return result +} diff --git a/huobao-source/frontend/app/components/HdVideoStudio.vue b/huobao-source/frontend/app/components/HdVideoStudio.vue index 97d4466..96afdd5 100644 --- a/huobao-source/frontend/app/components/HdVideoStudio.vue +++ b/huobao-source/frontend/app/components/HdVideoStudio.vue @@ -24,6 +24,7 @@

以上选择仅用于当前镜头。{{ selected.activeTask ? routingLabel(selected.activeTask) : '尚未提交高清任务' }}

+

支持横屏 16:9 与竖屏 9:16，竖屏自动交换宽高。4K 制作耗时较长。拼接采用所选视频中的最高分辨率；未重制的低清镜头只会缩放，不会增加真实细节。

原版

当前用于导出
diff --git a/huobao-source/frontend/app/components/TaskProgress.vue b/huobao-source/frontend/app/components/TaskProgress.vue new file mode 100644 index 0000000..37e80f9 --- /dev/null +++ b/huobao-source/frontend/app/components/TaskProgress.vue @@ -0,0 +1,64 @@ + + + + + diff --git a/huobao-source/frontend/app/views/drama/episode.vue b/huobao-source/frontend/app/views/drama/episode.vue index 198c8b7..d8dc8e5 100644 --- a/huobao-source/frontend/app/views/drama/episode.vue +++ b/huobao-source/frontend/app/views/drama/episode.vue @@ -765,6 +765,7 @@
+
{{ t('episode.vid.history') }} @@ -1092,6 +1093,7 @@ · #{{ row.id }}
+
{{ mapError(row.errorMsg) }}
@@ -2434,6 +2436,7 @@ const genTaskRows = computed(() => { const taskRows = genTasks.value.map(t => ({ key: `task-${t.id}`, kind: t.type, // image | video + sourceTask: t, id: t.id, targetLabel: genTaskTargetLabel(t), provider: t.provider || '', diff --git a/integration-work/browser-test/verify_progress_live.cjs b/integration-work/browser-test/verify_progress_live.cjs new file mode 100644 index 0000000..533aff7 --- /dev/null +++ b/integration-work/browser-test/verify_progress_live.cjs @@ -0,0 +1,18 @@ +const {chromium}=require('playwright-core'),fs=require('node:fs'),assert=require('node:assert/strict'); +(async()=>{ + const root='D:/web/dj',base=JSON.parse(fs.readFileSync(root+'/huobao-data/last-launch.json')).api.replace('/api/v1',''); + const browser=await chromium.launch({channel:'chrome',headless:true}); + try{ + const page=await browser.newPage({viewport:{width:1600,height:1100}}),errors=[]; + page.on('pageerror',e=>errors.push(e.message)); + await page.addInitScript(()=>localStorage.setItem('huobao:tours',JSON.stringify(['episode']))); + await page.route('**/api/**',r=>['GET','HEAD'].includes(r.request().method())?r.continue():r.abort()); + await page.goto(base+'/drama/6/episode/1',{waitUntil:'networkidle'}); + await page.getByText('视频制作',{exact:true}).first().click(); + const panel=page.locator('.video-task-inspector > .task-progress');await panel.waitFor(); + await panel.scrollIntoViewIfNeeded(); + const text=await panel.innerText();assert.match(text,/已提交/);assert.match(text,/采样|解码|编码|处理|保存/); + await page.screenshot({path:root+'/integration-work/progress-installed.png'}); + assert.deepEqual(errors,[]);console.log('Installed live panel:',text); + }finally{await browser.close()} +})().catch(e=>{console.error(e.stack);process.exit(1)}); diff --git a/integration-work/browser-test/verify_progress_ui.cjs b/integration-work/browser-test/verify_progress_ui.cjs new file mode 100644 index 0000000..fbc2abd --- /dev/null +++ b/integration-work/browser-test/verify_progress_ui.cjs @@ -0,0 +1,39 @@ +const {chromium}=require('playwright-core'),assert=require('node:assert/strict'); +(async()=>{ + const browser=await chromium.launch({channel:'chrome',headless:true}); + try { + const page=await browser.newPage({viewport:{width:1600,height:1100}}),errors=[]; + page.on('pageerror',e=>errors.push(e.message)); + await page.addInitScript(()=>localStorage.setItem('huobao:tours',JSON.stringify(['episode']))); + let phase='sampling'; + await page.route('**/api/**',async route=>{ + if(!['GET','HEAD'].includes(route.request().method()))return route.abort(); + if(route.request().url().includes('/generation-tasks')){ + const response=await route.fetch(),body=await response.json(); + const task=body.data.tasks.find(t=>t.id===237);assert.ok(task); + const progress={phase,observedAt:Date.now(),historySamples:3,historyMinSeconds:1200,historyMaxSeconds:1800, + ...(phase==='sampling'?{step:8,totalSteps:20,stepUpdatedAt:Date.now(),samplingMinSeconds:600,samplingMaxSeconds:900}:{}), + ...(phase==='queued'?{queueAhead:2,queuePosition:2}:{})}; + if(phase==='stale'){progress.phase='sampling';progress.observedAt=Date.now()-120000} + task.status='processing';task.stage=['queued','downloading'].includes(phase)?phase:'running';task.params=JSON.stringify({...JSON.parse(task.params||'{}'),progress}); + return route.fulfill({response,json:body}); + } + return route.continue(); + }); + async function open(){ + await page.goto('http://127.0.0.1:18938/drama/6/episode/1',{waitUntil:'networkidle'}); + await page.getByText('视频制作',{exact:true}).first().click(); + const panel=page.locator('.video-task-inspector > .task-progress');await panel.waitFor();return panel; + } + let panel=await open();let text=await panel.innerText(); + assert.match(text,/8 \/ 20 步/);assert.match(text,/采样预计还需约 10–15 分钟/); + assert.match(text,/后续解码、编码和保存另计/);assert.match(text,/同显卡、同规格通常耗时约 20–30 分钟/); + await panel.scrollIntoViewIfNeeded();await page.screenshot({path:'D:/web/dj/integration-work/progress-ui-sampling.png'}); + await page.setViewportSize({width:1280,height:850});assert.equal(await panel.evaluate(e=>e.scrollWidth<=e.clientWidth+2),true); + phase='queued';panel=await open();assert.match(await panel.innerText(),/前面还有 2 个任务/); + phase='decoding';panel=await open();assert.match(await panel.innerText(),/解码中/);assert.equal(await panel.locator('progress').count(),0); + phase='stale';panel=await open();assert.match(await panel.innerText(),/进度更新暂时延迟/);assert.doesNotMatch(await panel.innerText(),/采样预计还需/); + phase='downloading';panel=await open();assert.match(await panel.innerText(),/视频生成已结束/);assert.doesNotMatch(await panel.innerText(),/采样预计还需|读取显卡进度/); + assert.deepEqual(errors,[]);console.log('PASS: sampling-only ETA, exact-spec history, queue count, decoding, stale suppression, 1280px layout; no writes'); + }finally{await browser.close()} +})().catch(e=>{console.error(e.stack);process.exit(1)}); diff --git a/integration-work/comfy_progress.py b/integration-work/comfy_progress.py new file mode 100644 index 0000000..3ea2ec5 --- /dev/null +++ b/integration-work/comfy_progress.py @@ -0,0 +1,169 @@ +"""Read-only ComfyUI telemetry, embedded in the existing private normalizers. + +Only the requesting Huobao job's client connection is observed. No queue mutation +or image previews are used. Sampling ETA is never presented as whole-video ETA. +""" +import asyncio as _hp_asyncio +import json as _hp_json +import time as _hp_time +import math as _hp_math +import statistics as _hp_stats +from urllib.parse import quote as _hp_quote +import httpx as _hp_httpx +import websockets as _hp_ws + +_hp_watchers = {} +_hp_histories = {} + + +def _hp_phase(kind): + kind = str(kind).lower() + if 'sampler' in kind: return 'sampling' + if 'decode' in kind: return 'decoding' + if 'savevideo' in kind or 'combine' in kind or 'createvideo' in kind: return 'encoding' + if 'loader' in kind or kind == 'loadimage': return 'loading' + if 'caption' in kind: return 'postprocessing' + if 'encode' in kind or 'tovideo' in kind: return 'conditioning' + return 'running' + + +def _hp_signature(graph): + # Keep graph topology, model weights, image count, dimensions, FPS, duration, + # steps and sampler. Exclude content, random seeds and output filenames. + omitted = {'text', 'prompt', 'positive', 'negative', 'seed', 'noise_seed', + 'filename_prefix', 'image', 'audio', 'video', 'file', 'path'} + result = {} + for node, value in graph.items(): + inputs = {} + for key, item in value.get('inputs', {}).items(): + if key in omitted and not isinstance(item, list): continue + if any(word in key.lower() for word in ['prompt', 'text', 'caption']) and not isinstance(item, list): continue + inputs[key] = item + result[node] = {'class_type': value.get('class_type'), 'inputs': inputs} + return _hp_json.dumps(result, sort_keys=True, separators=(',', ':')) + + +def _hp_history_estimate(graph, history): + signature = _hp_signature(graph) + durations = [] + for entry in history.values(): + status = entry.get('status', {}) + prompt = entry.get('prompt', []) + if status.get('status_str') != 'success' or not status.get('completed') or len(prompt) < 3: continue + previous = prompt[2] + if not isinstance(previous, dict) or _hp_signature(previous) != signature: continue + times = {}; cached = [] + for event in status.get('messages', []): + if len(event) < 2 or not isinstance(event[1], dict): continue + times[event[0]] = event[1].get('timestamp') + if event[0] == 'execution_cached': cached = event[1].get('nodes', []) + if any('sampler' in str(previous.get(str(n), {}).get('class_type', '')).lower() for n in cached): continue + start, end = times.get('execution_start'), times.get('execution_success') + if not isinstance(start, (int, float)) or not isinstance(end, (int, float)): continue + duration = (end - start) / 1000 + if 2 <= duration <= 86400: durations.append(duration) + result = {'historySamples': len(durations)} + if len(durations) >= 2: + # A range, not a guarantee; model loading and shared GPU contention vary. + result.update(historyMinSeconds=round(min(durations) * .8), historyMaxSeconds=round(max(durations) * 1.2)) + return result + + +def _hp_event(state, message, timestamp): + data = message.get('data', {}) + if not isinstance(data, dict): return + event = message.get('type') + if data.get('prompt_id') not in (None, state['prompt']): return + if event == 'executing': + node = data.get('node') + if node is None: return + node = str(node) + if node != state.get('node'): + state.update(node=node, samples=[], value=None, total=None) + state['phase'] = _hp_phase(state['classes'].get(node)) + state['updated'] = timestamp + elif event == 'progress' and data.get('prompt_id') == state['prompt']: + value, total = data.get('value'), data.get('max') + if not all(type(x) in (int, float) and _hp_math.isfinite(x) for x in (value, total)): return + if not 0 <= value <= total <= 1000000 or total <= 0: return + node = str(data.get('node', state.get('node', ''))) + if node != state.get('node') or total != state.get('total') or value < (state.get('value') or 0): + state.update(node=node, samples=[]) + if state.get('value') != value or not state['samples']: + state['samples'] = (state['samples'] + [(timestamp, value)])[-8:] + state.update(value=value, total=total, updated=timestamp, phase=_hp_phase(state['classes'].get(node))) + + +def _hp_snapshot(state, timestamp): + result = {'phase': state.get('phase', 'running')} + if not state.get('connected') or timestamp - state.get('updated', 0) > 600: + return {'phase': 'running'} + if state.get('value') is None: return result + result.update(step=state['value'], totalSteps=state['total'], stepUpdatedAt=round(state['updated'] * 1000)) + samples = state['samples'] + if state.get('phase') == 'sampling' and len(samples) >= 3: + seconds = samples[-1][0] - samples[0][0]; steps = samples[-1][1] - samples[0][1] + if seconds >= 5 and steps > 0 and state['value'] < state['total']: + speed = seconds / steps + if timestamp - state['updated'] <= max(30, speed * 2.5): + remaining = max(1, speed * (state['total'] - state['value']) - (timestamp - state['updated'])) + result.update(samplingMinSeconds=round(remaining * .8), samplingMaxSeconds=round(remaining * 1.3)) + return result + + +async def _hp_watch(state, base, headers, client_id): + uri = base.replace('https://', 'wss://').replace('http://', 'ws://') + '/ws?clientId=' + _hp_quote(client_id, safe='') + while _hp_time.time() - state['requested'] < 120: + try: + async with _hp_ws.connect(uri, additional_headers=headers, max_size=32 * 1024 * 1024, open_timeout=5) as ws: + state['connected'] = True + # This observer needs text progress only, never preview images. + await ws.send(_hp_json.dumps({'type':'feature_flags','data':{'supports_preview_metadata':True}})) + while _hp_time.time() - state['requested'] < 120: + try: message = await _hp_asyncio.wait_for(ws.recv(), 15) + except _hp_asyncio.TimeoutError: continue + if isinstance(message, str): + try: _hp_event(state, _hp_json.loads(message), _hp_time.time()) + except (ValueError, TypeError, AttributeError): pass + except _hp_asyncio.CancelledError: raise + except Exception: + await _hp_asyncio.sleep(5) + finally: state['connected'] = False + + +async def huobao_progress(provider, prompt_id, queue, headers): + """Failure of progress reporting must never fail or resubmit a generation.""" + timestamp = _hp_time.time() + running = queue.get('queue_running', []); pending = queue.get('queue_pending', []) + row = next((x for x in running + pending if len(x) > 3 and x[1] == prompt_id), None) + if row is None: return None + result = {'observedAt': round(timestamp * 1000), 'phase': 'queued' if row in pending else 'running'} + if row in pending: + result.update(queuePosition=pending.index(row) + 1, queueAhead=len(running) + pending.index(row)) + try: + previous = _hp_watchers.get(provider.id) + if previous and (previous['prompt'] != prompt_id and row in running or previous['task'].done()): + previous['task'].cancel(); _hp_watchers.pop(provider.id, None); previous = None + if row in running: + client_id = row[3].get('client_id') + # Huobao's normalizer generates unique canvas-* client IDs. Never + # replace a human's browser websocket or guess another job's state. + if isinstance(client_id, str) and client_id.startswith('canvas-'): + if previous is None: + previous = {'prompt':prompt_id, 'classes':{str(k):v.get('class_type') for k,v in row[2].items()}, + 'requested':timestamp, 'samples':[], 'connected':False} + previous['task'] = _hp_asyncio.create_task(_hp_watch(previous, provider.base_url, headers, client_id)) + _hp_watchers[provider.id] = previous + previous['requested'] = timestamp + if previous['prompt'] == prompt_id: result.update(_hp_snapshot(previous, timestamp)) + cached = _hp_histories.get(provider.id) + if cached is None or timestamp - cached[0] > 90: + async with _hp_httpx.AsyncClient(timeout=5) as client: + response = await client.get(provider.base_url + '/history', params={'max_items':80}, headers=headers) + response.raise_for_status(); history = response.json() + _hp_histories[provider.id] = (timestamp, history) + else: history = cached[1] + result.update(_hp_history_estimate(row[2], history)) + except Exception: + pass + return result diff --git a/integration-work/general-normalizer-panels.py b/integration-work/general-normalizer-panels.py index a7d482f..1ef9193 100644 --- a/integration-work/general-normalizer-panels.py +++ b/integration-work/general-normalizer-panels.py @@ -643,9 +643,9 @@ async def huobao_job_status(job_id: str) -> dict[str, Any]: running = [row[1] for row in queue.get("queue_running", [])] pending = [row[1] for row in queue.get("queue_pending", [])] if prompt_id in running: - return {**result, "worker_state": "running"} + return {**result, "worker_state": "running", "progress": await huobao_progress(provider, prompt_id, queue, provider_headers(provider))} if prompt_id in pending: - return {**result, "worker_state": "queued", "queue_position": pending.index(prompt_id) + 1} + return {**result, "worker_state": "queued", "queue_position": pending.index(prompt_id) + 1, "progress": await huobao_progress(provider, prompt_id, queue, provider_headers(provider))} # A task may finish between the first history read and the queue read. fresh = await get_job(job_id) if fresh.get("status") != "running": @@ -654,3 +654,174 @@ async def huobao_job_status(job_id: str) -> dict[str, Any]: return {**fresh, "worker_state": "checking"} # Do not invent success/failure or submit again when a worker has lost history. return {**fresh, "worker_state": "missing"} + +# BEGIN HUOBAO READ-ONLY PROGRESS +"""Read-only ComfyUI telemetry, embedded in the existing private normalizers. + +Only the requesting Huobao job's client connection is observed. No queue mutation +or image previews are used. Sampling ETA is never presented as whole-video ETA. +""" +import asyncio as _hp_asyncio +import json as _hp_json +import time as _hp_time +import math as _hp_math +import statistics as _hp_stats +from urllib.parse import quote as _hp_quote +import httpx as _hp_httpx +import websockets as _hp_ws + +_hp_watchers = {} +_hp_histories = {} + + +def _hp_phase(kind): + kind = str(kind).lower() + if 'sampler' in kind: return 'sampling' + if 'decode' in kind: return 'decoding' + if 'savevideo' in kind or 'combine' in kind or 'createvideo' in kind: return 'encoding' + if 'loader' in kind or kind == 'loadimage': return 'loading' + if 'caption' in kind: return 'postprocessing' + if 'encode' in kind or 'tovideo' in kind: return 'conditioning' + return 'running' + + +def _hp_signature(graph): + # Keep graph topology, model weights, image count, dimensions, FPS, duration, + # steps and sampler. Exclude content, random seeds and output filenames. + omitted = {'text', 'prompt', 'positive', 'negative', 'seed', 'noise_seed', + 'filename_prefix', 'image', 'audio', 'video', 'file', 'path'} + result = {} + for node, value in graph.items(): + inputs = {} + for key, item in value.get('inputs', {}).items(): + if key in omitted and not isinstance(item, list): continue + if any(word in key.lower() for word in ['prompt', 'text', 'caption']) and not isinstance(item, list): continue + inputs[key] = item + result[node] = {'class_type': value.get('class_type'), 'inputs': inputs} + return _hp_json.dumps(result, sort_keys=True, separators=(',', ':')) + + +def _hp_history_estimate(graph, history): + signature = _hp_signature(graph) + durations = [] + for entry in history.values(): + status = entry.get('status', {}) + prompt = entry.get('prompt', []) + if status.get('status_str') != 'success' or not status.get('completed') or len(prompt) < 3: continue + previous = prompt[2] + if not isinstance(previous, dict) or _hp_signature(previous) != signature: continue + times = {}; cached = [] + for event in status.get('messages', []): + if len(event) < 2 or not isinstance(event[1], dict): continue + times[event[0]] = event[1].get('timestamp') + if event[0] == 'execution_cached': cached = event[1].get('nodes', []) + if any('sampler' in str(previous.get(str(n), {}).get('class_type', '')).lower() for n in cached): continue + start, end = times.get('execution_start'), times.get('execution_success') + if not isinstance(start, (int, float)) or not isinstance(end, (int, float)): continue + duration = (end - start) / 1000 + if 2 <= duration <= 86400: durations.append(duration) + result = {'historySamples': len(durations)} + if len(durations) >= 2: + # A range, not a guarantee; model loading and shared GPU contention vary. + result.update(historyMinSeconds=round(min(durations) * .8), historyMaxSeconds=round(max(durations) * 1.2)) + return result + + +def _hp_event(state, message, timestamp): + data = message.get('data', {}) + if not isinstance(data, dict): return + event = message.get('type') + if data.get('prompt_id') not in (None, state['prompt']): return + if event == 'executing': + node = data.get('node') + if node is None: return + node = str(node) + if node != state.get('node'): + state.update(node=node, samples=[], value=None, total=None) + state['phase'] = _hp_phase(state['classes'].get(node)) + state['updated'] = timestamp + elif event == 'progress' and data.get('prompt_id') == state['prompt']: + value, total = data.get('value'), data.get('max') + if not all(type(x) in (int, float) and _hp_math.isfinite(x) for x in (value, total)): return + if not 0 <= value <= total <= 1000000 or total <= 0: return + node = str(data.get('node', state.get('node', ''))) + if node != state.get('node') or total != state.get('total') or value < (state.get('value') or 0): + state.update(node=node, samples=[]) + if state.get('value') != value or not state['samples']: + state['samples'] = (state['samples'] + [(timestamp, value)])[-8:] + state.update(value=value, total=total, updated=timestamp, phase=_hp_phase(state['classes'].get(node))) + + +def _hp_snapshot(state, timestamp): + result = {'phase': state.get('phase', 'running')} + if not state.get('connected') or timestamp - state.get('updated', 0) > 600: + return {'phase': 'running'} + if state.get('value') is None: return result + result.update(step=state['value'], totalSteps=state['total'], stepUpdatedAt=round(state['updated'] * 1000)) + samples = state['samples'] + if state.get('phase') == 'sampling' and len(samples) >= 3: + seconds = samples[-1][0] - samples[0][0]; steps = samples[-1][1] - samples[0][1] + if seconds >= 5 and steps > 0 and state['value'] < state['total']: + speed = seconds / steps + if timestamp - state['updated'] <= max(30, speed * 2.5): + remaining = max(1, speed * (state['total'] - state['value']) - (timestamp - state['updated'])) + result.update(samplingMinSeconds=round(remaining * .8), samplingMaxSeconds=round(remaining * 1.3)) + return result + + +async def _hp_watch(state, base, headers, client_id): + uri = base.replace('https://', 'wss://').replace('http://', 'ws://') + '/ws?clientId=' + _hp_quote(client_id, safe='') + while _hp_time.time() - state['requested'] < 120: + try: + async with _hp_ws.connect(uri, additional_headers=headers, max_size=32 * 1024 * 1024, open_timeout=5) as ws: + state['connected'] = True + # This observer needs text progress only, never preview images. + await ws.send(_hp_json.dumps({'type':'feature_flags','data':{'supports_preview_metadata':True}})) + while _hp_time.time() - state['requested'] < 120: + try: message = await _hp_asyncio.wait_for(ws.recv(), 15) + except _hp_asyncio.TimeoutError: continue + if isinstance(message, str): + try: _hp_event(state, _hp_json.loads(message), _hp_time.time()) + except (ValueError, TypeError, AttributeError): pass + except _hp_asyncio.CancelledError: raise + except Exception: + await _hp_asyncio.sleep(5) + finally: state['connected'] = False + + +async def huobao_progress(provider, prompt_id, queue, headers): + """Failure of progress reporting must never fail or resubmit a generation.""" + timestamp = _hp_time.time() + running = queue.get('queue_running', []); pending = queue.get('queue_pending', []) + row = next((x for x in running + pending if len(x) > 3 and x[1] == prompt_id), None) + if row is None: return None + result = {'observedAt': round(timestamp * 1000), 'phase': 'queued' if row in pending else 'running'} + if row in pending: + result.update(queuePosition=pending.index(row) + 1, queueAhead=len(running) + pending.index(row)) + try: + previous = _hp_watchers.get(provider.id) + if previous and (previous['prompt'] != prompt_id and row in running or previous['task'].done()): + previous['task'].cancel(); _hp_watchers.pop(provider.id, None); previous = None + if row in running: + client_id = row[3].get('client_id') + # Huobao's normalizer generates unique canvas-* client IDs. Never + # replace a human's browser websocket or guess another job's state. + if isinstance(client_id, str) and client_id.startswith('canvas-'): + if previous is None: + previous = {'prompt':prompt_id, 'classes':{str(k):v.get('class_type') for k,v in row[2].items()}, + 'requested':timestamp, 'samples':[], 'connected':False} + previous['task'] = _hp_asyncio.create_task(_hp_watch(previous, provider.base_url, headers, client_id)) + _hp_watchers[provider.id] = previous + previous['requested'] = timestamp + if previous['prompt'] == prompt_id: result.update(_hp_snapshot(previous, timestamp)) + cached = _hp_histories.get(provider.id) + if cached is None or timestamp - cached[0] > 90: + async with _hp_httpx.AsyncClient(timeout=5) as client: + response = await client.get(provider.base_url + '/history', params={'max_items':80}, headers=headers) + response.raise_for_status(); history = response.json() + _hp_histories[provider.id] = (timestamp, history) + else: history = cached[1] + result.update(_hp_history_estimate(row[2], history)) + except Exception: + pass + return result diff --git a/integration-work/install_comfy_progress.py b/integration-work/install_comfy_progress.py new file mode 100644 index 0000000..48bb436 --- /dev/null +++ b/integration-work/install_comfy_progress.py @@ -0,0 +1,40 @@ +"""Run via ssh_task.py after uploading comfy_progress.py. No GPU workers restart.""" +import datetime, pathlib, subprocess, json +payload = pathlib.Path('/home/ps/huobao-progress-update/comfy_progress.py').read_text() +marker = '# BEGIN HUOBAO READ-ONLY PROGRESS\n' +compile(payload, 'comfy_progress.py', 'exec') +names = ['canvas-h3-prompt-normalizer-healthy-20261026', 'huobao-vdn-normalizer'] +backup = pathlib.Path('/home/ps/huobao-progress-update') / ('backup-' + datetime.datetime.now().strftime('%Y%m%d-%H%M%S')) +backup.mkdir(mode=0o700) +changed = [] +for name in names: + path = pathlib.Path('/home/ps/huobao-gpu-selection') / (name + '.candidate.py') + original = path.read_text(); source = original.split(marker)[0] + for old,new in [ + ('return {**result, "worker_state": "running"}', 'return {**result, "worker_state": "running", "progress": await huobao_progress(provider, prompt_id, queue, provider_headers(provider))}'), + ('return {**result, "worker_state": "queued", "queue_position": pending.index(prompt_id) + 1}', 'return {**result, "worker_state": "queued", "queue_position": pending.index(prompt_id) + 1, "progress": await huobao_progress(provider, prompt_id, queue, provider_headers(provider))}')]: + if old in source: + assert source.count(old) == 1 + source = source.replace(old,new) + else: assert new in source + source = source.rstrip() + '\n\n' + marker + payload + compile(source, str(path), 'exec') + (backup / (name + '.py')).write_text(original) + # Preserve the inode of the individually bind-mounted source file. + path.write_text(source); changed.append(name) +for name,path in [('huobao-model-bridge','/home/ps/huobao-model-bridge/bridge.py'),('huobao-vdn-bridge','/home/ps/huobao-vdn/bridge/bridge.py')]: + path = pathlib.Path(path); original = path.read_text(); source = original + marker_line = ' task["progress"] = result.get("progress")' + if marker_line not in source: + lines = source.splitlines() + indexes = [i for i,line in enumerate(lines) if line.startswith(' task = {"id": jid, "status": state, "duration": duration,')] + assert len(indexes) == 1 + lines.insert(indexes[0]+1,marker_line); source = '\n'.join(lines)+'\n' + compile(source,str(path),'exec'); (backup / (name+'.py')).write_text(original) + path.write_text(source); changed.append(name) +for name in changed: + subprocess.run(['docker','restart','--time','15',name],check=True,capture_output=True) + d=json.loads(subprocess.check_output(['docker','inspect',name]))[0] + assert d['State']['Running'], name + print(name,'running') +print('backup',str(backup)) diff --git a/integration-work/test_comfy_progress.py b/integration-work/test_comfy_progress.py new file mode 100644 index 0000000..c2a84d4 --- /dev/null +++ b/integration-work/test_comfy_progress.py @@ -0,0 +1,46 @@ +import unittest +from comfy_progress import _hp_event, _hp_snapshot, _hp_history_estimate + +class ProgressTests(unittest.TestCase): + def state(self): + return {'prompt':'one','classes':{'12':'SamplerCustomAdvanced','13':'VAEDecode'},'samples':[], 'connected':True} + def event(self,state,value,t,node='12',prompt='one',maximum=20): + _hp_event(state,{'type':'progress','data':{'prompt_id':prompt,'node':node,'value':value,'max':maximum}},t) + def test_job_isolation_and_node_reset(self): + s=self.state();self.event(s,1,100);self.event(s,2,110,prompt='other') + self.assertEqual(s['value'],1) + self.event(s,2,110);self.event(s,3,120) + self.assertEqual(_hp_snapshot(s,120)['samplingMinSeconds'],136) + _hp_event(s,{'type':'executing','data':{'prompt_id':'one','node':'13'}},125) + r=_hp_snapshot(s,125);self.assertEqual(r['phase'],'decoding');self.assertNotIn('step',r) + def test_stale_disconnect_reset_no_false_countdown(self): + s=self.state() + for i in range(1,4):self.event(s,i,100+i*10) + self.assertNotIn('samplingMinSeconds',_hp_snapshot(s,200)) + self.event(s,1,201) + self.assertNotIn('samplingMinSeconds',_hp_snapshot(s,201)) + s['connected']=False + self.assertEqual(_hp_snapshot(s,202),{'phase':'running'}) + def test_sampling_completion_is_not_video_completion(self): + s=self.state() + for i in [18,19,20]:self.event(s,i,100+i*10) + r=_hp_snapshot(s,300) + self.assertEqual(r['phase'],'sampling');self.assertEqual(r['step'],20) + self.assertNotIn('samplingMinSeconds',r) + def test_history_exact_settings_and_cache_exclusion(self): + graph={'1':{'class_type':'SamplerCustomAdvanced','inputs':{'steps':20,'noise_seed':123}}, + '2':{'class_type':'MiniMaxH3ReferenceToVideo','inputs':{'width':1344,'height':768,'text':'private'}}} + def history(duration=1000,change=None,cached=False,failed=False): + import copy + other=copy.deepcopy(graph);other['1']['inputs']['noise_seed']=456;other['2']['inputs']['text']='another' + if change:other['2']['inputs'].update(change) + return {'prompt':[1,'id',other], 'status':{'status_str':'error' if failed else 'success','completed':True, + 'messages':[['execution_start',{'timestamp':1000}],['execution_cached',{'nodes':['1'] if cached else []}], + ['execution_success',{'timestamp':1000+duration*1000}]]}} + r=_hp_history_estimate(graph,{'a':history()}) + self.assertEqual(r,{'historySamples':1}) + r=_hp_history_estimate(graph,{'a':history(),'b':history(1200),'c':history(5,cached=True), + 'd':history(500,change={'width':2560}),'e':history(2,failed=True)}) + self.assertEqual(r,{'historySamples':2,'historyMinSeconds':800,'historyMaxSeconds':1440}) + +if __name__=='__main__':unittest.main() diff --git a/integration-work/test_gpu_replacement_references.py b/integration-work/test_gpu_replacement_references.py index 7bb619c..563a095 100644 --- a/integration-work/test_gpu_replacement_references.py +++ b/integration-work/test_gpu_replacement_references.py @@ -14,6 +14,12 @@ source = re.sub(r'^import (\w+) from "(better-sqlite3|sharp)";', r'const \1 = re needle = 'target.hostname === "192.168.110.111" && ["18792", "18794"].includes(target.port)' assert source.count(needle) == 1 source = source.replace(needle, 'target.hostname === "127.0.0.1" && target.port === "18935"') +source, count = re.subn(r'(url\d*)\.hostname === "192\.168\.110\.111" && \["18792", "18794"\]\.includes\(\1.port\)', + r'\1.hostname === "127.0.0.1" && \1.port === "18935"', source) +assert count == 1 +# Fake IDs are private to the loopback fixture. +source, count = re.subn(r'&& !!taskRouting\(\{ provider: config\d*\.provider, taskId \}\)', '&& !!taskId', source) +assert count == 1 (data / 'backend.mjs').write_text(source, 'utf-8') events = []; reject = False; process = None @@ -35,7 +41,8 @@ class Mock(BaseHTTPRequestHandler): else: self.reply(200, {'task_id': 'new-' + str(len(events)), 'routing': {'gpu': body['gpu'], 'port': 8191, 'providerId': 'gpu3', 'legacy': False}}) def do_GET(self): - self.reply(200, {'task': {'id': self.path.split('/')[-1], 'status': 'running', 'worker_state': 'running'}}) + self.reply(200, {'task': {'id': self.path.split('/')[-1], 'status': 'running', 'worker_state': 'running', + 'progress': {'phase':'sampling','step':4,'totalSteps':20,'observedAt':round(time.time()*1000),'samplingMinSeconds':120,'samplingMaxSeconds':180}}}) server = ThreadingHTTPServer(('127.0.0.1', 18935), Mock) threading.Thread(target=server.serve_forever, daemon=True).start() @@ -105,6 +112,13 @@ try: image = events[before][1]['content'][1] assert image['role'] == 'first_frame' and image['image_url']['url'].startswith('data:image/') assert events[before][1]['content'] == events[before+2][1]['content'] + for _ in range(140): + row=sql('SELECT params FROM sys_task WHERE id=?',(result['id'],))[0] + progress=json.loads(row['params']).get('progress') + if progress: break + time.sleep(.1) + assert progress['step']==4 and progress['totalSteps']==20 + assert progress['samplingMaxSeconds']==180 print(json.dumps({'passed':True,'inline_reference_preflight':True,'first_frame':True,'cancel_only_selected_shot':True, 'preserve_old_on_missing_image':True,'preserve_old_on_validation_failure':True,'fixture':str(data)})) finally: diff --git a/integration-work/test_task_progress.cjs b/integration-work/test_task_progress.cjs new file mode 100644 index 0000000..ae8938f --- /dev/null +++ b/integration-work/test_task_progress.cjs @@ -0,0 +1,11 @@ +const fs=require('node:fs'),assert=require('node:assert/strict'),ts=require('../huobao-source/backend/node_modules/typescript'); +(async()=>{ + const source=fs.readFileSync('huobao-source/backend/src/services/task-progress.ts','utf8'); + const code=ts.transpileModule(source,{compilerOptions:{module:ts.ModuleKind.ESNext,target:ts.ScriptTarget.ES2022}}).outputText; + const {parseTaskProgress}=await import('data:text/javascript;base64,'+Buffer.from(code).toString('base64')); + assert.equal(parseTaskProgress({phase:'invented',observedAt:123}),null); + assert.equal(parseTaskProgress({phase:'sampling',observedAt:Infinity}),null); + assert.deepEqual(parseTaskProgress({phase:'sampling',observedAt:123,step:3,totalSteps:20,api_key:'secret',historySamples:NaN,samplingMaxSeconds:-1}), + {phase:'sampling',observedAt:123,step:3,totalSteps:20}); + console.log('PASS: unknown phases, non-finite/negative values and unapproved payload fields rejected'); +})().catch(e=>{console.error(e);process.exit(1)}); diff --git a/integration-work/vdn-normalizer-panels.py b/integration-work/vdn-normalizer-panels.py index d6f061c..7a3b75d 100644 --- a/integration-work/vdn-normalizer-panels.py +++ b/integration-work/vdn-normalizer-panels.py @@ -656,9 +656,9 @@ async def huobao_job_status(job_id: str) -> dict[str, Any]: running = [row[1] for row in queue.get("queue_running", [])] pending = [row[1] for row in queue.get("queue_pending", [])] if prompt_id in running: - return {**result, "worker_state": "running"} + return {**result, "worker_state": "running", "progress": await huobao_progress(provider, prompt_id, queue, provider_headers(provider))} if prompt_id in pending: - return {**result, "worker_state": "queued", "queue_position": pending.index(prompt_id) + 1} + return {**result, "worker_state": "queued", "queue_position": pending.index(prompt_id) + 1, "progress": await huobao_progress(provider, prompt_id, queue, provider_headers(provider))} # A task may finish between the first history read and the queue read. fresh = await get_job(job_id) if fresh.get("status") != "running": @@ -667,3 +667,174 @@ async def huobao_job_status(job_id: str) -> dict[str, Any]: return {**fresh, "worker_state": "checking"} # Do not invent success/failure or submit again when a worker has lost history. return {**fresh, "worker_state": "missing"} + +# BEGIN HUOBAO READ-ONLY PROGRESS +"""Read-only ComfyUI telemetry, embedded in the existing private normalizers. + +Only the requesting Huobao job's client connection is observed. No queue mutation +or image previews are used. Sampling ETA is never presented as whole-video ETA. +""" +import asyncio as _hp_asyncio +import json as _hp_json +import time as _hp_time +import math as _hp_math +import statistics as _hp_stats +from urllib.parse import quote as _hp_quote +import httpx as _hp_httpx +import websockets as _hp_ws + +_hp_watchers = {} +_hp_histories = {} + + +def _hp_phase(kind): + kind = str(kind).lower() + if 'sampler' in kind: return 'sampling' + if 'decode' in kind: return 'decoding' + if 'savevideo' in kind or 'combine' in kind or 'createvideo' in kind: return 'encoding' + if 'loader' in kind or kind == 'loadimage': return 'loading' + if 'caption' in kind: return 'postprocessing' + if 'encode' in kind or 'tovideo' in kind: return 'conditioning' + return 'running' + + +def _hp_signature(graph): + # Keep graph topology, model weights, image count, dimensions, FPS, duration, + # steps and sampler. Exclude content, random seeds and output filenames. + omitted = {'text', 'prompt', 'positive', 'negative', 'seed', 'noise_seed', + 'filename_prefix', 'image', 'audio', 'video', 'file', 'path'} + result = {} + for node, value in graph.items(): + inputs = {} + for key, item in value.get('inputs', {}).items(): + if key in omitted and not isinstance(item, list): continue + if any(word in key.lower() for word in ['prompt', 'text', 'caption']) and not isinstance(item, list): continue + inputs[key] = item + result[node] = {'class_type': value.get('class_type'), 'inputs': inputs} + return _hp_json.dumps(result, sort_keys=True, separators=(',', ':')) + + +def _hp_history_estimate(graph, history): + signature = _hp_signature(graph) + durations = [] + for entry in history.values(): + status = entry.get('status', {}) + prompt = entry.get('prompt', []) + if status.get('status_str') != 'success' or not status.get('completed') or len(prompt) < 3: continue + previous = prompt[2] + if not isinstance(previous, dict) or _hp_signature(previous) != signature: continue + times = {}; cached = [] + for event in status.get('messages', []): + if len(event) < 2 or not isinstance(event[1], dict): continue + times[event[0]] = event[1].get('timestamp') + if event[0] == 'execution_cached': cached = event[1].get('nodes', []) + if any('sampler' in str(previous.get(str(n), {}).get('class_type', '')).lower() for n in cached): continue + start, end = times.get('execution_start'), times.get('execution_success') + if not isinstance(start, (int, float)) or not isinstance(end, (int, float)): continue + duration = (end - start) / 1000 + if 2 <= duration <= 86400: durations.append(duration) + result = {'historySamples': len(durations)} + if len(durations) >= 2: + # A range, not a guarantee; model loading and shared GPU contention vary. + result.update(historyMinSeconds=round(min(durations) * .8), historyMaxSeconds=round(max(durations) * 1.2)) + return result + + +def _hp_event(state, message, timestamp): + data = message.get('data', {}) + if not isinstance(data, dict): return + event = message.get('type') + if data.get('prompt_id') not in (None, state['prompt']): return + if event == 'executing': + node = data.get('node') + if node is None: return + node = str(node) + if node != state.get('node'): + state.update(node=node, samples=[], value=None, total=None) + state['phase'] = _hp_phase(state['classes'].get(node)) + state['updated'] = timestamp + elif event == 'progress' and data.get('prompt_id') == state['prompt']: + value, total = data.get('value'), data.get('max') + if not all(type(x) in (int, float) and _hp_math.isfinite(x) for x in (value, total)): return + if not 0 <= value <= total <= 1000000 or total <= 0: return + node = str(data.get('node', state.get('node', ''))) + if node != state.get('node') or total != state.get('total') or value < (state.get('value') or 0): + state.update(node=node, samples=[]) + if state.get('value') != value or not state['samples']: + state['samples'] = (state['samples'] + [(timestamp, value)])[-8:] + state.update(value=value, total=total, updated=timestamp, phase=_hp_phase(state['classes'].get(node))) + + +def _hp_snapshot(state, timestamp): + result = {'phase': state.get('phase', 'running')} + if not state.get('connected') or timestamp - state.get('updated', 0) > 600: + return {'phase': 'running'} + if state.get('value') is None: return result + result.update(step=state['value'], totalSteps=state['total'], stepUpdatedAt=round(state['updated'] * 1000)) + samples = state['samples'] + if state.get('phase') == 'sampling' and len(samples) >= 3: + seconds = samples[-1][0] - samples[0][0]; steps = samples[-1][1] - samples[0][1] + if seconds >= 5 and steps > 0 and state['value'] < state['total']: + speed = seconds / steps + if timestamp - state['updated'] <= max(30, speed * 2.5): + remaining = max(1, speed * (state['total'] - state['value']) - (timestamp - state['updated'])) + result.update(samplingMinSeconds=round(remaining * .8), samplingMaxSeconds=round(remaining * 1.3)) + return result + + +async def _hp_watch(state, base, headers, client_id): + uri = base.replace('https://', 'wss://').replace('http://', 'ws://') + '/ws?clientId=' + _hp_quote(client_id, safe='') + while _hp_time.time() - state['requested'] < 120: + try: + async with _hp_ws.connect(uri, additional_headers=headers, max_size=32 * 1024 * 1024, open_timeout=5) as ws: + state['connected'] = True + # This observer needs text progress only, never preview images. + await ws.send(_hp_json.dumps({'type':'feature_flags','data':{'supports_preview_metadata':True}})) + while _hp_time.time() - state['requested'] < 120: + try: message = await _hp_asyncio.wait_for(ws.recv(), 15) + except _hp_asyncio.TimeoutError: continue + if isinstance(message, str): + try: _hp_event(state, _hp_json.loads(message), _hp_time.time()) + except (ValueError, TypeError, AttributeError): pass + except _hp_asyncio.CancelledError: raise + except Exception: + await _hp_asyncio.sleep(5) + finally: state['connected'] = False + + +async def huobao_progress(provider, prompt_id, queue, headers): + """Failure of progress reporting must never fail or resubmit a generation.""" + timestamp = _hp_time.time() + running = queue.get('queue_running', []); pending = queue.get('queue_pending', []) + row = next((x for x in running + pending if len(x) > 3 and x[1] == prompt_id), None) + if row is None: return None + result = {'observedAt': round(timestamp * 1000), 'phase': 'queued' if row in pending else 'running'} + if row in pending: + result.update(queuePosition=pending.index(row) + 1, queueAhead=len(running) + pending.index(row)) + try: + previous = _hp_watchers.get(provider.id) + if previous and (previous['prompt'] != prompt_id and row in running or previous['task'].done()): + previous['task'].cancel(); _hp_watchers.pop(provider.id, None); previous = None + if row in running: + client_id = row[3].get('client_id') + # Huobao's normalizer generates unique canvas-* client IDs. Never + # replace a human's browser websocket or guess another job's state. + if isinstance(client_id, str) and client_id.startswith('canvas-'): + if previous is None: + previous = {'prompt':prompt_id, 'classes':{str(k):v.get('class_type') for k,v in row[2].items()}, + 'requested':timestamp, 'samples':[], 'connected':False} + previous['task'] = _hp_asyncio.create_task(_hp_watch(previous, provider.base_url, headers, client_id)) + _hp_watchers[provider.id] = previous + previous['requested'] = timestamp + if previous['prompt'] == prompt_id: result.update(_hp_snapshot(previous, timestamp)) + cached = _hp_histories.get(provider.id) + if cached is None or timestamp - cached[0] > 90: + async with _hp_httpx.AsyncClient(timeout=5) as client: + response = await client.get(provider.base_url + '/history', params={'max_items':80}, headers=headers) + response.raise_for_status(); history = response.json() + _hp_histories[provider.id] = (timestamp, history) + else: history = cached[1] + result.update(_hp_history_estimate(row[2], history)) + except Exception: + pass + return result diff --git a/integration-work/verify_progress_ui.py b/integration-work/verify_progress_ui.py new file mode 100644 index 0000000..76b90dc --- /dev/null +++ b/integration-work/verify_progress_ui.py @@ -0,0 +1,22 @@ +"""Serve built frontend with a read-only proxy to verify progress; no DB writes.""" +import http.server,json,pathlib,subprocess,threading,urllib.request,urllib.error +root=pathlib.Path(__file__).resolve().parent.parent +public=root/'huobao-source/frontend/.output/public' +upstream=json.loads((root/'huobao-data/last-launch.json').read_text())['api'].removesuffix('/api/v1') +class ReadOnly(http.server.SimpleHTTPRequestHandler): + def __init__(self,*args,**kw):super().__init__(*args,directory=str(public),**kw) + def log_message(self,*args):pass + def do_GET(self): + if self.path.startswith(('/api/','/static/')): + try: + with urllib.request.urlopen(upstream+self.path,timeout=15) as r: + data=r.read();self.send_response(r.status);self.send_header('Content-Type',r.headers.get('Content-Type','application/json'));self.end_headers();self.wfile.write(data) + except urllib.error.HTTPError as e:self.send_error(e.code) + except (ConnectionAbortedError, BrokenPipeError):pass + else: + if not pathlib.Path(self.translate_path(self.path)).exists():self.path='/index.html' + super().do_GET() +server=http.server.ThreadingHTTPServer(('127.0.0.1',18938),ReadOnly) +threading.Thread(target=server.serve_forever,daemon=True).start() +try:subprocess.run(['node',str(root/'integration-work/browser-test/verify_progress_ui.cjs')],cwd=root,check=True) +finally:server.shutdown() diff --git a/model-bridge/bridge.py b/model-bridge/bridge.py index b2c5b64..4a9608e 100644 --- a/model-bridge/bridge.py +++ b/model-bridge/bridge.py @@ -538,6 +538,7 @@ def create_app(config): result = await upstream("GET", "/v1/huobao/jobs/" + quote(jid, safe="")) state = result.get("status", "running") task = {"id": jid, "status": state, "duration": duration, "provider_id": result.get("providerId"), "routing": worker_routing(result.get("providerId")), "worker_state": result.get("worker_state"), "queue_position": result.get("queue_position")} + task["progress"] = result.get("progress") if state == "succeeded": output = next((o for o in result.get("outputs", []) if o.get("kind") == "video"), None) if not output: diff --git a/model-bridge/build_upgrade.py b/model-bridge/build_upgrade.py index 09b5e7e..e65ec50 100644 --- a/model-bridge/build_upgrade.py +++ b/model-bridge/build_upgrade.py @@ -31,7 +31,7 @@ for directory in ('prompts', 'skills'): shutil.copytree(root/'huobao-source/backend/workspace'/directory, stage/'resources/workspace-template'/directory,dirs_exist_ok=True) manifest={'upstream_version':'4.0.11','source_commit':'b48e6b2312166e25b1ff141832d2e82e5d63c142', - 'custom_features':['persistent-data','local-video-pool','multi-gpu-images','vdn-hd-versions','per-model-gpu-selection','fixed-gpu-models','cancel-and-regenerate','per-shot-routing','direct-comfy-panels','confirmed-worker-labels','asset-image-references','durable-local-task-tracking','cross-episode-continuity','tail-frame-dependency-queue','storyboard-timeline-validation','gpu-replacement-inline-references'], + 'custom_features':['persistent-data','local-video-pool','multi-gpu-images','vdn-hd-versions','per-model-gpu-selection','fixed-gpu-models','cancel-and-regenerate','per-shot-routing','direct-comfy-panels','confirmed-worker-labels','asset-image-references','durable-local-task-tracking','cross-episode-continuity','tail-frame-dependency-queue','storyboard-timeline-validation','gpu-replacement-inline-references','comfy-live-progress-and-estimates'], 'archive_sha256':hashlib.sha256(archive.read_bytes()).hexdigest()} (stage/'resources/huobao-local-build.json').write_text(json.dumps(manifest,indent=2)) print('Staged 4.0.11 with persistent data and VDN HD:',manifest['archive_sha256']) diff --git a/model-bridge/test_bridge.py b/model-bridge/test_bridge.py index bd21cf7..67a7d08 100644 --- a/model-bridge/test_bridge.py +++ b/model-bridge/test_bridge.py @@ -21,6 +21,7 @@ class BridgeTests(unittest.TestCase): task=self.client.get('/v2/query/video_generation/'+video,headers=self.headers).json()['task'] self.assertEqual(task['worker_state'],state) self.assertEqual(task['status'],'running') + self.assertEqual(task['progress'],{'phase':'sampling','step':4,'totalSteps':20,'observedAt':123456}) image=self.client.get('/v1/images/task/image-job',headers=self.headers).json() self.assertEqual(image['worker_state'],state) self.assertEqual(image['status'],'processing') @@ -131,7 +132,7 @@ class BridgeTests(unittest.TestCase): if request.url.path == "/v1/jobs/image-job/outputs/0": return httpx.Response(200, content=b"test-image-bytes", headers={"content-type": "image/png"}) if request.url.path in ("/v1/huobao/jobs/test-job", "/v1/jobs/test-job/cancel"): - return httpx.Response(200, json={"status": self.job_status, "error": "quality gate failed", "providerId": "gpu3", "worker_state":getattr(self,'worker_state','running'), + return httpx.Response(200, json={"status": self.job_status, "error": "quality gate failed", "providerId": "gpu3", "worker_state":getattr(self,'worker_state','running'), "progress":{'phase':'sampling','step':4,'totalSteps':20,'observedAt':123456}, "outputs": [{"index": 0, "kind": "image"}, {"index": 1, "kind": "video"}]}) if request.url.path == "/v1/jobs/test-job/outputs/1": return httpx.Response(200, content=b"test-video-bytes", headers={"content-type": "video/mp4"})