nodes/Artificial-Sweetener-SimpleSyrup
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(sampling): preserve native trajectories for CD noise reuse

Browse Source 
Capture the nearest existing source model evaluation without inserting sampling steps, retain its actual noise coordinate, and recover residuals with stable native scaling.

Observe source and receiver states at ComfyUI's shared boundaries, discover compatible named/custom source nodes, and report graph node identities and selected samplers in failures. Admit the complete sampler catalog through CD and tiled routes with regression coverage for future additions.

Verify the isolated CD-only tree: 4322 Python tests passed, 48 skipped; 115 frontend tests passed; format, lint, strict typing, architecture, and test governance passed. Contextual Attention remains outside this commit.
This commit is contained in:
Artificial Sweetener
2026-10-06 01:25:06 -04:00
parent 027efb8eef
commit f89b436a09
49 changed files with 1099 additions and 342 deletions
Download Patch File Download Diff File Expand all files Collapse all files
governance/architecture/soft_reviews.toml
+1 -1
View File
@@ -1,6 +1,6 @@
schema_version = 1
review_by = 2027-03-31
fingerprint = "sha256:a71fe87163eb7585fafbca08b585954be79abd884e3a1df38bc041c2bc934764"
fingerprint = "sha256:766e064dd709d770a9628931ef37920d9207fc1a993b89e55db153c378937236"
cohesive_paths = [
"simple_syrup/masking/prompt_segs_with_sam_service.py",
simple_syrup/domain/latent_upscale.py
+3
View File
@@ -1,3 +1,6 @@
# SimpleSyrup - workflow-focused ComfyUI extensions for image generation
# Copyright (C) 2026 Artificial Sweetener and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Define bounded learned scaling stages and verified format recommendations."""
from __future__ import annotations
simple_syrup/domain/noise_reuse.py
+46 -15
View File
@@ -1,3 +1,7 @@
# SimpleSyrup - workflow-focused ComfyUI extensions for image generation
# Copyright (C) 2026 Artificial Sweetener and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Describe native trajectory captures and independent finishing controls."""
from __future__ import annotations
@@ -47,30 +51,57 @@ class NoiseReuseOptions:
return float(torch.tensor(target * fraction, dtype=torch.float32))
@dataclass(frozen=True, slots=True)
class CapturedSamplingState:
"""Associate a requested level with a real observed state and recovered noise."""
requested_coordinate: float
actual_coordinate: float
state: torch.Tensor
residual: torch.Tensor
def __post_init__(self) -> None:
"""Require finite positive coordinates and matching real tensor layouts."""
if not all(
math.isfinite(value) and value > 0
for value in (self.requested_coordinate, self.actual_coordinate)
):
raise ValueError("Saved noise levels must be finite and greater than zero.")
if self.state.shape != self.residual.shape:
raise ValueError(
"Saved noisy state and recovered noise must have one shape."
)
if not bool(torch.isfinite(self.state).all()) or not bool(
torch.isfinite(self.residual).all()
):
raise ValueError("The original sampler saved invalid noise values.")
@dataclass(frozen=True, slots=True)
class SamplingProvenance:
"""Carry the source identity and real model-space states beside external latents."""
latent_format: str
clean: torch.Tensor
states: tuple[tuple[float, torch.Tensor], ...] = ()
captures: tuple[CapturedSamplingState, ...] = ()
schedule: torch.Tensor | None = None
residuals: tuple[tuple[float, torch.Tensor], ...] = ()
graph_context: str = ""
def capture_at(self, coordinate: float) -> CapturedSamplingState:
"""Require an observation explicitly associated with this receiving request."""
for capture in self.captures:
if capture.requested_coordinate == coordinate:
return capture
raise ValueError(
f"The original sampling run did not save noise for the requested level "
f"{coordinate:.9g}. Queue the connected workflow again with Noise Reuse "
"Options enabled."
)
def state_at(self, coordinate: float) -> torch.Tensor:
"""Require an exact requested schedule coordinate instead of approximating."""
for sigma, state in self.states:
if sigma == coordinate:
return state
raise ValueError(
f"Source sampling did not capture sigma {coordinate:.9g}. "
"Queue the connected workflow again with Noise Reuse Options enabled."
)
"""Return the real noisy state associated with the requested level."""
return self.capture_at(coordinate).state
def residual_at(self, coordinate: float) -> torch.Tensor:
"""Return noise separated with the originating model's own parameterization."""
self.state_at(coordinate)
for sigma, residual in self.residuals:
if sigma == coordinate:
return residual
raise ValueError("Source capture is missing its native recovered residual.")
return self.capture_at(coordinate).residual
simple_syrup/domain/spectral_noise.py
+3
View File
@@ -1,3 +1,6 @@
# SimpleSyrup - workflow-focused ComfyUI extensions for image generation
# Copyright (C) 2026 Artificial Sweetener and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Lift separated residuals in orthonormal spatial frequency coordinates."""
from __future__ import annotations
simple_syrup/nodes_v3/latent_from_image.py
+3
View File
@@ -1,3 +1,6 @@
# SimpleSyrup - workflow-focused ComfyUI extensions for image generation
# Copyright (C) 2026 Artificial Sweetener and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Expose original decoded-image latent recovery as a strict provenance bridge."""
from __future__ import annotations
simple_syrup/nodes_v3/noise_reuse_options.py
+3
View File
@@ -1,3 +1,6 @@
# SimpleSyrup - workflow-focused ComfyUI extensions for image generation
# Copyright (C) 2026 Artificial Sweetener and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Expose captured noise reuse and optional full-resolution inversion finishing."""
from __future__ import annotations
simple_syrup/nodes_v3/simple_upscale_latent.py
+3
View File
@@ -1,3 +1,6 @@
# SimpleSyrup - workflow-focused ComfyUI extensions for image generation
# Copyright (C) 2026 Artificial Sweetener and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Expose automatic learned and native latent scaling through one method control."""
from __future__ import annotations
simple_syrup/nodes_v3/trajectory_capture.py
+27 -2
View File
@@ -1,3 +1,7 @@
# SimpleSyrup - workflow-focused ComfyUI extensions for image generation
# Copyright (C) 2026 Artificial Sweetener and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Expose internal cache-keyed capture planning and handoff nodes."""
from __future__ import annotations
@@ -114,6 +118,17 @@ class TrajectoryCaptureModelV3(OptionsNodeBase):
COMFY_IO.String.Input(
"sampler_name", tooltip="Source solver identifier."
),
COMFY_IO.Sampler.Input(
"sampler",
optional=True,
tooltip="Connected source algorithm used by a custom sampler.",
),
COMFY_IO.String.Input(
"graph_context",
optional=True,
default="",
tooltip="Source and receiving node names for error messages.",
),
COMFY_IO.Custom(REQUEST_TYPE).Input(
"requests",
optional=True,
@@ -135,10 +150,20 @@ class TrajectoryCaptureModelV3(OptionsNodeBase):
@classmethod
def execute(
cls, model: Any, sampler_name: str, requests: tuple[float, ...] = ()
cls,
model: Any,
sampler_name: str,
requests: tuple[float, ...] = (),
sampler: Any = None,
graph_context: str = "",
) -> tuple[Any, TrajectoryCapture]:
"""Keep observer state scoped to this source and cache identity."""
capture = TrajectoryCapture(model, sampler_name, requests)
if not sampler_name and sampler is not None:
function = getattr(sampler, "sampler_function", None)
sampler_name = getattr(function, "__name__", type(sampler).__name__)
capture = TrajectoryCapture(
model, sampler_name, requests, graph_context=graph_context
)
return capture.prepare_model(), capture
simple_syrup/runtime/contextual_diffusion_sampling.py
+1 -3
View File
@@ -46,7 +46,6 @@ from .tiled_sampling_validation import (
LOGGER = get_logger(__name__)
SAMPLER_LABEL = "Contextual Diffusion"
UNIPC_SAMPLERS = frozenset({"uni_pc", "uni_pc_bh2"})
def sample_contextual_diffusion(
@@ -82,8 +81,6 @@ def sample_contextual_diffusion(
latent_tile_batch_size=controls.latent_context_batch_size,
)
controls.validate()
if sampler_name in UNIPC_SAMPLERS:
raise ValueError("Contextual Diffusion is not compatible with UniPC samplers.")
reject_unsupported_conditioning(
positive,
sampler_label=SAMPLER_LABEL,
@@ -166,6 +163,7 @@ def sample_contextual_diffusion(
noise=noise,
cfg=cfg,
sampler=sampler,
sampler_name=sampler_name,
sigmas=sigmas,
positive=positive,
negative=negative,
simple_syrup/runtime/detail_sampling.py
+1
View File
@@ -100,6 +100,7 @@ class DetailSampler:
noise=noise,
cfg=cfg,
sampler=sampler,
sampler_name=sampler_name,
sigmas=sigmas,
positive=positive,
negative=negative,
simple_syrup/runtime/guided_sampling.py
+16 -24
View File
@@ -16,6 +16,7 @@ from ..shared.logging import get_logger
from .noise_inversion import InversionModelFactory, invert_sampling_noise
from .noise_reuse_sampling import prepare_reused_noise
from .noise_reuse_scope import current_noise_reuse
from .noise_reuse_start import NoiseReuseStartVerification
LOGGER = get_logger(__name__)
@@ -27,6 +28,7 @@ def sample_with_optional_negative(
noise: torch.Tensor,
cfg: float,
sampler: Any,
sampler_name: str,
sigmas: torch.Tensor,
positive: Any,
negative: Any | None,
@@ -41,6 +43,7 @@ def sample_with_optional_negative(
"""Prepare optional source-derived noise and select the actual Comfy guider."""
reuse = current_noise_reuse()
start_verification: NoiseReuseStartVerification | None = None
if reuse is not None:
if noise_inversion is not None:
raise ValueError(
@@ -60,29 +63,13 @@ def sample_with_optional_negative(
noise_mask=noise_mask,
model_factory=inversion_model_factory,
)
original_callback = callback
def observe_start(
step: int, denoised: torch.Tensor, state: torch.Tensor, total: int
) -> None:
"""Check the actual first state to catch double-noising or schedule
mismatch.
"""
if step == 0 and noise_mask is None:
error = float((state.detach().float().cpu() - expected).abs().max())
if error > 2e-5:
raise ValueError(
f"Noise reuse starting state differs from native "
f"reconstruction: {error:.6g}."
)
LOGGER.info(
"Noise reuse starting state verified",
extra={"operation": "noise_reuse_start", "max_error": error},
)
if original_callback is not None:
original_callback(step, denoised, state, total)
callback = observe_start
start_verification = NoiseReuseStartVerification(
model.get_model_object("model_sampling.noise_scaling"),
expected,
graph_context=reuse.provenance.graph_context,
sampler_name=sampler_name,
)
model = start_verification.derive_model(model)
if noise_inversion is not None:
inversion = invert_sampling_noise(
@@ -100,7 +87,7 @@ def sample_with_optional_negative(
noise = inversion.noise.to(noise)
if negative is not None:
return cast(
samples = cast(
torch.Tensor,
comfy_sample.sample_custom(
model,
@@ -117,6 +104,9 @@ def sample_with_optional_negative(
seed=seed,
),
)
if start_verification is not None:
start_verification.require_verified()
return samples
comfy_samplers = import_module("comfy.samplers")
model_management = import_module("comfy.model_management")
@@ -132,6 +122,8 @@ def sample_with_optional_negative(
disable_pbar=disable_pbar,
seed=seed,
)
if start_verification is not None:
start_verification.require_verified()
LOGGER.debug(
"Positive-only ComfyUI guider selected",
extra={
simple_syrup/runtime/latent_upscale_catalog.py
+3
View File
@@ -1,3 +1,6 @@
# SimpleSyrup - workflow-focused ComfyUI extensions for image generation
# Copyright (C) 2026 Artificial Sweetener and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
"""List checksum-pinned SesquiLSR artifacts in Comfy's standard latent model folder."""
from __future__ import annotations
simple_syrup/runtime/latent_upscale_models.py
+3
View File
@@ -1,3 +1,6 @@
# SimpleSyrup - workflow-focused ComfyUI extensions for image generation
# Copyright (C) 2026 Artificial Sweetener and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Resolve trusted learned checkpoints through registered Comfy model paths."""
from __future__ import annotations
simple_syrup/runtime/mixture_of_diffusers_sampling.py
+1
View File
@@ -159,6 +159,7 @@ def sample_mixture_of_diffusers(
noise=noise,
cfg=cfg,
sampler=sampler,
sampler_name=sampler_name,
sigmas=sigmas,
positive=positive,
negative=negative,
simple_syrup/runtime/multidiffusion_sampling.py
+1 -9
View File
@@ -54,7 +54,6 @@ from .tiled_sampling_validation import (
LOGGER = get_logger(__name__)
SAMPLER_LABEL = "MultiDiffusion"
UNIPC_SAMPLERS = frozenset({"uni_pc", "uni_pc_bh2"})
def sample_multidiffusion(
@@ -92,7 +91,6 @@ def sample_multidiffusion(
latent_tile_height=latent_tile_height,
latent_tile_batch_size=latent_tile_batch_size,
)
_reject_unipc_sampler(sampler_name)
reject_unsupported_conditioning(
positive,
sampler_label=SAMPLER_LABEL,
@@ -161,6 +159,7 @@ def sample_multidiffusion(
noise=noise,
cfg=cfg,
sampler=sampler,
sampler_name=sampler_name,
sigmas=sigmas,
positive=positive,
negative=negative,
@@ -327,13 +326,6 @@ class MultiDiffusionModelWrapper:
return apply_model(args["input"], args["timestep"], **conditioning)
def _reject_unipc_sampler(sampler_name: str) -> None:
"""Reject UniPC samplers because MultiDiffusion is incompatible with them."""
if sampler_name in UNIPC_SAMPLERS:
raise ValueError("MultiDiffusion is not compatible with UniPC samplers.")
def _validate_supplied_plan(
plan: TiledDiffusionPlan,
latent_width: int,
simple_syrup/runtime/native_noise_scaling.py
+7 -3
View File
@@ -1,3 +1,7 @@
# SimpleSyrup - workflow-focused ComfyUI extensions for image generation
# Copyright (C) 2026 Artificial Sweetener and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Separate clean references and residuals with the model's actual parameterization."""
from __future__ import annotations
@@ -13,9 +17,9 @@ def noise_coefficients(
"""Validate affine noise mixing instead of assuming DDPM alpha/sigma units."""
sigma = clean.new_tensor(coordinate)
base = scaling.noise_scaling(sigma, torch.zeros_like(clean), clean, False)
amplitude = (
scaling.noise_scaling(sigma, torch.ones_like(clean), clean, False) - base
)
zero = torch.zeros_like(clean)
amplitude = scaling.noise_scaling(sigma, torch.ones_like(clean), zero, False)
amplitude = amplitude - scaling.noise_scaling(sigma, zero, zero, False)
if not isinstance(base, torch.Tensor) or not isinstance(amplitude, torch.Tensor):
raise TypeError("Native noise scaling must produce tensor coefficients.")
if base.shape != clean.shape or amplitude.shape != clean.shape:
simple_syrup/runtime/native_sampling_target.py
+48
View File
@@ -0,0 +1,48 @@
# SimpleSyrup - workflow-focused ComfyUI extensions for image generation
# Copyright (C) 2026 Artificial Sweetener and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Validate shared model scaling and partial-denoise receiver coordinates."""
from __future__ import annotations
import math
from importlib import import_module
from typing import Any
import torch
MINIMUM_PARTIAL_SIGMA = 0.0001
def validate_partial_sampling_target(
model: Any, sigmas: torch.Tensor, *, operation: str
) -> float:
"""Reject unsupported scaling and singular targets before model execution."""
sampling_types = import_module("comfy.model_sampling")
sampling = model.get_model_object("model_sampling")
if not isinstance(
sampling, (sampling_types.CONST, sampling_types.EPS)
) or isinstance(
sampling, (sampling_types.IMG_TO_IMG, sampling_types.IMG_TO_IMG_FLOW)
):
raise ValueError(f"{operation} requires a flow or EPS-compatible image model.")
if sigmas.ndim != 1 or len(sigmas) < 2 or not bool(torch.isfinite(sigmas).all()):
raise ValueError(f"{operation} requires a finite forward sampling schedule.")
target = float(sigmas[0])
if target <= MINIMUM_PARTIAL_SIGMA:
raise ValueError(
f"{operation} requires a positive partial-denoise start sigma."
)
if isinstance(sampling, sampling_types.CONST) and target >= 0.9999:
raise ValueError(
f"Flow {operation.lower()} requires denoise below the full-noise endpoint."
)
if isinstance(sampling, sampling_types.EPS):
maximum = float(sampling.sigma_max)
if target > maximum or math.isclose(target, maximum, rel_tol=1e-5):
raise ValueError(
f"{operation} requires partial denoise, below the model's "
"maximum sigma."
)
return target
simple_syrup/runtime/noise_inversion.py
+5 -35
View File
@@ -6,7 +6,6 @@
from __future__ import annotations
import math
import time
from collections.abc import Callable
from dataclasses import dataclass
@@ -22,6 +21,7 @@ from ..domain.inversion_solver import (
)
from ..domain.noise_inversion import InversionMethod, NoiseInversionOptions
from ..shared.logging import get_logger
from .native_sampling_target import validate_partial_sampling_target
from .spatial_tensor_projection import resize_spatial_tensor
from .tiled_sampling_validation import validate_tensor_shape
@@ -72,39 +72,6 @@ def _source_in_model_space(model: Any, latent: torch.Tensor) -> torch.Tensor:
return source.detach().float().cpu()
def validate_inversion_target(model: Any, sigmas: torch.Tensor) -> float:
"""Reject unsupported scaling and singular targets before model execution."""
sampling_types = import_module("comfy.model_sampling")
sampling = model.get_model_object("model_sampling")
if not isinstance(
sampling, (sampling_types.CONST, sampling_types.EPS)
) or isinstance(
sampling, (sampling_types.IMG_TO_IMG, sampling_types.IMG_TO_IMG_FLOW)
):
raise ValueError(
"Noise inversion requires a flow or EPS-compatible image model."
)
if sigmas.ndim != 1 or len(sigmas) < 2 or not bool(torch.isfinite(sigmas).all()):
raise ValueError("Noise inversion requires a finite forward sampling schedule.")
target = float(sigmas[0])
if target <= INVERSION_START_SIGMA:
raise ValueError(
"Noise inversion requires a positive partial-denoise start sigma."
)
if isinstance(sampling, sampling_types.CONST) and target >= 0.9999:
raise ValueError(
"Flow noise inversion requires denoise below the full-noise endpoint."
)
if isinstance(sampling, sampling_types.EPS):
maximum = float(sampling.sigma_max)
if target > maximum or math.isclose(target, maximum, rel_tol=1e-5):
raise ValueError(
"Noise inversion requires partial denoise, below the model's "
"maximum sigma."
)
return target
def invert_sampling_noise(
*,
model: Any,
@@ -133,7 +100,9 @@ def invert_sampling_noise(
raise ValueError(
"Noise inversion source must contain finite floating-point values."
)
target = validate_inversion_target(model, forward_sigmas)
target = validate_partial_sampling_target(
model, forward_sigmas, operation="Noise inversion"
)
coarse_target = target * options.coarse_target_fraction
if coarse_target <= INVERSION_START_SIGMA:
raise ValueError(
@@ -202,6 +171,7 @@ def invert_sampling_noise(
noise=torch.zeros_like(stage_latent),
cfg=cfg,
sampler=comfy_samplers.KSAMPLER(invert),
sampler_name=options.method,
sigmas=schedule,
positive=positive,
negative=negative,
simple_syrup/runtime/noise_reuse_finishing.py
+5
View File
@@ -1,3 +1,7 @@
# SimpleSyrup - workflow-focused ComfyUI extensions for image generation
# Copyright (C) 2026 Artificial Sweetener and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Advance an already reconstructed state through full-resolution inversion
finishing.
"""
@@ -82,6 +86,7 @@ def finish_reused_state(
noise=torch.zeros_like(latent),
cfg=cfg,
sampler=import_module("comfy.samplers").KSAMPLER(integrate),
sampler_name=options.finishing_method,
sigmas=schedule,
positive=positive,
negative=negative,
simple_syrup/runtime/noise_reuse_graph.py
+47 -17
View File
@@ -1,3 +1,7 @@
# SimpleSyrup - workflow-focused ComfyUI extensions for image generation
# Copyright (C) 2026 Artificial Sweetener and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Resolve strict latent lineage and inject cache-keyed upstream capture
dependencies.
"""
@@ -15,16 +19,10 @@ from .comfy_graph_provenance import (
parse_graph_link,
trace_vae_decode_provenance,
)
SOURCE_SAMPLERS = frozenset(
{
"KSampler",
"KSamplerAdvanced",
"SimpleSyrup.KSampler",
"SimpleSyrup.KSamplerExtras",
"SimpleSyrup.KSamplerContextualDiffusion",
"SimpleSyrup.KSamplerTiledDiffusion",
}
from .noise_reuse_source import (
capture_graph_context,
describe_graph_node,
resolve_source_sampling_graph,
)
@@ -56,7 +54,7 @@ def source_lineage(
kind, inputs = node.get("class_type"), node.get("inputs")
if not isinstance(inputs, Mapping):
raise ValueError("Noise reuse source node has missing inputs.")
if kind in SOURCE_SAMPLERS:
if resolve_source_sampling_graph(prompt, current[0]) is not None:
return current[0], json.dumps(list(reversed(geometry)))
if kind == "SimpleSyrup.LatentFromImage":
decoded = trace_vae_decode_provenance(prompt, inputs.get("image"), registry)
@@ -93,7 +91,8 @@ def source_lineage(
continue
raise ValueError(
f"Noise reuse cannot resolve latent provenance through "
f"{kind}. Use a direct sampler latent, Simple Upscale "
f"{describe_graph_node(prompt, current[0])}. "
"Use a direct sampler latent, Simple Upscale "
f"Latent, or Latent From Image."
)
raise ValueError("Noise reuse requires a graph-connected source sampler latent.")
@@ -155,7 +154,11 @@ def rewrite_noise_reuse_graph(prompt: dict[str, Any], registry: NodeRegistry) ->
)
except ValueError as error:
if receiver:
failures[node_id] = str(error)
failures[node_id] = (
f"Noise Reuse Options on receiving sampler "
f"{describe_graph_node(prompt, node_id)} cannot reach a usable "
f"source sampler. {error}"
)
for source in sorted({consumer.source for consumer in consumers}):
_inject_source(
prompt,
@@ -183,6 +186,11 @@ def _inject_source(
its result.
"""
source_inputs = prompt[source]["inputs"]
source_graph = resolve_source_sampling_graph(prompt, source)
if source_graph is None:
raise ValueError(
f"Source {describe_graph_node(prompt, source)} is not a sampler."
)
request: list[object] | None = None
for consumer in consumers:
if consumer.receiver is None:
@@ -196,7 +204,7 @@ def _inject_source(
values.update(
options=receiver["options"],
denoise=receiver.get("denoise", 1.0),
source_latent=source_inputs["latent_image"],
source_latent=source_graph.latent_input,
geometry_json=consumer.geometry_json,
)
if request is not None:
@@ -207,17 +215,39 @@ def _inject_source(
}
request = [request_id, 0]
capture_id = _unique_id(prompt, f"{source}.trajectory_capture")
model_inputs = prompt[source_graph.model_owner]["inputs"]
if source_graph.model_owner != source:
guider_id = _unique_id(prompt, f"{source}.trajectory_guider")
guider_node = prompt[source_graph.model_owner].copy()
model_inputs = model_inputs.copy()
guider_node["inputs"] = model_inputs
prompt[guider_id] = guider_node
guider_link = parse_graph_link(source_inputs["guider"])
if guider_link is None:
raise ValueError("Source sampler has an invalid guider connection.")
source_inputs["guider"] = [guider_id, guider_link[1]]
values = {
"model": source_inputs["model"],
"sampler_name": source_inputs["sampler_name"],
"model": model_inputs["model"],
"sampler_name": source_graph.sampler_name,
"graph_context": capture_graph_context(
prompt,
source,
[
consumer.receiver
for consumer in consumers
if consumer.receiver is not None
],
),
}
if source_graph.sampler is not None:
values["sampler"] = source_graph.sampler
if request is not None:
values["requests"] = request
prompt[capture_id] = {
"class_type": "SimpleSyrup.TrajectoryCaptureModel",
"inputs": values,
}
source_inputs["model"] = [capture_id, 0]
model_inputs["model"] = [capture_id, 0]
attach_id = _unique_id(prompt, f"{source}.sampling_provenance")
for node in prompt.values():
inputs = node.get("inputs", {})
simple_syrup/runtime/noise_reuse_prompt_handler.py
+3
View File
@@ -1,3 +1,6 @@
# SimpleSyrup - workflow-focused ComfyUI extensions for image generation
# Copyright (C) 2026 Artificial Sweetener and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Register capture coordination before Comfy validates or executes source sampling."""
from __future__ import annotations
simple_syrup/runtime/noise_reuse_sampling.py
+10 -3
View File
@@ -1,3 +1,7 @@
# SimpleSyrup - workflow-focused ComfyUI extensions for image generation
# Copyright (C) 2026 Artificial Sweetener and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Prepare captured residuals for the shared native sampler starting-noise input."""
from __future__ import annotations
@@ -9,7 +13,8 @@ import torch
from ..domain.spectral_noise import lift_residual
from ..shared.logging import get_logger
from .native_noise_scaling import noise_coefficients, recover_residual
from .noise_inversion import InversionModelFactory, validate_inversion_target
from .native_sampling_target import validate_partial_sampling_target
from .noise_inversion import InversionModelFactory
from .noise_reuse_finishing import finish_reused_state
from .noise_reuse_scope import NoiseReuseExecution
@@ -40,7 +45,7 @@ def prepare_reused_noise(
"batches."
)
execution.consumed = True
target = validate_inversion_target(model, sigmas)
target = validate_partial_sampling_target(model, sigmas, operation="Noise reuse")
coordinate = execution.options.capture_coordinate(target)
provenance = execution.provenance
if type(model.model.latent_format).__name__ != provenance.latent_format:
@@ -48,7 +53,8 @@ def prepare_reused_noise(
"Noise reuse source and receiving model use different latent formats."
)
scaling = model.get_model_object("model_sampling")
residual = provenance.residual_at(coordinate)
capture = provenance.capture_at(coordinate)
residual = capture.residual
lifted = lift_residual(residual, fresh.detach().float().cpu())
clean = model.model.process_latent_in(latent).detach().float().cpu()
if execution.options.finishing_steps:
@@ -75,6 +81,7 @@ def prepare_reused_noise(
extra={
"operation": "noise_reuse",
"capture_sigma": coordinate,
"actual_capture_sigma": capture.actual_coordinate,
"target_sigma": target,
"source_shape": tuple(residual.shape),
"target_shape": tuple(lifted.shape),
simple_syrup/runtime/noise_reuse_scope.py
+3
View File
@@ -1,3 +1,6 @@
# SimpleSyrup - workflow-focused ComfyUI extensions for image generation
# Copyright (C) 2026 Artificial Sweetener and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Own one request-scoped noise handoff across existing sampler routing."""
from __future__ import annotations
simple_syrup/runtime/noise_reuse_source.py
+100
View File
@@ -0,0 +1,100 @@
# SimpleSyrup - workflow-focused ComfyUI extensions for image generation
# Copyright (C) 2026 Artificial Sweetener and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Adapt source sampling graphs through their shared model and latent contracts."""
from __future__ import annotations
from collections.abc import Mapping
from dataclasses import dataclass
from typing import Any
from .comfy_graph_provenance import parse_graph_link
@dataclass(frozen=True, slots=True)
class SourceSamplingGraph:
"""Describe the exact graph dependencies needed to observe one source sampler."""
model_owner: str
latent_input: object
sampler_name: object
sampler: object | None = None
def resolve_source_sampling_graph(
prompt: Mapping[str, Any], node_id: str
) -> SourceSamplingGraph | None:
"""Recognize named and custom samplers by their host-facing sampling inputs."""
node = prompt.get(node_id)
if not isinstance(node, Mapping):
return None
inputs = node.get("inputs")
if not isinstance(inputs, Mapping) or "latent_image" not in inputs:
return None
if "model" in inputs and "sampler_name" in inputs:
return SourceSamplingGraph(
node_id, inputs["latent_image"], inputs["sampler_name"]
)
if "sampler" not in inputs or "sigmas" not in inputs:
return None
owner = node_id
if "model" not in inputs:
guider = parse_graph_link(inputs.get("guider"))
producer = prompt.get(guider[0]) if guider is not None else None
if not isinstance(producer, Mapping):
return None
assert guider is not None
guider_inputs = producer.get("inputs")
if not isinstance(guider_inputs, Mapping) or "model" not in guider_inputs:
raise ValueError(
"The source custom sampler's guider must expose its connected model "
"so Noise Reuse Options can observe that source run."
)
owner = guider[0]
link = parse_graph_link(inputs["sampler"])
selector = prompt.get(link[0]) if link is not None else None
selector_inputs = (
selector.get("inputs", {}) if isinstance(selector, Mapping) else {}
)
name = (
selector_inputs.get("sampler_name", "")
if isinstance(selector_inputs, Mapping)
else ""
)
return SourceSamplingGraph(owner, inputs["latent_image"], name, inputs["sampler"])
def describe_graph_node(prompt: Mapping[str, Any], node_id: str) -> str:
"""Name user titles, node types and graph IDs together in actionable errors."""
node = prompt.get(node_id, {})
if not isinstance(node, Mapping):
return f"missing node {node_id}"
kind = str(node.get("class_type", "unknown node"))
metadata = node.get("_meta", {})
title = metadata.get("title") if isinstance(metadata, Mapping) else None
return (
f"'{title}' ({kind}, node {node_id})"
if isinstance(title, str) and title
else f"{kind} (node {node_id})"
)
def capture_graph_context(
prompt: Mapping[str, Any], source: str, receivers: list[str]
) -> str:
"""Explain which downstream option reaches back into the source graph."""
if not receivers:
return (
"Latent format discovery is observing source "
f"{describe_graph_node(prompt, source)}."
)
downstream = "; ".join(
describe_graph_node(prompt, node_id) for node_id in receivers
)
return (
f"Noise Reuse Options on receiving sampler {downstream} reaches back into "
f"source sampler {describe_graph_node(prompt, source)} "
"to save its original noise."
)
simple_syrup/runtime/noise_reuse_start.py
+94
View File
@@ -0,0 +1,94 @@
# SimpleSyrup - workflow-focused ComfyUI extensions for image generation
# Copyright (C) 2026 Artificial Sweetener and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Verify reused starting states at native scaling before any sampler algorithm runs."""
from __future__ import annotations
from collections.abc import Callable
from typing import Any
import torch
from ..shared.logging import get_logger
from .model_patcher_mutations import ModelCallableObjectPatchMutation
from .patcher_lifecycle import PATCHER_LIFECYCLE
LOGGER = get_logger(__name__)
class NoiseReuseStartVerification:
"""Check the first native scaling result without changing callbacks or tensors."""
def __init__(
self,
original: Callable[..., torch.Tensor],
expected: torch.Tensor,
*,
graph_context: str,
sampler_name: str,
) -> None:
"""Retain one receiver's expected state and actionable graph identity."""
self.original = original
self.expected = expected
self.graph_context = graph_context
self.sampler_name = sampler_name
self.verified = False
def __call__(
self,
sigma: torch.Tensor,
noise: torch.Tensor,
latent_image: torch.Tensor,
max_denoise: bool = False,
) -> torch.Tensor:
"""Return native scaling unchanged after validating its first result."""
result = self.original(sigma, noise, latent_image, max_denoise)
if not self.verified:
if result.shape != self.expected.shape:
raise ValueError(self._message("produced a different latent shape"))
error = float((result.detach().float().cpu() - self.expected).abs().max())
if not bool(torch.isfinite(result).all()) or error > 2e-5:
raise ValueError(
self._message(
f"changed the reused starting latent (error {error:.6g})"
)
)
self.verified = True
LOGGER.info(
"Noise reuse starting state verified",
extra={
"operation": "noise_reuse_start",
"sampler_name": self.sampler_name,
"graph_context": self.graph_context,
"max_error": error,
},
)
return result
def require_verified(self) -> None:
"""Reject receiver execution paths that bypass native starting-noise scaling."""
if not self.verified:
raise ValueError(
self._message(
"is not supported by noise reuse on this execution path: "
"it did not expose native starting-noise scaling"
)
)
def _message(self, reason: str) -> str:
"""Identify the graph traversal and selected receiving algorithm plainly."""
return (
f"{self.graph_context} The receiving sampler '{self.sampler_name}' "
f"{reason}. Choose a sampler node that uses "
"ComfyUI's native sampling boundary."
)
def derive_model(self, model: Any) -> Any:
"""Install verification through the authoritative MODEL lifecycle."""
return PATCHER_LIFECYCLE.derive_model(
model,
(ModelCallableObjectPatchMutation("model_sampling.noise_scaling", self),),
operation="noise_reuse_start",
)
simple_syrup/runtime/regional_multidiffusion_sampling.py
+1 -11
View File
@@ -44,7 +44,6 @@ from .tiled_sampling_validation import (
LOGGER = get_logger(__name__)
SAMPLER_LABEL = "Regional MultiDiffusion"
UNIPC_SAMPLERS = frozenset({"uni_pc", "uni_pc_bh2"})
@dataclass(frozen=True)
@@ -84,7 +83,6 @@ def sample_regional_multidiffusion(
denoise=denoise,
global_prompt_weight=global_prompt_weight,
)
_reject_unipc_sampler(sampler_name)
if not regions:
raise ValueError("Regional MultiDiffusion requires at least one region.")
reject_unsupported_conditioning(
@@ -155,6 +153,7 @@ def sample_regional_multidiffusion(
noise=noise,
cfg=cfg,
sampler=sampler,
sampler_name=sampler_name,
sigmas=sigmas,
positive=positive,
negative=negative,
@@ -283,15 +282,6 @@ def _validate_global_prompt_weight(global_prompt_weight: float) -> None:
raise ValueError("global_prompt_weight must be between 0.0 and 1.0.")
def _reject_unipc_sampler(sampler_name: str) -> None:
"""Reject UniPC samplers because MultiDiffusion is incompatible with them."""
if sampler_name in UNIPC_SAMPLERS:
raise ValueError(
"Regional MultiDiffusion is not compatible with UniPC samplers."
)
def _sampling_callback(
model: Any,
steps: int,
simple_syrup/runtime/sesqui_latent_upscale.py
+3
View File
@@ -1,3 +1,6 @@
# SimpleSyrup - workflow-focused ComfyUI extensions for image generation
# Copyright (C) 2026 Artificial Sweetener and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Adapt SesquiLSR inference to Comfy external latent layouts and device management."""
from __future__ import annotations
simple_syrup/runtime/trajectory_capture.py
+160 -100
View File
@@ -1,68 +1,56 @@
"""Observe one upstream sampler through Comfy's per-model sampler wrapper."""
# SimpleSyrup - workflow-focused ComfyUI extensions for image generation
# Copyright (C) 2026 Artificial Sweetener and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Observe native source sampling through solver-independent Comfy boundaries."""
from __future__ import annotations
from copy import copy
import math
from dataclasses import dataclass, field
from importlib import import_module
from typing import Any
import torch
from ..domain.noise_reuse import PROVENANCE_KEY, SamplingProvenance
from ..domain.noise_reuse import (
PROVENANCE_KEY,
CapturedSamplingState,
SamplingProvenance,
)
from ..shared.logging import get_logger
from .model_patcher_mutations import ModelKeyedWrapperMutation
from .native_noise_scaling import recover_residual
from .patcher_lifecycle import PATCHER_LIFECYCLE
LOGGER = get_logger(__name__)
SUPPORTED_CAPTURE_SAMPLERS = frozenset({"euler", "euler_ancestral"})
def insert_capture_coordinates(
sigmas: torch.Tensor, coordinates: tuple[float, ...]
) -> torch.Tensor:
"""Insert exact float32 boundaries without replacing native scheduler points."""
if sigmas.ndim != 1 or len(sigmas) < 2 or not bool(torch.isfinite(sigmas).all()):
raise ValueError(
"Source capture requires a finite descending sampling schedule."
)
if not bool((sigmas[:-1] > sigmas[1:]).all()):
raise ValueError(
"Source capture requires strictly descending sigma boundaries."
)
if float(sigmas[-1]) != 0:
raise ValueError(
"Source capture requires a schedule ending at zero noise; "
"disable leftover noise on the originating sampler."
)
result = sigmas
for value in sorted(set(coordinates), reverse=True):
coordinate = sigmas.new_tensor(value)
if not 0 < float(coordinate) < float(sigmas[0]):
raise ValueError(
f"Requested source sigma {value:.9g} lies outside the "
f"source sampling schedule."
)
if not bool((result == coordinate).any()):
index = int((result > coordinate).sum())
result = torch.cat((result[:index], coordinate.reshape(1), result[index:]))
return result
@dataclass(frozen=True, slots=True)
class ObservedSamplingState:
"""Retain one real model-space observation without manufacturing a timestep."""
coordinate: float
state: torch.Tensor
@dataclass(slots=True)
class TrajectoryCapture:
"""Keep capture storage local to an injected, cache-keyed MODEL derivation."""
"""Own bounded source observations without changing its solver or schedule."""
model: Any
sampler_name: str
coordinates: tuple[float, ...]
states: dict[float, torch.Tensor] = field(default_factory=dict)
graph_context: str = ""
states: dict[float, ObservedSamplingState] = field(default_factory=dict)
clean: torch.Tensor | None = None
schedule: torch.Tensor | None = None
_active: bool = False
def __post_init__(self) -> None:
"""Store the exact float32 coordinates used by native sampler schedules."""
"""Normalize requests to the receiving native float32 coordinate precision."""
if any(not math.isfinite(value) or value <= 0 for value in self.coordinates):
raise ValueError("Requested noise levels must be finite and above zero.")
self.coordinates = tuple(
sorted(
{float(torch.tensor(value)) for value in self.coordinates}, reverse=True
@@ -70,12 +58,14 @@ class TrajectoryCapture:
)
def prepare_model(self) -> Any:
"""Clone only this source MODEL and leave unrelated samplers untouched."""
if self.coordinates and self.sampler_name not in SUPPORTED_CAPTURE_SAMPLERS:
raise ValueError(
"Noise reuse source capture supports Euler and Euler "
"ancestral; select a supported source sampler."
)
"""Install one observer for every solver using Comfy's sampling contract."""
try:
return self._prepare_model()
except (TypeError, ValueError) as error:
raise type(error)(self._failure_message(error)) from error
def _prepare_model(self) -> Any:
"""Derive the capture model through the authoritative patcher lifecycle."""
if not self.coordinates:
return self.model
wrappers = import_module("comfy.patcher_extension").WrappersMP
@@ -87,6 +77,11 @@ class TrajectoryCapture:
"simple_syrup.trajectory_capture",
self.observe,
),
ModelKeyedWrapperMutation(
wrappers.PREDICT_NOISE,
"simple_syrup.trajectory_state",
self.observe_prediction,
),
),
operation="SimpleSyrup trajectory capture",
)
@@ -94,54 +89,117 @@ class TrajectoryCapture:
def observe(
self, executor: Any, *arguments: Any, **keyword_arguments: Any
) -> torch.Tensor:
"""Wrap the individual native solver instance and record its pre-update x."""
local = copy(executor.class_obj)
solver = local.sampler_function
"""Scope observations to one unchanged native sampler execution and endpoint."""
try:
return self._observe(executor, *arguments, **keyword_arguments)
except (TypeError, ValueError) as error:
raise type(error)(self._failure_message(error)) from error
def _observe(
self, executor: Any, *arguments: Any, **keyword_arguments: Any
) -> torch.Tensor:
"""Retain actual states and release observation scope on every exit."""
schedule = (
arguments[1] if len(arguments) > 1 else keyword_arguments.get("sigmas")
)
if not isinstance(schedule, torch.Tensor) or schedule.ndim != 1:
raise ValueError("The original sampler did not provide its noise schedule.")
if not bool(torch.isfinite(schedule).all()) or bool((schedule < 0).any()):
raise ValueError("The original sampler provided an invalid noise schedule.")
self.states.clear()
self.clean = None
def recorded(
model: Any, state: torch.Tensor, sigmas: torch.Tensor, **kwargs: Any
) -> torch.Tensor:
"""Use native solver randomness and preserve its original callback."""
schedule = insert_capture_coordinates(sigmas, self.coordinates)
self.schedule = schedule.detach().float().cpu().clone()
original_callback = kwargs.get("callback")
def observe_state(record: dict[str, Any]) -> None:
"""Capture actual noisy state, rejecting churned or ambiguous
coordinates.
"""
coordinate = float(schedule[int(record["i"])])
if coordinate in self.coordinates:
if float(record.get("sigma_hat", record["sigma"])) != coordinate:
raise ValueError(
"Noise reuse cannot capture churned source states."
)
self.states[coordinate] = record["x"].detach().float().cpu().clone()
if original_callback is not None:
original_callback(record)
kwargs["callback"] = observe_state
result: torch.Tensor = solver(model, state, schedule, **kwargs)
self.schedule = schedule.detach().float().cpu().clone()
self._active = True
try:
result = executor(*arguments, **keyword_arguments)
if not isinstance(result, torch.Tensor):
raise TypeError("The original sampler did not return a latent tensor.")
self.clean = result.detach().float().cpu().clone()
return result
finally:
self._active = False
local.sampler_function = recorded
isolated = copy(executor)
isolated.original = local.sample
result: torch.Tensor = isolated(*arguments, **keyword_arguments)
def observe_prediction(
self, executor: Any, *arguments: Any, **keyword_arguments: Any
) -> torch.Tensor:
"""Save real input states at actual solver coordinates, including substeps."""
if self._active:
state = arguments[0] if arguments else keyword_arguments.get("x")
sigma = (
arguments[1]
if len(arguments) > 1
else keyword_arguments.get("timestep")
)
if not isinstance(state, torch.Tensor) or not isinstance(
sigma, torch.Tensor
):
raise TypeError("The original sampler did not expose its noisy latent.")
values = sigma.detach().float().flatten()
if values.numel() == 0 or not bool(torch.isfinite(values).all()):
raise ValueError(
"The original sampler reported an invalid noise level."
)
if not bool((values == values[0]).all()):
raise ValueError(
"Noise reuse requires one source noise level per batch."
)
coordinate = float(values[0])
if coordinate > 0:
self._record(state, coordinate)
result = executor(*arguments, **keyword_arguments)
if not isinstance(result, torch.Tensor):
raise TypeError(
"The original sampler's model returned an invalid prediction."
)
return result
def _record(self, state: torch.Tensor, coordinate: float) -> None:
"""Prefer exact matches, then the closest real level with deterministic ties."""
selected: torch.Tensor | None = None
for requested in self.coordinates:
previous = self.states.get(requested)
rank = (abs(coordinate - requested), -coordinate)
if previous is not None and rank >= (
abs(previous.coordinate - requested),
-previous.coordinate,
):
continue
if selected is None:
selected = state.detach().float().cpu().clone()
if not bool(torch.isfinite(selected).all()):
raise ValueError(
"The original sampler produced an invalid noisy latent."
)
self.states[requested] = ObservedSamplingState(coordinate, selected)
def attach(self, latent: dict[str, Any]) -> dict[str, Any]:
"""Associate captured states with their matching external clean sampler
output.
"""
"""Surface capture failures with the original graph identities and algorithm."""
try:
return self._attach(latent)
except (TypeError, ValueError) as error:
raise type(error)(self._failure_message(error)) from error
def _failure_message(self, error: Exception) -> str:
"""Identify the graph reach-through and selected algorithm in plain English."""
LOGGER.error(
"Source noise handoff failed",
extra={
"sampler": self.sampler_name,
"graph_context": self.graph_context,
"operation": "trajectory_capture",
"reason": str(error),
},
)
return (
f"{self.graph_context} Chosen source sampler: '{self.sampler_name}'. "
f"{error}"
).strip()
def _attach(self, latent: dict[str, Any]) -> dict[str, Any]:
"""Associate observations with their source output and parameterization."""
samples = latent.get("samples")
if not isinstance(samples, torch.Tensor):
raise TypeError(
"Source sampler output must contain a latent samples tensor."
)
raise TypeError("The original sampler output must contain a latent tensor.")
actual = self.model.model.process_latent_in(samples).detach().float().cpu()
if self.coordinates:
if (
@@ -150,42 +208,44 @@ class TrajectoryCapture:
or set(self.states) != set(self.coordinates)
):
raise ValueError(
"Source trajectory capture is incomplete; run the full "
"connected workflow again."
f"The original sampler '{self.sampler_name}' "
"is not supported by noise reuse on this execution path: "
"it finished without exposing the original noisy latent. "
"Run the complete connected workflow again; if this repeats, "
"that source node must expose ComfyUI's sampling boundaries."
)
if not torch.allclose(actual, self.clean, atol=1e-5, rtol=1e-5):
raise ValueError(
"Captured trajectory endpoint does not match the source "
"sampler output."
"Saved noise belongs to a different original sampler output."
)
actual = self.clean
scaling = (
self.model.get_model_object("model_sampling") if self.coordinates else None
)
captures = tuple(
CapturedSamplingState(
requested,
observed.coordinate,
observed.state,
recover_residual(scaling, actual, observed.state, observed.coordinate),
)
for requested, observed in self.states.items()
)
result = latent.copy()
result[PROVENANCE_KEY] = SamplingProvenance(
type(self.model.model.latent_format).__name__,
actual,
tuple(
(coordinate, self.states[coordinate]) for coordinate in self.coordinates
),
captures,
self.schedule,
tuple(
(
coordinate,
recover_residual(
self.model.get_model_object("model_sampling"),
actual,
self.states[coordinate],
coordinate,
),
)
for coordinate in self.coordinates
),
self.graph_context,
)
LOGGER.info(
"Sampling provenance attached",
extra={
"operation": "trajectory_capture",
"sampler": self.sampler_name,
"coordinates": self.coordinates,
"requested_coordinates": self.coordinates,
"actual_coordinates": tuple(c.actual_coordinate for c in captures),
"source_shape": tuple(actual.shape),
},
)
simple_syrup/services/ksampler_sampling_service.py
+4
View File
@@ -89,6 +89,7 @@ class KSamplerSamplingService:
noise=noise,
cfg=cfg,
sampler=sampler,
sampler_name=sampler_name,
sigmas=sigmas,
positive=positive,
negative=negative,
@@ -106,6 +107,7 @@ class KSamplerSamplingService:
noise=noise,
cfg=cfg,
sampler=sampler,
sampler_name=sampler_name,
sigmas=sigmas,
positive=positive,
negative=negative,
@@ -152,6 +154,7 @@ class KSamplerSamplingService:
noise: torch.Tensor,
cfg: float,
sampler: Any,
sampler_name: str,
sigmas: torch.Tensor,
positive: Any,
negative: Any,
@@ -173,6 +176,7 @@ class KSamplerSamplingService:
noise=noise[index : index + 1],
cfg=cfg,
sampler=sampler,
sampler_name=sampler_name,
sigmas=sigmas,
positive=select_conditioning(positive, index),
negative=select_conditioning(negative, index),
simple_syrup/services/latent_provenance_service.py
+3
View File
@@ -1,3 +1,6 @@
# SimpleSyrup - workflow-focused ComfyUI extensions for image generation
# Copyright (C) 2026 Artificial Sweetener and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Recover original latents from image graph provenance without pixel encoding."""
from __future__ import annotations
simple_syrup/services/latent_upscale_service.py
+3
View File
@@ -1,3 +1,6 @@
# SimpleSyrup - workflow-focused ComfyUI extensions for image generation
# Copyright (C) 2026 Artificial Sweetener and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Own latent upscale method selection, provenance preservation and mask geometry."""
from __future__ import annotations
simple_syrup/services/noise_capture_request_service.py
+8 -2
View File
@@ -1,3 +1,7 @@
# SimpleSyrup - workflow-focused ComfyUI extensions for image generation
# Copyright (C) 2026 Artificial Sweetener and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Resolve upstream capture requests from the receiving sampler's live settings."""
from __future__ import annotations
@@ -9,7 +13,7 @@ import torch
from ..domain.sampler_options import SamplerOptions
from ..runtime.comfy_latent_normalization import COMFY_LATENT_NORMALIZER
from ..runtime.noise_inversion import validate_inversion_target
from ..runtime.native_sampling_target import validate_partial_sampling_target
from ..runtime.sampling_schedulers import SchedulerView, calculate_sigmas
@@ -54,7 +58,9 @@ class NoiseCaptureRequestService:
sigmas = calculate_sigmas(
model, scheduler, sampler_name, steps, denoise, view=view
)
target = validate_inversion_target(model, sigmas)
target = validate_partial_sampling_target(
model, sigmas, operation="Noise reuse"
)
coordinate = options.noise_reuse.capture_coordinate(target)
return tuple(sorted({*previous, coordinate}, reverse=True))
simple_syrup/services/sampler_options_sampling_service.py
+4 -9
View File
@@ -17,7 +17,7 @@ from ..domain.attention_coupling_request import (
from ..domain.noise_reuse import PROVENANCE_KEY, SamplingProvenance
from ..domain.sampler_options import SamplerOptions, TilingOptions
from ..runtime import sampling_schedulers
from ..runtime.noise_inversion import validate_inversion_target
from ..runtime.native_sampling_target import validate_partial_sampling_target
from ..runtime.noise_reuse_scope import NoiseReuseExecution, noise_reuse_scope
from ..runtime.tiled_sampling_validation import validate_sampling_controls
from ..shared.logging import get_logger
@@ -265,13 +265,6 @@ class SamplerOptionsSamplingService:
latent_tile_height=tiling.height if tiling is not None else 16,
latent_tile_batch_size=tiling.batch_size if tiling is not None else 1,
)
incompatible_unipc = options.contextual_diffusion is not None or (
tiling is not None and tiling.diffusion_mode == "multidiffusion"
)
if incompatible_unipc and arguments["sampler_name"] in {"uni_pc", "uni_pc_bh2"}:
raise ValueError(
"Tiling and Contextual Diffusion do not support UniPC samplers."
)
if options.noise_inversion is not None:
view = (
sampling_schedulers.SchedulerView(tiling.width, tiling.height)
@@ -286,4 +279,6 @@ class SamplerOptionsSamplingService:
denoise=arguments["denoise"],
view=view,
)
validate_inversion_target(arguments["model"], sigmas)
validate_partial_sampling_target(
arguments["model"], sigmas, operation="Noise inversion"
)
tests/comfy_integration/test_noise_reuse_graph.py
+68
View File
@@ -1,3 +1,7 @@
# SimpleSyrup - workflow-focused ComfyUI extensions for image generation
# Copyright (C) 2026 Artificial Sweetener and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Prove receiver planning, strict image provenance and source cache dependencies."""
from copy import deepcopy
@@ -163,3 +167,67 @@ def test_receiving_settings_are_live_source_cache_dependencies() -> None:
assert request["options"] == ["reuse", 0]
capture = prompt["source.trajectory_capture"]["inputs"]
assert capture["requests"] == ["receive.noise_capture_request", 0]
@pytest.mark.parametrize(
"kind",
[
"SimpleSyrup.KSamplerAttentionCoupling",
"SimpleSyrup.KSamplerAttentionCouplingContextual",
"SimpleSyrup.KSamplerAttentionCouplingTiled",
"SimpleSyrup.KSamplerPromptByRegion",
"SimpleSyrup.KSamplerPromptByTiledRegion",
"ExternalCompatibleSampler",
],
)
def test_sampling_contract_discovers_regional_and_external_sources(kind: str) -> None:
"""Capture new sampler nodes naturally when they expose the native contract."""
prompt = graph()
prompt["source"]["class_type"] = kind
prompt["source"]["inputs"]["sampler_name"] = "er_sde"
prompt["source"]["_meta"] = {"title": "Original pajamas"}
prompt["receive"]["_meta"] = {"title": "Upscale pajamas"}
rewrite_noise_reuse_graph(prompt, {})
capture = prompt["source.trajectory_capture"]["inputs"]
assert capture["sampler_name"] == "er_sde"
assert "Original pajamas" in capture["graph_context"]
assert "Upscale pajamas" in capture["graph_context"]
assert kind in capture["graph_context"]
@pytest.mark.parametrize("advanced", [False, True])
def test_custom_sampling_capture_preserves_sampler_and_guider_consumers(
advanced: bool,
) -> None:
"""Clone only the source guider branch while observing any connected solver."""
prompt = graph()
source = prompt["source"]
source["class_type"] = "SamplerCustomAdvanced" if advanced else "SamplerCustom"
source["inputs"].pop("sampler_name")
source["inputs"].update(sampler=["select", 0], sigmas=["sigmas", 0])
prompt["select"] = {
"class_type": "KSamplerSelect",
"inputs": {"sampler_name": "dpmpp_2m"},
}
prompt["guidance"] = {
"class_type": "CFGGuider",
"inputs": {"model": ["loader", 0], "cfg": 4.0},
}
original_guider = deepcopy(prompt["guidance"])
if advanced:
source["inputs"].pop("model")
source["inputs"]["guider"] = ["guidance", 0]
rewrite_noise_reuse_graph(prompt, {})
assert prompt["guidance"] == original_guider
assert source["inputs"]["sampler"] == ["select", 0]
capture = prompt["source.trajectory_capture"]["inputs"]
assert capture["sampler_name"] == "dpmpp_2m"
assert capture["sampler"] == ["select", 0]
if advanced:
assert source["inputs"]["guider"] == ["source.trajectory_guider", 0]
assert prompt["source.trajectory_guider"]["inputs"]["model"] == [
"source.trajectory_capture",
0,
]
else:
assert source["inputs"]["model"] == ["source.trajectory_capture", 0]
tests/media/test_latent_upscale.py
+3
View File
@@ -1,3 +1,6 @@
# SimpleSyrup - workflow-focused ComfyUI extensions for image generation
# Copyright (C) 2026 Artificial Sweetener and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Verify native geometry, capture preservation and learned format selection."""
from importlib import import_module
tests/media/test_sesqui_latent_adapter.py
+3
View File
@@ -1,3 +1,6 @@
# SimpleSyrup - workflow-focused ComfyUI extensions for image generation
# Copyright (C) 2026 Artificial Sweetener and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Exercise actual learned inference in Comfy's external latent units and layouts."""
from importlib import import_module
tests/models/patching/test_patcher_lifecycle_policy.py
+1 -1
View File
@@ -35,7 +35,7 @@ APPROVED_VALUE_CLONES = Counter(
("simple_syrup/domain/spectral_noise.py", "spatial_dct(fresh)"): 1,
(
"simple_syrup/runtime/trajectory_capture.py",
"record['x'].detach().float().cpu()",
"state.detach().float().cpu()",
): 1,
(
"simple_syrup/runtime/trajectory_capture.py",
tests/node_api/test_latent_reuse_nodes.py
+3
View File
@@ -1,3 +1,6 @@
# SimpleSyrup - workflow-focused ComfyUI extensions for image generation
# Copyright (C) 2026 Artificial Sweetener and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Verify the public latent reuse node contracts and v3 registration."""
from typing import Any
tests/regional_generation/regional/test_regional_multidiffusion_sampling_runtime.py
+6 -5
View File
@@ -74,19 +74,20 @@ class FakeSampler:
return None
def test_sample_rejects_unipc_before_sampler_resolution(
def test_unipc_reaches_the_native_sampler_resolution_boundary(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""UniPC sampler names fail before ComfyUI sampler lookup."""
"""Admit UniPC through the same native boundary as other offered solvers."""
def fail_resolve_sampler(_sampler_name: str) -> FakeSampler:
"""Fail if sampler resolution is reached."""
"""Mark native resolution without evaluating the external host."""
raise AssertionError("UniPC rejection should happen before sampler resolution")
assert _sampler_name == "uni_pc"
raise ValueError("native resolver reached")
monkeypatch.setattr(sampling_samplers, "resolve_sampler", fail_resolve_sampler)
with pytest.raises(ValueError, match="not compatible with UniPC"):
with pytest.raises(ValueError, match="native resolver reached"):
regional_multidiffusion_sampling.sample_regional_multidiffusion(
model=FakeModel(),
seed=1,
tests/sampling/test_guided_noise_inversion.py
+2
View File
@@ -135,6 +135,7 @@ def test_inversion_and_forward_sampling_keep_the_same_guidance_branches(
noise=random_noise,
cfg=7.5,
sampler=object(),
sampler_name="euler",
sigmas=torch.tensor([0.5, 0]),
positive=positive,
negative=negative,
@@ -175,6 +176,7 @@ def test_absent_inversion_preserves_supplied_noise_without_extra_execution(
noise=noise,
cfg=7.5,
sampler=object(),
sampler_name="euler",
sigmas=torch.tensor([0.5, 0]),
positive=[],
negative=negative,
tests/sampling/test_guided_sampling.py
+1
View File
@@ -79,6 +79,7 @@ def _arguments(negative: object | None) -> dict[str, Any]:
"noise": torch.zeros((1, 4, 2, 2)),
"cfg": 7.5,
"sampler": object(),
"sampler_name": "euler",
"sigmas": torch.tensor([1.0, 0.0]),
"positive": [[torch.ones((1, 1, 1)), {}]],
"negative": negative,
tests/sampling/test_multidiffusion_sampling.py
+6 -5
View File
@@ -538,19 +538,20 @@ def test_model_wrapper_rejects_unsupported_conditioning() -> None:
)
def test_sample_rejects_unipc_before_sampler_resolution(
def test_unipc_reaches_the_native_sampler_resolution_boundary(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""UniPC sampler names fail before ComfyUI sampler lookup."""
"""Admit UniPC through the same native boundary as other offered solvers."""
def fail_resolve_sampler(_sampler_name: str) -> FakeSampler:
"""Fail if sampler resolution is reached."""
"""Mark the native resolver boundary without evaluating a toy model."""
raise AssertionError("UniPC rejection should happen before sampler resolution")
assert _sampler_name == "uni_pc"
raise ValueError("native resolver reached")
monkeypatch.setattr(sampling_samplers, "resolve_sampler", fail_resolve_sampler)
with pytest.raises(ValueError, match="not compatible with UniPC"):
with pytest.raises(ValueError, match="native resolver reached"):
multidiffusion_sampling.sample_multidiffusion(
model=FakeModel(),
seed=1,
tests/sampling/test_noise_capture_request.py
+10 -2
View File
@@ -1,3 +1,7 @@
# SimpleSyrup - workflow-focused ComfyUI extensions for image generation
# Copyright (C) 2026 Artificial Sweetener and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Verify receiving prediction views and model-aware schedule geometry."""
from typing import Any, TypedDict
@@ -40,7 +44,9 @@ def test_cd_coordinate_ignores_unresolved_scale_and_coalesces_finish_requests(
monkeypatch.setattr(requests, "calculate_sigmas", schedule)
monkeypatch.setattr(
requests, "validate_inversion_target", lambda model, sigmas: float(sigmas[0])
requests,
"validate_partial_sampling_target",
lambda model, sigmas, **kwargs: float(sigmas[0]),
)
arguments: CaptureArguments = {
"model": object(),
@@ -94,7 +100,9 @@ def test_flux2_planning_normalizes_empty_layout_before_scale(
monkeypatch.setattr(COMFY_LATENT_NORMALIZER, "normalize", normalize)
monkeypatch.setattr(requests, "calculate_sigmas", schedule)
monkeypatch.setattr(
requests, "validate_inversion_target", lambda model, sigmas: float(sigmas[0])
requests,
"validate_partial_sampling_target",
lambda model, sigmas, **kwargs: float(sigmas[0]),
)
result = requests.NoiseCaptureRequestService().coordinates(
model=object(),
tests/sampling/test_noise_reuse.py
+29 -4
View File
@@ -1,10 +1,18 @@
# SimpleSyrup - workflow-focused ComfyUI extensions for image generation
# Copyright (C) 2026 Artificial Sweetener and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Verify captured coefficient identity, coordinate selection and option isolation."""
import pytest
import torch
from simple_syrup.domain.noise_inversion import NoiseInversionOptions
from simple_syrup.domain.noise_reuse import NoiseReuseOptions, SamplingProvenance
from simple_syrup.domain.noise_reuse import (
CapturedSamplingState,
NoiseReuseOptions,
SamplingProvenance,
)
from simple_syrup.domain.sampler_options import (
SamplerOptions,
append_sampler_capability,
@@ -64,9 +72,26 @@ def test_options_branch_without_mutating_prefix_and_reject_full_inversion() -> N
def test_missing_capture_does_not_fall_back_to_fresh_noise() -> None:
"""Require the requested actual coordinate instead of the closest checkpoint."""
"""Require a saved request while retaining its independently observed coordinate."""
state = torch.zeros(1, 4, 2, 2)
provenance = SamplingProvenance("SDXL", state, ((0.5, state),))
provenance = SamplingProvenance(
"SDXL", state, (CapturedSamplingState(0.5, 0.5, state, state),)
)
assert provenance.state_at(0.5) is state
with pytest.raises(ValueError, match="did not capture"):
with pytest.raises(ValueError, match="did not save"):
provenance.state_at(0.4)
def test_noise_amplitude_survives_large_clean_reference() -> None:
"""Avoid subtracting near-identical large values to measure noise amplitude."""
from types import SimpleNamespace
from simple_syrup.runtime.native_noise_scaling import noise_coefficients
clean = torch.full((1, 4, 2, 2), 1e10)
scaling = SimpleNamespace(
noise_scaling=lambda sigma, noise, reference, maximum: reference + sigma * noise
)
base, amplitude = noise_coefficients(scaling, clean, 0.5)
assert torch.equal(base, clean)
assert torch.equal(amplitude, torch.full_like(clean, 0.5))
tests/sampling/test_noise_reuse_execution.py
+3
View File
@@ -1,3 +1,6 @@
# SimpleSyrup - workflow-focused ComfyUI extensions for image generation
# Copyright (C) 2026 Artificial Sweetener and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Verify native residual parameterization and independent full-resolution finishing."""
from types import SimpleNamespace
tests/sampling/test_noise_reuse_start.py
+64
View File
@@ -0,0 +1,64 @@
# SimpleSyrup - workflow-focused ComfyUI extensions for image generation
# Copyright (C) 2026 Artificial Sweetener and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Safeguard native starting-state verification independently of sampler callbacks."""
import pytest
import torch
from simple_syrup.runtime.noise_reuse_start import NoiseReuseStartVerification
def _native(
sigma: torch.Tensor,
noise: torch.Tensor,
latent: torch.Tensor,
maximum: bool,
) -> torch.Tensor:
"""Expose a native flow scaling boundary without sampler-specific machinery."""
del maximum
return sigma * noise + (1 - sigma) * latent
def _verification(expected: torch.Tensor) -> NoiseReuseStartVerification:
"""Attach recognizable workflow and selected-algorithm identity."""
return NoiseReuseStartVerification(
_native,
expected,
graph_context="Noise Reuse Options node 'Upscale' (ID 8) reaches Source (2).",
sampler_name="uni_pc",
)
def test_native_result_is_unchanged_and_only_initial_scaling_is_checked() -> None:
"""Permit algorithms to scale later states without rechecking the initial state."""
latent, noise, sigma = (
torch.ones(1, 4, 2, 2),
torch.full((1, 4, 2, 2), 3.0),
torch.tensor(0.4),
)
expected = _native(sigma, noise, latent, False)
verification = _verification(expected)
assert torch.equal(verification(sigma, noise, latent), expected)
verification.require_verified()
assert torch.equal(
verification(sigma, noise + 1, latent), _native(sigma, noise + 1, latent, False)
)
@pytest.mark.parametrize("failure", ("unused", "changed", "shape", "nonfinite"))
def test_failures_name_graph_node_and_selected_receiver(failure: str) -> None:
"""Name bypasses and double-noising independently of callback timing."""
expected = torch.ones(1, 4, 2, 2)
verification = _verification(expected)
with pytest.raises(ValueError, match="Upscale.*ID 8.*uni_pc"):
if failure == "unused":
verification.require_verified()
else:
value = torch.zeros_like(expected)
if failure == "shape":
value = torch.zeros(1, 4, 1, 1)
if failure == "nonfinite":
value.fill_(float("nan"))
verification(torch.tensor(0.4), value, value)
tests/sampling/test_pack_sampling_catalog_contract.py
+41
View File
@@ -112,3 +112,44 @@ def test_catalog_additions_propagate_without_node_edits(
for current, prior in zip(schema.inputs, previous.inputs, strict=True):
assert current.optional == prior.optional
assert getattr(current, "default", None) == getattr(prior, "default", None)
@pytest.mark.parametrize("prompt_control_available", [False, True])
def test_every_exported_latent_sampler_admits_source_capture(
monkeypatch: pytest.MonkeyPatch,
prompt_control_available: bool,
) -> None:
"""Enforce graph-capture support when registration adds a new sampling node."""
from simple_syrup.runtime.noise_reuse_graph import rewrite_noise_reuse_graph
from simple_syrup.runtime.noise_reuse_source import resolve_source_sampling_graph
monkeypatch.setattr(
nodes_v3, "prompt_control_is_available", lambda: prompt_control_available
)
discovered: list[str] = []
for node in sampling_nodes():
schema = node.define_schema()
if schema.outputs[0].io_type != "LATENT":
continue
inputs: dict[str, object] = {item.id: 0 for item in schema.inputs}
assert {"model", "latent_image", "sampler_name"} <= inputs.keys(), (
schema.node_id
)
inputs.update(
model=["loader", 0], latent_image=["empty", 0], sampler_name="er_sde"
)
prompt = {
"source": {"class_type": schema.node_id, "inputs": inputs},
"upscale": {
"class_type": "SimpleSyrup.SimpleUpscaleLatent",
"inputs": {"latent": ["source", 0], "method": "Auto"},
},
}
assert resolve_source_sampling_graph(prompt, "source") is not None, (
schema.node_id
)
rewrite_noise_reuse_graph(prompt, {})
assert inputs["model"] == ["source.trajectory_capture", 0], schema.node_id
assert inputs["sampler_name"] == "er_sde", schema.node_id
discovered.append(schema.node_id)
assert len(discovered) >= 9
tests/sampling/test_sampler_options_sampling_service.py
+9 -7
View File
@@ -27,7 +27,7 @@ from simple_syrup.nodes_v3.contextual_diffusion_options import (
ContextualDiffusionOptionsV3,
)
from simple_syrup.nodes_v3.tiling_options import TilingOptionsV3
from simple_syrup.runtime import sampling_schedulers
from simple_syrup.runtime import sampling_samplers, sampling_schedulers
from simple_syrup.services import sampler_options_sampling_service as routing
@@ -213,16 +213,18 @@ def test_invalid_inversion_target_fails_before_attention_preparation(
SamplerOptions(contextual_diffusion=ContextualDiffusionOptions()),
],
)
def test_unipc_conflicts_fail_before_sampling(
@pytest.mark.parametrize("sampler_name", sampling_samplers.available_samplers())
def test_every_sampler_keeps_the_selected_context_or_tiling_route(
routed_calls: list[tuple[str, dict[str, Any]]],
options: SamplerOptions,
sampler_name: str,
) -> None:
"""Do not silently bypass tiling for an incompatible sampling method."""
"""Discover additions and require their complete contextual and tiled routing."""
values = arguments()
values["sampler_name"] = "uni_pc"
with pytest.raises(ValueError, match="UniPC"):
routing.SamplerOptionsSamplingService().sample(**values, options=options)
assert not routed_calls
values["sampler_name"] = sampler_name
routing.SamplerOptionsSamplingService().sample(**values, options=options)
assert len(routed_calls) == 1
assert routed_calls[0][1]["sampler_name"] == sampler_name
@pytest.mark.parametrize("cd_first", [False, True])
tests/sampling/test_trajectory_capture.py
+227 -84
View File
@@ -1,6 +1,8 @@
"""Exercise capture through native Euler solver states and verified endpoint
identity.
"""
# SimpleSyrup - workflow-focused ComfyUI extensions for image generation
# Copyright (C) 2026 Artificial Sweetener and contributors
# SPDX-License-Identifier: AGPL-3.0-or-later
"""Prove observational capture preserves native sampling and real coordinates."""
from types import SimpleNamespace
from typing import Any
@@ -10,112 +12,253 @@ import torch
from comfy.k_diffusion.sampling import sample_euler
from simple_syrup.domain.noise_reuse import PROVENANCE_KEY, SamplingProvenance
from simple_syrup.runtime.trajectory_capture import (
TrajectoryCapture,
insert_capture_coordinates,
)
from simple_syrup.runtime.sampling_samplers import available_samplers, resolve_sampler
from simple_syrup.runtime.trajectory_capture import TrajectoryCapture
class NativeSampler:
"""Exercise the same sampler-instance method binding as native Comfy."""
class PredictionModel:
"""Emulate the native prediction boundary without loading model weights."""
def __init__(self) -> None:
"""Keep native solver ownership local to the sampler instance."""
self.sampler_function: Any = sample_euler
def __init__(self, capture: TrajectoryCapture | None = None) -> None:
"""Retain the optional observer around the same deterministic prediction."""
self.capture = capture
def sample(self, *args: Any) -> torch.Tensor:
"""Invoke the currently owned solver using a deterministic denoised
prediction.
"""
def __call__(
self, state: torch.Tensor, sigma: torch.Tensor, **kwargs: Any
) -> torch.Tensor:
"""Run the source observation before its unchanged denoising prediction."""
if self.capture is not None:
return self.capture.observe_prediction(self.predict, state, sigma, **kwargs)
return self.predict(state, sigma, **kwargs)
def model(
state: torch.Tensor, sigma: torch.Tensor, **kwargs: Any
) -> torch.Tensor:
"""Return a clean zero reference through the native model boundary."""
return torch.zeros_like(state)
result: torch.Tensor = self.sampler_function(model, *args, disable=True)
return result
def predict(
self, state: torch.Tensor, sigma: torch.Tensor, **kwargs: Any
) -> torch.Tensor:
"""Return a zero endpoint with a known linear Euler trajectory."""
return torch.zeros_like(state)
class Executor:
"""Represent Comfy's external sampler-wrapper invocation contract."""
"""Represent Comfy's sampler wrapper while retaining the original algorithm."""
def __init__(self) -> None:
"""Retain one externally owned native sampler."""
self.class_obj = NativeSampler()
self.original: Any = self.class_obj.sample
def __init__(self, capture: TrajectoryCapture | None = None) -> None:
"""Observe only model calls, leaving callbacks and the solver untouched."""
self.model = PredictionModel(capture)
def __call__(self, *args: Any, **kwargs: Any) -> torch.Tensor:
"""Bind replacement sampler methods like the native WrapperExecutor does."""
result: torch.Tensor = self.original(*args, **kwargs)
def __call__(
self, state: torch.Tensor, schedule: torch.Tensor, **kwargs: Any
) -> torch.Tensor:
"""Execute the real Euler implementation on the supplied schedule."""
result: torch.Tensor = sample_euler(
self.model, state, schedule, disable=True, **kwargs
)
return result
def test_native_euler_captures_actual_state_at_inserted_coordinate() -> None:
"""Detect accidental capture of the denoised prediction instead of noisy x."""
format_type = type("Wan21", (), {})
model = SimpleNamespace(
model=SimpleNamespace(
latent_format=format_type(), process_latent_in=lambda x: x
)
)
model.get_model_object = lambda name: SimpleNamespace(
noise_scaling=lambda sigma, noise, clean, maximum: clean + sigma * noise
)
capture = TrajectoryCapture(model, "euler", (0.6, 0.3))
initial = torch.ones(1, 16, 1, 3, 4)
result = capture.observe(Executor(), initial, torch.tensor([1.0, 0.5, 0.0]))
assert torch.allclose(capture.states[float(torch.tensor(0.6))], initial * 0.6)
assert torch.allclose(capture.states[float(torch.tensor(0.3))], initial * 0.3)
attached = capture.attach({"samples": result})
provenance = attached[PROVENANCE_KEY]
assert isinstance(provenance, SamplingProvenance)
assert provenance.latent_format == "Wan21"
assert torch.equal(provenance.clean, result)
assert torch.allclose(provenance.residual_at(float(torch.tensor(0.6))), initial)
assert initial.sum() > 0 and result.sum() == 0
with pytest.raises(ValueError, match="endpoint"):
capture.attach({"samples": result + 1})
def test_capture_uses_native_clean_endpoint_before_external_format_roundoff() -> None:
"""Preserve the original trajectory's reference across lossy format conversions."""
initial = torch.ones(1, 16, 1, 3, 4)
model = SimpleNamespace(
def model_surface(roundoff: float = 0.0) -> Any:
"""Provide native latent format and additive noise parameterization."""
return SimpleNamespace(
model=SimpleNamespace(
latent_format=type("Wan21", (), {})(),
process_latent_in=lambda value: value + 1e-7,
process_latent_in=lambda value: value + roundoff,
),
get_model_object=lambda name: SimpleNamespace(
noise_scaling=lambda sigma, noise, clean, maximum: clean + sigma * noise
),
)
capture = TrajectoryCapture(model, "euler", (0.5,))
result = capture.observe(Executor(), initial, torch.tensor([1.0, 0.5, 0.0]))
def test_native_euler_captures_nearest_real_coordinate_without_inserting_steps() -> (
None
):
"""Distinguish requested levels from actually observed source states."""
capture = TrajectoryCapture(model_surface(), "euler", (0.6, 0.3))
initial = torch.ones(1, 16, 1, 3, 4)
schedule = torch.tensor([1.0, 0.5, 0.0])
result = capture.observe(Executor(capture), initial, schedule)
assert capture.schedule is not None
assert torch.equal(capture.schedule, schedule)
for observation in capture.states.values():
assert observation.coordinate == 0.5
assert torch.equal(observation.state, initial * 0.5)
attached = capture.attach({"samples": result})
provenance = attached[PROVENANCE_KEY]
assert isinstance(provenance, SamplingProvenance)
assert provenance.latent_format == "Wan21"
assert torch.equal(provenance.clean, result)
for request in capture.coordinates:
assert provenance.capture_at(request).actual_coordinate == 0.5
assert torch.equal(provenance.residual_at(request), initial)
assert initial.sum() > 0 and result.sum() == 0
with pytest.raises(ValueError, match="different original sampler output"):
capture.attach({"samples": result + 1})
def test_native_boundary_capture_preserves_source_result_and_callback() -> None:
"""Retain the source's exact schedule, endpoint and original callback sequence."""
initial = torch.ones(1, 4, 3, 4)
schedule = torch.tensor([1.0, 0.5, 0.0])
callbacks: list[int] = []
reference = Executor()(initial.clone(), schedule)
capture = TrajectoryCapture(model_surface(), "euler", (0.5,))
def callback(record: dict[str, Any]) -> None:
"""Retain the original solver's callback order and count."""
callbacks.append(int(record["i"]))
output = capture.observe(
Executor(capture), initial.clone(), schedule, callback=callback
)
assert torch.equal(output, reference)
assert callbacks == [0, 1]
assert capture.states[0.5].coordinate == 0.5
assert torch.equal(capture.states[0.5].state, initial * 0.5)
assert torch.equal(schedule, torch.tensor([1.0, 0.5, 0.0]))
def test_format_only_capture_returns_the_original_model() -> None:
"""Keep Auto latent-format provenance free of numerical sampling changes."""
model = object()
assert TrajectoryCapture(model, "er_sde", ()).prepare_model() is model
def test_capture_uses_native_clean_endpoint_before_external_format_roundoff() -> None:
"""Preserve the original trajectory's reference across lossy format conversions."""
initial = torch.ones(1, 16, 1, 3, 4)
capture = TrajectoryCapture(model_surface(1e-7), "euler", (0.5,))
result = capture.observe(Executor(capture), initial, torch.tensor([1.0, 0.5, 0.0]))
provenance = capture.attach({"samples": result})[PROVENANCE_KEY]
assert isinstance(provenance, SamplingProvenance)
assert torch.equal(provenance.clean, result)
assert torch.equal(provenance.residual_at(0.5), initial)
def test_coordinate_insertion_keeps_boundaries_and_rejects_invalid_ranges() -> None:
"""Avoid changing source schedule points or silently accepting unavailable
noise levels.
"""
original = torch.tensor([1.0, 0.7, 0.4, 0.0])
schedule = insert_capture_coordinates(original, (0.6, 0.2))
assert torch.equal(schedule, torch.tensor([1.0, 0.7, 0.6, 0.4, 0.2, 0.0]))
assert torch.equal(original, torch.tensor([1.0, 0.7, 0.4, 0.0]))
assert torch.equal(insert_capture_coordinates(schedule, (0.6,)), schedule)
with pytest.raises(ValueError, match="outside"):
insert_capture_coordinates(original, (1.5,))
with pytest.raises(ValueError, match="leftover noise"):
insert_capture_coordinates(torch.tensor([1.0, 0.7, 0.4]), (0.6,))
def test_ties_prefer_higher_noise_and_snapshots_are_immutable() -> None:
"""Make nearest selection independent of evaluation order and tensor mutation."""
capture = TrajectoryCapture(model_surface(), "external", (0.5, 0.25))
state = torch.ones(1, 4, 3, 4)
def source(*args: Any) -> torch.Tensor:
"""Observe a tie followed by an exact match in an external algorithm."""
capture.observe_prediction(
PredictionModel().predict, state, torch.tensor([0.25])
)
state.mul_(3)
capture.observe_prediction(
PredictionModel().predict, state, torch.tensor([0.75])
)
state.zero_()
return state
capture.observe(source, state, torch.tensor([1.0, 0.0]))
assert capture.states[0.5].coordinate == 0.75
assert capture.states[0.5].state.mean() == 3
assert capture.states[0.25].coordinate == 0.25
assert capture.states[0.25].state.mean() == 1
def test_incomplete_capture_and_unsupported_solver_fail_closed() -> None:
"""Keep unavailable data from masquerading as usable source noise."""
capture = TrajectoryCapture(object(), "dpmpp_2m", (0.5,))
with pytest.raises(ValueError, match="Euler"):
capture.prepare_model()
def test_incomplete_capture_fails_closed_with_chosen_algorithm() -> None:
"""Reject a source that bypasses the native model-prediction contract."""
capture = TrajectoryCapture(model_surface(), "external_unobserved", (0.5,))
with pytest.raises(
ValueError, match="external_unobserved.*is not supported.*noisy latent"
):
capture.attach({"samples": torch.ones(1, 4, 3, 4)})
@pytest.mark.parametrize("coordinate", [0.0, -1.0, float("nan"), float("inf")])
def test_invalid_capture_requests_fail_before_sampling(coordinate: float) -> None:
"""Reject undefined native noise levels before installing observers."""
with pytest.raises(ValueError, match="finite and above zero"):
TrajectoryCapture(model_surface(), "euler", (coordinate,))
class SolverModel(PredictionModel):
"""Supply native sampler metadata and CFG hooks around a controlled model."""
def __init__(self, capture: TrajectoryCapture) -> None:
"""Model the external guider boundary shared by all advertised algorithms."""
super().__init__(capture)
sampling = SimpleNamespace(
sigma_max=torch.tensor(1.0),
sigma_min=torch.tensor(0.01),
noise_scaling=lambda sigma, noise, clean: clean + sigma * noise,
calculate_denoised=lambda sigma, eps, x: x - sigma * eps,
percent_to_sigma=lambda fraction: 1.0 - fraction,
)
self.inner_model = SimpleNamespace(
inner_model=SimpleNamespace(
device=torch.device("cpu"),
model_sampling=sampling,
diffusion_model=SimpleNamespace(),
),
model_patcher=SimpleNamespace(get_model_object=lambda name: sampling),
)
def predict(
self, state: torch.Tensor, sigma: torch.Tensor, **kwargs: Any
) -> torch.Tensor:
"""Run native post-CFG hooks to expose conditional and unconditional results."""
denoised = torch.full_like(state, 0.2)
for hook in kwargs.get("model_options", {}).get(
"sampler_post_cfg_function", ()
):
denoised = hook(
{
"denoised": denoised,
"uncond_denoised": denoised,
"cond_denoised": denoised,
}
)
return denoised
@pytest.mark.parametrize("sampler_name", available_samplers())
@pytest.mark.filterwarnings(
"ignore:To copy construct from a tensor:UserWarning:comfy.k_diffusion.sampling"
)
@pytest.mark.filterwarnings(
"ignore:__array_wrap__ must accept context:DeprecationWarning:"
"comfy.k_diffusion.deis"
)
def test_every_advertised_algorithm_captures_real_source_noise(
sampler_name: str,
) -> None:
"""Discover newly added algorithms automatically and require executable capture."""
capture = TrajectoryCapture(model_surface(), sampler_name, (0.4,))
model = SolverModel(capture)
sampler = resolve_sampler(sampler_name)
sample = vars(sampler)["sampler_function"]
initial = torch.randn(1, 4, 8, 8, generator=torch.Generator().manual_seed(19))
schedule = torch.tensor([1.0, 0.5, 0.0])
if sampler_name in {"uni_pc", "uni_pc_bh2"}:
schedule = torch.tensor([1.0, 0.5, 0.1, 0.0])
supplied_schedule = schedule.clone()
def execute(state: torch.Tensor, sigmas: torch.Tensor) -> torch.Tensor:
"""Invoke each genuine solver with its own declared options and fixed seed."""
result: torch.Tensor = sample(
model,
state,
sigmas,
extra_args={"seed": 42, "model_options": {"transformer_options": {}}},
callback=None,
disable=True,
**vars(sampler)["extra_options"],
)
return result
with torch.random.fork_rng(devices=[]):
torch.manual_seed(43)
reference = execute(initial.clone(), schedule.clone())
torch.manual_seed(43)
output = capture.observe(execute, initial.clone(), schedule)
observation = capture.states[float(torch.tensor(0.4))]
assert observation.coordinate > 0
assert observation.state.shape == initial.shape
assert torch.isfinite(observation.state).all()
assert capture.clean is not None
assert capture.schedule is not None
assert torch.equal(capture.clean, output.float().cpu())
assert torch.equal(output, reference)
assert torch.equal(capture.schedule, supplied_schedule)