PR4: param_schema centralization + FP8 preflight hardening

src/streamdiffusion/acceleration/tensorrt/runtime_engines/unet_engine.py

@@ -55,10 +55,39 @@ def __init__(self, filepath: str, stream: "cuda.Stream", use_cuda_graph: bool =
         self._fio_cache_len: int = -1  # -1 = not yet initialized
 
         # Sub-phase 5.6: kvo/fio cache inputs are persistent, address-stable,
-        # TRT-contiguous tensors (see models/utils.py create_kvo_cache/create_fi_cache),
-        # so Engine.infer() can bind TensorRT directly to them instead of copying into
-        # its own staging buffer every frame. Rebuilt only when the kvo/fio lazy-init
-        # branches below fire (cache length change), never per-frame.
+        # TRT-contiguous tensors in steady state (models/utils.py
+        # create_kvo_cache/create_fi_cache), so Engine.infer() can bind
+        # TensorRT directly to them instead of copying into its own staging
+        # buffer every frame. `_staging_action` falls back to a copy on
+        # non-contiguous/dtype-mismatch/mis-aligned inputs, and to
+        # bind_and_reset on a pointer change — see utilities.py. Rebuilt only
+        # when the kvo/fio lazy-init branches below fire (cache length
+        # change), never per-frame.
+        #
+        # ControlNet input_control_* residuals are deliberately, PERMANENTLY
+        # EXCLUDED here (reverted from Phase-2 D2 / f8ff50f) and stay on the
+        # DtoD-copy staging path. Two reasons, confirmed against code, not
+        # just bisection:
+        #   1. The residual source toggles between the ControlNet engine's
+        #      live output and idle cached-dummy-zero tensors
+        #      (_add_controlnet_residuals vs _add_cached_dummy_inputs below),
+        #      and a missed bind_and_reset on that toggle silently pins the
+        #      UNet's CUDA graph to whichever buffer was bound at capture
+        #      time.
+        #   2. copy_() runs on the engine stream and orders the ControlNet
+        #      engine's residual write before the UNet graph's read of it; a
+        #      zero-copy bind drops that ordering guarantee with no
+        #      replacement synchronization (see the "copy_() executes on
+        #      engine stream to guarantee ordering" log line in utilities.py).
+        # Re-enabling zero-copy for these names reproduced the "ControlNet
+        # produces no visual change" regression on the rig for both a
+        # single-scale (Canny) and ramped-scale (scribble) config, even
+        # though single-ControlNet residuals are themselves contiguous,
+        # 256-aligned, persistent buffers (controlnet_module.py
+        # build_unet_hook single-CN path) — i.e. the failure is NOT explained
+        # by the copy-fallback path ever firing; it is the ordering guarantee
+        # above. Do not re-add these names to _zero_copy_names without a
+        # rig-verified CUDA-event ordering fix.
         self._zero_copy_names: FrozenSet[str] = frozenset()
 
         self._shape_dict: Dict[str, Any] = {}

src/streamdiffusion/acceleration/tensorrt/utilities.py

@@ -385,8 +385,11 @@ def _apply_gpu_profile_to_config(
     # + EDGE_MASK_CONVOLUTIONS — the sources that produce valid SM_120 kernels.
     # TRT 10.16 exposes TacticSource as an int enum (not IntFlag), so the bitmask
     # is built via (1 << int(source)). No-op on Ada/Ampere.
+    # CUBLAS/CUBLAS_LT are removed from TacticSource in TRT 11 (cuBLAS tactics
+    # dropped entirely) — gate on hasattr so that's a deliberate, logged branch
+    # rather than a silently-swallowed AttributeError.
     if gpu_profile.compute_capability >= (12, 0):
-        try:
+        if hasattr(trt.TacticSource, "CUBLAS"):
             sources = (
                 (1 << int(trt.TacticSource.CUBLAS))
                 | (1 << int(trt.TacticSource.CUBLAS_LT))
@@ -397,8 +400,11 @@ def _apply_gpu_profile_to_config(
             logger.info(
                 "[TRT Config] tactic sources = CUBLAS|CUBLAS_LT|JIT_CONV|EDGE_MASK (CUDNN excluded for SM_120+)"
             )
-        except (AttributeError, TypeError) as e:
-            logger.debug(f"[TRT Config] set_tactic_sources not available: {e}")
+        else:
+            logger.info(
+                "[TRT Config] TRT >=11: CUBLAS/CUBLAS_LT removed from TacticSource — "
+                "default tactic sources already exclude cuDNN/cuBLAS, nothing to scope"
+            )
 
     # max_num_tactics: cap profiling candidates per layer to reduce build time.
     # Available since TRT 10.x; -1 (default) lets TRT decide heuristically. 64 is a
@@ -558,12 +564,28 @@ def _staging_action(
 
     Returns one of:
       "copy"           - copy into self.tensors[name] (today's behavior; always safe)
+      "copy_and_reset" - copy into self.tensors[name], but reset the CUDA graph first
+                         because this name was previously bound zero-copy into a live
+                         graph, which still reads the old bound address rather than
+                         self.tensors[name]
       "bind"           - skip the copy; bind TensorRT directly to the caller's tensor
       "bind_and_reset" - skip the copy and bind directly, but reset the CUDA graph
                          first because the caller's address changed while a graph
                          built from the old address was still live
+
+    A pointer that is not 256-byte aligned falls back to "copy": TensorRT's
+    setTensorAddress contract requires ≥256-byte alignment, and real torch CUDA
+    allocations are always ≥512-byte aligned, so this guard is a defensive
+    invariant against config drift rather than a path exercised in production.
     """
-    if name not in zero_copy_names or not is_contiguous or not dtype_match:
+    if name not in zero_copy_names or not is_contiguous or not dtype_match or cur_ptr % 256 != 0:
+        # Falling back to copy. If this name was bound zero-copy into a LIVE graph,
+        # that graph still reads the stale bound address, not self.tensors[name] —
+        # force one reset so it re-captures reading the staging buffer. prev_ptr is
+        # only non-None for names actually bound before, so plain copy-path inputs
+        # (never in zero_copy_names) never trip this.
+        if graph_exists and prev_ptr is not None:
+            return "copy_and_reset"
         return "copy"
     if graph_exists and cur_ptr != prev_ptr:
         return "bind_and_reset"
@@ -999,12 +1021,8 @@ def get_input_profile_bounds(self, name: str):
         except Exception:
             return None
 
-    def activate(self, reuse_device_memory=None):
-        if reuse_device_memory:
-            self.context = self.engine.create_execution_context_without_device_memory()
-            self.context.device_memory = reuse_device_memory
-        else:
-            self.context = self.engine.create_execution_context()
+    def activate(self):
+        self.context = self.engine.create_execution_context()
 
         # Attach per-layer profiler when STREAMDIFFUSION_PROFILE_TRT is set.
         # Requires engines built with profiling_verbosity=DETAILED for meaningful names.
@@ -1191,8 +1209,14 @@ def infer(self, feed_dict, stream, use_cuda_graph=False, zero_copy_names: frozen
                         cur_ptr,
                         graph_exists,
                     )
-                    if action == "copy":
+                    if action in ("copy", "copy_and_reset"):
                         self.tensors[name].copy_(buf)
+                        if action == "copy_and_reset":
+                            needs_reset = True
+                            # Next frame this name has no prior bind, so
+                            # _staging_action sees prev_ptr=None and returns
+                            # plain "copy" — converges after one reset.
+                            self._bound_ptrs.pop(name, None)
                     else:
                         bind_ptrs[name] = cur_ptr
                         if action == "bind_and_reset":
@@ -1525,7 +1549,6 @@ def export_onnx(
             onnx_path,
             export_params=True,
             opset_version=onnx_opset,
-            do_constant_folding=True,
             input_names=model_data.get_input_names(),
             output_names=model_data.get_output_names(),
             dynamic_axes=model_data.get_dynamic_axes(),
@@ -1577,15 +1600,17 @@ def optimize_onnx(
 ):
     import os
 
-    # Check if external data files exist (indicating external data format was used)
     onnx_dir = os.path.dirname(onnx_path)
-    external_data_files = [f for f in os.listdir(onnx_dir) if f.endswith(".pb")]
-    uses_external_data = len(external_data_files) > 0
+    # Inspect TensorProto.data_location on the raw (unloaded) model rather than
+    # scanning the directory for ".pb" filenames — load_external_data_for_model
+    # resets data_location back to DEFAULT once external data is loaded, so this
+    # check must happen before loading.
+    onnx_model = onnx.load(onnx_path, load_external_data=False)
+    uses_external_data = any(onnx.external_data_helper.uses_external_data(t) for t in onnx_model.graph.initializer)
 
     if uses_external_data:
-        logger.info(f"Optimizing ONNX with external data (found: {external_data_files})")
-        # Load model with external data
-        onnx_model = onnx.load(onnx_path, load_external_data=True)
+        logger.info("Optimizing ONNX with external data")
+        onnx.external_data_helper.load_external_data_for_model(onnx_model, onnx_dir)
         onnx_opt_graph = model_data.optimize(onnx_model)
 
         # Create output directory
@@ -1610,8 +1635,7 @@ def optimize_onnx(
         logger.info("ONNX optimization complete with external data")
 
     else:
-        # Standard optimization for smaller models
-        onnx_model = onnx.load(onnx_path)
+        # No external data to load — the model loaded above is already complete.
         onnx_opt_graph = model_data.optimize(onnx_model)
 
         onnx.save(onnx_opt_graph, onnx_opt_path)

src/streamdiffusion/config.py

@@ -1,16 +1,16 @@
-import logging
-import os
-import sys
-import yaml
 import json
+import logging
 from pathlib import Path
 from typing import Any, Dict, List, Tuple, Union
 
 import yaml
 
+from .param_schema import DEFAULTS
+
 
 logger = logging.getLogger(__name__)
 
+
 def load_config(config_path: Union[str, Path]) -> Dict[str, Any]:
     """Load StreamDiffusion configuration from YAML or JSON file"""
     config_path = Path(config_path)
@@ -96,7 +96,7 @@ def create_wrapper_from_config(config: Dict[str, Any], **overrides) -> Any:
         seed_blend_config = final_config["seed_blending"]
         wrapper.update_stream_params(
             seed_list=seed_blend_config.get("seed_list", []),
-            interpolation_method=seed_blend_config.get("interpolation_method", "linear"),
+            seed_interpolation_method=seed_blend_config.get("interpolation_method", "linear"),
         )
 
     return wrapper
@@ -106,7 +106,7 @@ def _extract_wrapper_params(config: Dict[str, Any]) -> Dict[str, Any]:
     """Extract parameters for StreamDiffusionWrapper.__init__() from config"""
     param_map = {
         "model_id_or_path": config.get("model_id", "stabilityai/sd-turbo"),
-        "t_index_list": config.get("t_index_list", [0, 16, 32, 45]),
+        "t_index_list": config.get("t_index_list", list(DEFAULTS["t_index_list"])),
         "lora_dict": config.get("lora_dict"),
         "mode": config.get("mode", "img2img"),
         "output_type": config.get("output_type", "pil"),
@@ -128,12 +128,12 @@ def _extract_wrapper_params(config: Dict[str, Any]) -> Dict[str, Any]:
         "similar_filter_sleep_fraction": config.get("similar_filter_sleep_fraction", 0.025),
         "use_denoising_batch": config.get("use_denoising_batch", True),
         "cfg_type": config.get("cfg_type", "self"),
-        "seed": config.get("seed", 2),
-        "use_safety_checker": config.get("use_safety_checker", False),
+        "seed": config.get("seed", DEFAULTS["seed"]),
+        "use_safety_checker": config.get("use_safety_checker", DEFAULTS["use_safety_checker"]),
         "skip_diffusion": config.get("skip_diffusion", False),
         "engine_dir": config.get("engine_dir", "engines"),
-        "normalize_prompt_weights": config.get("normalize_prompt_weights", True),
-        "normalize_seed_weights": config.get("normalize_seed_weights", True),
+        "normalize_prompt_weights": config.get("normalize_prompt_weights", DEFAULTS["normalize_prompt_weights"]),
+        "normalize_seed_weights": config.get("normalize_seed_weights", DEFAULTS["normalize_seed_weights"]),
         "scheduler": config.get("scheduler", "lcm"),
         "sampler": config.get("sampler", "normal"),
         "compile_engines_only": config.get("compile_engines_only", False),
@@ -161,12 +161,12 @@ def _extract_wrapper_params(config: Dict[str, Any]) -> Dict[str, Any]:
 
     param_map["use_cached_attn"] = config.get("use_cached_attn", False)
 
-    param_map["cache_maxframes"] = config.get("cache_maxframes", 1)
-    param_map["cache_interval"] = config.get("cache_interval", 1)
+    param_map["cache_maxframes"] = config.get("cache_maxframes", DEFAULTS["cache_maxframes"])
+    param_map["cache_interval"] = config.get("cache_interval", DEFAULTS["cache_interval"])
     # cn_cache_interval: ControlNet residual reuse interval.
     # 1 (default) = disabled, run CN every frame.
     # N > 1 = run CN once every N frames; reuse residuals between (control latency = N-1 frames).
-    param_map["cn_cache_interval"] = config.get("cn_cache_interval", 1)
+    param_map["cn_cache_interval"] = config.get("cn_cache_interval", DEFAULTS["cn_cache_interval"])
 
     # Feature Injection (StreamV2V §3.4.2) — requires use_cached_attn=True
     if "use_feature_injection" not in config:
@@ -178,8 +178,8 @@ def _extract_wrapper_params(config: Dict[str, Any]) -> Dict[str, Any]:
     param_map["use_feature_injection"] = config.get("use_feature_injection", True)
     # fi_strength: blend weight α (thesis §3.4.2 Eq 3.2 α=0.75; default matches thesis).
     # fi_threshold: cosine-similarity gate below which injection is skipped (default 0.98).
-    param_map["fi_strength"] = config.get("fi_strength", 0.75)
-    param_map["fi_threshold"] = config.get("fi_threshold", 0.98)
+    param_map["fi_strength"] = config.get("fi_strength", DEFAULTS["fi_strength"])
+    param_map["fi_threshold"] = config.get("fi_threshold", DEFAULTS["fi_threshold"])
     # max_cache_maxframes: allocation cap for the KVO/FI cache ring buffers (VRAM).
     # cache_maxframes is the live logical write window; this is the hard upper bound.
     param_map["max_cache_maxframes"] = config.get("max_cache_maxframes", 4)
@@ -206,10 +206,10 @@ def _extract_prepare_params(config: Dict[str, Any]) -> Dict[str, Any]:
     """Extract parameters for wrapper.prepare() from config"""
     prepare_params = {
         "prompt": config.get("prompt", ""),
-        "negative_prompt": config.get("negative_prompt", ""),
-        "num_inference_steps": config.get("num_inference_steps", 50),
-        "guidance_scale": config.get("guidance_scale", 1.2),
-        "delta": config.get("delta", 1.0),
+        "negative_prompt": config.get("negative_prompt", DEFAULTS["negative_prompt"]),
+        "num_inference_steps": config.get("num_inference_steps", DEFAULTS["num_inference_steps"]),
+        "guidance_scale": config.get("guidance_scale", DEFAULTS["guidance_scale"]),
+        "delta": config.get("delta", DEFAULTS["delta"]),
     }
 
     # Handle prompt blending configuration
@@ -335,67 +335,6 @@ def _prepare_single_hook_config(hook_config: Dict[str, Any], hook_type: str) ->
     }
 
 
-def _validate_pipeline_hook_configs(config: Dict[str, Any]) -> None:
-    """Validate pipeline hook configurations following ControlNet/IPAdapter validation pattern"""
-    hook_types = ["image_preprocessing", "image_postprocessing", "latent_preprocessing", "latent_postprocessing"]
-
-    for hook_type in hook_types:
-        if hook_type in config:
-            hook_config = config[hook_type]
-            if not isinstance(hook_config, dict):
-                raise ValueError(f"_validate_config: '{hook_type}' must be a dictionary")
-
-            # Validate enabled field
-            if "enabled" in hook_config:
-                enabled = hook_config["enabled"]
-                if not isinstance(enabled, bool):
-                    raise ValueError(f"_validate_config: '{hook_type}.enabled' must be a boolean")
-
-            # Validate processors field
-            if "processors" in hook_config:
-                processors = hook_config["processors"]
-                if not isinstance(processors, list):
-                    raise ValueError(f"_validate_config: '{hook_type}.processors' must be a list")
-
-                for i, processor in enumerate(processors):
-                    if not isinstance(processor, dict):
-                        raise ValueError(f"_validate_config: '{hook_type}.processors[{i}]' must be a dictionary")
-
-                    # Validate processor type (required)
-                    if "type" not in processor:
-                        raise ValueError(
-                            f"_validate_config: '{hook_type}.processors[{i}]' missing required 'type' field"
-                        )
-
-                    if not isinstance(processor["type"], str):
-                        raise ValueError(f"_validate_config: '{hook_type}.processors[{i}].type' must be a string")
-
-                    # Validate enabled field (optional, defaults to True)
-                    if "enabled" in processor:
-                        enabled = processor["enabled"]
-                        if not isinstance(enabled, bool):
-                            raise ValueError(
-                                f"_validate_config: '{hook_type}.processors[{i}].enabled' must be a boolean"
-                            )
-
-                    # Validate order field (optional)
-                    if "order" in processor:
-                        order = processor["order"]
-                        if not isinstance(order, int):
-                            raise ValueError(
-                                f"_validate_config: '{hook_type}.processors[{i}].order' must be an integer"
-                            )
-
-                    # Validate params field (optional, coerce None to empty dict)
-                    if "params" in processor:
-                        if processor["params"] is None:
-                            processor["params"] = {}
-                        elif not isinstance(processor["params"], dict):
-                            raise ValueError(
-                                f"_validate_config: '{hook_type}.processors[{i}].params' must be a dictionary"
-                            )
-
-
 def create_prompt_blending_config(
     base_config: Dict[str, Any],
     prompt_list: List[Tuple[str, float]],
@@ -548,4 +487,4 @@ def _validate_config(config: Dict[str, Any]) -> None:
                     raise ValueError(f"_validate_config: Seed value {i} must be a non-negative integer")
 
                 if not isinstance(weight, (int, float)) or weight < 0:
-                    raise ValueError(f"_validate_config: Seed weight {i} must be a non-negative number")
+                    raise ValueError(f"_validate_config: Seed weight {i} must be a non-negative number")