Fix ZeroGPU request serialization and preserve error tracebacks

core/execution_plan.py

@@ -7,6 +7,7 @@ trying to keep several checkpoints resident on one ZeroGPU allocation.
 
 from __future__ import annotations
 
+import logging
 import random
 from collections.abc import Callable, Iterable, Sequence
 from dataclasses import dataclass
@@ -26,6 +27,8 @@ from core.settings import (
     MODEL_TYPE_MAP,
 )
 
+logger = logging.getLogger(__name__)
+
 MODE_SINGLE = "single"
 MODE_MODEL_PK = "model_pk"
 MODE_MULTI_INDEPENDENT = "multi_independent"
@@ -441,6 +444,12 @@ def execute_generation_plan(
                     summary.append("- ⏹️ 已取消：后续组合未执行，已保留成功结果。")
                     break
                 raise
+            logger.exception(
+                "生成任务失败（%s/%s，模型=%s）",
+                index + 1,
+                total,
+                item.inputs.get("model_display_name", "未知"),
+            )
             summary.append(f"- ❌ {item.caption}：{exc}")
 
     if not gallery:

core/pipelines/sd_image_pipeline.py

@@ -61,12 +61,10 @@ class SdImagePipeline(BasePipeline):
         else:
             return [model_display_name]
 
-    def _gpu_logic(self, ui_inputs: Dict, loras_string: str, workflow: Dict[str, Any], assembler: WorkflowAssembler, progress=gr.Progress(track_tqdm=True)):
+    def _gpu_logic(self, ui_inputs: Dict, loras_string: str, workflow: Dict[str, Any]):
         model_display_name = ui_inputs['model_display_name']
         succeeded = False
         try:
-            progress(0.4, desc="正在执行工作流…")
-
             initial_objects = {}
             decoded_images_tensor = WorkflowExecutor.execute_workflow(
                 workflow, initial_objects=initial_objects
@@ -306,16 +304,20 @@ class SdImagePipeline(BasePipeline):
             gpu_duration = estimate_gpu_duration(ui_inputs)
             progress(1.0, desc=f"模型已就绪，正在申请 GPU（预计上限 {gpu_duration} 秒）…")
 
+            # 取消信号和持有它的 assembler 留在 CPU 侧，不能跨 ZeroGPU 序列化边界。
+            gpu_inputs = {
+                key: value
+                for key, value in ui_inputs.items()
+                if key != "_cancel_event"
+            }
             results = self._execute_gpu_logic(
                 self._gpu_logic,
                 duration=gpu_duration,
                 default_duration=60,
                 task_name=f"ImageGen ({task_type})",
-                ui_inputs=ui_inputs,
+                ui_inputs=gpu_inputs,
                 loras_string=loras_string,
                 workflow=workflow,
-                assembler=assembler,
-                progress=progress
             )
 
             import json

tests/test_execution_plan.py

@@ -119,9 +119,38 @@ class ExecutionPlanTests(unittest.TestCase):
                 raise RuntimeError("显存不足")
             return ["one.png"]
 
-        gallery, summary = execute_generation_plan(plan, generate)
+        with self.assertLogs("core.execution_plan", level="ERROR") as logs:
+            gallery, summary = execute_generation_plan(plan, generate)
         self.assertEqual(gallery, [("one.png", "模型一")])
         self.assertIn("模型二：显存不足", summary)
+        self.assertIn("RuntimeError: 显存不足", logs.output[0])
+
+    def test_failed_job_logs_original_traceback_without_input_payload(self):
+        plan = [
+            PlannedGeneration(
+                {
+                    "model_display_name": "example-model",
+                    "positive_prompt": "do-not-log-this-input",
+                },
+                "测试任务",
+            )
+        ]
+
+        def generate(_inputs, _progress):
+            raise TypeError("cannot pickle '_thread.lock' object")
+
+        with (
+            self.assertLogs("core.execution_plan", level="ERROR") as logs,
+            self.assertRaisesRegex(ExecutionPlanError, "cannot pickle"),
+        ):
+            execute_generation_plan(plan, generate)
+
+        self.assertEqual(len(logs.records), 1)
+        self.assertIs(logs.records[0].exc_info[0], TypeError)
+        self.assertIn("Traceback (most recent call last)", logs.output[0])
+        self.assertIn("in generate", logs.output[0])
+        self.assertIn("模型=example-model", logs.output[0])
+        self.assertNotIn("do-not-log-this-input", logs.output[0])
 
     def test_executor_stops_before_next_case_after_cancel(self):
         cancellation = threading.Event()

tests/test_gpu_boundary.py

@@ -0,0 +1,197 @@
+from __future__ import annotations
+
+import importlib.util
+import pickle
+import sys
+import tempfile
+import threading
+import types
+import unittest
+from pathlib import Path
+from unittest import mock
+
+from core.task_scheduler import TaskCancelledError
+
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+class _LockedProgress:
+    def __init__(self, *args, **kwargs):
+        self.lock = threading.Lock()
+        self.updates = []
+
+    def __call__(self, value, desc=None):
+        self.updates.append((value, desc))
+
+
+class _Assembler:
+    def __init__(self, after_assembly=None):
+        self.after_assembly = after_assembly
+
+    def assemble(self, values):
+        self.ui_values = values
+        if self.after_assembly:
+            self.after_assembly()
+        return {"1": {"class_type": "TestNode", "inputs": {"seed": values["seed"]}}}
+
+
+def _module(name, **attributes):
+    module = types.ModuleType(name)
+    module.__dict__.update(attributes)
+    return module
+
+
+class GpuBoundaryTests(unittest.TestCase):
+    """执行真实 pipeline 调用链，只替换下载、推理依赖和 GPU 传输。"""
+
+    def setUp(self):
+        output_dir = self.enterContext(tempfile.TemporaryDirectory())
+        self.payloads = []
+        self.assemblers = []
+        self.after_assembly = None
+        self.download = mock.Mock()
+        self.release_models = mock.Mock()
+        self.execute_workflow = mock.Mock(
+            return_value=types.SimpleNamespace(shape=(0,))
+        )
+
+        def gpu_decorator(**_options):
+            def decorate(function):
+                def run(*args, **kwargs):
+                    # 真正序列化入参，不能像 UI 烟测那样直接跳过生成调用。
+                    args, kwargs = pickle.loads(pickle.dumps((args, kwargs)))
+                    self.payloads.append(kwargs)
+                    return function(*args, **kwargs)
+
+                return run
+
+            return decorate
+
+        def create_assembler(*_args, **_kwargs):
+            assembler = _Assembler(self.after_assembly)
+            self.assemblers.append(assembler)
+            return assembler
+
+        prepared_keys = (
+            "temp_files_to_clean", "active_loras_for_gpu", "active_loras_for_meta",
+            "active_controlnets", "active_anima_controlnets", "active_diffsynth_controlnets",
+            "active_ipadapters", "active_flux1_ipadapters", "active_sd3_ipadapters",
+            "active_styles", "active_reference_latents", "active_hidream_o1_reference",
+            "active_conditioning",
+        )
+        base_name = "core.pipelines.base_pipeline"
+        pipeline_name = "core.pipelines.sd_image_pipeline"
+        stubs = {
+            "gradio": _module("gradio", Error=RuntimeError, Progress=_LockedProgress),
+            "spaces": _module("spaces", GPU=gpu_decorator),
+            "torch": _module("torch"),
+            "imageio": _module("imageio"),
+            "numpy": _module("numpy"),
+            "core.model_manager": _module(
+                "core.model_manager",
+                model_manager=types.SimpleNamespace(ensure_models_downloaded=self.download),
+                release_loaded_models=self.release_models,
+            ),
+            "core.workflow_assembler": _module(
+                "core.workflow_assembler", WorkflowAssembler=create_assembler
+            ),
+            "imagegen_utils.app_utils": _module(
+                "imagegen_utils.app_utils", sanitize_prompt=lambda value: value
+            ),
+            "core.pipelines.pipeline_input_processor": _module(
+                "core.pipelines.pipeline_input_processor",
+                process_pipeline_inputs=lambda *_: {key: [] for key in prepared_keys},
+            ),
+            "core.pipelines.workflow_executor": _module(
+                "core.pipelines.workflow_executor",
+                WorkflowExecutor=types.SimpleNamespace(execute_workflow=self.execute_workflow),
+            ),
+        }
+        specs = []
+        for name in (base_name, pipeline_name):
+            spec = importlib.util.spec_from_file_location(
+                name, ROOT / "core" / "pipelines" / f"{name.rsplit('.', 1)[1]}.py"
+            )
+            stubs[name] = importlib.util.module_from_spec(spec)
+            specs.append(spec)
+
+        self.enterContext(mock.patch.dict(sys.modules, stubs))
+        for spec in specs:
+            spec.loader.exec_module(stubs[spec.name])
+        self.module = stubs[pipeline_name]
+        self.module.OUTPUT_DIR = output_dir
+        self.pipeline = self.module.SdImagePipeline()
+
+    def inputs(self, **overrides):
+        values = {
+            "task_type": "txt2img",
+            "model_display_name": "circlestone-labs/Anima-Turbo-v1.0",
+            "positive_prompt": "测试图片",
+            "negative_prompt": "",
+            "seed": 42,
+            "batch_size": 1,
+            "num_inference_steps": 4,
+            "guidance_scale": 1.0,
+            "sampler": "euler",
+            "scheduler": "simple",
+            "width": 64,
+            "height": 64,
+            "denoise": 1.0,
+        }
+        values.update(overrides)
+        return values
+
+    def test_repeated_ui_calls_serialize_and_preserve_model_release(self):
+        for release in (True, False):
+            with self.subTest(release=release):
+                cancellation = threading.Event()
+                values = self.inputs(
+                    _cancel_event=cancellation, _release_models_after_run=release
+                )
+                progress = _LockedProgress()
+                self.assertEqual(self.pipeline.run(values, progress), [])
+
+                payload = self.payloads[-1]
+                self.assertEqual(set(payload), {"ui_inputs", "loras_string", "workflow"})
+                self.assertNotIn("_cancel_event", payload["ui_inputs"])
+                self.assertEqual(payload["ui_inputs"]["_release_models_after_run"], release)
+                self.assertIs(values["_cancel_event"], cancellation)
+                self.assertIs(self.assemblers[-1].ui_values["_cancel_event"], cancellation)
+                self.assertTrue(progress.updates)
+
+        self.assertEqual(len(self.payloads), 2)
+        self.release_models.assert_called_once_with()
+
+    def test_request_without_cancel_event_serializes(self):
+        self.pipeline.run(self.inputs(), _LockedProgress())
+        self.assertNotIn("_cancel_event", self.payloads[0]["ui_inputs"])
+        self.release_models.assert_not_called()
+
+    def test_cancellation_after_download_stays_on_cpu(self):
+        cancellation = threading.Event()
+        self.download.side_effect = lambda *args, **kwargs: cancellation.set()
+        with self.assertRaises(TaskCancelledError):
+            self.pipeline.run(self.inputs(_cancel_event=cancellation), _LockedProgress())
+        self.assertEqual(self.payloads, [])
+        self.execute_workflow.assert_not_called()
+
+    def test_cancellation_after_assembly_stays_on_cpu(self):
+        cancellation = threading.Event()
+        self.after_assembly = cancellation.set
+        with self.assertRaises(TaskCancelledError):
+            self.pipeline.run(self.inputs(_cancel_event=cancellation), _LockedProgress())
+        self.assertEqual(self.payloads, [])
+        self.execute_workflow.assert_not_called()
+
+    def test_gpu_failure_still_releases_model_state(self):
+        self.execute_workflow.side_effect = RuntimeError("sampling failed")
+        with self.assertRaisesRegex(RuntimeError, "sampling failed"):
+            self.pipeline.run(
+                self.inputs(_cancel_event=threading.Event()), _LockedProgress()
+            )
+        self.release_models.assert_called_once_with()
+
+
+if __name__ == "__main__":
+    unittest.main()