From b9eb3575715c2cec17b37826ba6636ed2dffe6b6 Mon Sep 17 00:00:00 2001 From: magicwang1111 Date: Wed, 28 Oct 2026 11:41:02 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=E6=81=A2=E5=A4=8D=E8=AE=B0?= =?UTF-8?q?=E5=BD=95=E5=AD=97=E6=AE=B5=E7=BC=BA=E5=A4=B1=E5=AF=BC=E8=87=B4?= =?UTF-8?q?=E5=8E=86=E5=8F=B2=E4=BB=BB=E5=8A=A1=E5=88=97=E8=A1=A8=E4=B8=BA?= =?UTF-8?q?=E7=A9=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tests/browser_recovery_history.cjs | 63 ++++++++++++++++++++++++++++++ tests/test_api_recovery.py | 2 + tests/test_history_store.py | 20 ++++++++++ utils/history_store.py | 18 +++++++++ 4 files changed, 103 insertions(+) create mode 100644 tests/browser_recovery_history.cjs diff --git a/tests/browser_recovery_history.cjs b/tests/browser_recovery_history.cjs new file mode 100644 index 0000000..8f3d61c --- /dev/null +++ b/tests/browser_recovery_history.cjs @@ -0,0 +1,63 @@ +const assert = require('node:assert/strict'); +const { spawnSync } = require('node:child_process'); +const path = require('node:path'); +const { chromium } = require(process.env.PLAYWRIGHT_MODULE || 'playwright'); + +const fixture = spawnSync(process.env.COMFY_TEST_PYTHON || 'python', ['-c', ` +import importlib.util,json,sys,tempfile +from pathlib import Path +root=Path.cwd() +sys.path.insert(0,str(root.parents[1])) +from comfy_execution.jobs import get_all_jobs,normalize_history_item +spec=importlib.util.spec_from_file_location('history_store',root/'utils/history_store.py') +module=importlib.util.module_from_spec(spec); spec.loader.exec_module(module) +with tempfile.TemporaryDirectory() as directory: + store=module.HistoryStore(str(Path(directory)/'history.db')) + old={'prompt':[0,'recovered',{}, {'create_time':1000},[]], 'user_id':'a', + 'api_recovery':{'task':{'state':'pending'}},'outputs':{}, + 'status':{'status_str':'error','completed':False,'messages':[ + ['execution_start',{'timestamp':1000}], + ['execution_error',{'node_id':'7','node_type':'APIRecovery', + 'exception_type':'WorkerInterrupted','exception_message':'Interrupted','timestamp':1100}]]}} + store.save('recovered',old,10) + store.save('previous',{'prompt':[1,'previous',{}, {'create_time':900},[]], + 'user_id':'a','outputs':{},'status':{'status_str':'success','completed':True,'messages':[]}},10) + history=store.query(owner_id='a',for_jobs=True) + jobs,total=get_all_jobs([],[],history,limit=200,offset=0) + print(json.dumps({'page':{'jobs':jobs,'pagination':{'offset':0,'limit':200,'total':total,'has_more':False}}, + 'detail':normalize_history_item('recovered',store.query(prompt_id='recovered')['recovered'],include_outputs=True)})) +`], { cwd: path.resolve(__dirname, '..'), encoding: 'utf8' }); +assert.equal(fixture.status, 0, fixture.stderr); +const payload = JSON.parse(fixture.stdout); + +(async () => { + const browser = await chromium.launch({ headless: true, executablePath: process.env.PLAYWRIGHT_CHROMIUM_EXECUTABLE || undefined }); + try { + const page = await browser.newPage(); + let broken = true; + await page.route('**/api/jobs?*', route => { + const body = structuredClone(payload.page); + if (broken) delete body.jobs.find(job => job.id === 'recovered').execution_error.traceback; + return route.fulfill({ json: body }); + }); + await page.route('**/api/jobs/recovered', route => route.fulfill({ json: payload.detail })); + await page.goto(process.env.COMFY_TEST_URL || 'http://127.0.0.1:8200'); + await page.waitForFunction(() => window.comfyAPI?.api?.api); + const read = () => page.evaluate(async () => { + const { api } = await import('/scripts/api.js'); + return (await api.getHistory()).map(job => job.id); + }); + assert.deepEqual(await read(), [], 'Missing traceback should reproduce the empty history page'); + broken = false; + assert.deepEqual((await read()).sort(), ['previous', 'recovered']); + const detail = await page.evaluate(async () => { + const { api } = await import('/scripts/api.js'); + return api.getJobDetail('recovered'); + }); + assert.equal(detail.id, 'recovered'); + assert.deepEqual(detail.execution_error.traceback, []); + console.log('PASS: real frontend reproduces empty history and accepts repaired legacy recovery list/detail'); + } finally { + await browser.close(); + } +})().catch(error => { console.error(error); process.exitCode = 1; }); diff --git a/tests/test_api_recovery.py b/tests/test_api_recovery.py index d58f203..5f663b9 100644 --- a/tests/test_api_recovery.py +++ b/tests/test_api_recovery.py @@ -69,6 +69,8 @@ class ApiRecoveryTests(unittest.TestCase): visible = self.history.query(owner_id="user-a")["job-1"] self.assertEqual("error", visible["status"]["status_str"]) self.assertFalse(visible["status"]["completed"]) + self.assertEqual([], visible["status"]["messages"][1][1]["traceback"]) + self.assertIsInstance(visible["prompt"][3]["create_time"], int) self.assertEqual({}, self.history.query(owner_id="user-b")) videos = [ref for output in visible["outputs"].values() for ref in output.get("video", [])] self.assertEqual(1, len(videos)) diff --git a/tests/test_history_store.py b/tests/test_history_store.py index a1ab127..5ee7775 100644 --- a/tests/test_history_store.py +++ b/tests/test_history_store.py @@ -1,7 +1,9 @@ import importlib.util import os +import sqlite3 import tempfile import unittest +from contextlib import closing from pathlib import Path @@ -116,6 +118,24 @@ class HistoryStoreTests(unittest.TestCase): list(self.store.query(owner_id="user-a", offset=1, max_items=2, for_jobs=True)), ) + def test_legacy_recovery_is_readable_without_rewriting_or_deleting_history(self): + item = self.item("user-a", 1) + item["api_recovery"] = {"task": {"state": "pending"}} + item["status"] = {"status_str": "error", "messages": [ + ["execution_start", {"timestamp": 123}], + ["execution_error", {"node_type": "APIRecovery", "exception_message": "Interrupted"}], + ]} + self.store.save("recovered", item, 10) + self.store.save("previous", self.item("user-a", 2), 10) + for rows in (self.store.load(10), self.store.query(), self.store.query(for_jobs=True)): + self.assertEqual({"recovered", "previous"}, set(rows)) + recovered = rows["recovered"] + self.assertEqual([], recovered["status"]["messages"][1][1]["traceback"]) + self.assertEqual(123, recovered["prompt"][3]["create_time"]) + with closing(sqlite3.connect(self.database)) as db: + stored = db.execute("SELECT data FROM history WHERE prompt_id='recovered'").fetchone()[0] + self.assertNotIn('traceback', stored) + if __name__ == "__main__": unittest.main() diff --git a/utils/history_store.py b/utils/history_store.py index 493779c..107cd8e 100644 --- a/utils/history_store.py +++ b/utils/history_store.py @@ -5,6 +5,20 @@ import time from contextlib import closing +def _repair_recovery_fields(item: dict) -> None: + # Earlier recovery records omitted fields required by the frontend Jobs schema. + if not item.get("api_recovery"): + return + for event, data in (item.get("status") or {}).get("messages", []): + if event == "execution_start" and data.get("timestamp") is not None: + if item["prompt"][3].get("create_time") is None: + item["prompt"][3]["create_time"] = data["timestamp"] + if event == "execution_error" and data.get("node_type") == "APIRecovery": + data.setdefault("traceback", []) + data.setdefault("current_inputs", {}) + data.setdefault("current_outputs", []) + + class HistoryStore: """Persist completed ComfyUI history entries in a small SQLite database.""" @@ -55,6 +69,7 @@ class HistoryStore: item = json.loads(data) if not isinstance(item, dict): continue + _repair_recovery_fields(item) item["user_id"] = owner_id history[prompt_id] = item return history @@ -122,6 +137,7 @@ class HistoryStore: for current_prompt_id, current_owner_id, data in rows: item = json.loads(data) if isinstance(item, dict): + _repair_recovery_fields(item) item["user_id"] = current_owner_id result[current_prompt_id] = item return result @@ -176,6 +192,7 @@ class HistoryStore: ["execution_start", {"timestamp": int(task["submitted_at"] * 1000)}], ["execution_error", {"node_id": task["node_id"], "node_type": "APIRecovery", "exception_type": "WorkerInterrupted", "exception_message": "服务器中断，工作流未完整执行。", + "traceback": [], "current_inputs": {}, "current_outputs": [], "timestamp": int(time.time() * 1000)}], ]}, } @@ -189,6 +206,7 @@ class HistoryStore: if videos: output["video"] = videos item.setdefault("outputs", {})["api_recovery_" + task["id"]] = output + _repair_recovery_fields(item) connection.execute(""" INSERT INTO history(prompt_id,owner_id,data) VALUES(?,?,?) ON CONFLICT(prompt_id) DO UPDATE SET data=excluded.data