fix(ltx): Soften Schedule — coverage log follows the anchor's schedule

Review P3: when runtime sigmas were present but malformed/empty while a
valid SIGMAS input was wired, behavior fell back to the wired anchor but
the coverage log still read the malformed runtime object and could
report "first 0 of 0 steps". Bind cov to whichever schedule the anchor
actually came from, and extend the fallback to a zero-start runtime
schedule (unusable) instead of going inert while a wired schedule is
available. Tooltip now states the wired input's runtime-fallback role.

Adds wiring tests for the denoise-mask hook: runtime-anchor coverage,
malformed/zero-start runtime fallback to wired, and warn-once inertness.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
malkuthro
2026-07-19 21:37:23 +02:00
co-authored by Claude Fable 5
parent cabaec4107
commit 0fd477759d
2 changed files with 79 additions and 8 deletions
+14 -8
View File
@@ -136,7 +136,8 @@ class LTXKeyframeSoftenSchedule:
"optional": {
"sigmas": ("SIGMAS", {
"tooltip": "Optional — connect your sampler's SIGMAS to show the "
"exact steps on the 'info' output.",
"exact steps on the 'info' output; also used as the "
"fallback schedule when the run doesn't supply sigmas.",
}),
},
}
@@ -171,7 +172,8 @@ class LTXKeyframeSoftenSchedule:
prev_fn = m.model_options.get("denoise_mask_function")
# Fallback from the connected SIGMAS (if wired). Used when a run omits
# extra_options["sigmas"] — gives the wired input a real runtime role,
# extra_options["sigmas"] or supplies an unusable one — gives the wired
# input a real runtime role,
# not just the info readout. Keep the WHOLE schedule, not only the anchor:
# the coverage log needs the real step count to stay honest (a wired
# 16-step schedule must report "of 16 steps", not a synthetic 1).
@@ -195,16 +197,20 @@ class LTXKeyframeSoftenSchedule:
# different sampler schedule — so no stale cached anchor, and a run
# that momentarily lacks runtime sigmas can never latch the node
# permanently off (the old cache-0.0 path). Falls back to the wired
# SIGMAS when the runtime doesn't supply them.
# SIGMAS when the runtime doesn't supply usable ones.
rt = (extra_options or {}).get("sigmas")
ref = None
cov = None # the schedule the anchor came from — coverage log follows it
if rt is not None:
try:
ref = float(rt[0])
cov = rt
except Exception:
ref = None
if ref is None:
ref = connected_ref
if ref is None or ref <= 0.0:
# Runtime sigmas absent or unusable (malformed/empty/zero-start):
# anchor AND coverage both come from the wired schedule instead.
ref, cov = connected_ref, connected_sigmas
if not ref or ref <= 0.0:
# No usable reference sigma from the run and none wired: the node
# is patched but inert this run. Warn ONCE so a silent no-op on a
@@ -228,9 +234,9 @@ class LTXKeyframeSoftenSchedule:
if not logged[0]:
logged[0] = True
nested = getattr(denoise_mask, "is_nested", False)
# Report coverage against the schedule actually in force: the
# runtime one, else the wired one we anchored to.
cov = rt if rt is not None else connected_sigmas
# cov is the schedule the anchor came from (see above), so a
# malformed runtime object can't be reported while the wired
# schedule is the one actually in force.
try:
count, n_steps = steps_in_range(cov, soften_range)
except Exception:
@@ -159,3 +159,68 @@ def test_steps_flat_top_curve_spans_many_steps():
def test_steps_short_or_empty():
assert steps([1.0], 0.5) == (0, 0)
assert steps([], 0.5) == (0, 0)
# --- denoise_mask_function wiring: connected-SIGMAS fallback + honest coverage log
class _FakeModelPatcher:
"""Stub of the ModelPatcher surface patch() touches."""
def __init__(self):
self.model_options = {}
self.mask_fn = None
def clone(self):
return self
def set_model_denoise_mask_function(self, fn):
self.mask_fn = fn
WIRED = [1.0, 0.9, 0.8, 0.6, 0.3, 0.0] # 5 steps; top 25% covers the first 3
def _patched_fn(sigmas=WIRED):
node = _m.LTXKeyframeSoftenSchedule()
fake = _FakeModelPatcher()
node.patch(fake, max_soften=0.5, soften_range=0.25, sigmas=sigmas)
return fake.mask_fn
def test_runtime_sigmas_anchor_and_log_coverage(caplog):
# a valid runtime schedule wins over the wired one, for anchor AND log
fn = _patched_fn()
with caplog.at_level("INFO", logger=_m.log.name):
out = fn(1.0, 0.2, extra_options={"sigmas": [1.0, 0.5, 0.0]})
assert out == pytest.approx(soften(0.2, 0.5)) # progress 0 -> full gain
assert "first 1 of 2 steps" in caplog.text
def test_malformed_runtime_sigmas_fall_back_to_wired_for_anchor_and_log(caplog):
# runtime sigmas present but empty: behavior anchors to the WIRED schedule,
# and the coverage log must report the wired steps too — not "0 of 0".
fn = _patched_fn()
with caplog.at_level("INFO", logger=_m.log.name):
out = fn(1.0, 0.2, extra_options={"sigmas": []})
assert out == pytest.approx(soften(0.2, 0.5))
assert "first 3 of 5 steps" in caplog.text
assert "0 of 0" not in caplog.text
def test_zero_start_runtime_sigmas_fall_back_to_wired(caplog):
# a degenerate runtime schedule (starts at 0) is unusable -> wired fallback
fn = _patched_fn()
with caplog.at_level("INFO", logger=_m.log.name):
out = fn(1.0, 0.2, extra_options={"sigmas": [0.0, 0.0]})
assert out == pytest.approx(soften(0.2, 0.5))
assert "first 3 of 5 steps" in caplog.text
def test_no_usable_sigmas_anywhere_warns_once_and_passes_through(caplog):
fn = _patched_fn(sigmas=None)
with caplog.at_level("WARNING", logger=_m.log.name):
out1 = fn(1.0, 0.2, extra_options={})
out2 = fn(1.0, 0.2, extra_options={})
assert out1 == 0.2 and out2 == 0.2 # mask untouched
assert caplog.text.count("soften inactive") == 1