Files

7994 lines
359 KiB
Python

_EVENT_HORIZON_CASCADE_CACHE = {
"latent": None,
"frames": None,
"segment_index": -1
}
import copy
import csv
import json
import math
import os
import re
import inspect
import importlib
import time
import hashlib
from pathlib import Path
from datetime import datetime
from .core.enums import (
DEBUG_MODES,
DEBUG_BASIC,
TECH_TEXT,
TECH_LATENT,
TECH_IMAGE,
TECH_NOISE,
TECH_CONDITIONING,
TECH_DELTA,
FORMULA_ROLES_CODE_PASS_6,
RELATION_TYPES_CODE_PASS_6,
EQ_UNKNOWN,
SPACE_TEXT,
SPACE_LATENT,
SPACE_IMAGE,
SPACE_NOISE,
SPACE_CONDITIONING,
SPACE_DELTA,
ROLE_UNKNOWN,
ROLE_OUTCOME_PREVIOUS,
ROLE_OUTCOME_NEXT,
ROLE_OBSERVED_BEHAVIOR,
SEV_LOW,
SEV_INFO,
SEV_MEDIUM,
CONFLICT_ROLE_UNKNOWN,
CONFLICT_NO_READER_FOUND,
CONFLICT_EMPTY_RELATION,
CONFLICT_INVALID_SIGNAL_ID,
CONFLICT_DELTA_UNAVAILABLE,
CONFLICT_BOUNDARY_SHAPE_MISMATCH,
CONFLICT_SAMPLER_BOUNDARY_SHAPE_MISMATCH,
CONFLICT_SAMPLER_DELTA_UNAVAILABLE,
CONFLICT_SAMPLER_METADATA_MISSING,
CONFLICT_NO_STABLE_ANCHORS,
CONFLICT_NO_ACTIVE_CHANGES,
CONFLICT_NO_CONTACT_RULE,
CONFLICT_NO_ENDPOINT,
CONFLICT_NO_FORBIDDEN_DRIFT,
CONFLICT_NOISE_MISSING,
CONFLICT_NOISE_SHAPE_UNKNOWN,
CONFLICT_NOISE_READER_FAILED,
CONFLICT_NOISE_LATENT_SHAPE_MISMATCH,
CONFLICT_ALPHA_ROUTE_MISSING,
CONFLICT_ALPHA_PARTIAL_INPUT,
CONFLICT_FROZEN_ROUTE_MISSING,
CONFLICT_FROZEN_PARTIAL_OBSERVABILITY,
CONFLICT_PASSTHROUGH_OUTPUT_MISSING,
CONFLICT_PACKET_BRANCH_CREATED,
CONFLICT_REPORT_SAVE_FAILED,
REL_GUIDES,
REL_CONSTRAINS,
REL_EXPANDS_TO,
REL_TRANSFORMS_INTO,
)
from .core.packet import (
make_event_packet,
ensure_packet,
record_stage,
add_signal,
add_projection,
add_relation,
add_conflict,
get_latest_signals,
get_projection_ids_for_signal,
signal_exists,
)
from .core.signal import make_event_signal
from .core.relation import make_event_relation
from .core.conflict import make_conflict
from .core.event_sampler import EventSamplerCore, EventSamplerWindow, EventSamplerResult
from .reports.markdown_report import build_markdown_report
from .resolvers.role_resolver import resolve_role
from .resolvers.operator_registry import OperatorRegistry
from .resolvers.s_resolver import build_sstate_from_packet
from .utils.tensor_stats import compute_tensor_delta, extract_latent_samples, safe_shape
from .utils.frozen_helpers import build_input_signatures, build_passthrough_status, score_observability, collect_shared_targets, now_run_id
from .adapters.wan.wan_adapter import apply_wan_adapter
EVENT_HORIZON_RUNTIME_VERSION = "0.1.1-r59"
EVENT_HORIZON_RUNTIME_NAME = "Event Horizon R59 Strategy Math Native Loop"
EVENT_HORIZON_BODY_VERSION = "0.1-r59"
def _event_json_safe(value, depth=0):
if depth > 8:
return str(type(value).__name__)
if value is None or isinstance(value, (str, int, bool)):
return value
if isinstance(value, float):
return value if math.isfinite(value) else str(value)
if isinstance(value, Path):
return str(value)
if isinstance(value, dict):
return {str(k): _event_json_safe(v, depth + 1) for k, v in value.items()}
if isinstance(value, (list, tuple)):
return [_event_json_safe(v, depth + 1) for v in value]
return str(value)
def _read_signal(packet, technical_type, representation_space, raw_ref, source_stage, manual_role, route_id, reader_operator, metadata=None, route_position=""):
packet = ensure_packet(packet)
role = resolve_role(
technical_type,
position_context={"stage_name": source_stage, "route_position": route_position},
manual_role=manual_role,
)
conflict_ids = []
signal = make_event_signal(
technical_type=technical_type,
formula_role=role,
representation_space=representation_space,
raw_ref=raw_ref,
source_stage=source_stage,
created_by=f"EventRead{technical_type.title()}",
reader_operator=reader_operator,
route_id=route_id,
position_context={"route_position": route_position},
metadata=metadata or {},
)
projection = OperatorRegistry().read(signal)
if role == ROLE_UNKNOWN:
conflict = make_conflict(
CONFLICT_ROLE_UNKNOWN,
severity=SEV_LOW,
involved_signal_ids=[signal["id"]],
stage_position=source_stage,
suspected_cause="RoleResolver could not assign a formula role.",
observed_symptom=f"{technical_type} resolved to UnknownRole.",
suggested_response="Provide a manual role or improve position_context.",
)
packet = add_conflict(packet, conflict)
conflict_ids.append(conflict["id"])
if projection.get("metadata", {}).get("warning") == CONFLICT_NO_READER_FOUND or projection.get("operator_name") == "UnknownReader":
conflict = make_conflict(
CONFLICT_NO_READER_FOUND,
severity=SEV_MEDIUM,
involved_signal_ids=[signal["id"]],
involved_projection_ids=[projection["id"]],
stage_position=source_stage,
suspected_cause="OperatorRegistry found no reader for this signal.",
observed_symptom=f"No reader for technical type {technical_type}.",
suggested_response="Add/register a ReaderOperator for this technical type.",
)
packet = add_conflict(packet, conflict)
conflict_ids.append(conflict["id"])
packet = add_signal(packet, signal)
packet = add_projection(packet, projection)
packet = record_stage(
packet,
stage_name=f"EventRead{technical_type.title()}",
action="READ_SIGNAL",
observed_behavior=f"{technical_type} read as {role}",
output_signal_ids=[signal["id"]],
projection_ids=[projection["id"]],
conflict_ids=conflict_ids,
formula_note=f"{technical_type} projected into EVENT_SPACE",
)
return packet, signal, projection, conflict_ids
def _parse_id_list(value):
text = str(value or "").strip()
if not text:
return []
if text.lower() in ("latest", "auto"):
return []
return [x.strip() for x in text.split(",") if x.strip()]
def _event_core_list_input_images():
try:
import folder_paths
input_dir = folder_paths.get_input_directory()
if not os.path.isdir(input_dir):
return ["none"]
valid_ext = {".png", ".jpg", ".jpeg", ".webp", ".bmp", ".tiff"}
files = []
for name in os.listdir(input_dir):
path = os.path.join(input_dir, name)
if os.path.isfile(path) and os.path.splitext(name)[1].lower() in valid_ext:
files.append(name)
files = sorted(files)
return ["none"] + files if files else ["none"]
except Exception:
return ["none"]
def _event_core_list_output_folders():
try:
import folder_paths
output_dir = folder_paths.get_output_directory()
entries = ["default", r"D:\AI NSFW\VID", r"D:\AI NSFW\PIC"]
if os.path.isdir(output_dir):
for root, dirs, files in os.walk(output_dir):
rel = os.path.relpath(root, output_dir)
if rel != ".":
entries.append(rel.replace("\\", "/"))
if len(entries) >= 100:
break
return sorted(set(entries))
except Exception:
return ["default", r"D:\AI NSFW\VID", r"D:\AI NSFW\PIC"]
def _event_core_preferred_media_dirs(media_type="video"):
media_type = str(media_type or "video").lower()
if media_type == "video":
folder_names = ["VID"]
elif media_type == "image":
folder_names = ["PIC"]
elif media_type == "report":
folder_names = ["reports"]
else:
folder_names = [media_type]
roots = [Path(r"D:\AI NSFW"), Path(r"D:/AI NSFW")]
candidates = []
for root in roots:
for name in folder_names:
candidates.append(root / name)
return candidates
class EventDebugPing:
@classmethod
def INPUT_TYPES(cls):
return {"required": {"message": ("STRING", {"default": "ping"})}}
RETURN_TYPES = ("STRING",)
RETURN_NAMES = ("message",)
FUNCTION = "run"
CATEGORY = "Event Equality/Core"
def run(self, message):
return (f"Event Debug Ping: {message}",)
class EventInitPacket:
@classmethod
def INPUT_TYPES(cls):
return {"required": {"debug_mode": (DEBUG_MODES, {"default": DEBUG_BASIC})}}
RETURN_TYPES = ("EVENT_PACKET", "STRING")
RETURN_NAMES = ("event_packet", "summary")
FUNCTION = "run"
CATEGORY = "Event Equality/Core"
def run(self, debug_mode):
packet = make_event_packet(metadata={"debug_mode": debug_mode})
packet = record_stage(
packet,
stage_name="EventInitPacket",
action="INIT_PACKET",
observed_behavior="EventPacket created with empty RouteMemory",
metadata={"debug_mode": debug_mode},
)
return (packet, build_markdown_report(packet))
class EventReadText:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"event_packet": ("EVENT_PACKET",),
"text": ("STRING", {"default": "", "multiline": True}),
"source_stage": ("STRING", {"default": "manual_text"}),
"manual_role": (FORMULA_ROLES_CODE_PASS_6, {"default": "AUTO"}),
}
}
RETURN_TYPES = ("EVENT_PACKET", "STRING")
RETURN_NAMES = ("event_packet", "summary")
FUNCTION = "run"
CATEGORY = "Event Equality/Readers"
def run(self, event_packet, text, source_stage, manual_role):
packet, signal, projection, conflict_ids = _read_signal(
event_packet, TECH_TEXT, SPACE_TEXT, text, source_stage,
manual_role, "text_strategy_route", "TextStrategyReader",
metadata={"text_length": len(str(text or ""))},
)
sem = projection.get("semantic_summary", {})
return (packet, f"Created {signal['id']} as TEXT/{signal['formula_role']}; projection={projection['id']}; words={sem.get('word_count', 0)}; conflicts={conflict_ids}")
class EventReadLatent:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"event_packet": ("EVENT_PACKET",),
"latent": ("LATENT",),
"source_stage": ("STRING", {"default": "manual_latent"}),
"manual_role": (FORMULA_ROLES_CODE_PASS_6, {"default": "AUTO"}),
"route_position": ("STRING", {"default": ""}),
}
}
RETURN_TYPES = ("EVENT_PACKET", "STRING")
RETURN_NAMES = ("event_packet", "summary")
FUNCTION = "run"
CATEGORY = "Event Equality/Readers"
def run(self, event_packet, latent, source_stage, manual_role, route_position):
packet, signal, projection, conflict_ids = _read_signal(
event_packet, TECH_LATENT, SPACE_LATENT, latent, source_stage,
manual_role, "latent_route", "LatentEventReader",
route_position=route_position,
)
num = projection.get("numeric_summary", {})
return (packet, f"Created {signal['id']} as LATENT/{signal['formula_role']}; projection={projection['id']}; shape={num.get('shape')}; conflicts={conflict_ids}")
class EventReadImage:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"event_packet": ("EVENT_PACKET",),
"image": ("IMAGE",),
"source_stage": ("STRING", {"default": "manual_image"}),
"manual_role": (FORMULA_ROLES_CODE_PASS_6, {"default": "AUTO"}),
"route_position": ("STRING", {"default": ""}),
}
}
RETURN_TYPES = ("EVENT_PACKET", "STRING")
RETURN_NAMES = ("event_packet", "summary")
FUNCTION = "run"
CATEGORY = "Event Equality/Readers"
def run(self, event_packet, image, source_stage, manual_role, route_position):
packet, signal, projection, conflict_ids = _read_signal(
event_packet, TECH_IMAGE, SPACE_IMAGE, image, source_stage,
manual_role, "image_route", "ImageOutcomeReader",
route_position=route_position,
)
num = projection.get("numeric_summary", {})
return (packet, f"Created {signal['id']} as IMAGE/{signal['formula_role']}; projection={projection['id']}; shape={num.get('shape')}; conflicts={conflict_ids}")
class EventReadNoise:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"event_packet": ("EVENT_PACKET",),
"noise": ("LATENT",),
"source_stage": ("STRING", {"default": "manual_noise"}),
"manual_role": (FORMULA_ROLES_CODE_PASS_6, {"default": "AUTO"}),
}
}
RETURN_TYPES = ("EVENT_PACKET", "STRING")
RETURN_NAMES = ("event_packet", "summary")
FUNCTION = "run"
CATEGORY = "Event Equality/Readers"
def run(self, event_packet, noise, source_stage, manual_role):
raw_noise = noise.get("samples") if isinstance(noise, dict) and "samples" in noise else noise
packet, signal, projection, conflict_ids = _read_signal(
event_packet, TECH_NOISE, SPACE_NOISE, raw_noise, source_stage,
manual_role, "noise_route", "NoisePossibilityReader",
metadata={"input_socket": "LATENT_as_noise_carrier"},
)
num = projection.get("numeric_summary", {})
return (packet, f"Created {signal['id']} as NOISE/{signal['formula_role']}; projection={projection['id']}; shape={num.get('shape')}; conflicts={conflict_ids}")
class EventReadConditioning:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"event_packet": ("EVENT_PACKET",),
"conditioning": ("CONDITIONING",),
"source_stage": ("STRING", {"default": "manual_conditioning"}),
"manual_role": (FORMULA_ROLES_CODE_PASS_6, {"default": "AUTO"}),
}
}
RETURN_TYPES = ("EVENT_PACKET", "STRING")
RETURN_NAMES = ("event_packet", "summary")
FUNCTION = "run"
CATEGORY = "Event Equality/Readers"
def run(self, event_packet, conditioning, source_stage, manual_role):
packet, signal, projection, conflict_ids = _read_signal(
event_packet, TECH_CONDITIONING, SPACE_CONDITIONING, conditioning, source_stage,
manual_role, "conditioning_route", "ConditioningStrategyReader",
)
num = projection.get("numeric_summary", {})
return (packet, f"Created {signal['id']} as CONDITIONING/{signal['formula_role']}; projection={projection['id']}; length={num.get('length')}; conflicts={conflict_ids}")
class EventStructuredStrategy:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"event_packet": ("EVENT_PACKET",),
"main_strategy": ("STRING", {"default": "", "multiline": True}),
"stable_anchors": ("STRING", {"default": "", "multiline": True}),
"active_changes": ("STRING", {"default": "", "multiline": True}),
"motion_axis": ("STRING", {"default": "", "multiline": True}),
"contact_rule": ("STRING", {"default": "", "multiline": True}),
"reciprocal_reaction": ("STRING", {"default": "", "multiline": True}),
"deformation_rule": ("STRING", {"default": "", "multiline": True}),
"material_rule": ("STRING", {"default": "", "multiline": True}),
"support_rule": ("STRING", {"default": "", "multiline": True}),
"endpoint": ("STRING", {"default": "", "multiline": True}),
"loop_rule": ("STRING", {"default": "", "multiline": True}),
"forbidden_drift": ("STRING", {"default": "", "multiline": True}),
"negative_additions": ("STRING", {"default": "", "multiline": True}),
"priority_notes": ("STRING", {"default": "", "multiline": True}),
"strategy_strength": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 10.0, "step": 0.1}),
"source_stage": ("STRING", {"default": "StructuredStrategy"}),
}
}
RETURN_TYPES = ("EVENT_PACKET", "STRING")
RETURN_NAMES = ("event_packet", "strategy_summary")
FUNCTION = "run"
CATEGORY = "Event Equality/Strategy"
def run(
self,
event_packet,
main_strategy,
stable_anchors,
active_changes,
motion_axis,
contact_rule,
reciprocal_reaction,
deformation_rule,
material_rule,
support_rule,
endpoint,
loop_rule,
forbidden_drift,
negative_additions,
priority_notes,
strategy_strength,
source_stage,
):
packet = ensure_packet(event_packet)
structured = {
"main_strategy": main_strategy,
"stable_anchors": stable_anchors,
"active_changes": active_changes,
"motion_axis": motion_axis,
"contact_rule": contact_rule,
"reciprocal_reaction": reciprocal_reaction,
"deformation_rule": deformation_rule,
"material_rule": material_rule,
"support_rule": support_rule,
"endpoint": endpoint,
"loop_rule": loop_rule,
"forbidden_drift": forbidden_drift,
"negative_additions": negative_additions,
"priority_notes": priority_notes,
"strategy_strength": strategy_strength,
}
combined_text = "\\n".join([
str(main_strategy or ""),
str(stable_anchors or ""),
str(active_changes or ""),
str(motion_axis or ""),
str(contact_rule or ""),
str(reciprocal_reaction or ""),
str(deformation_rule or ""),
str(material_rule or ""),
str(support_rule or ""),
str(endpoint or ""),
str(loop_rule or ""),
str(forbidden_drift or ""),
str(negative_additions or ""),
str(priority_notes or ""),
])
packet, signal, projection, conflict_ids = _read_signal(
packet,
TECH_TEXT,
SPACE_TEXT,
combined_text,
source_stage,
"StrategyCurrent",
"structured_strategy_route",
"TextStrategyReader",
metadata={
"structured_strategy": structured,
"text_length": len(combined_text),
},
)
# Strategy completeness warnings.
checks = [
(stable_anchors, CONFLICT_NO_STABLE_ANCHORS, "Stable anchors are empty.", "Add what must remain fixed/preserved."),
(active_changes, CONFLICT_NO_ACTIVE_CHANGES, "Active changes are empty.", "Add what should move/change."),
(contact_rule, CONFLICT_NO_CONTACT_RULE, "Contact rule is empty.", "Add contact/alignment/interaction rule if relevant."),
(endpoint, CONFLICT_NO_ENDPOINT, "Endpoint is empty.", "Add the intended end state/equilibrium."),
(forbidden_drift, CONFLICT_NO_FORBIDDEN_DRIFT, "Forbidden drift is empty.", "Add what must not drift or change."),
]
for value, conflict_type, symptom, suggestion in checks:
if not str(value or "").strip():
conflict = make_conflict(
conflict_type,
severity=SEV_LOW,
involved_signal_ids=[signal["id"]],
involved_projection_ids=[projection["id"]],
stage_position=source_stage,
suspected_cause="StructuredStrategy field is empty.",
observed_symptom=symptom,
suggested_response=suggestion,
)
packet = add_conflict(packet, conflict)
conflict_ids.append(conflict["id"])
packet = record_stage(
packet,
stage_name="EventStructuredStrategy",
action="READ_STRUCTURED_STRATEGY",
observed_behavior="Structured strategy read as explicit StrategyCurrent",
output_signal_ids=[signal["id"]],
projection_ids=[projection["id"]],
conflict_ids=conflict_ids,
formula_note="Structured Strategy creates expectations, priority map and route hints",
metadata={
"strategy_strength": strategy_strength,
"has_priority_map": bool(projection.get("metadata", {}).get("priority_map")),
"has_route_hints": bool(projection.get("metadata", {}).get("route_hints")),
},
)
role_vector = projection.get("role_vector", {})
summary = (
f"StructuredStrategy signal={signal['id']} projection={projection['id']} "
f"main={role_vector.get('has_main_strategy')} anchors={role_vector.get('has_anchors')} "
f"active={role_vector.get('has_active_changes')} contact={role_vector.get('has_contact_rule')} "
f"endpoint={role_vector.get('has_endpoint')} forbidden_drift={role_vector.get('has_forbidden_drift')} "
f"conflicts={conflict_ids}"
)
return (packet, summary)
class EventRelationBuilder:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"event_packet": ("EVENT_PACKET",),
"relation_type": (RELATION_TYPES_CODE_PASS_6, {"default": "EQUALS_AS_EVENT"}),
"stage_name": ("STRING", {"default": "RelationBuilder"}),
"source_signal_ids": ("STRING", {"default": ""}),
"target_signal_ids": ("STRING", {"default": ""}),
"formula_meaning": ("STRING", {"default": ""}),
"local_strategy_id": ("STRING", {"default": ""}),
}
}
RETURN_TYPES = ("EVENT_PACKET", "STRING")
RETURN_NAMES = ("event_packet", "summary")
FUNCTION = "run"
CATEGORY = "Event Equality/Research"
def run(self, event_packet, relation_type, stage_name, source_signal_ids, target_signal_ids, formula_meaning, local_strategy_id):
packet = ensure_packet(event_packet)
source_ids = _parse_id_list(source_signal_ids)
target_ids = _parse_id_list(target_signal_ids)
conflict_ids = []
if not source_ids and not target_ids:
latest = get_latest_signals(packet, 2)
if len(latest) >= 2:
source_ids = [latest[-2]["id"]]
target_ids = [latest[-1]["id"]]
invalid_ids = []
for sid in source_ids + target_ids:
if not signal_exists(packet, sid):
invalid_ids.append(sid)
if invalid_ids:
conflict = make_conflict(
CONFLICT_INVALID_SIGNAL_ID,
severity=SEV_MEDIUM,
involved_signal_ids=invalid_ids,
stage_position=stage_name,
suspected_cause="RelationBuilder received signal ids not present in EventPacket.",
observed_symptom=f"Invalid signal ids: {invalid_ids}",
suggested_response="Use ids from report or leave fields empty to use latest two signals.",
)
packet = add_conflict(packet, conflict)
conflict_ids.append(conflict["id"])
if not source_ids or not target_ids:
conflict = make_conflict(
CONFLICT_EMPTY_RELATION,
severity=SEV_MEDIUM,
involved_signal_ids=source_ids + target_ids,
stage_position=stage_name,
suspected_cause="RelationBuilder did not receive enough source/target signals.",
observed_symptom="Relation has empty source or target side.",
suggested_response="Create/read at least two signals or provide explicit ids.",
)
packet = add_conflict(packet, conflict)
conflict_ids.append(conflict["id"])
source_projection_ids = []
target_projection_ids = []
for sid in source_ids:
source_projection_ids.extend(get_projection_ids_for_signal(packet, sid))
for tid in target_ids:
target_projection_ids.extend(get_projection_ids_for_signal(packet, tid))
if not formula_meaning:
formula_meaning = f"{relation_type}: source signals relate to target signals"
if not local_strategy_id:
local_strategy_id = f"S0.{str(relation_type).lower()}"
relation = make_event_relation(
relation_type=relation_type,
source_signal_ids=source_ids,
target_signal_ids=target_ids,
source_projection_ids=source_projection_ids,
target_projection_ids=target_projection_ids,
operator_name="EventRelationBuilder",
formula_meaning=formula_meaning,
local_strategy_id=local_strategy_id,
equality_status=EQ_UNKNOWN,
metadata={"stage_name": stage_name},
)
relation["conflict_ids"] = conflict_ids[:]
packet = add_relation(packet, relation)
packet = record_stage(
packet,
stage_name="EventRelationBuilder",
action="CREATE_RELATION",
observed_behavior=f"Relation created: {relation_type}",
input_signal_ids=source_ids + target_ids,
relation_ids=[relation["id"]],
conflict_ids=conflict_ids,
formula_note=formula_meaning,
)
return (packet, f"Created relation {relation['id']} type={relation_type} sources={source_ids} targets={target_ids}; conflicts={conflict_ids}")
class EventSStateBuilder:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"event_packet": ("EVENT_PACKET",),
"position": ("STRING", {"default": "S0"}),
"relation_ids": ("STRING", {"default": ""}),
}
}
RETURN_TYPES = ("EVENT_PACKET", "STRING")
RETURN_NAMES = ("event_packet", "summary")
FUNCTION = "run"
CATEGORY = "Event Equality/Research"
def run(self, event_packet, position, relation_ids):
packet = ensure_packet(event_packet)
rel_ids = _parse_id_list(relation_ids)
if not rel_ids:
rel_ids = None
packet, sstate = build_sstate_from_packet(packet, position=position, active_relation_ids=rel_ids)
packet = record_stage(
packet,
stage_name="EventSStateBuilder",
action="BUILD_SSTATE",
observed_behavior="SState resolved from active relations",
input_signal_ids=sstate.get("active_signal_ids", []),
relation_ids=sstate.get("active_relation_ids", []),
sstate_ids=[sstate["id"]],
conflict_ids=sstate.get("conflict_ids", []),
formula_note="SState is relation-based, not signal-based",
)
return (packet, f"Created SState {sstate['id']} position={position}; active_relations={len(sstate.get('active_relation_ids', []))}; local_strategies={len(sstate.get('local_strategies', {}))}")
class EventNoiseStrategy:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"event_packet": ("EVENT_PACKET",),
"mode": (["READ_EXISTING", "GENERATE_FROM_LATENT"], {"default": "READ_EXISTING"}),
"latent_or_noise": ("LATENT",),
"source_stage": ("STRING", {"default": "EventNoiseStrategy"}),
"seed": ("INT", {"default": 0, "min": 0, "max": 0xffffffffffffffff}),
"noise_strength": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 100.0, "step": 0.01}),
"normalize_noise": ("BOOLEAN", {"default": False}),
"target_mean": ("FLOAT", {"default": 0.0, "min": -100.0, "max": 100.0, "step": 0.01}),
"target_std": ("FLOAT", {"default": 1.0, "min": 0.000001, "max": 100.0, "step": 0.01}),
}
}
RETURN_TYPES = ("LATENT", "EVENT_PACKET", "STRING")
RETURN_NAMES = ("noise_or_latent", "event_packet", "noise_report")
FUNCTION = "run"
CATEGORY = "Event Equality/Noise"
def run(self, event_packet, mode, latent_or_noise, source_stage, seed, noise_strength, normalize_noise, target_mean, target_std):
packet = ensure_packet(event_packet)
conflict_ids = []
raw_input = latent_or_noise
input_samples = extract_latent_samples(raw_input)
generated = False
noise_tensor = None
output_latent = latent_or_noise
try:
import torch
if mode == "GENERATE_FROM_LATENT":
if input_samples is None or not hasattr(input_samples, "shape"):
conflict = make_conflict(
CONFLICT_NOISE_MISSING,
severity=SEV_MEDIUM,
stage_position=source_stage,
suspected_cause="Cannot generate noise without tensor-like latent samples.",
observed_symptom="latent_or_noise has no tensor-like samples.",
suggested_response="Connect a valid LATENT object with samples.",
)
packet = add_conflict(packet, conflict)
conflict_ids.append(conflict["id"])
noise_tensor = input_samples
else:
generator = None
try:
generator = torch.Generator(device=input_samples.device)
generator.manual_seed(int(seed))
noise_tensor = torch.randn(
input_samples.shape,
dtype=input_samples.dtype,
device=input_samples.device,
generator=generator,
)
except Exception:
torch.manual_seed(int(seed))
noise_tensor = torch.randn_like(input_samples)
if normalize_noise and noise_tensor is not None:
nf = noise_tensor.float()
current_mean = nf.mean()
current_std = nf.std()
eps = 1e-12
noise_tensor = ((nf - current_mean) / max(float(current_std.item()), eps)) * float(target_std) + float(target_mean)
noise_tensor = noise_tensor.to(dtype=input_samples.dtype)
noise_tensor = noise_tensor * float(noise_strength)
output_latent = dict(latent_or_noise) if isinstance(latent_or_noise, dict) else {"samples": noise_tensor}
output_latent["samples"] = noise_tensor
generated = True
else:
# READ_EXISTING: treat latent samples as the existing noise carrier.
noise_tensor = input_samples
if noise_tensor is None or not hasattr(noise_tensor, "shape"):
conflict = make_conflict(
CONFLICT_NOISE_SHAPE_UNKNOWN,
severity=SEV_LOW,
stage_position=source_stage,
suspected_cause="Noise object is not tensor-like.",
observed_symptom="Noise shape could not be read.",
suggested_response="Use GENERATE_FROM_LATENT or connect a LATENT-like object.",
)
packet = add_conflict(packet, conflict)
conflict_ids.append(conflict["id"])
except Exception as e:
conflict = make_conflict(
CONFLICT_NOISE_READER_FAILED,
severity=SEV_MEDIUM,
stage_position=source_stage,
suspected_cause="EventNoiseStrategy failed during noise read/generation.",
observed_symptom=str(e),
suggested_response="Check torch availability and latent input format.",
)
packet = add_conflict(packet, conflict)
conflict_ids.append(conflict["id"])
noise_tensor = input_samples
packet, signal, projection, read_conflicts = _read_signal(
packet,
TECH_NOISE,
SPACE_NOISE,
noise_tensor,
source_stage,
"StrategyCandidate",
"noise_strategy_route",
"NoisePossibilityReader",
metadata={
"seed": seed,
"noise_strength": noise_strength,
"noise_mode": mode,
"normalize_noise": normalize_noise,
"target_mean": target_mean,
"target_std": target_std,
"generated_from_latent_shape": safe_shape(input_samples),
"generated": generated,
},
)
conflict_ids.extend(read_conflicts)
packet = record_stage(
packet,
stage_name="EventNoiseStrategy",
action="READ_NOISE_STRATEGY",
observed_behavior=f"Noise route registered as StrategyCandidate mode={mode}",
output_signal_ids=[signal["id"]],
projection_ids=[projection["id"]],
conflict_ids=conflict_ids,
formula_note="NOISE is a StrategyCandidate / possibility field for future sampler behavior",
metadata={
"mode": mode,
"seed": seed,
"noise_strength": noise_strength,
"generated": generated,
"normalize_noise": normalize_noise,
},
)
num = projection.get("numeric_summary", {})
summary = (
f"NoiseStrategy signal={signal['id']} projection={projection['id']} "
f"mode={mode} generated={generated} shape={num.get('shape')} "
f"mean={num.get('mean')} std={num.get('std')} norm={num.get('norm')} "
f"seed={seed} strength={noise_strength} conflicts={conflict_ids}"
)
return (output_latent, packet, summary)
class EventLatentBoundaryReader:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"event_packet": ("EVENT_PACKET",),
"latent_before": ("LATENT",),
"latent_after": ("LATENT",),
"stage_name": ("STRING", {"default": "LatentBoundary"}),
"position": ("STRING", {"default": "S_boundary"}),
"next_requirement_hint": ("STRING", {"default": ""}),
}
}
RETURN_TYPES = ("EVENT_PACKET", "STRING")
RETURN_NAMES = ("event_packet", "boundary_summary")
FUNCTION = "run"
CATEGORY = "Event Equality/Boundary"
def run(self, event_packet, latent_before, latent_after, stage_name, position, next_requirement_hint):
packet = ensure_packet(event_packet)
conflict_ids = []
# Create/read before latent as OutcomePrevious.
packet, before_signal, before_projection, before_conflicts = _read_signal(
packet,
TECH_LATENT,
SPACE_LATENT,
latent_before,
f"{stage_name}_before",
ROLE_OUTCOME_PREVIOUS,
"latent_before_boundary_route",
"LatentEventReader",
route_position="before",
)
conflict_ids.extend(before_conflicts)
# Create/read after latent as OutcomeNext.
packet, after_signal, after_projection, after_conflicts = _read_signal(
packet,
TECH_LATENT,
SPACE_LATENT,
latent_after,
f"{stage_name}_after",
ROLE_OUTCOME_NEXT,
"latent_after_boundary_route",
"LatentEventReader",
route_position="after",
)
conflict_ids.extend(after_conflicts)
before_samples = extract_latent_samples(latent_before)
after_samples = extract_latent_samples(latent_after)
delta, error = compute_tensor_delta(latent_before, latent_after)
delta_signal = None
delta_projection = None
if delta is None:
conflict_type = CONFLICT_BOUNDARY_SHAPE_MISMATCH if "shape mismatch" in str(error).lower() else CONFLICT_DELTA_UNAVAILABLE
conflict = make_conflict(
conflict_type,
severity=SEV_MEDIUM,
involved_signal_ids=[before_signal["id"], after_signal["id"]],
stage_position=stage_name,
suspected_cause="Could not compute delta between latent_before and latent_after.",
observed_symptom=str(error),
suggested_response="Check that both latents have compatible samples tensors.",
metadata={
"before_shape": safe_shape(before_samples),
"after_shape": safe_shape(after_samples),
},
)
packet = add_conflict(packet, conflict)
conflict_ids.append(conflict["id"])
source_ids = [before_signal["id"]]
source_proj_ids = [before_projection["id"]]
delta_id = None
delta_norm = None
relative_delta = None
else:
packet, delta_signal, delta_projection, delta_conflicts = _read_signal(
packet,
TECH_DELTA,
SPACE_DELTA,
delta,
f"{stage_name}_delta",
ROLE_OBSERVED_BEHAVIOR,
"delta_boundary_route",
"DeltaReader",
metadata={
"before_signal_id": before_signal["id"],
"after_signal_id": after_signal["id"],
"before_ref": before_samples,
"next_requirement_hint": next_requirement_hint,
},
)
conflict_ids.extend(delta_conflicts)
source_ids = [before_signal["id"], delta_signal["id"]]
source_proj_ids = [before_projection["id"], delta_projection["id"]]
delta_id = delta_signal["id"]
num = delta_projection.get("numeric_summary", {})
delta_norm = num.get("delta_norm")
relative_delta = num.get("relative_delta")
relation = make_event_relation(
relation_type=REL_TRANSFORMS_INTO,
source_signal_ids=source_ids,
target_signal_ids=[after_signal["id"]],
source_projection_ids=source_proj_ids,
target_projection_ids=[after_projection["id"]],
operator_name="EventLatentBoundaryReader",
formula_meaning="latent before plus observed delta becomes latent after",
local_strategy_id=f"{position}.latent_boundary",
equality_status=EQ_UNKNOWN,
metadata={
"stage_name": stage_name,
"boundary_type": "latent",
"next_requirement_hint": next_requirement_hint,
"delta_signal_id": delta_id,
"delta_norm": delta_norm,
"relative_delta": relative_delta,
},
)
relation["conflict_ids"] = conflict_ids[:]
packet = add_relation(packet, relation)
packet, sstate = build_sstate_from_packet(packet, position=position, active_relation_ids=[relation["id"]])
packet = record_stage(
packet,
stage_name="EventLatentBoundaryReader",
action="READ_BOUNDARY",
observed_behavior="Latent boundary read as before + delta = after",
input_signal_ids=[before_signal["id"], after_signal["id"]],
output_signal_ids=[x for x in [delta_id] if x],
projection_ids=[before_projection["id"], after_projection["id"]] + ([delta_projection["id"]] if delta_projection else []),
relation_ids=[relation["id"]],
sstate_ids=[sstate["id"]],
conflict_ids=conflict_ids,
formula_note="Outcome_before + ObservedBehavior_delta = S_boundary = Outcome_after",
metadata={
"delta_norm": delta_norm,
"relative_delta": relative_delta,
"next_requirement_hint": next_requirement_hint,
},
)
summary = (
f"Boundary {stage_name}: before={before_signal['id']} after={after_signal['id']} "
f"delta={delta_id} relation={relation['id']} sstate={sstate['id']} "
f"delta_norm={delta_norm} relative_delta={relative_delta} conflicts={conflict_ids}"
)
return (packet, summary)
class EventSamplerBoundaryReport:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"event_packet": ("EVENT_PACKET",),
"latent_before_sampler": ("LATENT",),
"latent_after_sampler": ("LATENT",),
"position": ("STRING", {"default": "S_sampler_summary"}),
"sampler_name": ("STRING", {"default": ""}),
"scheduler": ("STRING", {"default": ""}),
"steps": ("INT", {"default": 0, "min": 0, "max": 10000}),
"cfg": ("FLOAT", {"default": 0.0, "min": -1000.0, "max": 1000.0, "step": 0.1}),
"denoise": ("FLOAT", {"default": 0.0, "min": 0.0, "max": 1.0, "step": 0.01}),
"seed": ("INT", {"default": 0, "min": 0, "max": 0xffffffffffffffff}),
}
}
RETURN_TYPES = ("EVENT_PACKET", "STRING")
RETURN_NAMES = ("event_packet", "sampler_boundary_report")
FUNCTION = "run"
CATEGORY = "Event Equality/Boundary"
def run(self, event_packet, latent_before_sampler, latent_after_sampler, position, sampler_name, scheduler, steps, cfg, denoise, seed):
packet = ensure_packet(event_packet)
stage_name = "SamplerBoundaryReport"
conflict_ids = []
# Read before sampler latent as previous outcome.
packet, before_signal, before_projection, before_conflicts = _read_signal(
packet,
TECH_LATENT,
SPACE_LATENT,
latent_before_sampler,
"SamplerBoundary_before",
ROLE_OUTCOME_PREVIOUS,
"sampler_before_latent_route",
"LatentEventReader",
route_position="before_sampler",
)
conflict_ids.extend(before_conflicts)
# Read after sampler latent as next outcome.
packet, after_signal, after_projection, after_conflicts = _read_signal(
packet,
TECH_LATENT,
SPACE_LATENT,
latent_after_sampler,
"SamplerBoundary_after",
ROLE_OUTCOME_NEXT,
"sampler_after_latent_route",
"LatentEventReader",
route_position="after_sampler",
)
conflict_ids.extend(after_conflicts)
before_samples = extract_latent_samples(latent_before_sampler)
after_samples = extract_latent_samples(latent_after_sampler)
delta, error = compute_tensor_delta(latent_before_sampler, latent_after_sampler)
delta_signal = None
delta_projection = None
delta_id = None
delta_norm = None
relative_delta = None
if delta is None:
conflict_type = CONFLICT_SAMPLER_BOUNDARY_SHAPE_MISMATCH if "shape mismatch" in str(error).lower() else CONFLICT_SAMPLER_DELTA_UNAVAILABLE
conflict = make_conflict(
conflict_type,
severity=SEV_MEDIUM,
involved_signal_ids=[before_signal["id"], after_signal["id"]],
stage_position=stage_name,
suspected_cause="Could not compute sampler-level delta.",
observed_symptom=str(error),
suggested_response="Check that sampler input/output latents have compatible samples tensors.",
metadata={
"before_shape": safe_shape(before_samples),
"after_shape": safe_shape(after_samples),
"sampler_name": sampler_name,
"scheduler": scheduler,
"steps": steps,
"cfg": cfg,
"denoise": denoise,
"seed": seed,
},
)
packet = add_conflict(packet, conflict)
conflict_ids.append(conflict["id"])
source_ids = [before_signal["id"]]
source_proj_ids = [before_projection["id"]]
else:
packet, delta_signal, delta_projection, delta_conflicts = _read_signal(
packet,
TECH_DELTA,
SPACE_DELTA,
delta,
"SamplerBoundary_delta",
ROLE_OBSERVED_BEHAVIOR,
"sampler_delta_route",
"DeltaReader",
metadata={
"before_signal_id": before_signal["id"],
"after_signal_id": after_signal["id"],
"before_ref": before_samples,
"sampler_name": sampler_name,
"scheduler": scheduler,
"steps": steps,
"cfg": cfg,
"denoise": denoise,
"seed": seed,
"trajectory_requirement": {
"sampler_name": sampler_name,
"scheduler": scheduler,
"steps": steps,
"cfg": cfg,
"denoise": denoise,
"seed": seed,
},
},
)
conflict_ids.extend(delta_conflicts)
delta_id = delta_signal["id"]
source_ids = [before_signal["id"], delta_signal["id"]]
source_proj_ids = [before_projection["id"], delta_projection["id"]]
num = delta_projection.get("numeric_summary", {})
delta_norm = num.get("delta_norm")
relative_delta = num.get("relative_delta")
missing_metadata = []
if not sampler_name:
missing_metadata.append("sampler_name")
if not scheduler:
missing_metadata.append("scheduler")
if steps == 0:
missing_metadata.append("steps")
if missing_metadata:
conflict = make_conflict(
CONFLICT_SAMPLER_METADATA_MISSING,
severity=SEV_LOW,
involved_signal_ids=[before_signal["id"], after_signal["id"]],
stage_position=stage_name,
suspected_cause="Sampler metadata is incomplete.",
observed_symptom=f"Missing or default metadata fields: {missing_metadata}",
suggested_response="Fill sampler_name, scheduler and steps if available.",
metadata={"missing_metadata": missing_metadata},
)
packet = add_conflict(packet, conflict)
conflict_ids.append(conflict["id"])
relation = make_event_relation(
relation_type=REL_TRANSFORMS_INTO,
source_signal_ids=source_ids,
target_signal_ids=[after_signal["id"]],
source_projection_ids=source_proj_ids,
target_projection_ids=[after_projection["id"]],
operator_name="EventSamplerBoundaryReport",
formula_meaning="sampler transformed input latent into output latent through observed delta",
local_strategy_id=f"{position}.latent_transition",
equality_status=EQ_UNKNOWN,
metadata={
"stage_name": stage_name,
"boundary_type": "sampler_summary",
"sampler_name": sampler_name,
"scheduler": scheduler,
"steps": steps,
"cfg": cfg,
"denoise": denoise,
"seed": seed,
"delta_signal_id": delta_id,
"delta_norm": delta_norm,
"relative_delta": relative_delta,
"trajectory_requirement": {
"sampler_name": sampler_name,
"scheduler": scheduler,
"steps": steps,
"cfg": cfg,
"denoise": denoise,
"seed": seed,
},
},
)
relation["conflict_ids"] = conflict_ids[:]
packet = add_relation(packet, relation)
packet, sstate = build_sstate_from_packet(packet, position=position, active_relation_ids=[relation["id"]])
packet = record_stage(
packet,
stage_name="EventSamplerBoundaryReport",
action="READ_SAMPLER_BOUNDARY",
observed_behavior="Sampler summary boundary read as latent_before + sampler_delta = latent_after",
input_signal_ids=[before_signal["id"], after_signal["id"]],
output_signal_ids=[x for x in [delta_id] if x],
projection_ids=[before_projection["id"], after_projection["id"]] + ([delta_projection["id"]] if delta_projection else []),
relation_ids=[relation["id"]],
sstate_ids=[sstate["id"]],
conflict_ids=conflict_ids,
formula_note="latent_before_sampler + sampler_delta = S_sampler_summary = latent_after_sampler",
metadata={
"sampler_name": sampler_name,
"scheduler": scheduler,
"steps": steps,
"cfg": cfg,
"denoise": denoise,
"seed": seed,
"delta_norm": delta_norm,
"relative_delta": relative_delta,
},
)
summary = (
f"Sampler boundary: before={before_signal['id']} after={after_signal['id']} "
f"delta={delta_id} relation={relation['id']} sstate={sstate['id']} "
f"sampler={sampler_name} scheduler={scheduler} steps={steps} cfg={cfg} denoise={denoise} "
f"delta_norm={delta_norm} relative_delta={relative_delta} conflicts={conflict_ids}"
)
return (packet, summary)
class EventBranchPacket:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"event_packet": ("EVENT_PACKET",),
"branch_label": ("STRING", {"default": "branch"}),
}
}
RETURN_TYPES = ("EVENT_PACKET", "STRING")
RETURN_NAMES = ("event_packet", "summary")
FUNCTION = "run"
CATEGORY = "Event Equality/Research"
def run(self, event_packet, branch_label):
packet = ensure_packet(event_packet)
packet_copy = copy.deepcopy(packet)
branch_id = now_run_id(prefix=str(branch_label or "branch"))
packet_copy.setdefault("metadata", {})["branch_id"] = branch_id
conflict = make_conflict(
CONFLICT_PACKET_BRANCH_CREATED,
severity=SEV_INFO,
stage_position="EventBranchPacket",
suspected_cause="Research EventPacket was explicitly branched.",
observed_symptom="A deepcopy branch was created to avoid mutable dict graph corruption.",
suggested_response="Use this branch independently from the source packet.",
metadata={"branch_id": branch_id, "branch_label": branch_label},
)
packet_copy = add_conflict(packet_copy, conflict)
packet_copy = record_stage(
packet_copy,
stage_name="EventBranchPacket",
action="PACKET_BRANCH_CREATED",
observed_behavior="EventPacket deep-copied for safe research branching",
conflict_ids=[conflict["id"]],
metadata={"branch_id": branch_id, "branch_label": branch_label},
)
return (packet_copy, f"Created EventPacket branch {branch_id}")
class EventSaveReportToFile:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"report": ("STRING", {"default": "", "multiline": True}),
"filename_prefix": ("STRING", {"default": "event_report"}),
"output_directory": ("STRING", {"default": ""}),
}
}
RETURN_TYPES = ("STRING", "STRING")
RETURN_NAMES = ("file_path", "report")
FUNCTION = "run"
CATEGORY = "Event Equality/Core"
def run(self, report, filename_prefix, output_directory):
safe_prefix = re.sub(r"[^a-zA-Z0-9_\\-]+", "_", str(filename_prefix or "event_report")).strip("_")
if not safe_prefix:
safe_prefix = "event_report"
if output_directory and str(output_directory).strip():
out_dir = Path(str(output_directory)).expanduser()
else:
out_dir = Path.cwd() / "output" / "event_equality_reports"
out_dir.mkdir(parents=True, exist_ok=True)
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
path = out_dir / f"{safe_prefix}_{timestamp}.md"
path.write_text(str(report or ""), encoding="utf-8")
return (str(path), report)
class EventCoreFrozen:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"mode": (["BASIC", "RESEARCH", "WAN"], {"default": "BASIC"}),
"report_detail": (["COMPACT", "STANDARD", "FULL"], {"default": "STANDARD"}),
"text": ("STRING", {"default": "", "multiline": True}),
"structured_strategy_text": ("STRING", {"default": "", "multiline": True}),
"sampler_name": ("STRING", {"default": ""}),
"scheduler": ("STRING", {"default": ""}),
"steps": ("INT", {"default": 0, "min": 0, "max": 10000}),
"cfg": ("FLOAT", {"default": 0.0, "min": -1000.0, "max": 1000.0, "step": 0.1}),
"denoise": ("FLOAT", {"default": 0.0, "min": 0.0, "max": 1.0, "step": 0.01}),
"seed": ("INT", {"default": 0, "min": 0, "max": 0xffffffffffffffff}),
"workflow_hint": ("STRING", {"default": "", "multiline": True}),
},
"optional": {
"latent_before": ("LATENT",),
"latent_after": ("LATENT",),
"image": ("IMAGE",),
"noise": ("LATENT",),
"conditioning": ("CONDITIONING",),
"decoded_image": ("IMAGE",),
}
}
RETURN_TYPES = ("EVENT_PACKET", "STRING", "STRING", "LATENT", "LATENT", "IMAGE", "LATENT", "CONDITIONING", "IMAGE")
RETURN_NAMES = (
"event_packet",
"report",
"short_summary",
"latent_before_out",
"latent_after_out",
"image_out",
"noise_out",
"conditioning_out",
"decoded_image_out",
)
FUNCTION = "run"
CATEGORY = "Event Equality/Core"
def _missing_route(self, packet, route_name, mode, severity=None):
sev = severity or (SEV_LOW if mode in ("RESEARCH", "WAN") else SEV_INFO)
conflict = make_conflict(
CONFLICT_FROZEN_ROUTE_MISSING,
severity=sev,
stage_position="EventCoreFrozen",
suspected_cause=f"Optional route {route_name} was not connected.",
observed_symptom=f"{route_name} route missing.",
suggested_response="Connect this input only if the diagnostic requires it.",
metadata={"route": route_name, "mode": mode},
)
return add_conflict(packet, conflict), conflict["id"]
def _read_structured_text(self, packet, structured_strategy_text):
# v0.1.1 accepts structured strategy as plain text, not parsing schema yet.
structured = {
"main_strategy": structured_strategy_text,
"stable_anchors": "",
"active_changes": "",
"motion_axis": "",
"contact_rule": "",
"reciprocal_reaction": "",
"deformation_rule": "",
"material_rule": "",
"support_rule": "",
"endpoint": "",
"loop_rule": "",
"forbidden_drift": "",
"negative_additions": "",
"priority_notes": "",
"temporal_scope": "",
"identity_anchors": "",
"camera_anchors": "",
"motion_region": "",
"stability_region": "",
"strategy_strength": 1.0,
}
return _read_signal(
packet,
TECH_TEXT,
SPACE_TEXT,
structured_strategy_text,
"Frozen_structured_strategy_text",
"StrategyCurrent",
"frozen_structured_strategy_route",
"TextStrategyReader",
metadata={"structured_strategy": structured, "text_length": len(str(structured_strategy_text or ""))},
)
def run(
self,
mode,
report_detail,
text,
structured_strategy_text,
sampler_name,
scheduler,
steps,
cfg,
denoise,
seed,
workflow_hint,
latent_before=None,
latent_after=None,
image=None,
noise=None,
conditioning=None,
decoded_image=None,
):
run_id = now_run_id(prefix="frozen")
packet = make_event_packet(metadata={
"created_by": "EventCoreFrozen",
"version": "0.1.1",
"run_id": run_id,
})
packet = record_stage(
packet,
stage_name="EventCoreFrozen",
action="INIT_FROZEN_NODE",
observed_behavior="Frozen node created local EventPacket",
metadata={"mode": mode, "report_detail": report_detail, "run_id": run_id},
)
relation_ids = []
conflict_ids = []
strategy_signal_ids = []
source_signal_ids = []
has_sampler_boundary = False
has_decode = False
has_structured = bool(str(structured_strategy_text or "").strip())
has_noise = noise is not None
input_signatures = build_input_signatures(
text=text,
structured_strategy_text=structured_strategy_text,
latent_before=latent_before,
latent_after=latent_after,
image=image,
noise=noise,
conditioning=conditioning,
decoded_image=decoded_image,
)
passthrough_status = build_passthrough_status(
latent_before=latent_before,
latent_after=latent_after,
image=image,
noise=noise,
conditioning=conditioning,
decoded_image=decoded_image,
)
# Missing-route diagnostics. These are not hard failures.
for route_name, present in [
("text", bool(str(text or "").strip()) or has_structured),
("latent_before", latent_before is not None),
("latent_after", latent_after is not None),
("image", image is not None),
("noise", noise is not None),
("conditioning", conditioning is not None),
("decoded_image", decoded_image is not None),
]:
if not present:
packet, cid = self._missing_route(packet, route_name, mode)
conflict_ids.append(cid)
# Text strategy.
if str(text or "").strip():
packet, sig, proj, conf = _read_signal(
packet,
TECH_TEXT,
SPACE_TEXT,
text,
"Frozen_text",
"StrategyCurrent",
"frozen_text_strategy_route",
"TextStrategyReader",
metadata={"text_length": len(str(text or ""))},
)
strategy_signal_ids.append(sig["id"])
conflict_ids.extend(conf)
# Structured strategy text.
if has_structured:
packet, sig, proj, conf = self._read_structured_text(packet, structured_strategy_text)
strategy_signal_ids.append(sig["id"])
conflict_ids.extend(conf)
# Source / image route.
if image is not None:
packet, sig, proj, conf = _read_signal(
packet,
TECH_IMAGE,
SPACE_IMAGE,
image,
"Frozen_image",
"OutcomePrevious",
"frozen_image_route",
"ImageOutcomeReader",
route_position="source",
)
source_signal_ids.append(sig["id"])
conflict_ids.extend(conf)
# Conditioning route.
conditioning_signal_id = None
if conditioning is not None:
packet, sig, proj, conf = _read_signal(
packet,
TECH_CONDITIONING,
SPACE_CONDITIONING,
conditioning,
"Frozen_conditioning",
"StrategyCarrier",
"frozen_conditioning_route",
"ConditioningStrategyReader",
)
conditioning_signal_id = sig["id"]
conflict_ids.extend(conf)
# Noise route.
noise_signal_id = None
if noise is not None:
raw_noise = noise.get("samples") if isinstance(noise, dict) and "samples" in noise else noise
packet, sig, proj, conf = _read_signal(
packet,
TECH_NOISE,
SPACE_NOISE,
raw_noise,
"Frozen_noise",
"StrategyCandidate",
"frozen_noise_route",
"NoisePossibilityReader",
metadata={"seed": seed, "noise_mode": "FROZEN_READ_EXISTING"},
)
noise_signal_id = sig["id"]
conflict_ids.extend(conf)
# Latent before / after + boundary.
before_signal_id = None
after_signal_id = None
before_projection_id = None
after_projection_id = None
if latent_before is not None:
packet, before_sig, before_proj, conf = _read_signal(
packet,
TECH_LATENT,
SPACE_LATENT,
latent_before,
"Frozen_latent_before",
"OutcomePrevious",
"frozen_latent_before_route",
"LatentEventReader",
route_position="before_sampler",
)
before_signal_id = before_sig["id"]
before_projection_id = before_proj["id"]
source_signal_ids.append(before_signal_id)
conflict_ids.extend(conf)
if latent_after is not None:
packet, after_sig, after_proj, conf = _read_signal(
packet,
TECH_LATENT,
SPACE_LATENT,
latent_after,
"Frozen_latent_after",
"OutcomeNext",
"frozen_latent_after_route",
"LatentEventReader",
route_position="after_sampler",
)
after_signal_id = after_sig["id"]
after_projection_id = after_proj["id"]
conflict_ids.extend(conf)
# Strategy relations.
for sid in strategy_signal_ids:
if conditioning_signal_id:
rel = make_event_relation(
relation_type=REL_GUIDES,
source_signal_ids=[sid],
target_signal_ids=[conditioning_signal_id],
source_projection_ids=get_projection_ids_for_signal(packet, sid),
target_projection_ids=get_projection_ids_for_signal(packet, conditioning_signal_id),
operator_name="EventCoreFrozen",
formula_meaning="strategy guides conditioning route",
local_strategy_id="S0_frozen.strategy_conditioning",
equality_status=EQ_UNKNOWN,
metadata={"frozen_auto_relation": True},
)
packet = add_relation(packet, rel)
relation_ids.append(rel["id"])
target_candidates = []
if after_signal_id:
target_candidates.append(after_signal_id)
target_candidates.extend(source_signal_ids)
for target_id in target_candidates:
if target_id == sid:
continue
rel = make_event_relation(
relation_type=REL_GUIDES,
source_signal_ids=[sid],
target_signal_ids=[target_id],
source_projection_ids=get_projection_ids_for_signal(packet, sid),
target_projection_ids=get_projection_ids_for_signal(packet, target_id),
operator_name="EventCoreFrozen",
formula_meaning="strategy makes a guiding claim about target route",
local_strategy_id="S0_frozen.strategy_target",
equality_status=EQ_UNKNOWN,
metadata={"frozen_auto_relation": True},
)
packet = add_relation(packet, rel)
relation_ids.append(rel["id"])
# Boundary relation if before/after latents exist.
if latent_before is not None and latent_after is not None and before_signal_id and after_signal_id:
before_samples = extract_latent_samples(latent_before)
delta, error = compute_tensor_delta(latent_before, latent_after)
source_ids = [before_signal_id]
source_proj_ids = [before_projection_id] if before_projection_id else []
delta_id = None
delta_norm = None
relative_delta = None
if delta is not None:
packet, delta_sig, delta_proj, conf = _read_signal(
packet,
TECH_DELTA,
SPACE_DELTA,
delta,
"Frozen_sampler_delta",
"ObservedBehaviorCurrent",
"frozen_delta_route",
"DeltaReader",
metadata={
"before_signal_id": before_signal_id,
"after_signal_id": after_signal_id,
"before_ref": before_samples,
"sampler_name": sampler_name,
"scheduler": scheduler,
"steps": steps,
"cfg": cfg,
"denoise": denoise,
"seed": seed,
},
)
conflict_ids.extend(conf)
delta_id = delta_sig["id"]
source_ids.append(delta_id)
source_proj_ids.append(delta_proj["id"])
delta_norm = delta_proj.get("numeric_summary", {}).get("delta_norm")
relative_delta = delta_proj.get("numeric_summary", {}).get("relative_delta")
else:
conflict = make_conflict(
CONFLICT_FROZEN_PARTIAL_OBSERVABILITY,
severity=SEV_LOW,
involved_signal_ids=[before_signal_id, after_signal_id],
stage_position="EventCoreFrozen",
suspected_cause="Frozen node could not compute latent delta.",
observed_symptom=str(error),
suggested_response="Check latent_before and latent_after shapes.",
metadata={"route": "sampler_boundary"},
)
packet = add_conflict(packet, conflict)
conflict_ids.append(conflict["id"])
rel = make_event_relation(
relation_type=REL_TRANSFORMS_INTO,
source_signal_ids=source_ids,
target_signal_ids=[after_signal_id],
source_projection_ids=source_proj_ids,
target_projection_ids=[after_projection_id] if after_projection_id else [],
operator_name="EventCoreFrozen",
formula_meaning="latent_before plus observed delta transforms into latent_after",
local_strategy_id="S0_frozen.sampler_boundary",
equality_status=EQ_UNKNOWN,
metadata={
"frozen_auto_relation": True,
"boundary_type": "sampler_summary",
"sampler_name": sampler_name,
"scheduler": scheduler,
"steps": steps,
"cfg": cfg,
"denoise": denoise,
"seed": seed,
"delta_signal_id": delta_id,
"delta_norm": delta_norm,
"relative_delta": relative_delta,
},
)
packet = add_relation(packet, rel)
relation_ids.append(rel["id"])
has_sampler_boundary = True
if noise_signal_id:
rel_noise = make_event_relation(
relation_type=REL_CONSTRAINS,
source_signal_ids=[noise_signal_id],
target_signal_ids=[after_signal_id],
source_projection_ids=get_projection_ids_for_signal(packet, noise_signal_id),
target_projection_ids=get_projection_ids_for_signal(packet, after_signal_id),
operator_name="EventCoreFrozen",
formula_meaning="noise possibility field constrains latent_after event target",
local_strategy_id="S0_frozen.noise_target",
equality_status=EQ_UNKNOWN,
metadata={"frozen_auto_relation": True},
)
packet = add_relation(packet, rel_noise)
relation_ids.append(rel_noise["id"])
# Decoded image route.
if decoded_image is not None:
packet, img_sig, img_proj, conf = _read_signal(
packet,
TECH_IMAGE,
SPACE_IMAGE,
decoded_image,
"Frozen_decoded_image",
"OutcomeNext",
"frozen_decoded_image_route",
"ImageOutcomeReader",
route_position="after_decode",
)
conflict_ids.extend(conf)
has_decode = True
if after_signal_id:
rel = make_event_relation(
relation_type=REL_EXPANDS_TO,
source_signal_ids=[after_signal_id],
target_signal_ids=[img_sig["id"]],
source_projection_ids=get_projection_ids_for_signal(packet, after_signal_id),
target_projection_ids=[img_proj["id"]],
operator_name="EventCoreFrozen",
formula_meaning="latent_after expands to decoded visible output",
local_strategy_id="S0_frozen.decode",
equality_status=EQ_UNKNOWN,
metadata={"frozen_auto_relation": True, "boundary_type": "decode"},
)
packet = add_relation(packet, rel)
relation_ids.append(rel["id"])
# Build SState.
if relation_ids:
packet, sstate = build_sstate_from_packet(packet, position="S0_frozen", active_relation_ids=relation_ids)
else:
packet, sstate = build_sstate_from_packet(packet, position="S0_frozen", active_relation_ids=[])
# Wan adapter only in WAN mode.
has_wan = False
if mode == "WAN":
packet, wan_result = apply_wan_adapter(
packet,
mode="BASIC",
hints={
"workflow_hint": workflow_hint or "EventCoreFrozen",
"has_high_sampler": False,
"has_low_sampler": False,
"has_temporal_module": False,
"has_lora_stack": False,
"has_accvid": False,
"has_lightx2v": False,
},
)
relation_ids.extend(wan_result.get("created_relation_ids", []))
conflict_ids.extend(wan_result.get("conflict_ids", []))
has_wan = True
observability = score_observability(
input_signatures,
has_structured=has_structured,
has_noise=has_noise,
has_sampler_boundary=has_sampler_boundary,
has_decode=has_decode,
has_wan=has_wan,
)
shared_targets = collect_shared_targets(packet)
packet.setdefault("metadata", {})["frozen"] = {
"enabled": True,
"mode": mode,
"report_detail": report_detail,
"run_id": run_id,
"semantic_mode": "simultaneous relational claims about one evolving event",
"event_program_status": {
"current_mode": "report_only",
"program_state": "built_for_audit_not_control",
"report_role": "visible audit/debug view of the internal Event Program",
"final_target_output": "image_or_video_or_latent_or_decoded_media",
"generation_control": "disabled_in_v0_1_1",
"passthrough": "exact",
"future_role": "generation_control_body_after_trace_and_correction_layers",
},
"input_signatures": input_signatures,
"passthrough_status": passthrough_status,
"observability": observability,
"semantic_relation_mechanics": {
"reading_mode": "simultaneous_event_claims",
"sstate_meaning": "intersection_of_active_relations",
"verdict_state": "EQ_UNKNOWN = relation recorded before judgment",
"shared_targets": shared_targets,
},
}
packet = record_stage(
packet,
stage_name="EventCoreFrozen",
action="BUILD_FROZEN_REPORT",
observed_behavior="Frozen node built internal Event Program audit report and returned exact passthrough objects",
input_signal_ids=list(packet.get("signals", {}).keys()),
relation_ids=relation_ids,
sstate_ids=[sstate["id"]] if sstate else [],
conflict_ids=conflict_ids,
formula_note="partial inputs -> positional roles -> simultaneous relations -> SState intersection -> Event Program audit report -> exact passthrough",
metadata={
"mode": mode,
"report_detail": report_detail,
"run_id": run_id,
"observability": observability,
},
)
report = build_markdown_report(packet)
short_summary = (
f"EventCoreFrozen v0.1.1-r2: mode={mode}, current=report_only, target=image/video, "
f"observability={observability.get('level')}:{observability.get('score')}, "
f"signals={len(packet.get('signals', {}))}, "
f"relations={len(packet.get('relations', {}))}, "
f"sstates={len(packet.get('sstates', {}))}, "
f"conflicts={len(packet.get('conflicts', {}))}, "
f"passthrough=exact"
)
return (
packet,
report,
short_summary,
latent_before,
latent_after,
image,
noise,
conditioning,
decoded_image,
)
class EventWanAdapter:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"event_packet": ("EVENT_PACKET",),
"wan_mode": (["BASIC", "HIGH_LOW", "RESEARCH"], {"default": "BASIC"}),
"workflow_hint": ("STRING", {"default": "", "multiline": True}),
"has_high_sampler": ("BOOLEAN", {"default": False}),
"has_low_sampler": ("BOOLEAN", {"default": False}),
"has_temporal_module": ("BOOLEAN", {"default": False}),
"has_lora_stack": ("BOOLEAN", {"default": False}),
"has_accvid": ("BOOLEAN", {"default": False}),
"has_lightx2v": ("BOOLEAN", {"default": False}),
}
}
RETURN_TYPES = ("EVENT_PACKET", "STRING")
RETURN_NAMES = ("event_packet", "wan_report")
FUNCTION = "run"
CATEGORY = "Event Equality/Adapters"
def run(
self,
event_packet,
wan_mode,
workflow_hint,
has_high_sampler,
has_low_sampler,
has_temporal_module,
has_lora_stack,
has_accvid,
has_lightx2v,
):
packet = ensure_packet(event_packet)
hints = {
"workflow_hint": workflow_hint,
"has_high_sampler": has_high_sampler,
"has_low_sampler": has_low_sampler,
"has_temporal_module": has_temporal_module,
"has_lora_stack": has_lora_stack,
"has_accvid": has_accvid,
"has_lightx2v": has_lightx2v,
}
packet, result = apply_wan_adapter(packet, mode=wan_mode, hints=hints)
packet = record_stage(
packet,
stage_name="EventWanAdapter",
action="APPLY_WAN_ADAPTER",
observed_behavior="WanAdapter annotated generic EventPacket with Wan route labels and diagnostics",
relation_ids=result.get("created_relation_ids", []),
conflict_ids=result.get("conflict_ids", []),
formula_note="WanAdapter labels first; it does not modify generation or generic relations.",
metadata={
"wan_mode": wan_mode,
"route_status": result.get("route_status", {}),
},
)
report = build_markdown_report(packet)
return (packet, report)
class EventCoreNodeAlpha:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"debug_mode": (DEBUG_MODES, {"default": DEBUG_BASIC}),
"adapter_mode": (["GENERIC", "WAN"], {"default": "GENERIC"}),
"strategy_mode": (["NONE", "FREE_TEXT", "STRUCTURED", "BOTH"], {"default": "FREE_TEXT"}),
"prompt_text": ("STRING", {"default": "", "multiline": True}),
"stable_anchors": ("STRING", {"default": "", "multiline": True}),
"active_changes": ("STRING", {"default": "", "multiline": True}),
"motion_axis": ("STRING", {"default": "", "multiline": True}),
"contact_rule": ("STRING", {"default": "", "multiline": True}),
"reciprocal_reaction": ("STRING", {"default": "", "multiline": True}),
"endpoint": ("STRING", {"default": "", "multiline": True}),
"forbidden_drift": ("STRING", {"default": "", "multiline": True}),
"read_sampler_boundary": ("BOOLEAN", {"default": False}),
"read_decode_boundary": ("BOOLEAN", {"default": False}),
"build_sstate": ("BOOLEAN", {"default": True}),
"sampler_name": ("STRING", {"default": ""}),
"scheduler": ("STRING", {"default": ""}),
"steps": ("INT", {"default": 0, "min": 0, "max": 10000}),
"cfg": ("FLOAT", {"default": 0.0, "min": -1000.0, "max": 1000.0, "step": 0.1}),
"denoise": ("FLOAT", {"default": 0.0, "min": 0.0, "max": 1.0, "step": 0.01}),
"seed": ("INT", {"default": 0, "min": 0, "max": 0xffffffffffffffff}),
},
"optional": {
"image": ("IMAGE",),
"latent_source": ("LATENT",),
"conditioning": ("CONDITIONING",),
"noise": ("LATENT",),
"latent_before_sampler": ("LATENT",),
"latent_after_sampler": ("LATENT",),
"decoded_image": ("IMAGE",),
}
}
RETURN_TYPES = ("EVENT_PACKET", "STRING", "STRING")
RETURN_NAMES = ("event_packet", "markdown_report", "short_summary")
FUNCTION = "run"
CATEGORY = "Event Equality/Core"
def run(
self,
debug_mode,
adapter_mode,
strategy_mode,
prompt_text,
stable_anchors,
active_changes,
motion_axis,
contact_rule,
reciprocal_reaction,
endpoint,
forbidden_drift,
read_sampler_boundary,
read_decode_boundary,
build_sstate,
sampler_name,
scheduler,
steps,
cfg,
denoise,
seed,
image=None,
latent_source=None,
conditioning=None,
noise=None,
latent_before_sampler=None,
latent_after_sampler=None,
decoded_image=None,
):
packet = make_event_packet(metadata={
"debug_mode": debug_mode,
"adapter_mode": adapter_mode,
"created_by": "EventCoreNodeAlpha",
"alpha_report_only": True,
})
packet = record_stage(
packet,
stage_name="EventCoreNodeAlpha",
action="INIT_ALPHA_BODY",
observed_behavior="EventCoreNodeAlpha created internal report-only EventPacket",
metadata={"debug_mode": debug_mode, "adapter_mode": adapter_mode, "strategy_mode": strategy_mode},
)
conflict_ids = []
relation_ids = []
strategy_signal_ids = []
source_signal_ids = []
sampler_relation_id = None
decode_relation_id = None
# Strategy route: free text.
if strategy_mode in ("FREE_TEXT", "BOTH") and str(prompt_text or "").strip():
packet, sig, proj, conf = _read_signal(
packet,
TECH_TEXT,
SPACE_TEXT,
prompt_text,
"Alpha_prompt_text",
"StrategyCurrent",
"alpha_prompt_route",
"TextStrategyReader",
metadata={"text_length": len(str(prompt_text or ""))},
)
strategy_signal_ids.append(sig["id"])
conflict_ids.extend(conf)
# Strategy route: compact structured strategy inside Alpha.
if strategy_mode in ("STRUCTURED", "BOTH"):
structured = {
"main_strategy": prompt_text,
"stable_anchors": stable_anchors,
"active_changes": active_changes,
"motion_axis": motion_axis,
"contact_rule": contact_rule,
"reciprocal_reaction": reciprocal_reaction,
"deformation_rule": "",
"material_rule": "",
"support_rule": "",
"endpoint": endpoint,
"loop_rule": "",
"forbidden_drift": forbidden_drift,
"negative_additions": "",
"priority_notes": "",
"strategy_strength": 1.0,
}
combined_text = "\\n".join([
str(prompt_text or ""),
str(stable_anchors or ""),
str(active_changes or ""),
str(motion_axis or ""),
str(contact_rule or ""),
str(reciprocal_reaction or ""),
str(endpoint or ""),
str(forbidden_drift or ""),
])
packet, sig, proj, conf = _read_signal(
packet,
TECH_TEXT,
SPACE_TEXT,
combined_text,
"Alpha_structured_strategy",
"StrategyCurrent",
"alpha_structured_strategy_route",
"TextStrategyReader",
metadata={"structured_strategy": structured, "text_length": len(combined_text)},
)
strategy_signal_ids.append(sig["id"])
conflict_ids.extend(conf)
if strategy_mode != "NONE" and not strategy_signal_ids:
conflict = make_conflict(
CONFLICT_ALPHA_ROUTE_MISSING,
severity=SEV_LOW,
stage_position="EventCoreNodeAlpha",
suspected_cause="Strategy mode requested but no strategy content was provided.",
observed_symptom="No strategy signal created.",
suggested_response="Provide prompt_text or choose strategy_mode=NONE.",
metadata={"route": "strategy"},
)
packet = add_conflict(packet, conflict)
conflict_ids.append(conflict["id"])
# Source image route.
if image is not None:
packet, sig, proj, conf = _read_signal(
packet,
TECH_IMAGE,
SPACE_IMAGE,
image,
"Alpha_source_image",
"OutcomePrevious",
"alpha_source_image_route",
"ImageOutcomeReader",
route_position="source",
)
source_signal_ids.append(sig["id"])
conflict_ids.extend(conf)
# Source latent route.
if latent_source is not None:
packet, sig, proj, conf = _read_signal(
packet,
TECH_LATENT,
SPACE_LATENT,
latent_source,
"Alpha_source_latent",
"StrategyCarrier",
"alpha_source_latent_route",
"LatentEventReader",
route_position="source",
)
source_signal_ids.append(sig["id"])
conflict_ids.extend(conf)
# Conditioning route.
conditioning_signal_id = None
if conditioning is not None:
packet, sig, proj, conf = _read_signal(
packet,
TECH_CONDITIONING,
SPACE_CONDITIONING,
conditioning,
"Alpha_conditioning",
"StrategyCarrier",
"alpha_conditioning_route",
"ConditioningStrategyReader",
)
conditioning_signal_id = sig["id"]
conflict_ids.extend(conf)
# Noise route.
noise_signal_id = None
if noise is not None:
raw_noise = noise.get("samples") if isinstance(noise, dict) and "samples" in noise else noise
packet, sig, proj, conf = _read_signal(
packet,
TECH_NOISE,
SPACE_NOISE,
raw_noise,
"Alpha_noise",
"StrategyCandidate",
"alpha_noise_route",
"NoisePossibilityReader",
metadata={"seed": seed, "noise_mode": "ALPHA_READ_EXISTING"},
)
noise_signal_id = sig["id"]
conflict_ids.extend(conf)
# Build simple strategy relations.
for sid in strategy_signal_ids:
if conditioning_signal_id:
rel = make_event_relation(
relation_type=REL_GUIDES,
source_signal_ids=[sid],
target_signal_ids=[conditioning_signal_id],
source_projection_ids=get_projection_ids_for_signal(packet, sid),
target_projection_ids=get_projection_ids_for_signal(packet, conditioning_signal_id),
operator_name="EventCoreNodeAlpha",
formula_meaning="strategy guides conditioning route",
local_strategy_id="S0_core_alpha.strategy_conditioning",
equality_status=EQ_UNKNOWN,
metadata={"alpha_auto_relation": True},
)
packet = add_relation(packet, rel)
relation_ids.append(rel["id"])
for source_id in source_signal_ids:
rel = make_event_relation(
relation_type=REL_GUIDES,
source_signal_ids=[sid],
target_signal_ids=[source_id],
source_projection_ids=get_projection_ids_for_signal(packet, sid),
target_projection_ids=get_projection_ids_for_signal(packet, source_id),
operator_name="EventCoreNodeAlpha",
formula_meaning="strategy guides source route interpretation",
local_strategy_id="S0_core_alpha.strategy_source",
equality_status=EQ_UNKNOWN,
metadata={"alpha_auto_relation": True},
)
packet = add_relation(packet, rel)
relation_ids.append(rel["id"])
# Sampler boundary route.
sampler_after_signal_id = None
if read_sampler_boundary:
if latent_before_sampler is not None and latent_after_sampler is not None:
packet, before_sig, before_proj, conf1 = _read_signal(
packet,
TECH_LATENT,
SPACE_LATENT,
latent_before_sampler,
"Alpha_sampler_before",
"OutcomePrevious",
"alpha_sampler_before_route",
"LatentEventReader",
route_position="before_sampler",
)
packet, after_sig, after_proj, conf2 = _read_signal(
packet,
TECH_LATENT,
SPACE_LATENT,
latent_after_sampler,
"Alpha_sampler_after",
"OutcomeNext",
"alpha_sampler_after_route",
"LatentEventReader",
route_position="after_sampler",
)
conflict_ids.extend(conf1 + conf2)
sampler_after_signal_id = after_sig["id"]
before_samples = extract_latent_samples(latent_before_sampler)
delta, error = compute_tensor_delta(latent_before_sampler, latent_after_sampler)
delta_id = None
delta_proj_id = None
source_ids = [before_sig["id"]]
source_proj_ids = [before_proj["id"]]
if delta is not None:
packet, delta_sig, delta_proj, conf3 = _read_signal(
packet,
TECH_DELTA,
SPACE_DELTA,
delta,
"Alpha_sampler_delta",
"ObservedBehaviorCurrent",
"alpha_sampler_delta_route",
"DeltaReader",
metadata={
"before_signal_id": before_sig["id"],
"after_signal_id": after_sig["id"],
"before_ref": before_samples,
"sampler_name": sampler_name,
"scheduler": scheduler,
"steps": steps,
"cfg": cfg,
"denoise": denoise,
"seed": seed,
},
)
conflict_ids.extend(conf3)
delta_id = delta_sig["id"]
delta_proj_id = delta_proj["id"]
source_ids.append(delta_id)
source_proj_ids.append(delta_proj_id)
delta_norm = delta_proj.get("numeric_summary", {}).get("delta_norm")
relative_delta = delta_proj.get("numeric_summary", {}).get("relative_delta")
else:
conflict = make_conflict(
CONFLICT_ALPHA_PARTIAL_INPUT,
severity=SEV_LOW,
involved_signal_ids=[before_sig["id"], after_sig["id"]],
stage_position="EventCoreNodeAlpha",
suspected_cause="Alpha sampler boundary could not compute delta.",
observed_symptom=str(error),
suggested_response="Check sampler before/after latent shapes.",
metadata={"route": "sampler_boundary"},
)
packet = add_conflict(packet, conflict)
conflict_ids.append(conflict["id"])
delta_norm = None
relative_delta = None
rel = make_event_relation(
relation_type=REL_TRANSFORMS_INTO,
source_signal_ids=source_ids,
target_signal_ids=[after_sig["id"]],
source_projection_ids=source_proj_ids,
target_projection_ids=[after_proj["id"]],
operator_name="EventCoreNodeAlpha",
formula_meaning="alpha sampler boundary: latent before plus delta becomes latent after",
local_strategy_id="S0_core_alpha.sampler_boundary",
equality_status=EQ_UNKNOWN,
metadata={
"alpha_auto_relation": True,
"boundary_type": "sampler_summary",
"sampler_name": sampler_name,
"scheduler": scheduler,
"steps": steps,
"cfg": cfg,
"denoise": denoise,
"seed": seed,
"delta_signal_id": delta_id,
"delta_norm": delta_norm,
"relative_delta": relative_delta,
},
)
packet = add_relation(packet, rel)
relation_ids.append(rel["id"])
sampler_relation_id = rel["id"]
if noise_signal_id:
rel_noise = make_event_relation(
relation_type=REL_CONSTRAINS,
source_signal_ids=[noise_signal_id],
target_signal_ids=[after_sig["id"]],
source_projection_ids=get_projection_ids_for_signal(packet, noise_signal_id),
target_projection_ids=[after_proj["id"]],
operator_name="EventCoreNodeAlpha",
formula_meaning="noise possibility field constrains sampler output route",
local_strategy_id="S0_core_alpha.noise_sampler",
equality_status=EQ_UNKNOWN,
metadata={"alpha_auto_relation": True},
)
packet = add_relation(packet, rel_noise)
relation_ids.append(rel_noise["id"])
else:
conflict = make_conflict(
CONFLICT_ALPHA_ROUTE_MISSING,
severity=SEV_LOW,
stage_position="EventCoreNodeAlpha",
suspected_cause="read_sampler_boundary enabled but one or both sampler latents are missing.",
observed_symptom="Sampler boundary route missing.",
suggested_response="Connect latent_before_sampler and latent_after_sampler or disable read_sampler_boundary.",
metadata={"route": "sampler_boundary"},
)
packet = add_conflict(packet, conflict)
conflict_ids.append(conflict["id"])
# Decode route.
if read_decode_boundary:
if latent_after_sampler is not None and decoded_image is not None:
if sampler_after_signal_id is None:
packet, lat_sig, lat_proj, conf = _read_signal(
packet,
TECH_LATENT,
SPACE_LATENT,
latent_after_sampler,
"Alpha_decode_latent",
"OutcomePrevious",
"alpha_decode_latent_route",
"LatentEventReader",
route_position="after_sampler",
)
conflict_ids.extend(conf)
sampler_after_signal_id = lat_sig["id"]
else:
lat_sig = packet["signals"][sampler_after_signal_id]
lat_proj_ids = get_projection_ids_for_signal(packet, sampler_after_signal_id)
packet, img_sig, img_proj, conf = _read_signal(
packet,
TECH_IMAGE,
SPACE_IMAGE,
decoded_image,
"Alpha_decoded_image",
"OutcomeNext",
"alpha_decoded_image_route",
"ImageOutcomeReader",
route_position="after_decode",
)
conflict_ids.extend(conf)
rel = make_event_relation(
relation_type=REL_EXPANDS_TO,
source_signal_ids=[sampler_after_signal_id],
target_signal_ids=[img_sig["id"]],
source_projection_ids=get_projection_ids_for_signal(packet, sampler_after_signal_id),
target_projection_ids=[img_proj["id"]],
operator_name="EventCoreNodeAlpha",
formula_meaning="latent after sampler expands to decoded image outcome",
local_strategy_id="S0_core_alpha.decode",
equality_status=EQ_UNKNOWN,
metadata={"alpha_auto_relation": True, "boundary_type": "decode"},
)
packet = add_relation(packet, rel)
relation_ids.append(rel["id"])
decode_relation_id = rel["id"]
else:
conflict = make_conflict(
CONFLICT_ALPHA_ROUTE_MISSING,
severity=SEV_LOW,
stage_position="EventCoreNodeAlpha",
suspected_cause="read_decode_boundary enabled but latent_after_sampler or decoded_image is missing.",
observed_symptom="Decode boundary route missing.",
suggested_response="Connect latent_after_sampler and decoded_image or disable read_decode_boundary.",
metadata={"route": "decode_boundary"},
)
packet = add_conflict(packet, conflict)
conflict_ids.append(conflict["id"])
if build_sstate:
active_rel_ids = relation_ids if relation_ids else None
packet, sstate = build_sstate_from_packet(packet, position="S0_core_alpha", active_relation_ids=active_rel_ids)
packet = record_stage(
packet,
stage_name="EventCoreNodeAlpha",
action="BUILD_ALPHA_SSTATE",
observed_behavior="Alpha body SState built from internal auto-relations",
input_signal_ids=sstate.get("active_signal_ids", []),
relation_ids=sstate.get("active_relation_ids", []),
sstate_ids=[sstate["id"]],
conflict_ids=conflict_ids,
formula_note="EventCoreNodeAlpha is report-only body: routes -> relations -> S0_core_alpha",
metadata={
"adapter_mode": adapter_mode,
"sampler_relation_id": sampler_relation_id,
"decode_relation_id": decode_relation_id,
},
)
# Adapter layer: label and diagnose after generic routes are built.
if adapter_mode == "WAN":
packet, wan_result = apply_wan_adapter(
packet,
mode="BASIC",
hints={
"workflow_hint": "EventCoreNodeAlpha",
"has_high_sampler": False,
"has_low_sampler": False,
"has_temporal_module": False,
"has_lora_stack": False,
"has_accvid": False,
"has_lightx2v": False,
},
)
relation_ids.extend(wan_result.get("created_relation_ids", []))
conflict_ids.extend(wan_result.get("conflict_ids", []))
packet = record_stage(
packet,
stage_name="EventCoreNodeAlpha",
action="BUILD_ALPHA_REPORT",
observed_behavior="Alpha report-only body assembled available routes",
conflict_ids=conflict_ids,
metadata={
"strategy_signals": strategy_signal_ids,
"source_signals": source_signal_ids,
"relation_ids": relation_ids,
"adapter_mode": adapter_mode,
},
)
report = build_markdown_report(packet)
short_summary = (
f"EventCoreNodeAlpha: signals={len(packet.get('signals', {}))} "
f"relations={len(packet.get('relations', {}))} "
f"sstates={len(packet.get('sstates', {}))} "
f"conflicts={len(packet.get('conflicts', {}))} "
f"adapter={adapter_mode}"
)
return (packet, report, short_summary)
class EventUNIReport:
@classmethod
def INPUT_TYPES(cls):
return {"required": {"event_packet": ("EVENT_PACKET",)}}
RETURN_TYPES = ("STRING", "EVENT_PACKET")
RETURN_NAMES = ("markdown_report", "event_packet")
FUNCTION = "run"
CATEGORY = "Event Equality/Core"
def run(self, event_packet):
packet = ensure_packet(event_packet)
packet = record_stage(
packet,
stage_name="EventUNIReport",
action="BUILD_REPORT",
observed_behavior="Markdown report built from EventPacket",
)
return (build_markdown_report(packet), packet)
def _event_core_cleanup_memory(label="cleanup"):
info = {
"label": label,
"gc": False,
"torch_cuda_empty_cache": False,
"torch_cuda_ipc_collect": False,
"error": None,
"barrier_policy": "preserve pass-through tensor/state; cleanup must not destroy StrategyCarrier",
"memory_before": _event_core_memory_snapshot(f"{label}_before"),
}
try:
import gc
gc.collect()
info["gc"] = True
try:
import torch
if torch.cuda.is_available():
torch.cuda.empty_cache()
info["torch_cuda_empty_cache"] = True
try:
torch.cuda.ipc_collect()
info["torch_cuda_ipc_collect"] = True
except Exception:
pass
except Exception as e:
info["error"] = f"torch cleanup skipped: {e}"
except Exception as e:
info["error"] = str(e)
info["memory_after"] = _event_core_memory_snapshot(f"{label}_after")
return info
def _event_core_memory_snapshot(label="snapshot"):
info = {
"label": str(label or "snapshot"),
"timestamp": datetime.now().isoformat(timespec="seconds"),
"perf_counter": float(time.perf_counter()),
}
try:
import psutil
process = psutil.Process(os.getpid())
mem = process.memory_info()
info["process_rss_mb"] = round(float(mem.rss) / (1024 * 1024), 3)
info["process_vms_mb"] = round(float(mem.vms) / (1024 * 1024), 3)
except Exception as e:
info["process_memory_error"] = str(e)
try:
import torch
cuda_available = bool(torch.cuda.is_available())
info["torch_cuda_available"] = cuda_available
if cuda_available:
info["cuda_device"] = int(torch.cuda.current_device())
info["cuda_allocated_mb"] = round(float(torch.cuda.memory_allocated()) / (1024 * 1024), 3)
info["cuda_reserved_mb"] = round(float(torch.cuda.memory_reserved()) / (1024 * 1024), 3)
info["cuda_max_allocated_mb"] = round(float(torch.cuda.max_memory_allocated()) / (1024 * 1024), 3)
info["cuda_max_reserved_mb"] = round(float(torch.cuda.max_memory_reserved()) / (1024 * 1024), 3)
except Exception as e:
info["cuda_memory_error"] = str(e)
return info
def _event_core_numeric_delta(before, after):
try:
if before is None or after is None:
return None
return round(float(after) - float(before), 6)
except Exception:
return None
def _event_core_memory_delta(before_snapshot, after_snapshot):
before = before_snapshot if isinstance(before_snapshot, dict) else {}
after = after_snapshot if isinstance(after_snapshot, dict) else {}
fields = (
"process_rss_mb",
"process_vms_mb",
"cuda_allocated_mb",
"cuda_reserved_mb",
"cuda_max_allocated_mb",
"cuda_max_reserved_mb",
)
delta = {}
for key in fields:
value = _event_core_numeric_delta(before.get(key), after.get(key))
if value is not None:
delta[key] = value
return delta
def _event_core_state_descriptor(value, name="state"):
desc = {
"name": str(name or "state"),
"type": type(value).__name__,
}
tensor = value
if isinstance(value, dict) and "samples" in value:
tensor = value.get("samples")
desc["carrier_type"] = "latent_dict"
try:
if hasattr(tensor, "shape"):
desc["shape"] = list(tensor.shape)
if hasattr(tensor, "dtype"):
desc["dtype"] = str(tensor.dtype)
if hasattr(tensor, "device"):
desc["device"] = str(tensor.device)
except Exception as e:
desc["descriptor_error"] = str(e)
return desc
class WanEventWorkflowCore:
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"primary_model": ("MODEL",),
"clip": ("CLIP",),
"vae": ("VAE",),
"source_image_file": (_event_core_list_input_images(),),
"positive_prompt": ("STRING", {"default": "", "multiline": True, "height": 180, "dynamicPrompts": False}),
"negative_prompt": ("STRING", {"default": "", "multiline": True, "height": 180, "dynamicPrompts": False}),
"event_strategy": ("STRING", {"default": "", "multiline": True, "height": 90}),
"generation_target": (["VIDEO", "IMAGE", "AUTO"], {"default": "VIDEO"}),
"terminal_mode": ("BOOLEAN", {"default": True}),
"enable_continuation_outputs": ("BOOLEAN", {"default": False}),
"execution_mode": (["RUN"], {"default": "RUN"}),
"branch_mode": (["AUTO", "SINGLE", "DUAL_HIGH_LOW"], {"default": "AUTO"}),
"cascade_count": ("INT", {"default": 1, "min": 1, "max": 5}),
"pause_after_cascade_1": ("BOOLEAN", {"default": False}),
"pause_after_cascade_2": ("BOOLEAN", {"default": False}),
"pause_after_cascade_3": ("BOOLEAN", {"default": False}),
"pause_after_cascade_4": ("BOOLEAN", {"default": False}),
"resume_frame_index": ("INT", {"default": -1, "min": -1, "max": 4096}),
"cascade_mode": (["SOLO_1", "CASCADE_2", "CASCADE_3", "CASCADE_4", "CASCADE_5"], {"default": "SOLO_1"}),
"frames_per_cascade": ("INT", {"default": 49, "min": 1, "max": 4096}),
"width": ("INT", {"default": 608, "min": 16, "max": 8192, "step": 8}),
"height": ("INT", {"default": 416, "min": 16, "max": 8192, "step": 8}),
"frames": ("INT", {"default": 49, "min": 1, "max": 4096}),
"batch_size": ("INT", {"default": 1, "min": 1, "max": 4096}),
"fps": ("INT", {"default": 16, "min": 1, "max": 240}),
"seed": ("INT", {"default": 359, "min": 0, "max": 0xffffffffffffffff}),
"sampler_name": ("STRING", {"default": "euler"}),
"scheduler": ("STRING", {"default": "simple"}),
"global_steps": ("INT", {"default": 4, "min": 0, "max": 10000}),
"primary_cfg": ("FLOAT", {"default": 1.0, "min": -1000.0, "max": 1000.0, "step": 0.01}),
"secondary_cfg": ("FLOAT", {"default": 1.0, "min": -1000.0, "max": 1000.0, "step": 0.01}),
"primary_start_step": ("INT", {"default": 0, "min": 0, "max": 10000}),
"primary_end_step": ("INT", {"default": 3, "min": 0, "max": 10000}),
"secondary_start_step": ("INT", {"default": 3, "min": 0, "max": 10000}),
"secondary_end_step": ("INT", {"default": 4, "min": 0, "max": 10000}),
"primary_sd3_shift": ("FLOAT", {"default": 8.0, "min": -1000.0, "max": 1000.0, "step": 0.01}),
"secondary_sd3_shift": ("FLOAT", {"default": 8.0, "min": -1000.0, "max": 1000.0, "step": 0.01}),
"decode_tile_size": ("INT", {"default": 512, "min": 64, "max": 8192, "step": 8}),
"decode_overlap": ("INT", {"default": 64, "min": 0, "max": 8192, "step": 8}),
"decode_temporal_size": ("INT", {"default": 32, "min": 1, "max": 4096}),
"decode_temporal_overlap": ("INT", {"default": 12, "min": 0, "max": 4096}),
"image_upscale_method": (["nearest-exact", "nearest", "bilinear", "area", "bicubic", "lanczos"], {"default": "nearest-exact"}),
"image_crop": (["disabled", "center"], {"default": "disabled"}),
"cleanup_timing": ([
"NONE", "BEFORE_GENERATION", "BETWEEN_SAMPLERS", "AFTER_GENERATION", "BEFORE_AND_AFTER", "ALL",
"none", "before_generation", "between_samplers", "after_generation", "before_and_after", "all",
], {"default": "ALL"}),
"stage_delay_seconds": ("FLOAT", {"default": 0.0, "min": 0.0, "max": 5.0, "step": 0.1}),
"use_custom_cleanup_nodes": ("BOOLEAN", {"default": True}),
"save_video": ("BOOLEAN", {"default": True}),
"video_format": (["video/h264-mp4", "video/h265-mp4", "image/webp", "image/gif"], {"default": "video/h264-mp4"}),
"force_vhs_video_combine": ("BOOLEAN", {"default": True}),
"save_frames": ("BOOLEAN", {"default": False}),
"save_report": ("BOOLEAN", {"default": True}),
"output_target": (["COMFY_OUTPUT"], {"default": "COMFY_OUTPUT"}),
"save_output_image": ("BOOLEAN", {"default": False}),
"save_prefix": ("STRING", {"default": "wansolo"}),
"output_folder_mode": (["DEFAULT", "PICKER", "CUSTOM"], {"default": "DEFAULT"}),
"output_folder": (["default"],),
"custom_output_folder": ("STRING", {"default": ""}),
"report_detail": (["COMPACT", "STANDARD", "FULL"], {"default": "STANDARD"}),
},
"optional": {
"secondary_model": ("MODEL",),
"image": ("IMAGE",),
"mask": ("MASK",),
}
}
# Terminal output node.
# IMPORTANT r15:
# No IMAGE/LATENT outputs here. IMAGE outputs caused ComfyUI/PreviewImage-style
# behavior and made the node look like an image-producing endpoint.
# Video/image files are saved internally; strings are the final visible outputs.
RETURN_TYPES = ("STRING", "STRING", "STRING", "STRING")
RETURN_NAMES = (
"status",
"saved_video_path",
"saved_report_path",
"report",
)
FUNCTION = "run"
CATEGORY = "Event Equality/Event Horizon"
OUTPUT_NODE = True
def _placeholder_image(self, width=64, height=64):
try:
import torch
return torch.zeros((1, int(height), int(width), 3), dtype=torch.float32)
except Exception:
return None
def _ensure_preview_image(self, image, fallback_width=64, fallback_height=64):
if image is None:
return self._placeholder_image(fallback_width, fallback_height)
try:
import torch
if hasattr(image, "dim"):
if image.dim() == 3:
return image.unsqueeze(0)
return image
except Exception:
pass
return image
def _representative_preview_frame(self, image, fallback_width=64, fallback_height=64, mode="last"):
"""
PreviewImage saves every image in a batch. A decoded video is usually an IMAGE batch
shaped [frames, H, W, C]. Sending it directly to PreviewImage creates 65+ PNG files.
This function reduces any IMAGE batch to one representative frame for UI preview.
"""
img = self._ensure_preview_image(image, fallback_width, fallback_height)
if img is None:
return self._placeholder_image(fallback_width, fallback_height)
try:
if hasattr(img, "dim") and img.dim() == 4 and img.shape[0] > 1:
idx = 0 if str(mode).lower() == "first" else int(img.shape[0]) - 1
return img[idx:idx + 1]
except Exception:
pass
return img
def _load_image_from_upload(self, source_image_file, records):
if not source_image_file or str(source_image_file).lower() in ("none", "null", ""):
records.append({"stage": "source_image_upload", "status": "skipped", "reason": "no source_image_file"})
return None
try:
import folder_paths
import numpy as np
import torch
from PIL import Image, ImageOps, ImageSequence
image_path = folder_paths.get_annotated_filepath(source_image_file)
img = Image.open(image_path)
frames = []
for frame in ImageSequence.Iterator(img):
frame = ImageOps.exif_transpose(frame)
if frame.mode == "I":
frame = frame.point(lambda i: i * (1 / 255))
frame = frame.convert("RGB")
arr = np.array(frame).astype(np.float32) / 255.0
frames.append(torch.from_numpy(arr)[None,])
break
if not frames:
raise RuntimeError("No frame loaded from source_image_file")
out = torch.cat(frames, dim=0)
records.append({"stage": "source_image_upload", "status": "ok", "file": str(source_image_file), "shape": list(out.shape)})
return out
except Exception as e:
records.append({"stage": "source_image_upload", "status": "failed", "file": str(source_image_file), "error": str(e)})
return None
def _make_ui_previews(self, source_preview, result_preview, save_prefix,
records, include_result_preview=False):
"""
r15: no internal PreviewImage calls.
Reason:
PreviewImage writes PNG files. The main terminal node must not create image
outputs or preview dumps during video generation. Source selection is handled
by the image picker widget; final result is saved by VHS/video export.
"""
records.append({
"stage": "preview_ui",
"status": "disabled",
"preview_policy": "no_internal_previewimage_calls_in_terminal_node",
"reason": "avoid PNG dumps and avoid making video workflow behave like image output",
})
return []
def _save_pause_frames_to_temp(self, frames):
import folder_paths
import os
import random
from PIL import Image
import numpy as np
ui_images = []
try:
temp_dir = folder_paths.get_temp_directory()
total_f = frames.shape[0]
# Wan2.1 valid frames must be 4*k + 1.
# We want up to 7 valid frames from the tail.
valid_indices = []
for i in range(total_f):
# i is 0-indexed. The frame count is i+1.
if (i + 1 - 1) % 4 == 0:
valid_indices.append(i)
# Take the last 7 valid indices
selected_indices = valid_indices[-7:] if len(valid_indices) >= 7 else valid_indices
prefix = "cascade_preview_" + str(random.randint(100000, 999999))
for idx, frame_idx in enumerate(selected_indices):
frame = frames[frame_idx]
img_array = (255. * frame.cpu().numpy()).clip(0, 255).astype(np.uint8)
img = Image.fromarray(img_array)
filename = f"{prefix}_{idx}.png"
filepath = os.path.join(temp_dir, filename)
img.save(filepath)
ui_images.append({
"filename": filename,
"subfolder": "",
"type": "temp",
"resume_index": frame_idx + 1 # 1-indexed for the UI resume_frame_index
})
except Exception as e:
print(f"Failed to save pause frames: {e}")
return ui_images
def _import_comfy_nodes(self):
return importlib.import_module("nodes")
def _get_node_class(self, class_name):
comfy_nodes = self._import_comfy_nodes()
cls = getattr(comfy_nodes, class_name, None)
if cls is None and hasattr(comfy_nodes, "NODE_CLASS_MAPPINGS"):
cls = comfy_nodes.NODE_CLASS_MAPPINGS.get(class_name)
if cls is None:
raise RuntimeError(f"ComfyUI node class {class_name} not found")
return cls
def _call_node_method(self, class_name, method_names, *args, **kwargs):
cls = self._get_node_class(class_name)
inst = cls()
if isinstance(method_names, str):
method_names = [method_names]
fn_name = getattr(cls, "FUNCTION", None) or getattr(inst, "FUNCTION", None)
candidate_names = []
for name in method_names:
if name not in candidate_names:
candidate_names.append(name)
if fn_name and fn_name not in candidate_names:
candidate_names.append(fn_name)
for fallback in ("run", "execute", "process", "generate", "encode", "decode", "sample", "patch", "upscale", "save_images", "combine_video", "combine"):
if fallback not in candidate_names:
candidate_names.append(fallback)
last_error = None
tried = []
for name in candidate_names:
if hasattr(inst, name):
tried.append(name)
method = getattr(inst, name)
try:
return method(*args, **kwargs)
except TypeError as e:
last_error = e
try:
sig = inspect.signature(method)
filtered = {k: v for k, v in kwargs.items() if k in sig.parameters}
return method(*args, **filtered)
except Exception as e2:
last_error = e2
except Exception as e:
last_error = e
raise RuntimeError(
f"Could not call {class_name}. tried={tried}, requested={method_names}, "
f"class_FUNCTION={fn_name}, last_error={last_error}"
)
def _describe_raw_output(self, obj, max_items=80):
try:
attrs = []
for name in dir(obj):
if name.startswith("__"):
continue
attrs.append(name)
if len(attrs) >= max_items:
break
return {
"type": f"{type(obj).__module__}.{type(obj).__name__}",
"repr": str(obj)[:500],
"dir_sample": attrs,
"has_dict": hasattr(obj, "__dict__"),
"dict_keys": list(getattr(obj, "__dict__", {}).keys())[:max_items] if hasattr(obj, "__dict__") else [],
}
except Exception as e:
return {"type": str(type(obj)), "describe_error": str(e)}
def _node_output_to_sequence(self, result, preferred_names=None, stage="node_output", records=None):
preferred_names = preferred_names or []
if isinstance(result, tuple) or isinstance(result, list):
return list(result)
if isinstance(result, dict):
if "result" in result:
return self._node_output_to_sequence(result["result"], preferred_names, stage, records)
values = []
for name in preferred_names:
if name in result:
values.append(result[name])
return values if values else list(result.values())
values = []
for name in preferred_names:
if hasattr(result, name):
try:
values.append(getattr(result, name))
except Exception:
pass
if values:
if records is not None:
records.append({"stage": f"{stage}_node_output_unwrapped", "status": "ok", "mode": "preferred_attrs", "names": preferred_names})
return values
for attr in ("result", "results", "value", "values", "output", "outputs", "data"):
if hasattr(result, attr):
try:
obj = getattr(result, attr)
seq = self._node_output_to_sequence(obj, preferred_names, stage, records)
if seq:
if records is not None:
records.append({"stage": f"{stage}_node_output_unwrapped", "status": "ok", "mode": attr})
return seq
except Exception:
pass
try:
d = getattr(result, "__dict__", None)
if isinstance(d, dict) and d:
values = []
for name in preferred_names:
if name in d:
values.append(d[name])
if values:
return values
# common private storages in API output wrappers
for storage_name in ("_values", "_outputs", "_data", "_result", "_results"):
if storage_name in d:
try:
return self._node_output_to_sequence(d[storage_name], preferred_names, stage, records)
except Exception:
pass
plain = [v for k, v in d.items() if not k.startswith("_") and not callable(v)]
if plain:
return plain
private_plain = [v for k, v in d.items() if not callable(v)]
if private_plain:
return private_plain
except Exception:
pass
indexed = []
try:
for i in range(8):
try:
indexed.append(result[i])
except Exception:
break
if indexed:
if records is not None:
records.append({"stage": f"{stage}_node_output_unwrapped", "status": "ok", "mode": "getitem"})
return indexed
except Exception:
pass
try:
seq = list(result)
if seq:
if records is not None:
records.append({"stage": f"{stage}_node_output_unwrapped", "status": "ok", "mode": "iter"})
return seq
except Exception:
pass
desc = self._describe_raw_output(result)
if records is not None:
records.append({"stage": f"{stage}_node_output_unwrap_failed", "status": "failed", "raw_output": desc})
raise RuntimeError(f"{stage} returned unrecognized NodeOutput/container: {desc}")
def _first_output(self, result, preferred_names=None, stage="node_output", records=None):
seq = self._node_output_to_sequence(result, preferred_names=preferred_names, stage=stage, records=records)
if not seq:
raise RuntimeError(f"{stage} returned no outputs")
return seq[0]
def _extract_tensors_from_obj(self, obj, prefix="root", max_items=16):
tensors = []
try:
if hasattr(obj, "detach") and hasattr(obj, "shape"):
tensors.append((prefix, obj))
return tensors
if isinstance(obj, dict):
for k, v in obj.items():
if len(tensors) >= max_items:
break
tensors.extend(self._extract_tensors_from_obj(v, f"{prefix}.{k}", max_items=max_items-len(tensors)))
return tensors
if isinstance(obj, (list, tuple)):
for i, v in enumerate(obj):
if len(tensors) >= max_items:
break
tensors.extend(self._extract_tensors_from_obj(v, f"{prefix}[{i}]", max_items=max_items-len(tensors)))
return tensors
except Exception:
pass
return tensors
def _finite_guard(self, obj, records, stage, strict=True):
"""
Detect NaN/Inf tensors before black-video save.
Returns True if all discovered tensors are finite.
"""
try:
import torch
tensors = self._extract_tensors_from_obj(obj)
if not tensors:
records.append({"stage": stage, "status": "no_tensor_found"})
return True
all_ok = True
reports = []
for name, t in tensors:
try:
td = t.detach()
finite = torch.isfinite(td)
total = int(td.numel())
finite_count = int(finite.sum().item()) if total else 0
nonfinite_count = total - finite_count
finite_ratio = float(finite_count / total) if total else 1.0
reports.append({
"tensor": name,
"shape": list(td.shape),
"dtype": str(td.dtype),
"device": str(td.device),
"total": total,
"nonfinite_count": nonfinite_count,
"finite_ratio": finite_ratio,
})
if nonfinite_count > 0:
all_ok = False
except Exception as e:
reports.append({"tensor": name, "status": "inspect_failed", "error": str(e)})
records.append({"stage": stage, "status": "ok" if all_ok else "nonfinite_detected", "reports": reports})
if strict and not all_ok:
raise RuntimeError(f"{stage}: non-finite tensor values detected; refusing to save black/corrupt video")
return all_ok
except Exception as e:
records.append({"stage": stage, "status": "failed", "error": str(e)})
if strict:
raise
return False
def _encode_text(self, clip, text, records=None, label="TextEncode"):
result = self._call_node_method("CLIPTextEncode", ["encode"], clip=clip, text=str(text or ""))
conditioning = self._first_output(result, preferred_names=["conditioning", "CONDITIONING"], stage=f"Event{label}")
if records is not None:
self._event_universal_stage_math(
records,
f"Event{label}",
input_state=text,
output_state=conditioning,
observed_behavior="text prompt encoded into conditioning strategy",
formula_role="TEXT/CLIP -> CONDITIONING as NumericStrategy",
route_id=f"route_{str(label).lower()}",
next_requirement="WanImageToVideo and sampler require conditioning compatible with model route",
control_mode="REPORT_ONLY",
metadata={"text_length": len(str(text or ""))},
)
return conditioning
def _scale_image(self, image, width, height, method, crop, records):
if image is None:
records.append({"stage": "EventImageScaleStart", "status": "skipped", "reason": "no image"})
return None
try:
result = self._call_node_method(
"ImageScale",
["upscale"],
image=image,
upscale_method=str(method or "nearest-exact"),
width=int(width),
height=int(height),
crop=str(crop or "disabled"),
)
scaled = self._first_output(result, preferred_names=["image", "images", "IMAGE"], stage="EventImageScaleStart", records=records)
records.append({"stage": "EventImageScaleStart", "status": "ok", "width": width, "height": height})
self._event_universal_stage_math(
records,
"EventImageScaleStart",
input_state=image,
output_state=scaled,
observed_behavior="source image resized/cropped into model-compatible source outcome",
formula_role="IMAGE SourceAnchor -> IMAGE scaled OutcomePrevious",
route_id="route_source_image",
next_requirement="WanImageToVideo requires scaled image, width, height, frame count and conditioning",
control_mode="REPORT_ONLY",
metadata={"width": int(width), "height": int(height), "method": str(method or "nearest-exact"), "crop": str(crop or "disabled")},
)
return scaled
except Exception as e:
records.append({"stage": "EventImageScaleStart", "status": "failed_no_fake_success", "error": str(e)})
raise
def _apply_sd3_shift(self, model, shift, label, records):
try:
shift_value = float(shift or 0.0)
except Exception:
shift_value = 0.0
if abs(shift_value) < 1e-9:
records.append({
"stage": f"EventModelShift_{label}",
"status": "skipped_zero_shift",
"shift": shift_value,
"formula": "shift=0 -> original model passthrough; do not apply ModelSamplingSD3 patch",
})
self._event_universal_stage_math(
records,
f"EventModelShift_{label}",
input_state=model,
output_state=model,
observed_behavior="shift value is zero; model route passes through unchanged",
formula_role="MODEL Operator -> MODEL Operator passthrough",
route_id=f"route_model_shift_{label}",
next_requirement="sampler requires model route compatible with current branch",
control_mode="REPORT_ONLY",
metadata={"shift": shift_value, "applied": False},
)
return model
try:
result = self._call_node_method("ModelSamplingSD3", ["patch"], model=model, shift=shift_value)
patched = self._first_output(result, preferred_names=["model", "MODEL"], stage=f"EventModelShift_{label}", records=records)
records.append({"stage": f"EventModelShift_{label}", "status": "ok", "shift": shift_value, "formula": "model + shift behavior = EventSingularity_model_shift = shifted model"})
self._event_universal_stage_math(
records,
f"EventModelShift_{label}",
input_state=model,
output_state=patched,
observed_behavior="ModelSamplingSD3 shift patches model route for sampler branch",
formula_role="MODEL Operator + shift behavior -> MODEL shifted Operator",
route_id=f"route_model_shift_{label}",
next_requirement="sampler requires shifted model route compatible with branch window",
control_mode="REPORT_ONLY",
metadata={"shift": shift_value, "applied": True},
)
return patched
except Exception as e:
records.append({"stage": f"event_model_shift_{label}", "status": "failed", "shift": shift_value, "error": str(e)})
raise
def _wan_image_to_video(self, positive, negative, vae, start_image, width, height, frames, batch_size, records):
try:
result = self._call_node_method(
"WanImageToVideo",
["encode", "generate", "process", "run"],
positive=positive,
negative=negative,
vae=vae,
clip_vision_output=None,
start_image=start_image,
width=int(width),
height=int(height),
length=int(frames),
batch_size=int(batch_size),
)
except Exception as first_error:
records.append({"stage": "EventWanImageToVideoSeed_first_signature_failed", "status": "retrying", "error": str(first_error)})
result = self._call_node_method(
"WanImageToVideo",
["generate", "process", "run", "encode"],
positive=positive,
negative=negative,
vae=vae,
clip_vision_output=None,
image=start_image,
start_image=start_image,
width=int(width),
height=int(height),
frames=int(frames),
length=int(frames),
batch_size=int(batch_size),
)
seq = self._node_output_to_sequence(
result,
preferred_names=["positive", "negative", "latent", "conditioning", "conditioning_positive", "conditioning_negative", "samples"],
stage="EventWanImageToVideoSeed",
records=records,
)
if len(seq) < 3:
raise RuntimeError(
f"WanImageToVideo returned fewer than 3 usable outputs after unwrapping: "
f"count={len(seq)}, raw={self._describe_raw_output(result)}"
)
records.append({
"stage": "EventWanImageToVideoSeed",
"status": "ok",
"width": width,
"height": height,
"frames": frames,
"formula": "source image + conditioning + VAE behavior = EventSingularity_i2v = video latent seed",
"event_node": "EventWanImageToVideo",
"unwrapped_output_count": len(seq),
})
self._event_universal_stage_math(
records,
"EventWanImageToVideoSeed",
input_state=start_image,
output_state=seq[2],
observed_behavior="scaled source image, positive conditioning, VAE and dimensions created initial video latent seed",
formula_role="IMAGE + CONDITIONING + VAE -> LATENT seed OutcomePrevious",
route_id="route_wan_i2v_seed",
next_requirement="High sampler requires latent seed as previous outcome and noise-enabled first denoise window",
control_mode="REPORT_ONLY",
metadata={"width": int(width), "height": int(height), "frames": int(frames), "batch_size": int(batch_size)},
)
return seq[0], seq[1], seq[2]
def _low_level_sampler_operation(self, *, model, positive, negative, latent, seed, steps, cfg, sampler_name, scheduler, start_at_step, end_at_step, add_noise, return_leftover_noise):
result = self._call_node_method(
"KSamplerAdvanced",
["sample"],
model=model,
add_noise=add_noise,
noise_seed=int(seed),
steps=int(steps),
cfg=float(cfg),
sampler_name=str(sampler_name or "euler"),
scheduler=str(scheduler or "simple"),
positive=positive,
negative=negative,
latent_image=latent,
start_at_step=int(start_at_step),
end_at_step=int(end_at_step),
return_with_leftover_noise=return_leftover_noise,
)
return self._first_output(result, preferred_names=["latent", "samples", "LATENT"], stage="EventSamplerLowLevel")
def _tensor_from_latent_like(self, obj):
try:
if isinstance(obj, dict):
if "samples" in obj:
return obj["samples"]
for v in obj.values():
t = self._tensor_from_latent_like(v)
if t is not None:
return t
if hasattr(obj, "detach") and hasattr(obj, "shape"):
return obj
if isinstance(obj, (list, tuple)) and obj:
return self._tensor_from_latent_like(obj[0])
except Exception:
return None
return None
def _math_tensor_summary(self, obj, records, stage, reference=None, strict=False):
"""
r34 measurement-only math report.
Does not modify tensors. Records finite status, norm, mean/std, optional delta and relative delta.
"""
try:
import torch
t = self._tensor_from_latent_like(obj)
if t is None:
rec = {"stage": stage, "status": "unavailable", "reason": "no_tensor_found"}
records.append(rec)
return rec
tf = t.detach().float()
finite_mask = torch.isfinite(tf)
finite_ok = bool(finite_mask.all().item())
finite_ratio = float(finite_mask.float().mean().item()) if tf.numel() else 0.0
safe = torch.nan_to_num(tf, nan=0.0, posinf=0.0, neginf=0.0)
rec = {
"stage": stage,
"status": "ok" if finite_ok else "nonfinite",
"shape": list(tf.shape),
"dtype": str(t.dtype) if hasattr(t, "dtype") else str(type(t)),
"device": str(t.device) if hasattr(t, "device") else "unknown",
"finite_ok": finite_ok,
"finite_ratio": finite_ratio,
"mean": float(safe.mean().item()) if safe.numel() else 0.0,
"std": float(safe.std().item()) if safe.numel() > 1 else 0.0,
"min": float(safe.min().item()) if safe.numel() else 0.0,
"max": float(safe.max().item()) if safe.numel() else 0.0,
"norm": float(torch.linalg.vector_norm(safe).item()) if safe.numel() else 0.0,
}
if reference is not None:
rt = self._tensor_from_latent_like(reference)
if rt is not None:
rf = rt.detach().float()
rsafe = torch.nan_to_num(rf, nan=0.0, posinf=0.0, neginf=0.0)
if rsafe.shape == safe.shape:
delta = safe - rsafe
ref_norm = torch.linalg.vector_norm(rsafe).item() if rsafe.numel() else 0.0
delta_norm = torch.linalg.vector_norm(delta).item() if delta.numel() else 0.0
rec.update({
"delta_norm": float(delta_norm),
"reference_norm": float(ref_norm),
"relative_delta": float(delta_norm / (ref_norm + 1e-12)),
"delta_mean": float(delta.mean().item()) if delta.numel() else 0.0,
"delta_std": float(delta.std().item()) if delta.numel() > 1 else 0.0,
"delta_min": float(delta.min().item()) if delta.numel() else 0.0,
"delta_max": float(delta.max().item()) if delta.numel() else 0.0,
})
else:
rec.update({
"delta_status": "shape_mismatch",
"reference_shape": list(rf.shape),
})
records.append(rec)
if strict and not finite_ok:
raise RuntimeError(f"{stage} contains NaN/Inf; finite_ratio={finite_ratio}")
return rec
except Exception as e:
rec = {"stage": stage, "status": "failed", "error": str(e)}
records.append(rec)
if strict:
raise
return rec
def _frame_motion_math(self, frames, records, stage):
"""
r34: measure temporal frame-to-frame motion after decode.
Works on decoded frame tensors [T,H,W,C] or similar without modifying frames.
"""
try:
import torch
t = self._tensor_from_latent_like(frames)
if t is None:
records.append({"stage": stage, "status": "unavailable", "reason": "no_tensor_found"})
return None
tf = torch.nan_to_num(t.detach().float(), nan=0.0, posinf=0.0, neginf=0.0)
if tf.dim() < 2 or tf.shape[0] < 2:
records.append({"stage": stage, "status": "unavailable", "reason": "not_enough_frames", "shape": list(tf.shape)})
return None
delta = tf[1:] - tf[:-1]
flat = delta.reshape(delta.shape[0], -1)
norms = torch.linalg.vector_norm(flat, dim=1)
mean_abs = delta.abs().reshape(delta.shape[0], -1).mean(dim=1)
avg = float(norms.mean().item())
mx = float(norms.max().item())
mn = float(norms.min().item())
std = float(norms.std().item()) if norms.numel() > 1 else 0.0
spike_ratio = float(mx / (avg + 1e-12))
cv_ratio = float(std / (avg + 1e-12))
p25 = float(torch.quantile(norms, 0.25).item()) if norms.numel() else 0.0
p50 = float(torch.quantile(norms, 0.50).item()) if norms.numel() else 0.0
p75 = float(torch.quantile(norms, 0.75).item()) if norms.numel() else 0.0
p90 = float(torch.quantile(norms, 0.90).item()) if norms.numel() else 0.0
p95 = float(torch.quantile(norms, 0.95).item()) if norms.numel() else 0.0
iqr = float(p75 - p25)
p95_to_p50_ratio = float(p95 / (p50 + 1e-12))
# Consecutive delta direction stability.
cosine_mean = None
reversal_ratio = None
if flat.shape[0] >= 2:
a = flat[1:]
b = flat[:-1]
dot = (a * b).sum(dim=1)
denom = torch.linalg.vector_norm(a, dim=1) * torch.linalg.vector_norm(b, dim=1) + 1e-12
cos = dot / denom
cosine_mean = float(torch.nan_to_num(cos, nan=0.0).mean().item())
reversal_ratio = float((cos < -0.05).float().mean().item())
# Frame-to-frame motion acceleration/jerk proxy on norm trajectory.
jerk_abs_mean = 0.0
jerk_ratio = 0.0
if norms.numel() >= 2:
d1 = norms[1:] - norms[:-1]
jerk_abs_mean = float(d1.abs().mean().item())
jerk_ratio = float(jerk_abs_mean / (avg + 1e-12))
# Heuristic stability score for run-to-run comparisons (observer-only).
rev_component = max(0.0, 1.0 - float(reversal_ratio if reversal_ratio is not None else 0.5))
spike_component = 1.0 / (1.0 + max(0.0, spike_ratio - 1.0))
cv_component = 1.0 / (1.0 + cv_ratio)
jerk_component = 1.0 / (1.0 + jerk_ratio)
stability_score = float(
0.40 * rev_component
+ 0.25 * spike_component
+ 0.20 * cv_component
+ 0.15 * jerk_component
)
if stability_score >= 0.72:
motion_profile = "stable"
elif stability_score >= 0.55:
motion_profile = "mixed"
else:
motion_profile = "volatile"
rec = {
"stage": stage,
"status": "ok",
"shape": list(tf.shape),
"frame_delta_count": int(delta.shape[0]),
"frame_delta_norm_mean": avg,
"frame_delta_norm_std": std,
"frame_delta_norm_min": mn,
"frame_delta_norm_max": mx,
"frame_delta_norm_p25": p25,
"frame_delta_norm_p50": p50,
"frame_delta_norm_p75": p75,
"frame_delta_norm_p90": p90,
"frame_delta_norm_p95": p95,
"frame_delta_norm_iqr": iqr,
"frame_delta_norm_cv_ratio": cv_ratio,
"frame_delta_p95_to_p50_ratio": p95_to_p50_ratio,
"frame_delta_spike_ratio": spike_ratio,
"frame_delta_abs_mean": float(mean_abs.mean().item()) if mean_abs.numel() else 0.0,
"frame_delta_abs_max": float(mean_abs.max().item()) if mean_abs.numel() else 0.0,
"frame_delta_cosine_mean": cosine_mean,
"frame_delta_reversal_ratio": reversal_ratio,
"frame_delta_jerk_abs_mean": jerk_abs_mean,
"frame_delta_jerk_ratio": jerk_ratio,
"frame_motion_stability_score": stability_score,
"frame_motion_profile": motion_profile,
"frame_motion_score_formula": "0.40*(1-reversal)+0.25*(1/(1+max(0,spike-1)))+0.20*(1/(1+cv))+0.15*(1/(1+jerk_ratio))",
"frame_motion_score_note": "heuristic observer-only ranking metric for run-to-run comparison",
}
records.append(rec)
return rec
except Exception as e:
rec = {"stage": stage, "status": "failed", "error": str(e)}
records.append(rec)
return rec
def _dual_branch_delta_coupling_math(self, *, delta_high, delta_low, records, active_branch_mode, cascade_count):
"""
Observer-only branch coupling math.
Compares high/low sampler deltas to expose alignment and energy split diagnostics.
"""
rec = {
"stage": "EventMathDualBranchDeltaCoupling",
"active_branch_mode": str(active_branch_mode or ""),
"cascade_count": int(cascade_count or 1),
}
try:
import math
import torch
if str(active_branch_mode or "") != "DUAL_HIGH_LOW":
rec.update({"status": "skipped", "reason": "single_branch_mode"})
records.append(rec)
return rec
high_t = self._tensor_from_latent_like(delta_high)
low_t = self._tensor_from_latent_like(delta_low)
if high_t is None or low_t is None:
rec.update({
"status": "unavailable",
"reason": "missing_delta_tensor",
"high_present": bool(high_t is not None),
"low_present": bool(low_t is not None),
})
records.append(rec)
return rec
high_f = torch.nan_to_num(high_t.detach().float().reshape(-1), nan=0.0, posinf=0.0, neginf=0.0)
low_f = torch.nan_to_num(low_t.detach().float().reshape(-1), nan=0.0, posinf=0.0, neginf=0.0)
if high_f.numel() < 1 or low_f.numel() < 1:
rec.update({"status": "unavailable", "reason": "empty_delta"})
records.append(rec)
return rec
n = int(min(high_f.numel(), low_f.numel()))
high_f = high_f[:n]
low_f = low_f[:n]
max_elements = 200000
stride = max(1, n // max_elements) if n > max_elements else 1
if stride > 1:
high_f = high_f[::stride]
low_f = low_f[::stride]
sample_count = int(high_f.numel())
high_norm = float(torch.linalg.vector_norm(high_f).item()) if sample_count else 0.0
low_norm = float(torch.linalg.vector_norm(low_f).item()) if sample_count else 0.0
dot = float((high_f * low_f).sum().item()) if sample_count else 0.0
denom = (high_norm * low_norm) + 1e-12
cosine = float(dot / denom) if sample_count else 0.0
cosine = max(-1.0, min(1.0, cosine))
angle_deg = float(math.degrees(math.acos(cosine)))
if cosine > 0.20:
direction_relation = "aligned"
elif cosine < -0.20:
direction_relation = "opposed"
else:
direction_relation = "orthogonal_mixed"
energy_total = high_norm + low_norm
high_energy_fraction = float(high_norm / (energy_total + 1e-12))
low_energy_fraction = float(low_norm / (energy_total + 1e-12))
low_to_high_norm_ratio = float(low_norm / (high_norm + 1e-12))
alignment_score = float(0.5 * (1.0 + cosine))
refinement_coupling_score = float(alignment_score * low_energy_fraction)
rec.update({
"status": "ok",
"sample_count": sample_count,
"sample_stride": int(stride),
"high_delta_norm": high_norm,
"low_delta_norm": low_norm,
"low_to_high_norm_ratio": low_to_high_norm_ratio,
"delta_cosine_alignment": cosine,
"delta_alignment_angle_deg": angle_deg,
"direction_relation": direction_relation,
"high_energy_fraction": high_energy_fraction,
"low_energy_fraction": low_energy_fraction,
"alignment_score_01": alignment_score,
"refinement_coupling_score_01": refinement_coupling_score,
"formula": "ObservedBehavior(high) and ObservedBehavior(low) are compared as coupled branch deltas over a shared latent basis.",
})
records.append(rec)
return rec
except Exception as e:
rec.update({"status": "failed", "error": str(e)})
records.append(rec)
return rec
def _cascade_boundary_math(self, previous_frames, next_frames, records, segment_index):
try:
import torch
a = self._tensor_from_latent_like(previous_frames)
b = self._tensor_from_latent_like(next_frames)
if a is None or b is None:
records.append({"stage": "EventMathCascadeBoundary", "status": "unavailable", "segment_index": int(segment_index)})
return None
af = torch.nan_to_num(a.detach().float(), nan=0.0, posinf=0.0, neginf=0.0)
bf = torch.nan_to_num(b.detach().float(), nan=0.0, posinf=0.0, neginf=0.0)
if af.shape[0] < 1 or bf.shape[0] < 1:
records.append({"stage": "EventMathCascadeBoundary", "status": "unavailable", "segment_index": int(segment_index), "reason": "empty_frames"})
return None
last_prev = af[-1]
first_next = bf[0]
if last_prev.shape != first_next.shape:
records.append({
"stage": "EventMathCascadeBoundary",
"status": "shape_mismatch",
"segment_index": int(segment_index),
"prev_shape": list(last_prev.shape),
"next_shape": list(first_next.shape),
})
return None
d = first_next - last_prev
rec = {
"stage": "EventMathCascadeBoundary",
"status": "ok",
"segment_index": int(segment_index),
"boundary_delta_norm": float(torch.linalg.vector_norm(d).item()),
"boundary_delta_abs_mean": float(d.abs().mean().item()),
"boundary_delta_min": float(d.min().item()),
"boundary_delta_max": float(d.max().item()),
}
records.append(rec)
self._event_universal_boundary_math(
records,
f"EventCascadeBoundary_{int(segment_index)}",
before_state=a[-1:] if hasattr(a, "__getitem__") else previous_frames,
after_state=b[:1] if hasattr(b, "__getitem__") else next_frames,
observed_behavior="previous cascade last frame connects to next cascade first frame",
route_id=f"route_cascade_boundary_{int(segment_index)}",
control_mode="REPORT_ONLY",
metadata=rec,
)
return rec
except Exception as e:
rec = {"stage": "EventMathCascadeBoundary", "status": "failed", "segment_index": int(segment_index), "error": str(e)}
records.append(rec)
return rec
def _event_core_body_init(self, packet, execution_records, run_id, route_name="wan_terminal_one_node"):
"""
r44 internal Event Core Body.
This is not a separate visual ComfyUI node. It is the internal body of the single Event Horizon node.
"""
packet = ensure_packet(packet)
body = {
"body_version": EVENT_HORIZON_BODY_VERSION,
"body_name": "One Node Event Core Body + Runtime Monitor Body",
"external_node": "Event Horizon",
"visual_node_policy": "single_external_node_internal_event_body",
"route_name": str(route_name),
"run_id": str(run_id),
"formula": "NodeInputState + NodeObservedBehavior = EventSingularity = NodeOutputState",
"body_layers": [
"EventPacket",
"FormulaReader",
"RoleResolver",
"RouteMemory",
"SState",
"InternalWanPipeline",
"SamplerBoundaries",
"DecodeBoundary",
"OutputBoundary",
"ReportBuilder",
],
"technical_wires": ["MODEL", "CLIP", "VAE", "IMAGE", "CONDITIONING", "LATENT", "DELTA", "FRAMES", "VIDEO"],
"s_wire": [],
"stage_math_records": [],
"boundary_records": [],
"live_route_timeline": [],
"runtime_monitor_records": [],
"local_sstates": [],
"event_conflicts": [],
"conflict_policy": "report_first_no_generation_abort_unless_technical_failure",
"runtime_layer_policy": "observer-only until equivalence or safety proof",
}
self._event_live_body = {
"run_id": str(run_id),
"route_name": str(route_name),
"started_at": datetime.now().isoformat(timespec="seconds"),
"start_perf_counter": float(time.perf_counter()),
"route_timeline": [],
"s_wire_live": [],
"runtime_monitor": [],
"local_sstates": [],
"conflicts": [],
"policy": "observer_only_no_generation_side_effects",
}
packet.setdefault("metadata", {})["event_core_body"] = body
execution_records.append({
"stage": "EventCoreBodyInit",
"status": "recorded",
"body_version": body["body_version"],
"body_name": body["body_name"],
"visual_node_policy": body["visual_node_policy"],
"formula": body["formula"],
"route_name": body["route_name"],
"message": "Event Core Body is internal to this one Event Horizon node; no manual Event graph required.",
})
packet = record_stage(
packet,
stage_name="EventCoreBody",
action="INIT_INTERNAL_BODY",
observed_behavior="Single Event Horizon node created internal EventPacket, S-Wire, RouteMemory, and Formula route body.",
metadata=body,
formula_note=body["formula"],
)
return packet
def _event_core_live_record(self, stage_name, record_type="stage", status="", formula_role="", route_id="", observed_behavior="", metadata=None):
"""
r49 live RouteMemory/S-Wire hook.
This is observer-only: it records route timing and memory snapshots without changing tensors or routing.
"""
live = getattr(self, "_event_live_body", None)
if not isinstance(live, dict):
return None
timeline = live.setdefault("route_timeline", [])
entry = {
"index": len(timeline),
"stage": str(stage_name or "unknown"),
"record_type": str(record_type or "stage"),
"status": str(status or ""),
"formula_role": str(formula_role or ""),
"route_id": str(route_id or ""),
"observed_behavior": str(observed_behavior or ""),
"timestamp": datetime.now().isoformat(timespec="seconds"),
"perf_counter": float(time.perf_counter()),
"memory": _event_core_memory_snapshot(str(stage_name or "unknown")),
"metadata": metadata or {},
}
timeline.append(entry)
live.setdefault("runtime_monitor", []).append({
"stage": entry["stage"],
"record_type": entry["record_type"],
"status": entry["status"],
"perf_counter": entry["perf_counter"],
"memory": entry["memory"],
})
if record_type in ("stage_math", "boundary_math", "manual"):
live.setdefault("s_wire_live", []).append({
"stage": entry["stage"],
"formula_role": entry["formula_role"],
"route_id": entry["route_id"],
"status": entry["status"],
})
return entry
def _event_core_body_collect_from_records(self, packet, execution_records):
"""
Builds a compact internal S-Wire / RouteMemory summary from execution_records.
This keeps r44 one-node: records are collected inside the same Event Horizon node.
"""
packet = ensure_packet(packet)
body = packet.setdefault("metadata", {}).setdefault("event_core_body", {})
s_wire = []
stage_math_records = []
boundary_records = []
for rec in execution_records or []:
stage = str(rec.get("stage", ""))
if stage.startswith("EventUniversalMath_"):
item = {
"stage": stage,
"stage_name": rec.get("stage_name", stage.replace("EventUniversalMath_", "")),
"formula_role": rec.get("formula_role", ""),
"route_id": rec.get("route_id", ""),
"observed_behavior": rec.get("observed_behavior", ""),
"control_mode": rec.get("control_mode", ""),
"status": rec.get("status", ""),
}
stage_math_records.append(item)
s_wire.append(item)
elif stage.startswith("EventUniversalBoundary_"):
item = {
"stage": stage,
"boundary_name": rec.get("boundary_name", stage.replace("EventUniversalBoundary_", "")),
"route_id": rec.get("route_id", ""),
"observed_behavior": rec.get("observed_behavior", ""),
"control_mode": rec.get("control_mode", ""),
"status": rec.get("status", ""),
}
boundary_records.append(item)
s_wire.append(item)
body["s_wire"] = s_wire
body["stage_math_records"] = stage_math_records
body["boundary_records"] = boundary_records
body["stage_math_count"] = len(stage_math_records)
body["boundary_count"] = len(boundary_records)
body["s_wire_count"] = len(s_wire)
live = getattr(self, "_event_live_body", {})
if isinstance(live, dict):
body["live_route_timeline"] = list(live.get("route_timeline", []))
body["runtime_monitor_records"] = list(live.get("runtime_monitor", []))
body["s_wire_live"] = list(live.get("s_wire_live", []))
body["live_route_count"] = len(body["live_route_timeline"])
body["runtime_monitor_count"] = len(body["runtime_monitor_records"])
if body["runtime_monitor_records"]:
first = body["runtime_monitor_records"][0].get("perf_counter")
last = body["runtime_monitor_records"][-1].get("perf_counter")
try:
observed_span = float(last) - float(first)
except Exception:
observed_span = None
body["runtime_monitor_summary"] = {
"record_count": len(body["runtime_monitor_records"]),
"observed_stage_span_seconds": round(observed_span, 6) if observed_span is not None else None,
"observer_only": True,
"has_process_memory": any("process_rss_mb" in (x.get("memory") or {}) for x in body["runtime_monitor_records"]),
"has_cuda_memory": any((x.get("memory") or {}).get("torch_cuda_available") for x in body["runtime_monitor_records"]),
}
body["one_node_integrity"] = {
"external_node_count": 1,
"internal_event_body": True,
"manual_event_graph_required": False,
"route_memory_kept_inside_node": True,
}
return packet
def _event_core_body_consistency_audit(self, execution_records, result_status="", saved_video_path="", failure_reason=""):
"""
r45 internal consistency audit.
This does not modify generation. It only checks whether the one-node Event Core Body route is internally coherent.
"""
records = list(execution_records or [])
stages = [str(r.get("stage", "")) for r in records]
required_exact = [
"EventCoreBodyInit",
"EventOneNodePolicy",
"EventUniversalPipelineMap",
]
required_prefix = [
"EventUniversalMath_EventTextEncodePositive",
"EventUniversalMath_EventTextEncodeNegative",
"EventUniversalMath_EventImageScaleStart",
"EventUniversalMath_EventWanImageToVideoSeed",
"EventUniversalMath_EventSamplerHigh",
"EventUniversalMath_EventSamplerLow",
"EventUniversalMath_EventVAEDecodeTiled",
]
expected_if_video = [
"EventUniversalMath_EventVideoSaveBegin",
"EventUniversalMath_EventVideoCombine",
]
def has_stage(name):
return name in stages
def has_prefix(prefix):
return any(s.startswith(prefix) for s in stages)
missing_exact = [x for x in required_exact if not has_stage(x)]
missing_prefix = [x for x in required_prefix if not has_prefix(x)]
if str(result_status or "").upper() == "VIDEO":
missing_video = [x for x in expected_if_video if not has_prefix(x)]
else:
missing_video = []
stage_math = [s for s in stages if s.startswith("EventUniversalMath_")]
boundary_math = [s for s in stages if s.startswith("EventUniversalBoundary_")]
math_tensor = [s for s in stages if s.startswith("EventMath_")]
duplicate_counts = {}
for s in stages:
duplicate_counts[s] = duplicate_counts.get(s, 0) + 1
duplicates = {k: v for k, v in duplicate_counts.items() if v > 1 and k not in ("EventHorizonStageDelay",)}
passed = not missing_exact and not missing_prefix and not missing_video
severity = "PASS" if passed else ("WARN" if str(result_status or "").upper() in ("FAILED", "ERROR") else "FAIL")
audit = {
"stage": "EventCoreBodyConsistencyAudit",
"status": "pass" if passed else "issues_found",
"severity": severity,
"formula": "One external node must preserve one internal Event Core Body route from input strategy to output state.",
"result_status": str(result_status or ""),
"saved_video_path": str(saved_video_path or ""),
"failure_reason": str(failure_reason or ""),
"checks": {
"one_external_node_policy_present": has_stage("EventOneNodePolicy"),
"event_core_body_initialized": has_stage("EventCoreBodyInit"),
"pipeline_map_present": has_stage("EventUniversalPipelineMap"),
"stage_math_count": len(stage_math),
"boundary_math_count": len(boundary_math),
"math_tensor_record_count": len(math_tensor),
"required_exact_missing": missing_exact,
"required_prefix_missing": missing_prefix,
"video_stage_missing": missing_video,
"duplicate_stage_names": duplicates,
},
"interpretation": (
"Internal Event Core Body route is coherent enough for next research step."
if passed else
"Internal Event Core Body route has missing stage records; inspect missing lists before adding new controls."
),
}
return audit
def _event_core_body_stage_order_audit(self, execution_records):
"""
r48: corrected order audit.
Boot/static records may appear before EventCoreBodyInit and must not block the gate.
This audit checks the runtime route order only:
Text -> Image -> Seed -> High -> Cleanup -> Low -> Decode -> Output
"""
stages = [str(r.get("stage", "")) for r in (execution_records or [])]
boot_required = [
"EventCoreBodyInit",
"EventOneNodePolicy",
"EventUniversalPipelineMap",
]
runtime_order_prefix = [
"EventUniversalMath_EventTextEncodePositive",
"EventUniversalMath_EventTextEncodeNegative",
"EventUniversalMath_EventImageScaleStart",
"EventUniversalMath_EventWanImageToVideoSeed",
"EventUniversalMath_EventModelShift_high",
"EventUniversalMath_EventSamplerHigh",
"EventUniversalMath_EventCleanupBetweenSamplers",
"EventUniversalMath_EventModelShift_low",
"EventUniversalMath_EventSamplerLow",
"EventUniversalMath_EventVAEDecodeTiled",
"EventMath_decoded_frame_motion",
"EventUniversalMath_EventVideoSaveBegin",
"EventUniversalMath_EventVideoCombine",
]
def first_index(prefix):
for i, s in enumerate(stages):
if s == prefix or s.startswith(prefix):
return i
return None
boot_presence = {name: (first_index(name) is not None) for name in boot_required}
order_items = []
last_idx = -1
violations = []
missing_runtime = []
for prefix in runtime_order_prefix:
idx = first_index(prefix)
order_items.append({"stage": prefix, "first_index": idx})
if idx is None:
missing_runtime.append(prefix)
continue
if idx < last_idx:
violations.append({
"stage": prefix,
"first_index": idx,
"previous_required_index": last_idx,
"reason": "runtime route stage appeared before an earlier runtime route stage",
})
last_idx = max(last_idx, idx)
# Missing runtime stages are reported here, but the completion gate still uses the
# consistency audit missing lists as the source of truth for blocking.
# This avoids double-blocking on optional/non-video branches.
return {
"stage": "EventCoreBodyStageOrderAudit",
"status": "pass" if not violations else "issues_found",
"formula": "Runtime route order should preserve Text -> Image -> Seed -> High -> Low -> Decode -> Output.",
"checked_stage_count": len(runtime_order_prefix),
"present_checked_count": sum(1 for x in order_items if x["first_index"] is not None),
"boot_presence": boot_presence,
"order_items": order_items,
"missing_runtime_route_stages": missing_runtime,
"violations": violations,
"audit_policy": "boot records are presence-checked but excluded from order violations",
}
def _event_core_body_completion_gate(self, audit, order_audit):
"""
r46: one final gate before tuning.
PASS means the internal one-node Event Core Body is coherent enough to start tuning/control work.
"""
checks = audit.get("checks", {}) if isinstance(audit, dict) else {}
missing_total = (
len(checks.get("required_exact_missing", []) or []) +
len(checks.get("required_prefix_missing", []) or []) +
len(checks.get("video_stage_missing", []) or [])
)
stage_math_count = int(checks.get("stage_math_count", 0) or 0)
one_node_ok = bool(checks.get("one_external_node_policy_present")) and bool(checks.get("event_core_body_initialized"))
order_ok = isinstance(order_audit, dict) and order_audit.get("status") == "pass"
pass_gate = (
one_node_ok and
missing_total == 0 and
stage_math_count >= 7 and
order_ok
)
return {
"stage": "EventCoreBodyCompletionGate",
"status": "PASS" if pass_gate else "BLOCKED",
"formula": "Tuning may begin only after the one-node Event Core Body is coherent, ordered, and complete enough.",
"one_node_ok": one_node_ok,
"stage_order_ok": order_ok,
"missing_total": missing_total,
"stage_math_count": stage_math_count,
"boundary_math_count": int(checks.get("boundary_math_count", 0) or 0),
"math_tensor_record_count": int(checks.get("math_tensor_record_count", 0) or 0),
"result_status": audit.get("result_status", "") if isinstance(audit, dict) else "",
"saved_video_path": audit.get("saved_video_path", "") if isinstance(audit, dict) else "",
"next_action": (
"Core body gate passed. Tuning/control work may begin on the next iteration."
if pass_gate else
"Core body gate blocked. Fix missing stages or order violations before tuning."
),
}
def _event_core_local_sstate_breakdown(self, packet, execution_records, result_status="", saved_video_path=""):
stages = [str(r.get("stage", "")) for r in (execution_records or [])]
def has(prefix):
return any(s == prefix or s.startswith(prefix) for s in stages)
def first_status(prefix):
for rec in execution_records or []:
stage = str(rec.get("stage", ""))
if stage == prefix or stage.startswith(prefix):
return str(rec.get("status", ""))
return "missing"
meta = packet.get("metadata", {}) if isinstance(packet, dict) else {}
wan_interface = meta.get("wan_workflow_interface", {}) if isinstance(meta.get("wan_workflow_interface", {}), dict) else {}
return [
{
"name": "S_text",
"formula_role": "StrategyCandidate carrier",
"granularity": "Stage",
"stage_present": has("EventUniversalMath_EventTextEncodePositive") or has("EventUniversalMath_EventTextEncodeNegative"),
"status": first_status("EventUniversalMath_EventTextEncodePositive"),
"contents": "prompt/negative prompt conditioning references; no raw prompt text stored here",
},
{
"name": "S_seed",
"formula_role": "Possibility field",
"granularity": "Stage",
"stage_present": has("EventUniversalMath_EventWanImageToVideoSeed"),
"status": first_status("EventUniversalMath_EventWanImageToVideoSeed"),
"contents": f"seed={wan_interface.get('seed', '')}, frames_per_cascade={wan_interface.get('frames', '')}",
},
{
"name": "S_high",
"formula_role": "OutcomeNext(high) = StrategyCarrier(low)",
"granularity": "Stage",
"stage_present": has("EventUniversalMath_EventSamplerHigh"),
"status": first_status("EventUniversalMath_EventSamplerHigh"),
"contents": "high latent output plus high delta/control records",
},
{
"name": "S_low",
"formula_role": "OutcomeNext(low)",
"granularity": "Stage",
"stage_present": has("EventUniversalMath_EventSamplerLow"),
"status": first_status("EventUniversalMath_EventSamplerLow"),
"contents": "low latent output plus low delta/control records",
},
{
"name": "S_decode",
"formula_role": "Translation state",
"granularity": "Stage",
"stage_present": has("EventUniversalMath_EventVAEDecodeTiled"),
"status": first_status("EventUniversalMath_EventVAEDecodeTiled"),
"contents": "VAE tile settings and decoded frame tensor reference",
},
{
"name": "S_output",
"formula_role": "Final Outcome",
"granularity": "Pipeline",
"stage_present": has("EventUniversalMath_EventVideoCombine"),
"status": str(result_status or ""),
"contents": f"saved_video_path={saved_video_path or ''}",
},
]
def _event_core_conflict_integration(self, packet, audit, order_audit, gate):
packet = ensure_packet(packet)
checks = audit.get("checks", {}) if isinstance(audit, dict) else {}
conflicts = []
def add_core_conflict(conflict_type, severity, stage_position, symptom, suggested, metadata=None):
conflict = make_conflict(
conflict_type,
severity=severity,
stage_position=stage_position,
suspected_cause="Event Core Body structural or formula-role audit found a route inconsistency.",
observed_symptom=symptom,
suggested_response=suggested,
metadata=metadata or {},
)
conflicts.append(conflict)
return add_conflict(packet, conflict)
for field, severity in (
("required_exact_missing", "CRITICAL"),
("required_prefix_missing", "BLOCKED"),
("video_stage_missing", "BLOCKED"),
):
for missing in checks.get(field, []) or []:
packet = add_core_conflict(
"EventCoreBodyMissingRecord",
severity,
str(missing),
f"{field}: {missing}",
"Restore the missing internal Event Core Body stage record before adding active controls.",
{"missing_field": field, "missing_stage": missing},
)
for violation in order_audit.get("violations", []) if isinstance(order_audit, dict) else []:
packet = add_core_conflict(
"EventCoreBodyStageOrderViolation",
"BLOCKED",
str(violation.get("stage", "EventCoreBodyStageOrderAudit")),
str(violation.get("reason", "runtime route order violation")),
"Fix runtime route ordering; boot/static records are presence-only and should not block order.",
violation,
)
if isinstance(gate, dict) and gate.get("status") != "PASS":
packet = add_core_conflict(
"EventCoreBodyCompletionGateBlocked",
"BLOCKED",
"EventCoreBodyCompletionGate",
f"CompletionGate={gate.get('status')}, missing_total={gate.get('missing_total')}, stage_order_ok={gate.get('stage_order_ok')}",
"Do not enable active tuning/cache/compile paths until the gate is PASS.",
gate,
)
return packet, conflicts
def _event_core_body_summary_record(self, audit, order_audit, gate, body=None):
checks = audit.get("checks", {}) if isinstance(audit, dict) else {}
body = body if isinstance(body, dict) else {}
return {
"stage": "EventCoreBodySummary",
"status": gate.get("status", "UNKNOWN") if isinstance(gate, dict) else "UNKNOWN",
"body_version": EVENT_HORIZON_BODY_VERSION,
"one_node_ok": gate.get("one_node_ok", False) if isinstance(gate, dict) else False,
"stage_order_ok": gate.get("stage_order_ok", False) if isinstance(gate, dict) else False,
"audit_gate": gate.get("status", "UNKNOWN") if isinstance(gate, dict) else "UNKNOWN",
"missing_total": gate.get("missing_total", 0) if isinstance(gate, dict) else 0,
"stage_math_count": gate.get("stage_math_count", 0) if isinstance(gate, dict) else 0,
"boundary_math_count": gate.get("boundary_math_count", 0) if isinstance(gate, dict) else 0,
"math_tensor_record_count": gate.get("math_tensor_record_count", 0) if isinstance(gate, dict) else 0,
"live_route_count": body.get("live_route_count", 0),
"runtime_monitor_count": body.get("runtime_monitor_count", 0),
"local_sstate_count": len(body.get("local_sstates", []) or []),
"event_conflict_count": len(body.get("event_conflicts", []) or []),
"required_exact_missing": checks.get("required_exact_missing", []),
"required_prefix_missing": checks.get("required_prefix_missing", []),
"video_stage_missing": checks.get("video_stage_missing", []),
"order_violations": order_audit.get("violations", []) if isinstance(order_audit, dict) else [],
"next_action": gate.get("next_action", "") if isinstance(gate, dict) else "",
}
def _event_core_body_report_card(self, audit):
checks = audit.get("checks", {}) if isinstance(audit, dict) else {}
return {
"stage": "EventCoreBodyReportCard",
"status": audit.get("status", "unknown") if isinstance(audit, dict) else "unknown",
"severity": audit.get("severity", "UNKNOWN") if isinstance(audit, dict) else "UNKNOWN",
"one_node_ok": bool(checks.get("one_external_node_policy_present")) and bool(checks.get("event_core_body_initialized")),
"stage_math_count": checks.get("stage_math_count", 0),
"boundary_math_count": checks.get("boundary_math_count", 0),
"missing_total": (
len(checks.get("required_exact_missing", []) or []) +
len(checks.get("required_prefix_missing", []) or []) +
len(checks.get("video_stage_missing", []) or [])
),
"next_action": (
"Continue only if EventCoreBodyCompletionGate is PASS on normal and cascade runs."
if audit.get("status") == "pass" else
"Do not add new controls yet; fix missing Event Core Body stages first."
) if isinstance(audit, dict) else "Audit unavailable.",
}
def _event_core_body_finalize(self, packet, execution_records, result_status, saved_video_path, failure_reason):
packet = self._event_core_body_collect_from_records(packet, execution_records)
body = packet.setdefault("metadata", {}).setdefault("event_core_body", {})
body["result_status"] = str(result_status or "")
body["saved_video_path"] = str(saved_video_path or "")
body["failure_reason"] = str(failure_reason or "")
body["finalized"] = True
audit = self._event_core_body_consistency_audit(
execution_records,
result_status=result_status,
saved_video_path=saved_video_path,
failure_reason=failure_reason,
)
body["consistency_audit"] = audit
execution_records.append(audit)
order_audit = self._event_core_body_stage_order_audit(execution_records)
body["stage_order_audit"] = order_audit
execution_records.append(order_audit)
gate = self._event_core_body_completion_gate(audit, order_audit)
body["completion_gate"] = gate
execution_records.append(gate)
packet, integrated_conflicts = self._event_core_conflict_integration(packet, audit, order_audit, gate)
body["event_conflicts"] = integrated_conflicts
execution_records.append({
"stage": "EventConflictIntegration",
"status": "recorded",
"conflict_count": len(integrated_conflicts),
"severity_counts": {
severity: sum(1 for c in integrated_conflicts if c.get("severity") == severity)
for severity in sorted({c.get("severity") for c in integrated_conflicts if c.get("severity")})
},
"formula": "Audit findings become EventConflict objects with stage and formula-role context.",
})
local_sstates = self._event_core_local_sstate_breakdown(packet, execution_records, result_status, saved_video_path)
body["local_sstates"] = local_sstates
execution_records.append({
"stage": "EventCoreBodyLocalSStateBreakdown",
"status": "recorded",
"sstate_count": len(local_sstates),
"sstates": local_sstates,
"formula": "Stage-level states are labeled by formula role rather than only technical stage name.",
})
if body.get("runtime_monitor_summary"):
execution_records.append({
"stage": "EventRuntimeMonitorSummary",
"status": "recorded",
**body.get("runtime_monitor_summary", {}),
"formula": "Runtime timing and memory are observer-only ObservedBehavior extensions.",
})
summary = self._event_core_body_summary_record(audit, order_audit, gate, body=body)
body["summary"] = summary
execution_records.append(summary)
execution_records.append(self._event_core_body_report_card(audit))
execution_records.append({
"stage": "EventCoreBodyFinalize",
"status": "recorded",
"stage_math_count": body.get("stage_math_count", 0),
"boundary_count": body.get("boundary_count", 0),
"s_wire_count": body.get("s_wire_count", 0),
"result_status": str(result_status or ""),
"saved_video_path": str(saved_video_path or ""),
"failure_reason": str(failure_reason or ""),
"one_node_integrity": body.get("one_node_integrity", {}),
})
packet = record_stage(
packet,
stage_name="EventCoreBody",
action="FINALIZE_INTERNAL_BODY",
observed_behavior="Internal one-node Event Core Body collected universal stage math, boundary math, S-Wire route, and final output state.",
metadata={
"stage_math_count": body.get("stage_math_count", 0),
"boundary_count": body.get("boundary_count", 0),
"s_wire_count": body.get("s_wire_count", 0),
"result_status": str(result_status or ""),
"saved_video_path": str(saved_video_path or ""),
"one_node_integrity": body.get("one_node_integrity", {}),
},
formula_note="single external node -> internal Event Core Body -> internal Wan route -> final video/report output",
)
return packet
def _event_universal_stage_math(self, records, stage_name, input_state=None, output_state=None,
observed_behavior="", formula_role="", route_id="", next_requirement="",
control_mode="REPORT_ONLY", metadata=None):
"""
r40 universal stage math record.
Common wrapper for every internal stage:
NodeInputState + NodeObservedBehavior = NodeSState = NodeOutputState
Report-first. Does not modify generation.
"""
metadata = metadata or {}
rec = {
"stage": f"EventUniversalMath_{stage_name}",
"status": "recorded",
"formula": "NodeInputState + NodeObservedBehavior = NodeSState = NodeOutputState",
"stage_name": str(stage_name),
"formula_role": str(formula_role or ""),
"route_id": str(route_id or ""),
"observed_behavior": str(observed_behavior or ""),
"next_requirement": str(next_requirement or ""),
"control_mode": str(control_mode or "REPORT_ONLY"),
"metadata": metadata,
}
try:
if input_state is not None:
t = self._tensor_from_latent_like(input_state)
if t is not None:
rec["input_state_shape"] = list(t.shape)
rec["input_state_dtype"] = str(getattr(t, "dtype", "unknown"))
else:
rec["input_state_type"] = type(input_state).__name__
if output_state is not None:
t = self._tensor_from_latent_like(output_state)
if t is not None:
rec["output_state_shape"] = list(t.shape)
rec["output_state_dtype"] = str(getattr(t, "dtype", "unknown"))
else:
rec["output_state_type"] = type(output_state).__name__
except Exception as e:
rec["summary_error"] = str(e)
records.append(rec)
self._event_core_live_record(
stage_name=stage_name,
record_type="stage_math",
status=rec.get("status", ""),
formula_role=rec.get("formula_role", ""),
route_id=rec.get("route_id", ""),
observed_behavior=rec.get("observed_behavior", ""),
metadata={"execution_record_index": len(records) - 1, "control_mode": rec.get("control_mode", "")},
)
return rec
def _safe_event_universal_stage_math(self, records, stage_name, **kwargs):
try:
return self._event_universal_stage_math(records, stage_name, **kwargs)
except Exception as e:
try:
records.append({
"stage": f"EventUniversalMath_{stage_name}",
"status": "failed_nonfatal",
"error": str(e),
"formula": "NodeInputState + NodeObservedBehavior = NodeSState = NodeOutputState",
})
except Exception:
pass
return None
def _event_universal_boundary_math(self, records, boundary_name, before_state=None, after_state=None,
observed_behavior="", route_id="", control_mode="REPORT_ONLY", metadata=None):
"""
r41 boundary record for transitions between stages.
It records:
StageBefore.OutputState + BoundaryObservedBehavior = StageAfter.InputState
"""
metadata = metadata or {}
rec = {
"stage": f"EventUniversalBoundary_{boundary_name}",
"status": "recorded",
"formula": "StageBefore.OutputState + BoundaryObservedBehavior = StageAfter.InputState",
"boundary_name": str(boundary_name),
"route_id": str(route_id or ""),
"observed_behavior": str(observed_behavior or ""),
"control_mode": str(control_mode or "REPORT_ONLY"),
"metadata": metadata,
}
try:
if before_state is not None:
t = self._tensor_from_latent_like(before_state)
if t is not None:
rec["before_shape"] = list(t.shape)
rec["before_dtype"] = str(getattr(t, "dtype", "unknown"))
else:
rec["before_type"] = type(before_state).__name__
if after_state is not None:
t = self._tensor_from_latent_like(after_state)
if t is not None:
rec["after_shape"] = list(t.shape)
rec["after_dtype"] = str(getattr(t, "dtype", "unknown"))
else:
rec["after_type"] = type(after_state).__name__
except Exception as e:
rec["summary_error"] = str(e)
records.append(rec)
self._event_core_live_record(
stage_name=boundary_name,
record_type="boundary_math",
status=rec.get("status", ""),
route_id=rec.get("route_id", ""),
observed_behavior=rec.get("observed_behavior", ""),
metadata={"execution_record_index": len(records) - 1, "control_mode": rec.get("control_mode", "")},
)
return rec
def _event_control_warning(self, records, mode, high_delta_strength, low_delta_strength):
try:
high = float(high_delta_strength)
low = float(low_delta_strength)
mode = str(mode or "OBSERVE_ONLY")
if mode == "OBSERVE_ONLY" and (abs(high - 1.0) > 1e-9 or abs(low - 1.0) > 1e-9):
records.append({
"stage": "EventMathControlWarning",
"status": "strength_ignored_in_observe_only",
"mode": mode,
"high_delta_strength": high,
"low_delta_strength": low,
"message": "high_delta_strength / low_delta_strength are recorded but not applied while math_control_mode is OBSERVE_ONLY",
})
except Exception as e:
records.append({
"stage": "EventMathControlWarning",
"status": "failed",
"error": str(e),
})
def _apply_latent_delta_control(
self,
latent_before,
latent_after,
branch_name,
records,
*,
strength_override=None,
step_index=None,
window_steps=None,
):
"""
Active math control used directly in the sampler transition path.
Formula:
controlled_after = latent_before + (latent_after - latent_before) * strength_runtime
where strength_runtime may include:
- requested branch strength
- StrategyCarrier high->low coupling multiplier
- per-step schedule factor inside the denoise window
"""
try:
mode = str(getattr(self, "_event_math_control_mode", "OBSERVE_ONLY") or "OBSERVE_ONLY")
strengths = getattr(self, "_event_delta_strengths", {}) or {}
branch_lower = str(branch_name or "").lower()
if strength_override is not None:
base_strength = float(strength_override)
elif branch_lower == "high" or branch_lower.endswith("_high") or "_high_" in branch_lower:
base_strength = float(strengths.get("high", 1.0))
elif branch_lower == "low" or branch_lower.endswith("_low") or "_low_" in branch_lower:
base_strength = float(strengths.get("low", 1.0))
else:
base_strength = float(strengths.get(str(branch_name), 1.0))
except Exception:
mode = "OBSERVE_ONLY"
base_strength = 1.0
branch_lower = str(branch_name or "").lower()
coupling = getattr(self, "_event_strategy_coupling", {}) or {}
coupling_multiplier = 1.0
try:
if "low" in branch_lower:
coupling_multiplier = float(coupling.get("low_strength_multiplier", 1.0) or 1.0)
except Exception:
coupling_multiplier = 1.0
scheduled_strength = float(base_strength * coupling_multiplier)
step_schedule_factor = 1.0
try:
if step_index is not None and window_steps is not None and int(window_steps) > 0:
progress = float((int(step_index) + 1) / max(1, int(window_steps)))
if "high" in branch_lower:
# High branch: slightly stronger earlier, softer toward window end.
step_schedule_factor = 1.10 - (0.20 * progress)
elif "low" in branch_lower:
# Low branch: slightly softer early, stronger toward detail refinement end.
step_schedule_factor = 0.90 + (0.20 * progress)
scheduled_strength = scheduled_strength * step_schedule_factor
except Exception:
step_schedule_factor = 1.0
strength_runtime = max(0.0, min(2.0, float(scheduled_strength)))
if mode not in ("LATENT_DELTA_SCALE", "DEEP_STEP_DELTA_CONTROL") or abs(strength_runtime - 1.0) < 1e-9:
records.append({
"stage": f"EventMathDeltaControl_{branch_name}",
"status": "bypass",
"mode": mode,
"base_strength": base_strength,
"coupling_multiplier": coupling_multiplier,
"step_schedule_factor": step_schedule_factor,
"strength_runtime": strength_runtime,
"step_index": int(step_index) if step_index is not None else None,
"window_steps": int(window_steps) if window_steps is not None else None,
"formula": "latent_after unchanged",
})
return latent_after
try:
import torch
before_t = self._tensor_from_latent_like(latent_before)
after_t = self._tensor_from_latent_like(latent_after)
if before_t is None or after_t is None:
records.append({
"stage": f"EventMathDeltaControl_{branch_name}",
"status": "unavailable",
"mode": mode,
"base_strength": base_strength,
"coupling_multiplier": coupling_multiplier,
"step_schedule_factor": step_schedule_factor,
"strength_runtime": strength_runtime,
"reason": "missing before/after tensor",
})
return latent_after
if before_t.shape != after_t.shape:
records.append({
"stage": f"EventMathDeltaControl_{branch_name}",
"status": "shape_mismatch",
"mode": mode,
"base_strength": base_strength,
"coupling_multiplier": coupling_multiplier,
"step_schedule_factor": step_schedule_factor,
"strength_runtime": strength_runtime,
"before_shape": list(before_t.shape),
"after_shape": list(after_t.shape),
})
return latent_after
before_f = before_t.detach().float()
after_f = after_t.detach().float()
controlled = before_f + (after_f - before_f) * float(strength_runtime)
controlled = controlled.to(dtype=after_t.dtype, device=after_t.device)
if isinstance(latent_after, dict) and "samples" in latent_after:
out = dict(latent_after)
out["samples"] = controlled
else:
out = controlled
records.append({
"stage": f"EventMathDeltaControl_{branch_name}",
"status": "applied",
"mode": mode,
"base_strength": base_strength,
"coupling_multiplier": coupling_multiplier,
"step_schedule_factor": step_schedule_factor,
"strength_runtime": strength_runtime,
"step_index": int(step_index) if step_index is not None else None,
"window_steps": int(window_steps) if window_steps is not None else None,
"formula": "controlled_after = latent_before + (latent_after - latent_before) * strength_runtime",
})
self._math_tensor_summary(
out,
records,
f"EventMath_{branch_name}_latent_after_controlled",
reference=latent_before,
strict=False,
)
return out
except Exception as e:
records.append({
"stage": f"EventMathDeltaControl_{branch_name}",
"status": "failed_passthrough",
"mode": mode,
"base_strength": base_strength,
"coupling_multiplier": coupling_multiplier,
"step_schedule_factor": step_schedule_factor,
"strength_runtime": strength_runtime,
"error": str(e),
})
return latent_after
def _update_strategy_coupling_from_high(self, latent_before, latent_after, branch_name, records):
"""
High->low StrategyCarrier coupling:
derive low-branch strength multiplier from high branch delta energy ratio.
"""
try:
import torch
before_t = self._tensor_from_latent_like(latent_before)
after_t = self._tensor_from_latent_like(latent_after)
if before_t is None or after_t is None:
records.append({
"stage": "EventMathStrategyCarrierCoupling",
"status": "unavailable",
"branch_name": str(branch_name or ""),
"reason": "missing_high_branch_tensor",
})
self._event_strategy_coupling = {"low_strength_multiplier": 1.0}
return
if before_t.shape != after_t.shape:
records.append({
"stage": "EventMathStrategyCarrierCoupling",
"status": "shape_mismatch",
"branch_name": str(branch_name or ""),
"before_shape": list(before_t.shape),
"after_shape": list(after_t.shape),
})
self._event_strategy_coupling = {"low_strength_multiplier": 1.0}
return
before_f = torch.nan_to_num(before_t.detach().float(), nan=0.0, posinf=0.0, neginf=0.0)
after_f = torch.nan_to_num(after_t.detach().float(), nan=0.0, posinf=0.0, neginf=0.0)
delta = after_f - before_f
before_norm = float(torch.linalg.vector_norm(before_f).item()) if before_f.numel() else 0.0
after_norm = float(torch.linalg.vector_norm(after_f).item()) if after_f.numel() else 0.0
delta_norm = float(torch.linalg.vector_norm(delta).item()) if delta.numel() else 0.0
# Important for I2V: high branch often starts from near-zero latent seed.
# Using before_norm alone would explode relative_delta and force artificial low-branch suppression.
baseline_norm = float(max(before_norm, after_norm, 1e-12))
relative_delta = float(delta_norm / baseline_norm)
if before_norm <= 1e-9 and after_norm > 0.0:
low_multiplier = 1.0
policy = "neutralize_low_strength_when_high_baseline_is_zero"
elif relative_delta > 1.35:
low_multiplier = max(0.75, min(1.0, 1.35 / (relative_delta + 1e-12)))
policy = "reduce_low_strength_after_energetic_high_branch"
elif relative_delta < 0.85:
low_multiplier = min(1.25, max(1.0, 0.85 / (relative_delta + 1e-12)))
policy = "raise_low_strength_after_soft_high_branch"
else:
low_multiplier = 1.0
policy = "keep_low_strength_neutral"
self._event_strategy_coupling = {
"low_strength_multiplier": float(low_multiplier),
"source_branch": str(branch_name or ""),
"relative_delta": float(relative_delta),
"delta_norm": float(delta_norm),
"before_norm": float(before_norm),
"after_norm": float(after_norm),
"baseline_norm": float(baseline_norm),
"policy": policy,
}
records.append({
"stage": "EventMathStrategyCarrierCoupling",
"status": "applied",
"source_branch": str(branch_name or ""),
"relative_delta": float(relative_delta),
"low_strength_multiplier": float(low_multiplier),
"before_norm": float(before_norm),
"after_norm": float(after_norm),
"baseline_norm": float(baseline_norm),
"policy": policy,
"formula": "StrategyCarrier(high) statistics modulate low branch control strength before refinement.",
})
except Exception as e:
self._event_strategy_coupling = {"low_strength_multiplier": 1.0}
records.append({
"stage": "EventMathStrategyCarrierCoupling",
"status": "failed",
"branch_name": str(branch_name or ""),
"error": str(e),
})
def _event_sampler_trace_config(self):
cfg = getattr(self, "_event_sampler_trace", {}) or {}
mode = str(cfg.get("mode", "OFF") or "OFF").upper()
try:
max_steps = int(cfg.get("max_steps", 64) or 64)
except Exception:
max_steps = 64
max_steps = max(1, min(65535, max_steps))
return mode, max_steps
def _event_sampler_step_trace_shadow(self, model, positive, negative, latent_before, latent_after_main, window, records):
"""
Optional sampler step-level shadow trace (observer-only).
It does not replace the active sampler path and does not mutate generation outputs.
"""
mode, max_steps = self._event_sampler_trace_config()
branch = str(getattr(window, "branch_name", "branch") or "branch")
if mode != "SHADOW_STEP_TRACE":
records.append({
"stage": f"EventSamplerStepTraceSummary_{branch}",
"status": "disabled",
"trace_mode": mode,
"max_steps": max_steps,
})
return None
try:
start = int(getattr(window, "start_at_step", 0) or 0)
end = int(getattr(window, "end_at_step", 0) or 0)
requested_steps = max(0, end - start)
if requested_steps <= 0:
records.append({
"stage": f"EventSamplerStepTraceSummary_{branch}",
"status": "unavailable",
"trace_mode": mode,
"reason": "empty_window",
"start_at_step": start,
"end_at_step": end,
})
return None
traced_end = min(end, start + max_steps)
shadow_latent = latent_before
traced = 0
records.append({
"stage": f"EventSamplerStepTraceBegin_{branch}",
"status": "recorded",
"trace_mode": mode,
"requested_steps": requested_steps,
"traced_steps_limit": max_steps,
"traced_step_range": [start, traced_end],
"formula": "step-level shadow trace: X(k+1) = sampler_step(X(k), constraints)",
})
for step_index in range(start, traced_end):
step_add_noise = str(window.add_noise) if step_index == start else "disable"
step_return_leftover = "enable"
if step_index == traced_end - 1 and traced_end == end:
step_return_leftover = str(getattr(window, "return_with_leftover_noise", "disable"))
self._math_tensor_summary(
shadow_latent,
records,
f"EventMath_{branch}_step_{step_index}_latent_before",
strict=False,
)
step_after = self._low_level_sampler_operation(
model=model,
positive=positive,
negative=negative,
latent=shadow_latent,
seed=int(window.seed),
steps=int(window.steps),
cfg=float(window.cfg),
sampler_name=str(window.sampler_name),
scheduler=str(window.scheduler),
start_at_step=int(step_index),
end_at_step=int(step_index + 1),
add_noise=step_add_noise,
return_leftover_noise=step_return_leftover,
)
self._math_tensor_summary(
step_after,
records,
f"EventMath_{branch}_step_{step_index}_latent_after",
reference=shadow_latent,
strict=False,
)
self._safe_event_universal_stage_math(
records,
f"EventSamplerStepTrace_{branch}_k{step_index}",
input_state=shadow_latent,
output_state=step_after,
observed_behavior=f"{branch} sampler step trace transformed latent at step {step_index}->{step_index + 1}",
formula_role="LATENT OutcomePrevious(step) + sampler step delta = LATENT OutcomeNext(step)",
route_id=f"route_sampler_step_trace_{branch}_{step_index}",
next_requirement="next denoise step requires latent output from current step",
control_mode="REPORT_ONLY",
metadata={
"branch": branch,
"step_index": int(step_index),
"window_start": start,
"window_end": end,
"shadow_trace_mode": mode,
"add_noise": step_add_noise,
"return_with_leftover_noise": step_return_leftover,
},
)
shadow_latent = step_after
traced += 1
compare_rec = self._math_tensor_summary(
latent_after_main,
records,
f"EventMath_{branch}_step_trace_vs_window_output",
reference=shadow_latent,
strict=False,
)
records.append({
"stage": f"EventSamplerStepTraceSummary_{branch}",
"status": "recorded",
"trace_mode": mode,
"requested_steps": requested_steps,
"traced_steps": traced,
"truncated": bool(traced_end < end),
"max_steps": max_steps,
"trace_vs_window_relative_delta": compare_rec.get("relative_delta") if isinstance(compare_rec, dict) else None,
"formula": "shadow step trace is observer-only and compared against main window output",
})
return True
except Exception as e:
records.append({
"stage": f"EventSamplerStepTraceSummary_{branch}",
"status": "failed_nonfatal",
"trace_mode": mode,
"max_steps": max_steps,
"error": str(e),
})
return None
def _event_sample_window_math_native(self, model, positive, negative, latent, window, records):
"""
Event-native active math sampler loop.
Runs one denoise step at a time and applies delta control per step.
"""
branch = str(getattr(window, "branch_name", "branch") or "branch")
start = int(getattr(window, "start_at_step", 0) or 0)
end = int(getattr(window, "end_at_step", 0) or 0)
window_steps = max(0, end - start)
event_records = [{
"stage": "event_sampler_begin",
"branch_name": branch,
"branch_role": getattr(window, "branch_role", ""),
"start_at_step": start,
"end_at_step": end,
"replacement_layer": "event_native_math_loop",
"step_loop": "native_math_active",
"math_control_mode": str(getattr(self, "_event_math_control_mode", "OBSERVE_ONLY")),
}]
if window_steps <= 0:
event_records.append({
"stage": "event_sampler_failed",
"status": "failed",
"branch_name": branch,
"branch_role": getattr(window, "branch_role", ""),
"error": "empty sampler window",
})
return EventSamplerResult(False, latent, None, window, event_records, error="empty sampler window")
current = latent
try:
for local_index, step_index in enumerate(range(start, end)):
step_add_noise = str(window.add_noise) if local_index == 0 else "disable"
step_return_leftover = str(getattr(window, "return_with_leftover_noise", "disable")) if local_index == (window_steps - 1) else "enable"
self._math_tensor_summary(
current,
records,
f"EventMath_{branch}_native_step_{step_index}_latent_before",
strict=False,
)
raw_after = self._low_level_sampler_operation(
model=model,
positive=positive,
negative=negative,
latent=current,
seed=int(window.seed),
steps=int(window.steps),
cfg=float(window.cfg),
sampler_name=str(window.sampler_name),
scheduler=str(window.scheduler),
start_at_step=int(step_index),
end_at_step=int(step_index + 1),
add_noise=step_add_noise,
return_leftover_noise=step_return_leftover,
)
self._math_tensor_summary(
raw_after,
records,
f"EventMath_{branch}_native_step_{step_index}_latent_after_raw",
reference=current,
strict=False,
)
controlled_after = self._apply_latent_delta_control(
current,
raw_after,
branch,
records,
step_index=local_index,
window_steps=window_steps,
)
self._finite_guard(
controlled_after,
records,
f"EventFiniteGuard_{branch}_native_step_{step_index}_latent_after",
strict=True,
)
self._math_tensor_summary(
controlled_after,
records,
f"EventMath_{branch}_native_step_{step_index}_latent_after_controlled",
reference=current,
strict=False,
)
self._safe_event_universal_stage_math(
records,
f"EventSamplerNativeStep_{branch}_k{step_index}",
input_state=current,
output_state=controlled_after,
observed_behavior=f"{branch} native math sampler step {step_index}->{step_index + 1} applied delta control in-loop",
formula_role="LATENT OutcomePrevious(step) + controlled sampler delta = LATENT OutcomeNext(step)",
route_id=f"route_sampler_native_step_{branch}_{step_index}",
next_requirement="next denoise step consumes controlled latent",
control_mode=str(getattr(self, "_event_math_control_mode", "OBSERVE_ONLY")),
metadata={
"branch": branch,
"step_index": int(step_index),
"local_index": int(local_index),
"window_start": int(start),
"window_end": int(end),
"add_noise": step_add_noise,
"return_with_leftover_noise": step_return_leftover,
},
)
current = controlled_after
event_records.append({
"stage": "event_sampler_end",
"status": "ok",
"branch_name": branch,
"branch_role": getattr(window, "branch_role", ""),
"step_loop": "native_math_active",
"executed_steps": int(window_steps),
})
return EventSamplerResult(True, latent, current, window, event_records)
except Exception as e:
event_records.append({
"stage": "event_sampler_failed",
"status": "failed",
"branch_name": branch,
"branch_role": getattr(window, "branch_role", ""),
"step_loop": "native_math_active",
"error": str(e),
})
return EventSamplerResult(False, latent, None, window, event_records, error=str(e))
def _event_sample_window(self, model, positive, negative, latent, window, records):
self._math_tensor_summary(latent, records, f"EventMath_{window.branch_name}_latent_before", strict=False)
mode = str(getattr(self, "_event_math_control_mode", "OBSERVE_ONLY") or "OBSERVE_ONLY")
# Preserve model-native generation path by default.
# Math in LATENT_DELTA_SCALE should guide/measure behavior, not replace denoising physics step-by-step.
use_native_math_loop = (mode == "DEEP_STEP_DELTA_CONTROL")
if mode == "LATENT_DELTA_SCALE":
records.append({
"stage": "EventMathSamplerPathPolicy",
"status": "native_sampler_preserved",
"math_control_mode": mode,
"native_step_loop_replacement": False,
"formula": "Math acts as semantic overlay/observer and post-window control, while sampler core stays model-native.",
})
elif mode == "DEEP_STEP_DELTA_CONTROL":
records.append({
"stage": "EventMathSamplerPathPolicy",
"status": "native_step_loop_active",
"math_control_mode": mode,
"native_step_loop_replacement": True,
"formula": "WARNING: Experimental deep-step delta control is active. High risk of noise.",
})
if use_native_math_loop:
result = self._event_sample_window_math_native(model, positive, negative, latent, window, records)
else:
core = EventSamplerCore(self._low_level_sampler_operation)
result = core.sample_window(model=model, positive=positive, negative=negative, latent=latent, window=window)
records.extend(result.event_records)
if not result.ok:
raise RuntimeError(result.error or f"EventSampler {window.branch_name} failed")
self._math_tensor_summary(result.latent_after, records, f"EventMath_{window.branch_name}_latent_after", reference=latent, strict=False)
if not use_native_math_loop:
self._event_sampler_step_trace_shadow(
model=model,
positive=positive,
negative=negative,
latent_before=latent,
latent_after_main=result.latent_after,
window=window,
records=records,
)
controlled_latent_after = self._apply_latent_delta_control(latent, result.latent_after, window.branch_name, records)
else:
records.append({
"stage": f"EventSamplerStepTraceSummary_{window.branch_name}",
"status": "skipped_for_native_math_loop",
"trace_mode": str(getattr(self, "_event_sampler_trace", {}).get("mode", "OFF")),
"reason": "native_math_loop_already_generated_step_level_records",
})
controlled_latent_after = result.latent_after
branch_name_lower = str(getattr(window, "branch_name", "") or "").lower()
if "high" in branch_name_lower and mode == "LATENT_DELTA_SCALE":
self._update_strategy_coupling_from_high(latent, controlled_latent_after, window.branch_name, records)
self._finite_guard(controlled_latent_after, records, f"EventFiniteGuard_{window.branch_name}_latent_after", strict=True)
self._event_universal_stage_math(
records,
f"EventSampler{str(window.branch_name).capitalize()}",
input_state=latent,
output_state=controlled_latent_after,
observed_behavior=f"{window.branch_name} sampler transformed latent through step window {window.start_at_step}->{window.end_at_step}",
formula_role="LATENT OutcomePrevious + sampler update = LATENT OutcomeNext",
route_id=f"route_sampler_{window.branch_name}",
next_requirement="next sampler stage or VAE decode requires event-compatible latent",
control_mode=mode,
metadata={
"branch": str(window.branch_name),
"start_at_step": int(window.start_at_step),
"end_at_step": int(window.end_at_step),
"add_noise": str(window.add_noise),
"return_with_leftover_noise": str(getattr(window, "return_with_leftover_noise", "")),
"sampler_execution_path": "native_math_loop" if use_native_math_loop else "boundary_replacement",
},
)
return controlled_latent_after, result
def _decode_tiled(self, vae, latent, tile_size, overlap, temporal_size, temporal_overlap, records):
try:
result = self._call_node_method(
"VAEDecodeTiled",
["decode"],
samples=latent,
vae=vae,
tile_size=int(tile_size),
overlap=int(overlap),
temporal_size=int(temporal_size),
temporal_overlap=int(temporal_overlap),
)
image = self._first_output(result, preferred_names=["image", "images", "IMAGE"], stage="EventVAEDecodeTiled", records=records)
self._finite_guard(image, records, "EventFiniteGuard_decoded_frames", strict=True)
records.append({"stage": "EventVAEDecodeTiled", "status": "ok", "formula": "final latent + tiled decode behavior = EventSingularity_decode = visible frames"})
self._event_universal_stage_math(
records,
"EventVAEDecodeTiled",
input_state=latent,
output_state=image,
observed_behavior="sampled latent decoded into visible image/frame tensor",
formula_role="LATENT OutcomeNext -> IMAGE/FRAMES VisibleOutcome",
route_id="route_decode",
next_requirement="Output/VHS requires visible frames to package final media",
control_mode="REPORT_ONLY",
metadata={"tile_size": int(tile_size), "overlap": int(overlap), "temporal_size": int(temporal_size), "temporal_overlap": int(temporal_overlap)},
)
return image
except Exception as e:
records.append({"stage": "EventVAEDecodeTiled", "status": "failed", "error": str(e)})
raise
def _run_custom_cleanup_chain(self, value, label, records):
current = value
try:
result = self._call_node_method(
"RAMCleanup",
["cleanup", "clean"],
anything=current,
clean_file_cache=True,
clean_processes=True,
clean_dlls=True,
retry_times=3,
)
current = self._first_output(result, preferred_names=["anything", "output"], stage=f"EventRAMCleanup_{label}", records=records)
records.append({"stage": f"EventRAMCleanup_{label}", "status": "ok", "formula": "memory before + cleanup behavior = EventSingularity_cleanup_ram = memory after"})
self._event_universal_stage_math(
records,
f"EventRAMCleanup_{label}",
input_state=value,
output_state=current,
observed_behavior="RAM cleanup executed while preserving pass-through value",
formula_role="MEMORY pressure + cleanup behavior -> pass-through state",
route_id=f"route_cleanup_ram_{label}",
next_requirement="next stage should receive same technical value with reduced memory pressure",
control_mode="REPORT_ONLY",
metadata={"cleanup_type": "RAMCleanup"},
)
except Exception as e:
records.append({"stage": f"event_ram_cleanup_{label}", "status": "fallback_python_cleanup", "error": str(e)})
_event_core_cleanup_memory(f"ram_cleanup_fallback_{label}")
try:
result = self._call_node_method(
"VRAMCleanup",
["cleanup", "clean"],
anything=current,
offload_model=False,
offload_cache=True,
)
current = self._first_output(result, preferred_names=["anything", "output"], stage=f"EventVRAMCleanup_{label}", records=records)
records.append({"stage": f"EventVRAMCleanup_{label}", "status": "ok", "formula": "vram before + cleanup behavior = EventSingularity_cleanup_vram = vram after"})
self._event_universal_stage_math(
records,
f"EventVRAMCleanup_{label}",
input_state=value,
output_state=current,
observed_behavior="VRAM cleanup executed while preserving pass-through value",
formula_role="VRAM pressure + cleanup behavior -> pass-through state",
route_id=f"route_cleanup_vram_{label}",
next_requirement="next stage should receive same technical value with reduced VRAM pressure",
control_mode="REPORT_ONLY",
metadata={"cleanup_type": "VRAMCleanup"},
)
except Exception as e:
records.append({"stage": f"event_vram_cleanup_{label}", "status": "fallback_python_cleanup", "error": str(e)})
_event_core_cleanup_memory(f"vram_cleanup_fallback_{label}")
return current
def _cleanup_mode_includes(self, cleanup_timing, phase_name):
mode = str(cleanup_timing or "NONE").upper()
phase = str(phase_name or "").upper()
matrix = {
"BEFORE_GENERATION": {"BEFORE_GENERATION", "BEFORE_AND_AFTER", "ALL"},
"BETWEEN_SAMPLERS": {"BETWEEN_SAMPLERS", "ALL"},
"AFTER_GENERATION": {"AFTER_GENERATION", "BEFORE_AND_AFTER", "ALL"},
}
return mode in matrix.get(phase, set())
def _run_branch_barrier(
self,
*,
phase_name,
cleanup_timing,
strategy_state,
strategy_label,
records,
cleanup_records,
barrier_records,
route_id,
stage_name,
formula_role,
observed_behavior,
next_requirement,
use_custom_cleanup_nodes,
use_custom_chain,
cleanup_label,
):
phase = str(phase_name or "").upper()
mode = str(cleanup_timing or "NONE").upper()
if not self._cleanup_mode_includes(mode, phase):
return strategy_state, None
preserved_before = _event_core_state_descriptor(strategy_state, strategy_label)
current_state = strategy_state
custom_cleanup_used = bool(use_custom_chain and use_custom_cleanup_nodes and strategy_state is not None)
if custom_cleanup_used:
current_state = self._run_custom_cleanup_chain(current_state, cleanup_label, records)
cleanup_info = _event_core_cleanup_memory(f"{cleanup_label}_python")
if isinstance(cleanup_records, list):
cleanup_records.append(cleanup_info)
preserved_after = _event_core_state_descriptor(current_state, strategy_label)
shape_same = preserved_before.get("shape") == preserved_after.get("shape")
dtype_same = preserved_before.get("dtype") == preserved_after.get("dtype")
type_same = preserved_before.get("type") == preserved_after.get("type")
strategy_state_preserved = bool(type_same and (shape_same or "shape" not in preserved_before) and (dtype_same or "dtype" not in preserved_before))
released_actions = []
if custom_cleanup_used:
released_actions.extend(["RAMCleanup pass-through", "VRAMCleanup pass-through"])
if cleanup_info.get("gc"):
released_actions.append("python gc.collect")
if cleanup_info.get("torch_cuda_empty_cache"):
released_actions.append("torch.cuda.empty_cache")
if cleanup_info.get("torch_cuda_ipc_collect"):
released_actions.append("torch.cuda.ipc_collect")
memory_before = cleanup_info.get("memory_before", {}) if isinstance(cleanup_info.get("memory_before", {}), dict) else {}
memory_after = cleanup_info.get("memory_after", {}) if isinstance(cleanup_info.get("memory_after", {}), dict) else {}
barrier_record = {
"barrier_phase": phase.lower(),
"cleanup_timing_requested": mode,
"observer_only": True,
"strategy_state_required": strategy_state is not None,
"strategy_state_preserved": strategy_state_preserved,
"preserved": {
"before": preserved_before,
"after": preserved_after,
},
"released": {
"actions": released_actions,
"python_cleanup": {
"gc": bool(cleanup_info.get("gc")),
"torch_cuda_empty_cache": bool(cleanup_info.get("torch_cuda_empty_cache")),
"torch_cuda_ipc_collect": bool(cleanup_info.get("torch_cuda_ipc_collect")),
},
},
"memory_before": memory_before,
"memory_after": memory_after,
"memory_delta": _event_core_memory_delta(memory_before, memory_after),
}
if cleanup_info.get("error"):
barrier_record["cleanup_error"] = str(cleanup_info.get("error"))
records.append({
"stage": f"EventBranchBarrier_{phase}",
"status": "recorded",
"cleanup_timing": mode,
"strategy_state_preserved": strategy_state_preserved,
"released_actions": released_actions,
})
if isinstance(barrier_records, list):
barrier_records.append(barrier_record)
self._event_universal_stage_math(
records,
stage_name,
input_state=strategy_state,
output_state=current_state,
observed_behavior=observed_behavior,
formula_role=formula_role,
route_id=route_id,
next_requirement=next_requirement,
control_mode="REPORT_ONLY",
metadata=barrier_record,
)
return current_state, barrier_record
def _resolve_output_dir(self, output_folder_mode, output_folder, custom_output_folder, subdir="event_equality_reports", output_target="USER_D_AI_NSFW", media_type="video"):
mode_target = str(output_target or "USER_D_AI_NSFW")
media_type = str(media_type or "video").lower()
def local_try_make_dir(path):
try:
Path(path).mkdir(parents=True, exist_ok=True)
return Path(path)
except Exception:
return None
# Preferred explicit user folder:
# D:\AI NSFW\VID for video, D:\AI NSFW\PIC for images.
if mode_target in ("USER_D_AI_NSFW", "USER_D_AI_AND_SFW", "DEFAULT_USER_PATH"):
for candidate in _event_core_preferred_media_dirs(media_type):
made = local_try_make_dir(candidate)
if made is not None:
return made
try:
import folder_paths
base_output = Path(folder_paths.get_output_directory())
except Exception:
base_output = Path.cwd() / "output"
mode = str(output_folder_mode or "DEFAULT")
if mode_target == "CUSTOM" and str(custom_output_folder or "").strip():
out_dir = Path(str(custom_output_folder)).expanduser()
elif mode_target == "PICKER" and str(output_folder or "").strip() and str(output_folder) != "default":
candidate = Path(str(output_folder))
out_dir = candidate if candidate.is_absolute() else base_output / candidate
elif mode_target == "COMFY_OUTPUT":
out_dir = base_output / (subdir or "")
elif mode == "CUSTOM" and str(custom_output_folder or "").strip():
out_dir = Path(str(custom_output_folder)).expanduser()
elif mode == "PICKER" and str(output_folder or "").strip() and str(output_folder) != "default":
candidate = Path(str(output_folder))
out_dir = candidate if candidate.is_absolute() else base_output / candidate
else:
out_dir = base_output / (subdir or "")
out_dir.mkdir(parents=True, exist_ok=True)
return out_dir
def _save_report_file(self, report, save_prefix, output_target, output_folder_mode, output_folder, custom_output_folder):
safe_prefix = re.sub(r"[^a-zA-Z0-9_\\-]+", "_", str(save_prefix or "Event Horizon")).strip("_") or "Event_Horizon"
out_dir = self._resolve_output_dir(output_folder_mode, output_folder, custom_output_folder, subdir="event_equality_reports", output_target=output_target, media_type="report")
out_dir.mkdir(parents=True, exist_ok=True)
text = "" if report is None else str(report)
if not text.strip():
text = (
"# Event Horizon Report Fallback\n\n"
f"- runtime_version: {EVENT_HORIZON_RUNTIME_VERSION}\n"
f"- runtime_name: {EVENT_HORIZON_RUNTIME_NAME}\n"
"- warning: build_markdown_report returned an empty report; fallback report was written to avoid a 0 KB file.\n"
)
path = out_dir / f"{safe_prefix}_{datetime.now().strftime('%Y%m%d_%H%M%S_%f')}.md"
tmp_path = path.with_suffix(path.suffix + ".tmp")
with open(tmp_path, "w", encoding="utf-8", newline="\n") as f:
f.write(text)
f.flush()
try:
os.fsync(f.fileno())
except Exception:
pass
size = tmp_path.stat().st_size if tmp_path.exists() else 0
if size <= 0:
raise RuntimeError(f"Report temp file is empty: {tmp_path}")
tmp_path.replace(path)
final_size = path.stat().st_size if path.exists() else 0
if final_size <= 0:
raise RuntimeError(f"Report final file is empty after replace: {path}")
return str(path)
def _rewrite_report_file(self, saved_report_path, report):
path = Path(str(saved_report_path or ""))
if not str(path):
return 0
text = "" if report is None else str(report)
tmp_path = path.with_suffix(path.suffix + ".tmp")
with open(tmp_path, "w", encoding="utf-8", newline="\n") as f:
f.write(text)
f.flush()
try:
os.fsync(f.fileno())
except Exception:
pass
tmp_path.replace(path)
return path.stat().st_size if path.exists() else 0
def _runtime_monitor_rows(self, packet):
meta = packet.get("metadata", {}) if isinstance(packet, dict) else {}
body = meta.get("event_core_body", {}) if isinstance(meta.get("event_core_body", {}), dict) else {}
records = body.get("runtime_monitor_records", []) if isinstance(body.get("runtime_monitor_records", []), list) else []
rows = []
first_perf = None
previous_perf = None
for index, rec in enumerate(records):
memory = rec.get("memory", {}) if isinstance(rec.get("memory", {}), dict) else {}
perf_raw = rec.get("perf_counter")
try:
perf = float(perf_raw)
except Exception:
perf = None
if first_perf is None and perf is not None:
first_perf = perf
elapsed = (perf - first_perf) if (perf is not None and first_perf is not None) else None
delta = (perf - previous_perf) if (perf is not None and previous_perf is not None) else None
if perf is not None:
previous_perf = perf
rows.append({
"index": index,
"stage": rec.get("stage", ""),
"record_type": rec.get("record_type", ""),
"status": rec.get("status", ""),
"elapsed_from_start_s": round(elapsed, 6) if elapsed is not None else "",
"delta_since_previous_s": round(delta, 6) if delta is not None else "",
"process_rss_mb": memory.get("process_rss_mb", ""),
"torch_cuda_available": memory.get("torch_cuda_available", ""),
"cuda_allocated_mb": memory.get("cuda_allocated_mb", ""),
"cuda_reserved_mb": memory.get("cuda_reserved_mb", ""),
"cuda_max_allocated_mb": memory.get("cuda_max_allocated_mb", ""),
"cuda_max_reserved_mb": memory.get("cuda_max_reserved_mb", ""),
})
return rows
def _runtime_motion_summary(self, packet):
"""
Extract one canonical motion-math summary from execution records for sidecar diffing.
Preference order:
1) EventMath_concatenated_frame_motion (cascade-level full output)
2) EventMath_decoded_frame_motion (single segment/final decode)
3) latest *frame_motion record as fallback
"""
meta = packet.get("metadata", {}) if isinstance(packet, dict) else {}
execution_records = meta.get("execution_records", []) if isinstance(meta.get("execution_records", []), list) else []
if not execution_records:
return {"motion_stage": "", "available": False}
preferred = None
fallback = None
for rec in execution_records:
if not isinstance(rec, dict):
continue
stage = str(rec.get("stage", "") or "")
if stage == "EventMath_concatenated_frame_motion":
preferred = rec
break
if stage == "EventMath_decoded_frame_motion" and preferred is None:
preferred = rec
if stage.endswith("_frame_motion") or stage == "EventMathCascadeBoundary":
fallback = rec
selected = preferred if isinstance(preferred, dict) else (fallback if isinstance(fallback, dict) else None)
if not isinstance(selected, dict):
return {"motion_stage": "", "available": False}
keys = [
"frame_delta_count",
"frame_delta_norm_mean",
"frame_delta_norm_std",
"frame_delta_norm_min",
"frame_delta_norm_max",
"frame_delta_norm_p25",
"frame_delta_norm_p50",
"frame_delta_norm_p75",
"frame_delta_norm_p90",
"frame_delta_norm_p95",
"frame_delta_norm_iqr",
"frame_delta_norm_cv_ratio",
"frame_delta_p95_to_p50_ratio",
"frame_delta_spike_ratio",
"frame_delta_abs_mean",
"frame_delta_abs_max",
"frame_delta_cosine_mean",
"frame_delta_reversal_ratio",
"frame_delta_jerk_abs_mean",
"frame_delta_jerk_ratio",
"frame_motion_stability_score",
"frame_motion_profile",
]
out = {
"motion_stage": str(selected.get("stage", "")),
"status": str(selected.get("status", "")),
"available": str(selected.get("status", "")) == "ok",
}
for key in keys:
if key in selected:
out[key] = selected.get(key)
return out
def _runtime_effective_cascade_count(self, packet):
meta = packet.get("metadata", {}) if isinstance(packet, dict) else {}
execution_records = meta.get("execution_records", []) if isinstance(meta.get("execution_records", []), list) else []
for rec in execution_records:
if not isinstance(rec, dict):
continue
stage = str(rec.get("stage", "") or "")
if stage in ("EventHorizonCascadeExecutionGate", "EventHorizonCascadeBegin"):
try:
return max(1, int(rec.get("cascade_count", 1)))
except Exception:
return 1
result = meta.get("result_status", {}) if isinstance(meta.get("result_status", {}), dict) else {}
try:
return max(1, int(result.get("cascade_count", 1)))
except Exception:
return 1
def _runtime_monitor_settings_signature(self, packet):
meta = packet.get("metadata", {}) if isinstance(packet, dict) else {}
wan = meta.get("wan_workflow_interface", {}) if isinstance(meta.get("wan_workflow_interface", {}), dict) else {}
result = meta.get("result_status", {}) if isinstance(meta.get("result_status", {}), dict) else {}
effective_cascade_count = self._runtime_effective_cascade_count(packet)
signature_source = {
"width": wan.get("width"),
"height": wan.get("height"),
"frames": wan.get("frames"),
"fps": wan.get("fps"),
"seed": wan.get("seed"),
"cascade_count": effective_cascade_count,
"branch_mode_active": wan.get("branch_mode_active"),
"ksampler_windows": wan.get("ksampler_windows"),
"cleanup_timing": wan.get("cleanup_timing"),
"math_controls": getattr(self, "_event_requested_math_controls", {}),
"sampler_trace": getattr(self, "_event_sampler_trace", {}),
"runtime_controls": getattr(self, "_event_requested_runtime_controls", {}),
"input_normalization_signature": (getattr(self, "_event_input_normalization", {}) or {}).get("normalized_signature", ""),
"runtime_aliases": self._event_runtime_aliases(),
"runtime_version": EVENT_HORIZON_RUNTIME_VERSION,
}
encoded = json.dumps(_event_json_safe(signature_source), sort_keys=True, ensure_ascii=True)
signature = hashlib.sha256(encoded.encode("utf-8")).hexdigest()[:24]
return signature, signature_source
def _find_previous_runtime_monitor_json(self, out_dir, current_json_path, settings_signature):
try:
current = Path(current_json_path).resolve()
candidates = sorted(
Path(out_dir).glob("*_runtime_monitor.json"),
key=lambda p: p.stat().st_mtime,
reverse=True,
)
for path in candidates:
try:
if path.resolve() == current:
continue
with open(path, "r", encoding="utf-8") as f:
payload = json.load(f)
if payload.get("settings_signature") == settings_signature:
return path, payload
except Exception:
continue
except Exception:
pass
return None, None
def _write_runtime_monitor_csv(self, csv_path, rows):
fieldnames = [
"index",
"stage",
"record_type",
"status",
"elapsed_from_start_s",
"delta_since_previous_s",
"process_rss_mb",
"torch_cuda_available",
"cuda_allocated_mb",
"cuda_reserved_mb",
"cuda_max_allocated_mb",
"cuda_max_reserved_mb",
]
with open(csv_path, "w", encoding="utf-8", newline="") as f:
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
for row in rows:
writer.writerow({key: row.get(key, "") for key in fieldnames})
def _write_runtime_monitor_diff_csv(self, diff_path, current_payload, previous_payload, previous_path):
def summary_value(payload, key, default=""):
summary = payload.get("runtime_monitor_summary", {}) if isinstance(payload, dict) else {}
return summary.get(key, default) if isinstance(summary, dict) else default
def motion_value(payload, key, default=""):
motion = payload.get("motion_summary", {}) if isinstance(payload, dict) else {}
return motion.get(key, default) if isinstance(motion, dict) else default
rows = []
if not previous_payload:
rows.append({
"metric": "baseline",
"current_value": current_payload.get("json_path", ""),
"previous_value": "",
"delta": "",
"note": "no previous runtime monitor sidecar with matching settings_signature",
})
else:
rows.append({
"metric": "baseline",
"current_value": current_payload.get("json_path", ""),
"previous_value": str(previous_path or ""),
"delta": "",
"note": "matched by settings_signature",
})
comparisons = [
("record_count", summary_value(current_payload, "record_count"), summary_value(previous_payload, "record_count")),
("observed_stage_span_seconds", summary_value(current_payload, "observed_stage_span_seconds"), summary_value(previous_payload, "observed_stage_span_seconds")),
("result_status", current_payload.get("result_status", ""), previous_payload.get("result_status", "")),
("completion_gate", current_payload.get("completion_gate", ""), previous_payload.get("completion_gate", "")),
("video_path", current_payload.get("saved_video_path", ""), previous_payload.get("saved_video_path", "")),
]
for metric, current_value, previous_value in comparisons:
delta = ""
try:
delta = round(float(current_value) - float(previous_value), 6)
except Exception:
delta = ""
rows.append({
"metric": metric,
"current_value": current_value,
"previous_value": previous_value,
"delta": delta,
"note": "",
})
motion_comparisons = [
("motion_stage", motion_value(current_payload, "motion_stage", ""), motion_value(previous_payload, "motion_stage", "")),
("motion_profile", motion_value(current_payload, "frame_motion_profile", ""), motion_value(previous_payload, "frame_motion_profile", "")),
("motion_stability_score", motion_value(current_payload, "frame_motion_stability_score", ""), motion_value(previous_payload, "frame_motion_stability_score", "")),
("motion_norm_mean", motion_value(current_payload, "frame_delta_norm_mean", ""), motion_value(previous_payload, "frame_delta_norm_mean", "")),
("motion_spike_ratio", motion_value(current_payload, "frame_delta_spike_ratio", ""), motion_value(previous_payload, "frame_delta_spike_ratio", "")),
("motion_reversal_ratio", motion_value(current_payload, "frame_delta_reversal_ratio", ""), motion_value(previous_payload, "frame_delta_reversal_ratio", "")),
("motion_cosine_mean", motion_value(current_payload, "frame_delta_cosine_mean", ""), motion_value(previous_payload, "frame_delta_cosine_mean", "")),
("motion_jerk_ratio", motion_value(current_payload, "frame_delta_jerk_ratio", ""), motion_value(previous_payload, "frame_delta_jerk_ratio", "")),
]
for metric, current_value, previous_value in motion_comparisons:
delta = ""
try:
delta = round(float(current_value) - float(previous_value), 6)
except Exception:
delta = ""
rows.append({
"metric": metric,
"current_value": current_value,
"previous_value": previous_value,
"delta": delta,
"note": "",
})
with open(diff_path, "w", encoding="utf-8", newline="") as f:
writer = csv.DictWriter(f, fieldnames=["metric", "current_value", "previous_value", "delta", "note"])
writer.writeheader()
for row in rows:
writer.writerow(row)
def _save_runtime_monitor_sidecars(self, packet, saved_report_path, saved_video_path, save_prefix):
if not saved_report_path:
return {"status": "skipped_no_report_path"}
report_path = Path(saved_report_path)
out_dir = report_path.parent
base = report_path.with_suffix("")
json_path = Path(str(base) + "_runtime_monitor.json")
csv_path = Path(str(base) + "_runtime_monitor.csv")
diff_path = Path(str(base) + "_runtime_monitor_diff.csv")
meta = packet.get("metadata", {}) if isinstance(packet, dict) else {}
body = meta.get("event_core_body", {}) if isinstance(meta.get("event_core_body", {}), dict) else {}
result = meta.get("result_status", {}) if isinstance(meta.get("result_status", {}), dict) else {}
gate = body.get("completion_gate", {}) if isinstance(body.get("completion_gate", {}), dict) else {}
settings_signature, signature_source = self._runtime_monitor_settings_signature(packet)
rows = self._runtime_monitor_rows(packet)
motion_summary = self._runtime_motion_summary(packet)
payload = {
"schema_version": "event_horizon.runtime_monitor.v2",
"runtime_version": EVENT_HORIZON_RUNTIME_VERSION,
"runtime_name": EVENT_HORIZON_RUNTIME_NAME,
"generated_at": datetime.now().isoformat(timespec="seconds"),
"save_prefix": str(save_prefix or ""),
"run_id": body.get("run_id", ""),
"settings_signature": settings_signature,
"settings_signature_source": signature_source,
"result_status": result.get("result_status", body.get("result_status", "")),
"completion_gate": gate.get("status", ""),
"saved_video_path": str(saved_video_path or result.get("saved_video_path", "")),
"report_path": str(saved_report_path),
"json_path": str(json_path),
"csv_path": str(csv_path),
"diff_path": str(diff_path),
"event_core_summary": body.get("summary", {}),
"runtime_monitor_summary": body.get("runtime_monitor_summary", {}),
"runtime_monitor_records": body.get("runtime_monitor_records", []),
"runtime_monitor_rows": rows,
"motion_summary": motion_summary,
"observer_only": True,
"formula": "Runtime timing and memory are ObservedBehavior extensions recorded without mutating generation.",
}
out_dir.mkdir(parents=True, exist_ok=True)
previous_path, previous_payload = self._find_previous_runtime_monitor_json(out_dir, json_path, settings_signature)
payload["previous_runtime_monitor_json"] = str(previous_path) if previous_path else ""
with open(json_path, "w", encoding="utf-8", newline="\n") as f:
json.dump(_event_json_safe(payload), f, ensure_ascii=False, indent=2)
f.write("\n")
self._write_runtime_monitor_csv(csv_path, rows)
self._write_runtime_monitor_diff_csv(diff_path, payload, previous_payload, previous_path)
return {
"status": "ok",
"schema_version": payload["schema_version"],
"json_path": str(json_path),
"csv_path": str(csv_path),
"diff_path": str(diff_path),
"previous_json_path": str(previous_path) if previous_path else "",
"settings_signature": settings_signature,
"record_count": len(rows),
"motion_stage": motion_summary.get("motion_stage", ""),
"motion_profile": motion_summary.get("frame_motion_profile", ""),
"motion_stability_score": motion_summary.get("frame_motion_stability_score", ""),
"observer_only": True,
}
def _save_image_attempt(self, image, save_prefix, records):
result = self._call_node_method("SaveImage", ["save_images"], images=image, filename_prefix=str(save_prefix or "wansolo"))
records.append({"stage": "EventSaveFrames", "status": "ok", "result": str(result)[:300]})
self._event_universal_stage_math(
records,
"EventSaveFrames",
input_state=image,
output_state=str(result)[:300],
observed_behavior="decoded frame/image tensor saved as image assets",
formula_role="IMAGE/FRAMES VisibleOutcome -> FILE output reference",
route_id="route_save_frames",
next_requirement="saved image assets should match decoded visible outcome",
control_mode="REPORT_ONLY",
metadata={"save_prefix": str(save_prefix or "wansolo")},
)
return str(result)
def _normalize_video_path_candidate(self, candidate):
if candidate is None:
return ""
c = str(candidate)
if not c:
return ""
exts = (".mp4", ".webp", ".gif", ".mov", ".mkv")
if not c.lower().endswith(exts):
return ""
# If absolute, return it even if the file existence check fails during Comfy async timing.
if os.path.isabs(c):
return c
try:
import folder_paths
output_dir = Path(folder_paths.get_output_directory())
except Exception:
output_dir = Path.cwd() / "output"
direct = output_dir / c
if direct.exists():
return str(direct)
# VHS sometimes gives only filename and subfolder separately; search by basename.
if output_dir.exists():
matches = list(output_dir.rglob(Path(c).name))
if matches:
return str(max(matches, key=lambda x: x.stat().st_mtime))
# As a last resort return the candidate, but report will show it was unresolved.
return c
def _extract_path_from_video_result(self, result, save_prefix, records, *args, **kwargs):
"""
r19: extract path only from standard VHS return / standard ComfyUI output.
"""
raw_description = str(result)[:1000]
try:
import folder_paths
output_dir = Path(folder_paths.get_output_directory())
except Exception:
output_dir = Path.cwd() / "output"
try:
if isinstance(result, dict):
gifs = result.get("ui", {}).get("gifs", []) or []
for item in gifs:
if isinstance(item, dict):
fullpath = item.get("fullpath")
path = self._normalize_video_path_candidate(fullpath)
if path:
records.append({"stage": "EventVideoCombine_extract_path", "status": "ok", "mode": "ui.gifs.fullpath", "path": path})
return path
filename = item.get("filename")
subfolder = item.get("subfolder", "")
if filename:
candidate = output_dir / str(subfolder) / str(filename)
path = self._normalize_video_path_candidate(str(candidate))
if path:
records.append({"stage": "EventVideoCombine_extract_path", "status": "ok", "mode": "ui.gifs.filename", "path": path})
return path
except Exception as e:
records.append({"stage": "EventVideoCombine_extract_path", "status": "ui_parse_failed", "error": str(e)})
candidates = []
def visit(obj, depth=0):
if obj is None or depth > 8:
return
if isinstance(obj, str):
candidates.append(obj)
return
if isinstance(obj, (list, tuple, set)):
for x in obj:
visit(x, depth + 1)
return
if isinstance(obj, dict):
for key in ("fullpath", "path", "filename", "file", "files", "result", "results", "ui", "gifs", "images"):
if key in obj:
visit(obj[key], depth + 1)
return
for attr in ("fullpath", "path", "filename", "file", "files", "result", "results", "value", "values", "output", "outputs", "data"):
if hasattr(obj, attr):
try:
visit(getattr(obj, attr), depth + 1)
except Exception:
pass
d = getattr(obj, "__dict__", None)
if isinstance(d, dict):
visit(d, depth + 1)
visit(result)
for ext in (".mp4", ".mov", ".mkv", ".webp", ".gif"):
for c in candidates:
if isinstance(c, str) and c.lower().endswith(ext):
path = self._normalize_video_path_candidate(c)
if path:
records.append({"stage": "EventVideoCombine_extract_path", "status": "ok", "mode": "recursive_standard_vhs", "path": path})
return path
newest = self._find_latest_standard_output_video(save_prefix, records)
if newest:
return newest
records.append({
"stage": "EventVideoCombine_extract_path",
"status": "not_found",
"raw_result_type": f"{type(result).__module__}.{type(result).__name__}",
"raw_result": raw_description,
"candidates": [str(x)[:200] for x in candidates[:20]],
})
return ""
def _get_first_available_node_class(self, class_names, records=None, stage="node_class_lookup"):
errors = {}
for class_name in class_names:
try:
cls = self._get_node_class(class_name)
if records is not None:
records.append({
"stage": stage,
"status": "found",
"class_name": class_name,
"class": str(cls),
})
return class_name, cls
except Exception as e:
errors[class_name] = str(e)
if records is not None:
records.append({
"stage": stage,
"status": "not_found",
"class_names": list(class_names),
"errors": errors,
})
raise RuntimeError(f"No candidate node class found for {class_names}; errors={errors}")
def _call_candidate_node_method(self, class_names, method_names, records=None, stage="candidate_node_call", *args, **kwargs):
last_error = None
tried = []
for class_name in class_names:
try:
# Check class exists first so report shows whether VHS is installed / mapped.
self._get_first_available_node_class([class_name], records=records, stage=f"{stage}_class_check")
tried.append(class_name)
return self._call_node_method(class_name, method_names, *args, **kwargs)
except Exception as e:
last_error = e
if records is not None:
records.append({
"stage": stage,
"status": "candidate_failed",
"class_name": class_name,
"error": str(e),
})
raise RuntimeError(f"All candidate node calls failed. tried={tried}; last_error={last_error}")
def _frames_to_uint8_numpy(self, image, records=None, stage="frames_to_uint8"):
try:
import numpy as np
frames = image
if not hasattr(frames, "detach"):
raise RuntimeError(f"{stage} expected torch-like IMAGE batch, got {type(frames)}")
arr = frames.detach().cpu().numpy()
if arr.ndim != 4:
raise RuntimeError(f"{stage} expected IMAGE batch [N,H,W,C], got shape={getattr(arr, 'shape', None)}")
finite = np.isfinite(arr)
nonfinite_count = int(arr.size - finite.sum())
if nonfinite_count > 0:
raise RuntimeError(f"{stage} refusing to encode non-finite frames: nonfinite_count={nonfinite_count}, shape={arr.shape}")
arr = (np.clip(arr, 0.0, 1.0) * 255).astype("uint8")
if records is not None:
records.append({"stage": stage, "status": "ok", "shape": list(arr.shape), "nonfinite_count": 0})
return arr
except Exception as e:
if records is not None:
records.append({"stage": stage, "status": "failed", "error": str(e)})
raise
def _fallback_animated_export(self, image, fps, save_prefix, video_format, records, output_target="USER_D_AI_NSFW", output_folder_mode="DEFAULT", output_folder="default", custom_output_folder=""):
"""
Emergency media writer if VHS is absent or incompatible.
Tries mp4 first, then animated WebP. It must physically write a file or return "".
"""
try:
arr = self._frames_to_uint8_numpy(image, records=records, stage="EventVideoFallback_frames_to_numpy")
output_dir = self._resolve_output_dir(output_folder_mode, output_folder, custom_output_folder, subdir="", output_target=output_target, media_type="video")
output_dir.mkdir(parents=True, exist_ok=True)
safe_prefix = re.sub(r"[^a-zA-Z0-9_\\-]+", "_", str(save_prefix or "wansolo")).strip("_") or "wansolo"
ts = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
# 1) imageio v3 / ffmpeg if available.
if "mp4" in str(video_format).lower():
try:
import imageio.v3 as iio
path = output_dir / f"{safe_prefix}_{ts}_fallback_imageio.mp4"
iio.imwrite(str(path), arr, fps=float(fps), codec="libx264")
if path.exists() and path.stat().st_size > 0:
records.append({"stage": "EventVideoFallbackImageIO", "status": "ok", "path": str(path), "size": path.stat().st_size})
return str(path)
records.append({"stage": "EventVideoFallbackImageIO", "status": "wrote_no_file_or_empty", "path": str(path)})
except Exception as e:
records.append({"stage": "EventVideoFallbackImageIO", "status": "failed", "error": str(e)})
# 2) OpenCV mp4 if available.
try:
import cv2
path = output_dir / f"{safe_prefix}_{ts}_fallback_cv2.mp4"
h, w = int(arr.shape[1]), int(arr.shape[2])
fourcc = cv2.VideoWriter_fourcc(*"mp4v")
writer = cv2.VideoWriter(str(path), fourcc, float(fps), (w, h))
if not writer.isOpened():
raise RuntimeError("cv2.VideoWriter did not open")
for frame in arr:
writer.write(cv2.cvtColor(frame, cv2.COLOR_RGB2BGR))
writer.release()
if path.exists() and path.stat().st_size > 0:
records.append({"stage": "EventVideoFallbackCV2", "status": "ok", "path": str(path), "size": path.stat().st_size})
return str(path)
records.append({"stage": "EventVideoFallbackCV2", "status": "wrote_no_file_or_empty", "path": str(path)})
except Exception as e:
records.append({"stage": "EventVideoFallbackCV2", "status": "failed", "error": str(e)})
# 3) Animated WebP fallback.
try:
from PIL import Image
path = output_dir / f"{safe_prefix}_{ts}_fallback.webp"
pil_frames = [Image.fromarray(frame) for frame in arr]
duration = int(1000 / max(1, float(fps)))
pil_frames[0].save(path, save_all=True, append_images=pil_frames[1:], duration=duration, loop=0, lossless=True)
if path.exists() and path.stat().st_size > 0:
records.append({"stage": "EventVideoFallbackWebP", "status": "ok", "path": str(path), "size": path.stat().st_size})
return str(path)
records.append({"stage": "EventVideoFallbackWebP", "status": "wrote_no_file_or_empty", "path": str(path)})
except Exception as e:
records.append({"stage": "EventVideoFallbackWebP", "status": "failed", "error": str(e)})
return ""
except Exception as e:
records.append({"stage": "EventVideoFallback", "status": "failed", "error": str(e)})
return ""
def _extract_vhs_ui_payload_from_records(self, records):
"""
r21: recover the VHS UI payload so ComfyUI can display the generated video
in the node UI without adding IMAGE outputs.
"""
try:
import ast
for rec in reversed(records or []):
if rec.get("stage") == "EventVideoCombine" and rec.get("raw_vhs_result"):
raw = rec.get("raw_vhs_result")
try:
payload = ast.literal_eval(raw)
except Exception:
payload = None
if isinstance(payload, tuple) and len(payload) > 0:
payload = payload[0]
if isinstance(payload, dict) and isinstance(payload.get("ui"), dict):
ui = payload.get("ui")
if "gifs" in ui:
return {"gifs": ui.get("gifs")}
if "images" in ui:
return {"images": ui.get("images")}
except Exception:
pass
return None
def _video_combine_attempt(self, image, fps, save_prefix, records, video_format="video/h264-mp4", force_vhs=True, output_target="COMFY_OUTPUT", output_folder_mode="DEFAULT", output_folder="default", custom_output_folder=""):
"""
r19: Standard VHS-only save.
This intentionally matches the working JSON:
VHS_VideoCombine receives filename_prefix, format, frame_rate, save_output=True.
It does NOT receive output_dir or custom path arguments.
VHS saves into the standard ComfyUI output folder.
"""
records.append({
"stage": "EventVideoSaveBegin",
"status": "start_standard_vhs",
"video_format": str(video_format),
"fps": fps,
"filename_prefix": str(save_prefix or "wansolo"),
"path_policy": "standard_comfyui_vhs_output_no_custom_output_dir",
"matches_working_json": True,
"loop_count": 0,
"pingpong": False,
})
self._event_universal_stage_math(
records,
"EventVideoSaveBegin",
input_state=image,
output_state=image,
observed_behavior="decoded frames prepared for VHS video combine without loop/pingpong modification",
formula_role="FRAMES VisibleOutcome -> FRAMES video input state",
route_id="route_video_save_begin",
next_requirement="VHS_VideoCombine receives ordered frames with frame_rate and save_output",
control_mode="REPORT_ONLY",
metadata={"video_format": str(video_format), "fps": float(fps), "loop_count": 0, "pingpong": False},
)
try:
self._frames_to_uint8_numpy(image, records=records, stage="EventVideoSave_validate_frames")
except Exception as e:
raise RuntimeError(f"Cannot save video because decoded frames are invalid: {e}")
try:
result = self._call_candidate_node_method(
["VHS_VideoCombine"],
["combine_video", "combine", "run", "execute"],
records=records,
stage="EventVideoCombineVHS",
images=image,
audio=None,
meta_batch=None,
vae=None,
frame_rate=float(fps),
loop_count=0,
filename_prefix=str(save_prefix or "wansolo"),
format=str(video_format or "video/h264-mp4"),
pingpong=False,
save_output=True,
pix_fmt="yuv420p",
crf=19,
save_metadata=True,
trim_to_audio=False,
)
path = self._extract_path_from_video_result(result, save_prefix, records)
if not path:
path = self._find_latest_standard_output_video(save_prefix, records)
records.append({
"stage": "EventVideoCombine",
"status": "ok" if path else "ok_no_path_returned",
"saved_video_path": path,
"raw_vhs_result": str(result)[:5000],
"formula": "decoded frames + VHS_VideoCombine(save_output=True) = saved video in standard ComfyUI output",
})
self._event_universal_stage_math(
records,
"EventVideoCombine",
input_state=image,
output_state=path or str(result)[:500],
observed_behavior="VHS_VideoCombine packaged decoded frames into video file/reference",
formula_role="FRAMES VisibleOutcome -> VIDEO file OutcomeFinal",
route_id="route_video_combine",
next_requirement="final report and UI preview should point to saved video path",
control_mode="REPORT_ONLY",
metadata={"video_format": str(video_format), "fps": float(fps), "saved_video_path": path or ""},
)
if path:
return path
raise RuntimeError(f"VHS_VideoCombine ran but no output path was found. raw={str(result)[:500]}")
except Exception as e:
records.append({"stage": "EventVideoCombine", "status": "failed_standard_vhs", "error": str(e)})
raise
def _find_latest_standard_output_video(self, save_prefix, records):
try:
import folder_paths
output_dir = Path(folder_paths.get_output_directory())
except Exception:
output_dir = Path.cwd() / "output"
safe_prefix = str(save_prefix or "wansolo")
found = []
if output_dir.exists():
for ext in (".mp4", ".webp", ".gif", ".mov", ".mkv"):
found.extend(output_dir.rglob(f"{safe_prefix}*{ext}"))
if found:
newest = max(found, key=lambda x: x.stat().st_mtime)
records.append({
"stage": "EventVideoCombine_standard_output_search",
"status": "ok",
"path": str(newest),
"output_dir": str(output_dir),
})
return str(newest)
records.append({
"stage": "EventVideoCombine_standard_output_search",
"status": "not_found",
"output_dir": str(output_dir),
"prefix": safe_prefix,
})
return ""
def _make_text_signal(self, packet, text, role, route, source_stage):
return _read_signal(
packet,
TECH_TEXT,
SPACE_TEXT,
text,
source_stage,
role,
route,
"TextStrategyReader",
metadata={"text_length": len(str(text or ""))},
)
def _fail_conflict(self, packet, stage, message, metadata=None):
conflict = make_conflict(
CONFLICT_FROZEN_PARTIAL_OBSERVABILITY,
severity=SEV_MEDIUM,
stage_position=stage,
suspected_cause="Terminal Wan workflow generation did not produce the requested target.",
observed_symptom=str(message),
suggested_response="Send Execution Records and ComfyUI traceback; next patch will fix exact incompatible stage.",
metadata=metadata or {},
)
packet = add_conflict(packet, conflict)
return packet, conflict["id"]
def _stage_delay(self, seconds, records, label):
try:
seconds = float(seconds or 0.0)
except Exception:
seconds = 0.0
if seconds <= 0:
return
try:
import time
records.append({"stage": "EventHorizonStageDelay", "status": "sleep", "seconds": seconds, "label": str(label)})
time.sleep(seconds)
except Exception as e:
records.append({"stage": "EventHorizonStageDelay", "status": "failed", "seconds": seconds, "label": str(label), "error": str(e)})
def _last_frame_image(self, frames, width=64, height=64):
return self._representative_preview_frame(frames, width, height, mode="last")
def _drop_first_frame_batch(self, frames, records, segment_index):
"""
r22 temporal continuity guard:
segment N starts from the last frame of segment N-1.
Keeping that first generated frame duplicates the previous terminal frame and can look like
a motion reset or vector reversal at the cascade boundary.
"""
try:
if hasattr(frames, "dim") and frames.dim() == 4 and frames.shape[0] > 1:
out = frames[1:]
records.append({
"stage": "EventHorizonCascadeDropFirstFrame",
"status": "ok",
"segment_index": int(segment_index),
"before_shape": list(frames.shape),
"after_shape": list(out.shape),
"reason": "remove duplicated source/continuation frame at cascade boundary",
})
return out
except Exception as e:
records.append({
"stage": "EventHorizonCascadeDropFirstFrame",
"status": "failed",
"segment_index": int(segment_index),
"error": str(e),
})
return frames
def _concat_frame_batches(self, batches, records):
valid = [b for b in batches if b is not None]
if not valid:
return None
if len(valid) == 1:
return valid[0]
try:
import torch
out = torch.cat(valid, dim=0)
records.append({
"stage": "EventHorizonCascadeFrameConcat",
"status": "ok",
"segments": len(valid),
"shape": list(out.shape) if hasattr(out, "shape") else str(type(out)),
})
return out
except Exception as e:
records.append({
"stage": "EventHorizonCascadeFrameConcat",
"status": "failed_using_last_segment_only",
"segments": len(valid),
"error": str(e),
})
return valid[-1]
def _run_event_horizon_segment_core(
self,
*,
segment_index,
source_image,
primary_model,
secondary_model,
clip,
vae,
positive_prompt,
negative_prompt,
active_branch_mode,
width,
height,
frames,
batch_size,
seed,
sampler_name,
scheduler,
global_steps,
primary_cfg,
secondary_cfg,
primary_start_step,
primary_end_step,
secondary_start_step,
secondary_end_step,
primary_sd3_shift,
secondary_sd3_shift,
decode_tile_size,
decode_overlap,
decode_temporal_size,
decode_temporal_overlap,
image_upscale_method,
image_crop,
cleanup_timing,
use_custom_cleanup_nodes,
stage_delay_seconds,
records,
cleanup_records,
barrier_records,
):
segment_label = f"cascade_{segment_index}"
records.append({
"stage": "EventHorizonCascadeSegmentBegin",
"status": "begin",
"segment_index": segment_index,
"frames": int(frames),
"formula": "source frame + segment transition = EventSingularity_segment = decoded frame batch",
})
positive = self._encode_text(clip, positive_prompt, records, label=f'{segment_label}_TextEncodePositive')
negative = self._encode_text(clip, negative_prompt, records, label=f'{segment_label}_TextEncodeNegative')
self._stage_delay(stage_delay_seconds, records, f"cascade_{segment_index}_after_text_encode")
scaled_image = self._scale_image(source_image, width, height, image_upscale_method, image_crop, records)
wan_positive, wan_negative, wan_latent = self._wan_image_to_video(
positive, negative, vae, scaled_image, width, height, frames, batch_size, records
)
self._stage_delay(stage_delay_seconds, records, f"cascade_{segment_index}_after_wan_image_to_video")
high_model = self._apply_sd3_shift(primary_model, primary_sd3_shift, f"{segment_label}_high", records)
high_window = EventSamplerWindow(
branch_name=f"{segment_label}_high",
branch_role="cascade_segment_high_motion_structure",
seed=int(seed) + int(segment_index) - 1,
steps=int(global_steps),
cfg=float(primary_cfg),
sampler_name=str(sampler_name),
scheduler=str(scheduler),
start_at_step=int(primary_start_step),
end_at_step=int(primary_end_step),
add_noise="enable",
return_with_leftover_noise="enable",
sd3_shift=float(primary_sd3_shift),
)
latent_after_high, high_result = self._event_sample_window(
high_model, wan_positive, wan_negative, wan_latent, high_window, records
)
self._stage_delay(stage_delay_seconds, records, f"cascade_{segment_index}_after_high_sampler")
delta_high, delta_high_err = compute_tensor_delta(wan_latent, latent_after_high)
if delta_high is None:
records.append({
"stage": f"EventMathDualBranchDeltaCoupling_{segment_label}",
"status": "high_delta_unavailable",
"segment_index": int(segment_index),
"error": str(delta_high_err),
})
delta_low = None
final_latent = latent_after_high
if active_branch_mode == "DUAL_HIGH_LOW" and secondary_model is not None:
latent_before_low = latent_after_high
latent_before_low, _ = self._run_branch_barrier(
phase_name="BETWEEN_SAMPLERS",
cleanup_timing=cleanup_timing,
strategy_state=latent_before_low,
strategy_label=f"{segment_label}_strategy_carrier_low",
records=records,
cleanup_records=cleanup_records,
barrier_records=barrier_records,
route_id=f"route_cleanup_{segment_label}",
stage_name=f"EventCleanup_{segment_label}_between_high_low",
formula_role="LATENT cascade high output + cleanup behavior -> LATENT cascade low input",
observed_behavior="Smart Branch Barrier between cascade high and low preserves StrategyCarrier while releasing disposable memory.",
next_requirement="cascade low sampler receives preserved high output latent",
use_custom_cleanup_nodes=use_custom_cleanup_nodes,
use_custom_chain=True,
cleanup_label=f"{segment_label}_between_high_low",
)
low_model = self._apply_sd3_shift(secondary_model, secondary_sd3_shift, f"{segment_label}_low", records)
low_window = EventSamplerWindow(
branch_name=f"{segment_label}_low",
branch_role="cascade_segment_low_detail_refinement",
seed=int(seed) + int(segment_index) - 1,
steps=int(global_steps),
cfg=float(secondary_cfg),
sampler_name=str(sampler_name),
scheduler=str(scheduler),
start_at_step=int(secondary_start_step),
end_at_step=int(secondary_end_step),
add_noise="disable",
return_with_leftover_noise="disable",
sd3_shift=float(secondary_sd3_shift),
)
latent_after_low, low_result = self._event_sample_window(
low_model, wan_positive, wan_negative, latent_before_low, low_window, records
)
self._stage_delay(stage_delay_seconds, records, f"cascade_{segment_index}_after_low_sampler")
delta_low, delta_low_err = compute_tensor_delta(latent_before_low, latent_after_low)
if delta_low is None:
records.append({
"stage": f"EventMathDualBranchDeltaCoupling_{segment_label}",
"status": "low_delta_unavailable",
"segment_index": int(segment_index),
"error": str(delta_low_err),
})
final_latent = latent_after_low
self._dual_branch_delta_coupling_math(
delta_high=delta_high,
delta_low=delta_low,
records=records,
active_branch_mode=active_branch_mode,
cascade_count=int(segment_index),
)
frames_out = self._decode_tiled(
vae, final_latent,
decode_tile_size, decode_overlap, decode_temporal_size, decode_temporal_overlap,
records
)
self._math_tensor_summary(frames_out, records, f"EventMath_cascade_{segment_index}_decoded_frames", strict=False)
self._frame_motion_math(frames_out, records, f"EventMath_cascade_{segment_index}_frame_motion")
self._stage_delay(stage_delay_seconds, records, f"cascade_{segment_index}_after_decode")
last_frame = self._last_frame_image(frames_out, width, height)
records.append({
"stage": "EventHorizonCascadeSegmentEnd",
"status": "ok",
"segment_index": segment_index,
"frames": int(frames),
"last_frame_for_next_segment": True,
})
return frames_out, last_frame, final_latent
def run(
self,
primary_model,
clip,
vae,
source_image_file,
positive_prompt,
negative_prompt,
event_strategy,
generation_target,
terminal_mode,
enable_continuation_outputs,
execution_mode,
branch_mode,
cascade_count,
cascade_mode,
frames_per_cascade,
width,
height,
frames,
batch_size,
fps,
seed,
sampler_name,
scheduler,
global_steps,
primary_cfg,
secondary_cfg,
primary_start_step,
primary_end_step,
secondary_start_step,
secondary_end_step,
primary_sd3_shift,
secondary_sd3_shift,
decode_tile_size,
decode_overlap,
decode_temporal_size,
decode_temporal_overlap,
image_upscale_method,
image_crop,
cleanup_timing,
stage_delay_seconds,
use_custom_cleanup_nodes,
save_video,
video_format,
force_vhs_video_combine,
save_frames,
save_report,
output_target,
save_output_image,
save_prefix,
output_folder_mode,
output_folder,
custom_output_folder,
report_detail,
pause_after_cascade_1=False,
pause_after_cascade_2=False,
pause_after_cascade_3=False,
pause_after_cascade_4=False,
resume_frame_index=-1,
secondary_model=None,
image=None,
mask=None,
):
run_id = now_run_id(prefix="terminal_wansolo")
self._event_strategy_coupling = {"low_strength_multiplier": 1.0}
cleanup_records = []
branch_barrier_records = []
execution_records = []
runtime_aliases = self._event_runtime_aliases()
execution_records.append({
"stage": "EventHorizonRuntimeVersion",
"status": "loaded",
"runtime_version": EVENT_HORIZON_RUNTIME_VERSION,
"runtime_name": EVENT_HORIZON_RUNTIME_NAME,
"math_report_enabled": True,
"expected_math_stages": [
"EventMath_high_latent_before",
"EventMath_high_latent_after",
"EventMath_low_latent_before",
"EventMath_low_latent_after",
"EventMathStrategyCarrierCoupling",
"EventMathDualBranchDeltaCoupling",
"EventMath_decoded_frame_motion",
"EventUniversalMath_EventTextEncodePositive",
"EventUniversalMath_EventTextEncodeNegative",
"EventUniversalMath_EventImageScaleStart",
"EventUniversalMath_EventWanImageToVideoSeed",
"EventUniversalMath_EventSamplerHigh",
"EventUniversalMath_EventCleanupBetweenSamplers",
"EventUniversalMath_EventSamplerLow",
"EventUniversalMath_EventVAEDecodeTiled",
"EventUniversalMath_EventVideoCombine",
"EventUniversalMath_EventModelShift_high",
"EventUniversalMath_EventModelShift_low",
"EventUniversalMath_EventCleanupBetweenSamplers",
"EventUniversalMath_EventVideoSaveBegin",
"EventUniversalBoundary_EventCascadeBoundary",
"EventCoreBodyInit",
"EventOneNodePolicy",
"EventCoreBodyConsistencyAudit",
"EventCoreBodyStageOrderAudit",
"EventCoreBodyCompletionGate",
"EventCoreBodySummary",
"EventCoreBodyReportCard",
"EventCoreBodyFinalize",
],
})
execution_records.append({
"stage": "EventWorkflowBinding",
"status": "recorded",
"runtime_class": self.__class__.__name__,
"runtime_version": EVENT_HORIZON_RUNTIME_VERSION,
"runtime_name": EVENT_HORIZON_RUNTIME_NAME,
"alias_count": len(runtime_aliases),
"aliases": runtime_aliases,
"compatibility": {
"EventHorizon": bool("EventHorizon" in runtime_aliases),
},
"formula": "The public workflow node alias resolves to one internal Event Core Body runtime implementation.",
})
input_normalization = getattr(self, "_event_input_normalization", None)
if isinstance(input_normalization, dict):
adjustments = input_normalization.get("adjustments", [])
reason_histogram = {}
for adj in adjustments:
if not isinstance(adj, dict):
continue
reason = str(adj.get("reason", "unknown") or "unknown")
reason_histogram[reason] = int(reason_histogram.get(reason, 0)) + 1
execution_records.append({
"stage": "EventInputNormalization",
"status": "recorded",
"normalized_values": input_normalization.get("normalized", {}),
"adjustment_count": len(adjustments),
"adjustment_reason_histogram": reason_histogram,
"normalized_signature": input_normalization.get("normalized_signature", ""),
"formula": "input configuration is normalized before stage execution to preserve deterministic route semantics",
})
if adjustments:
execution_records.append({
"stage": "EventInputNormalizationAdjustments",
"status": "recorded",
"adjustments": adjustments,
})
self._event_core_live_record(
"EventInputNormalization",
record_type="manual",
status="recorded",
formula_role="ConfigGuard",
route_id="route_input_normalization",
observed_behavior="Input values were normalized before generation.",
metadata={
"adjustment_count": len(adjustments),
"normalized_signature": input_normalization.get("normalized_signature", ""),
},
)
self._event_control_warning(
execution_records,
getattr(self, "_event_math_control_mode", "OBSERVE_ONLY"),
getattr(self, "_event_delta_strengths", {}).get("high", 1.0),
getattr(self, "_event_delta_strengths", {}).get("low", 1.0),
)
execution_records.append({
"stage": "EventMathControlSummary",
"status": "recorded",
"math_control_mode": getattr(self, "_event_math_control_mode", "OBSERVE_ONLY"),
"high_delta_strength_requested": getattr(self, "_event_delta_strengths", {}).get("high", 1.0),
"low_delta_strength_requested": getattr(self, "_event_delta_strengths", {}).get("low", 1.0),
"sampler_trace_mode": getattr(self, "_event_sampler_trace", {}).get("mode", "OFF"),
"sampler_trace_max_steps": getattr(self, "_event_sampler_trace", {}).get("max_steps", 64),
"active_generation_math_path": (
"semantic_overlay_native_sampler"
if str(getattr(self, "_event_math_control_mode", "OBSERVE_ONLY")).upper() == "LATENT_DELTA_SCALE"
else "native_step_loop" if str(getattr(self, "_event_math_control_mode", "OBSERVE_ONLY")).upper() == "DEEP_STEP_DELTA_CONTROL"
else "boundary_sampler_wrapper"
),
"strategy_carrier_coupling": "enabled_for_low_branch_without_step_replacement",
"precision_step": 0.0001,
"precision_round": 0.0001,
})
execution_records.append({
"stage": "EventUniversalPipelineMap",
"status": "recorded",
"formula": "Every internal stage is read as NodeInputState + NodeObservedBehavior = NodeSState = NodeOutputState",
"route": [
"EventTextEncodePositive",
"EventTextEncodeNegative",
"EventImageScaleStart",
"EventWanImageToVideoSeed",
"EventModelShiftHigh",
"EventSamplerHigh",
"EventCleanupBetweenSamplers",
"EventModelShiftLow",
"EventSamplerLow",
"EventVAEDecodeTiled",
"EventHorizonCascadeBoundary",
"EventVideoSaveBegin",
"EventVideoCombine",
"EventSaveReport",
"EventCleanupAfterGeneration",
],
"status_note": "r44 keeps the Event Core Body inside one Event Horizon node; EVENT_PACKET/S-Wire are internal runtime body, not a manual visual graph",
})
execution_records.append({
"stage": "EventOneNodePolicy",
"status": "active",
"external_visual_node": "Event Horizon",
"manual_event_graph_required": False,
"internal_body": [
"EventPacket",
"FormulaReader",
"RoleResolver",
"RouteMemory",
"SState",
"UniversalStageMath",
"UniversalBoundaryMath",
"ReportBuilder",
],
"reason": "Avoids manual ComfyUI workflow errors, wire-order drift, lost route memory, and mismatched Event logic between hidden stages.",
})
self._event_core_live_record(
"EventOneNodePolicy",
record_type="manual",
status="active",
formula_role="ArchitectureBoundary",
route_id="route_one_node_policy",
observed_behavior="One external node owns the internal Runtime Layer and Event Core Body.",
)
prompt_key_source = (
str(positive_prompt or "") + "\n---NEGATIVE---\n" + str(negative_prompt or "") +
f"\nclip={type(clip).__name__}:{id(clip)}"
)
encoder_cache_key = hashlib.sha256(prompt_key_source.encode("utf-8", errors="ignore")).hexdigest()[:24]
execution_records.append({
"stage": "EventRuntimeLayerProbes",
"status": "observer_only",
"runtime_monitor": "active",
"compile_guard": "observed_not_compiling",
"encoder_cache": "key_computed_cache_disabled_until_equivalence_proof",
"branch_barrier": "smart branch barrier records preserved StrategyCarrier and released memory actions per phase",
"lora_matrix_switch": "external_model_clip_route_observed_no_internal_switch",
"prompt_operator_panel": "prompt fields observed as StrategyCandidate carriers; no prompt rewrite",
"test_runner": "not active inside generation node; report data is structured for external runner",
"universal_input_normalization": "readers and RoleResolver remain internal; fixed Wan interface still current",
"encoder_cache_key_preview": encoder_cache_key,
"formula": "Forgotten runtime-layer ideas are present as internal observer records before active intervention.",
})
self._event_core_live_record(
"EventRuntimeLayerProbes",
record_type="manual",
status="observer_only",
formula_role="Infrastructure / ObservedBehavior extension",
route_id="route_runtime_layer_probes",
observed_behavior="Runtime layer probes were recorded without changing generation tensors.",
metadata={"encoder_cache_key_preview": encoder_cache_key},
)
saved_video_path = ""
saved_report_path = ""
video_ui_payload = None
generated_frames = None
generated_latent = None
source_preview = None
result_preview = None
result_status = "NONE"
failure_reason = ""
# r14d compatibility guard: old saved workflows or stale node widgets
# must not crash if output_target is missing during execution.
if "output_target" not in locals() or output_target is None:
output_target = "USER_D_AI_NSFW"
# Internal image picker: if external IMAGE socket is not connected, use source_image_file.
uploaded_image = None
if image is None:
uploaded_image = self._load_image_from_upload(source_image_file, execution_records)
if uploaded_image is not None:
image = uploaded_image
source_preview = self._representative_preview_frame(image, width, height, mode="first")
requested_video = generation_target in ("VIDEO", "AUTO")
requested_image = generation_target in ("IMAGE", "AUTO")
dual_available = secondary_model is not None
if branch_mode == "SINGLE":
active_branch_mode = "SINGLE"
elif branch_mode == "DUAL_HIGH_LOW":
active_branch_mode = "DUAL_HIGH_LOW" if dual_available else "SINGLE_FALLBACK_SECONDARY_MISSING"
else:
active_branch_mode = "DUAL_HIGH_LOW" if dual_available else "SINGLE"
mode_to_count = {
"SOLO_1": 1,
"CASCADE_2": 2,
"CASCADE_3": 3,
"CASCADE_4": 4,
"CASCADE_5": 5,
}
mode_count = int(mode_to_count.get(str(cascade_mode), int(cascade_count or 1)))
numeric_count = int(cascade_count or mode_count or 1)
requested_cascade_count = max(mode_count, numeric_count)
requested_cascade_count = max(1, min(5, requested_cascade_count))
frames = int(frames_per_cascade or frames)
packet = make_event_packet(metadata={
"created_by": "EventHorizonCascade",
"version": EVENT_HORIZON_RUNTIME_VERSION,
"run_id": run_id,
"node_role": "terminal_event_horizon_cascade",
})
packet = self._event_core_body_init(packet, execution_records, run_id, route_name="wan_terminal_one_node")
conflict_ids = []
relation_ids = []
signal_ids = []
packet = record_stage(
packet,
stage_name="WanEventWorkflowCore",
action="INIT_EVENT_HORIZON_CASCADE_WORKFLOW",
observed_behavior="Terminal-first Event Horizon workflow initialized",
metadata={
"run_id": run_id,
"generation_target": generation_target,
"cascade_mode": cascade_mode,
"cascade_count": requested_cascade_count,
"frames_per_cascade": frames,
"terminal_mode": terminal_mode,
"enable_continuation_outputs": enable_continuation_outputs,
"execution_mode": execution_mode,
"active_branch_mode": active_branch_mode,
},
)
packet["metadata"]["event_program_status"] = {
"current_mode": execution_mode,
"runtime_version": EVENT_HORIZON_RUNTIME_VERSION,
"runtime_name": EVENT_HORIZON_RUNTIME_NAME,
"math_report_enabled": True,
"terminal_mode": bool(terminal_mode),
"generation_target": generation_target,
"result_status": result_status,
"program_state": "terminal_save_first_generation_body",
"final_target_output": "saved_video_or_saved_frames",
"normal_outputs": "status_paths_report_only_no_image_sockets",
"continuation_outputs": "removed_from_main_terminal_node_in_r15_use_future_extractor_node",
"sampler_replacement": "EventSamplerCore boundary replacement; native step loop pending",
"no_fake_success_rule": "source image is never presented as successful VIDEO result",
}
packet["metadata"]["terminal_ui_model"] = {
"start_preview": "input image",
"result_preview": "generated frames/video representative frame if available",
"save_first": True,
"default_target": "VIDEO",
}
packet["metadata"]["wan_workflow_interface"] = {
"interface_variant": "terminal_B_no_lora_strings",
"source_workflow": "Event Horizon / Wan2.2 I2V Quant 14B",
"lora_policy": "LoRA is applied outside; node receives already patched MODEL/CLIP.",
"required_external": ["primary_model", "clip", "vae", "source_image_file or image for VIDEO"],
"optional_external": ["secondary_model", "mask", "external image socket"],
"width": width,
"height": height,
"frames": frames,
"batch_size": batch_size,
"fps": fps,
"seed": seed,
"generation_target": generation_target,
"video_format": video_format,
"force_vhs_video_combine": force_vhs_video_combine,
"output_target": output_target,
"output_folder_mode": output_folder_mode,
"output_folder": output_folder,
"custom_output_folder": custom_output_folder,
"preferred_video_dir": r"D:\AI NSFW\VID",
"preferred_image_dir": r"D:\AI NSFW\PIC",
"preview_policy": "source preview only; result last-frame only when continuation is enabled",
"event_strategy_note": "optional future control field; currently stored in Event Program report, not required for animation",
"execution_mode": execution_mode,
"branch_mode_requested": branch_mode,
"branch_mode_active": active_branch_mode,
"cleanup_timing": cleanup_timing,
"ksampler_replacement": "EventSamplerCore",
"ksampler_windows": {
"primary": {"start": primary_start_step, "end": primary_end_step, "add_noise": "enable", "return_leftover_noise": "enable"},
"secondary": {"start": secondary_start_step, "end": secondary_end_step, "add_noise": "disable", "return_leftover_noise": "disable"},
"global_steps": global_steps,
},
}
packet["metadata"]["wan_event_internal_topology"] = {
"universal_event_node_rule": "every internal stage must pass through an Event Horizon; relation center is Event Singularity",
"event_horizon_model": {
"EventHorizon": "boundary layer where a technical node input/output transition becomes Event-Horizon-aware",
"EventSingularity": "EventSingularity center where input state, observed behavior, and output state collapse into one relation point",
"formula": "NodeInputState + NodeObservedBehavior = EventSingularity = NodeOutputState"
},
"terminal_route": [
"EventTextEncodePositive",
"EventTextEncodeNegative",
"EventImageScaleStart",
"EventWanImageToVideoSeed",
"EventModelShiftHigh",
"EventSamplerHigh",
"EventCleanupBetweenSamplers",
"EventModelShiftLow",
"EventSamplerLow",
"EventVAEDecodeTiled",
"EventVideoCombine",
"EventSaveReport",
],
"internal_hidden_states": [
"conditioning_positive",
"conditioning_negative",
"wan_latent_seed",
"latent_before_high",
"latent_after_high",
"delta_high",
"latent_before_low",
"latent_after_low",
"delta_low",
"decoded_frames",
"video_file",
"S0_wan_terminal",
],
}
if str(positive_prompt or "").strip():
packet, sig, proj, conf = self._make_text_signal(packet, positive_prompt, "StrategyCurrent", "wan_positive_prompt_route", "EventTextEncodePositive")
signal_ids.append(sig["id"])
conflict_ids.extend(conf)
if str(negative_prompt or "").strip():
packet, sig, proj, conf = self._make_text_signal(packet, negative_prompt, "NegativeConstraint", "wan_negative_prompt_route", "EventTextEncodeNegative")
signal_ids.append(sig["id"])
conflict_ids.extend(conf)
if str(event_strategy or "").strip():
packet, sig, proj, conf = self._make_text_signal(packet, event_strategy, "EventProgramStrategy", "wan_event_strategy_route", "Wan_event_strategy")
signal_ids.append(sig["id"])
conflict_ids.extend(conf)
if image is not None:
packet, sig, proj, conf = _read_signal(
packet, TECH_IMAGE, SPACE_IMAGE, image, "EventSourcePreview", "SourceAnchor",
"wan_source_image_route", "ImageOutcomeReader", route_position="start_preview"
)
signal_ids.append(sig["id"])
conflict_ids.extend(conf)
if generation_target == "VIDEO" and image is None:
failure_reason = "VIDEO target requires source image for WanImageToVideo; no image was connected."
result_status = "FAILED"
packet, cid = self._fail_conflict(packet, "WanEventWorkflowCore/InputValidation", failure_reason, {"generation_target": generation_target})
conflict_ids.append(cid)
if secondary_model is None and branch_mode == "DUAL_HIGH_LOW":
conflict = make_conflict(
CONFLICT_FROZEN_ROUTE_MISSING,
severity=SEV_LOW,
stage_position="WanEventWorkflowCore",
suspected_cause="DUAL_HIGH_LOW requested but secondary_model was not connected.",
observed_symptom="secondary branch disabled; falling back to single branch.",
suggested_response="Connect secondary_model for Low branch, or set branch_mode=SINGLE/AUTO.",
metadata={"route": "secondary_model", "fallback": "single_branch"},
)
packet = add_conflict(packet, conflict)
conflict_ids.append(conflict["id"])
_, _ = self._run_branch_barrier(
phase_name="BEFORE_GENERATION",
cleanup_timing=cleanup_timing,
strategy_state=None,
strategy_label="before_generation_context",
records=execution_records,
cleanup_records=cleanup_records,
barrier_records=branch_barrier_records,
route_id="route_cleanup_before_generation",
stage_name="EventCleanupBeforeGeneration",
formula_role="MEMORY pressure -> MEMORY prepared state",
observed_behavior="Smart Branch Barrier before generation releases disposable memory without mutating generation tensors.",
next_requirement="generation starts with reduced stale memory pressure",
use_custom_cleanup_nodes=use_custom_cleanup_nodes,
use_custom_chain=False,
cleanup_label="before_generation",
)
if execution_mode in ("RUN", "TRY_EVENT_HORIZON_CASCADE", "TRY_WAN_SOLO_EXECUTION") and result_status != "FAILED":
execution_records.append({
"stage": "EventHorizonCascadeExecutionGate",
"status": "running",
"execution_mode": execution_mode,
"cascade_count": requested_cascade_count,
"frames_per_cascade": frames,
})
try:
if image is None:
raise RuntimeError("Event Horizon VIDEO route requires source image input.")
positive = self._encode_text(clip, positive_prompt, execution_records, label='TextEncodePositive')
negative = self._encode_text(clip, negative_prompt, execution_records, label='TextEncodeNegative')
execution_records.append({"stage": "EventTextEncode", "status": "ok", "formula": "prompt + CLIP behavior = S_text = conditioning"})
self._stage_delay(stage_delay_seconds, execution_records, "after_text_encode")
scaled_image = self._scale_image(image, width, height, image_upscale_method, image_crop, execution_records)
wan_positive, wan_negative, wan_latent = self._wan_image_to_video(
positive, negative, vae, scaled_image, width, height, frames, batch_size, execution_records
)
self._stage_delay(stage_delay_seconds, execution_records, "after_wan_image_to_video")
packet, wan_latent_sig, wan_latent_proj, conf = _read_signal(
packet, TECH_LATENT, SPACE_LATENT, wan_latent,
"EventWanImageToVideoSeed", "OutcomePrevious",
"wan_i2v_latent_seed_route", "LatentEventReader", route_position="wan_i2v_latent_seed"
)
conflict_ids.extend(conf)
high_model = self._apply_sd3_shift(primary_model, primary_sd3_shift, "high", execution_records)
high_window = EventSamplerWindow(
branch_name="high",
branch_role="coarse_motion_global_trajectory",
seed=int(seed),
steps=int(global_steps),
cfg=float(primary_cfg),
sampler_name=str(sampler_name),
scheduler=str(scheduler),
start_at_step=int(primary_start_step),
end_at_step=int(primary_end_step),
add_noise="enable",
return_with_leftover_noise="enable",
sd3_shift=float(primary_sd3_shift),
)
latent_after_high, high_result = self._event_sample_window(
high_model, wan_positive, wan_negative, wan_latent, high_window, execution_records
)
self._stage_delay(stage_delay_seconds, execution_records, "after_high_sampler")
packet["metadata"].setdefault("event_sampler_results", {})["high"] = high_result.to_metadata()
packet, high_sig, high_proj, conf = _read_signal(
packet, TECH_LATENT, SPACE_LATENT, latent_after_high,
"EventSamplerHigh_latent_after", "OutcomeNext",
"wan_high_after_route", "LatentEventReader", route_position="after_event_sampler_high"
)
conflict_ids.extend(conf)
delta_high, delta_err = compute_tensor_delta(wan_latent, latent_after_high)
if delta_high is not None:
packet, delta_sig, delta_proj, conf = _read_signal(
packet, TECH_DELTA, SPACE_DELTA, delta_high,
"EventSamplerHigh_delta", "ObservedBehaviorCurrent",
"wan_high_delta_route", "DeltaReader",
metadata={"before_signal_id": wan_latent_sig["id"], "after_signal_id": high_sig["id"], "before_ref": extract_latent_samples(wan_latent)}
)
conflict_ids.extend(conf)
rel = make_event_relation(
relation_type=REL_TRANSFORMS_INTO,
source_signal_ids=[wan_latent_sig["id"], delta_sig["id"]],
target_signal_ids=[high_sig["id"]],
source_projection_ids=[wan_latent_proj["id"], delta_proj["id"]],
target_projection_ids=[high_proj["id"]],
operator_name="EventSamplerCore",
formula_meaning="EventSamplerHigh transforms Wan latent seed into high/noise motion latent",
local_strategy_id="S0_wan_terminal.high_sampler",
equality_status=EQ_UNKNOWN,
metadata={"branch": "high", "sampler_replacement": "EventSamplerCore"},
)
packet = add_relation(packet, rel)
relation_ids.append(rel["id"])
else:
execution_records.append({"stage": "EventSamplerHigh_delta", "status": "unavailable", "error": str(delta_err)})
final_latent = latent_after_high
delta_low = None
if active_branch_mode == "DUAL_HIGH_LOW" and secondary_model is not None:
latent_before_low = latent_after_high
latent_before_low, _ = self._run_branch_barrier(
phase_name="BETWEEN_SAMPLERS",
cleanup_timing=cleanup_timing,
strategy_state=latent_before_low,
strategy_label="strategy_carrier_low",
records=execution_records,
cleanup_records=cleanup_records,
barrier_records=branch_barrier_records,
route_id="route_cleanup_between_samplers",
stage_name="EventCleanupBetweenSamplers",
formula_role="LATENT high output + cleanup behavior -> LATENT low input",
observed_behavior="Smart Branch Barrier between samplers preserves high latent StrategyCarrier while releasing disposable memory.",
next_requirement="low sampler receives preserved high output latent",
use_custom_cleanup_nodes=use_custom_cleanup_nodes,
use_custom_chain=True,
cleanup_label="between_high_low",
)
packet, low_before_sig, low_before_proj, conf = _read_signal(
packet, TECH_LATENT, SPACE_LATENT, latent_before_low,
"EventSamplerLow_latent_before", "OutcomePrevious",
"wan_low_before_route", "LatentEventReader", route_position="before_event_sampler_low"
)
conflict_ids.extend(conf)
low_model = self._apply_sd3_shift(secondary_model, secondary_sd3_shift, "low", execution_records)
low_window = EventSamplerWindow(
branch_name="low",
branch_role="refinement_detail_stabilization",
seed=int(seed),
steps=int(global_steps),
cfg=float(secondary_cfg),
sampler_name=str(sampler_name),
scheduler=str(scheduler),
start_at_step=int(secondary_start_step),
end_at_step=int(secondary_end_step),
add_noise="disable",
return_with_leftover_noise="disable",
sd3_shift=float(secondary_sd3_shift),
)
latent_after_low, low_result = self._event_sample_window(
low_model, wan_positive, wan_negative, latent_before_low, low_window, execution_records
)
self._stage_delay(stage_delay_seconds, execution_records, "after_low_sampler")
packet["metadata"].setdefault("event_sampler_results", {})["low"] = low_result.to_metadata()
packet, low_after_sig, low_after_proj, conf = _read_signal(
packet, TECH_LATENT, SPACE_LATENT, latent_after_low,
"EventSamplerLow_latent_after", "OutcomeNext",
"wan_low_after_route", "LatentEventReader", route_position="after_event_sampler_low"
)
conflict_ids.extend(conf)
delta_low, delta_err = compute_tensor_delta(latent_before_low, latent_after_low)
if delta_low is not None:
packet, delta2_sig, delta2_proj, conf = _read_signal(
packet, TECH_DELTA, SPACE_DELTA, delta_low,
"EventSamplerLow_delta", "ObservedBehaviorCurrent",
"wan_low_delta_route", "DeltaReader",
metadata={"before_signal_id": low_before_sig["id"], "after_signal_id": low_after_sig["id"], "before_ref": extract_latent_samples(latent_before_low)}
)
conflict_ids.extend(conf)
rel = make_event_relation(
relation_type=REL_TRANSFORMS_INTO,
source_signal_ids=[low_before_sig["id"], delta2_sig["id"]],
target_signal_ids=[low_after_sig["id"]],
source_projection_ids=[low_before_proj["id"], delta2_proj["id"]],
target_projection_ids=[low_after_proj["id"]],
operator_name="EventSamplerCore",
formula_meaning="EventSamplerLow refines high latent into low/detail final latent",
local_strategy_id="S0_wan_terminal.low_sampler",
equality_status=EQ_UNKNOWN,
metadata={"branch": "low", "sampler_replacement": "EventSamplerCore"},
)
packet = add_relation(packet, rel)
relation_ids.append(rel["id"])
else:
execution_records.append({"stage": "EventSamplerLow_delta", "status": "unavailable", "error": str(delta_err)})
final_latent = latent_after_low
self._dual_branch_delta_coupling_math(
delta_high=delta_high,
delta_low=delta_low,
records=execution_records,
active_branch_mode=active_branch_mode,
cascade_count=requested_cascade_count,
)
generated_latent = final_latent
generated_frames = self._decode_tiled(
vae, generated_latent,
decode_tile_size, decode_overlap, decode_temporal_size, decode_temporal_overlap,
execution_records
)
self._math_tensor_summary(generated_frames, execution_records, "EventMath_decoded_frames", strict=False)
self._frame_motion_math(generated_frames, execution_records, "EventMath_decoded_frame_motion")
self._stage_delay(stage_delay_seconds, execution_records, "after_decode")
packet, img_sig, img_proj, conf = _read_signal(
packet, TECH_IMAGE, SPACE_IMAGE, generated_frames,
"EventVAEDecodeTiled_frames", "VisibleOutcome",
"wan_decoded_frames_route", "ImageOutcomeReader", route_position="decoded_frames"
)
conflict_ids.extend(conf)
# Event Horizon extension.
self._pause_flag_triggered = False
if resume_frame_index != -1 and _EVENT_HORIZON_CASCADE_CACHE["latent"] is not None:
# Resuming from a pause
generated_latent = _EVENT_HORIZON_CASCADE_CACHE["latent"]
generated_frames = _EVENT_HORIZON_CASCADE_CACHE["frames"]
# Trim the latent tensor in the temporal dimension T.
# For Wan2.1, T = (frames - 1) // 4 + 1
target_t = max(1, (resume_frame_index - 1) // 4 + 1)
if generated_latent['samples'].shape[2] > target_t:
generated_latent['samples'] = generated_latent['samples'][:, :, :target_t, :, :]
if generated_frames is not None and generated_frames.shape[0] > resume_frame_index:
generated_frames = generated_frames[:resume_frame_index, :, :, :]
segment_batches = [generated_frames]
current_cascade_image = self._last_frame_image(generated_frames, width, height)
start_segment = _EVENT_HORIZON_CASCADE_CACHE["segment_index"] + 1
execution_records.append({"stage": "EventHorizonCascadeResume", "status": "resumed", "resume_frame_index": resume_frame_index, "start_segment": start_segment})
else:
segment_batches = [generated_frames]
current_cascade_image = self._last_frame_image(generated_frames, width, height)
start_segment = 2
# Handle pause at cascade 1
if pause_after_cascade_1 and start_segment == 2 and resume_frame_index == -1:
_EVENT_HORIZON_CASCADE_CACHE["latent"] = generated_latent
_EVENT_HORIZON_CASCADE_CACHE["frames"] = generated_frames
_EVENT_HORIZON_CASCADE_CACHE["segment_index"] = 1
execution_records.append({"stage": "EventHorizonCascadePause_1", "status": "paused"})
self._pause_flag_triggered = True
# Skip the cascade loop
requested_cascade_count = 1
if requested_cascade_count >= start_segment:
execution_records.append({
"stage": "EventHorizonCascadeBegin",
"status": "begin",
"cascade_count": requested_cascade_count,
"frames_per_cascade": frames,
})
for segment_index in range(start_segment, requested_cascade_count + 1):
next_frames, current_cascade_image, generated_latent = self._run_event_horizon_segment_core(
segment_index=segment_index,
source_image=current_cascade_image,
primary_model=primary_model,
secondary_model=secondary_model,
clip=clip,
vae=vae,
positive_prompt=positive_prompt,
negative_prompt=negative_prompt,
active_branch_mode=active_branch_mode,
width=width,
height=height,
frames=frames,
batch_size=batch_size,
seed=seed,
sampler_name=sampler_name,
scheduler=scheduler,
global_steps=global_steps,
primary_cfg=primary_cfg,
secondary_cfg=secondary_cfg,
primary_start_step=primary_start_step,
primary_end_step=primary_end_step,
secondary_start_step=secondary_start_step,
secondary_end_step=secondary_end_step,
primary_sd3_shift=primary_sd3_shift,
secondary_sd3_shift=secondary_sd3_shift,
decode_tile_size=decode_tile_size,
decode_overlap=decode_overlap,
decode_temporal_size=decode_temporal_size,
decode_temporal_overlap=decode_temporal_overlap,
image_upscale_method=image_upscale_method,
image_crop=image_crop,
cleanup_timing=cleanup_timing,
use_custom_cleanup_nodes=use_custom_cleanup_nodes,
stage_delay_seconds=stage_delay_seconds,
records=execution_records,
cleanup_records=cleanup_records,
barrier_records=branch_barrier_records,
)
if segment_index > 1:
next_frames = self._drop_first_frame_batch(next_frames, execution_records, segment_index)
self._cascade_boundary_math(segment_batches[-1], next_frames, execution_records, segment_index)
segment_batches.append(next_frames)
pause_flags = {
2: pause_after_cascade_2,
3: pause_after_cascade_3,
4: pause_after_cascade_4
}
if pause_flags.get(segment_index, False):
_EVENT_HORIZON_CASCADE_CACHE["latent"] = generated_latent
_EVENT_HORIZON_CASCADE_CACHE["frames"] = self._concat_frame_batches(segment_batches, execution_records)
_EVENT_HORIZON_CASCADE_CACHE["segment_index"] = segment_index
execution_records.append({"stage": f"EventHorizonCascadePause_{segment_index}", "status": "paused"})
self._pause_flag_triggered = True
break
generated_frames = self._concat_frame_batches(segment_batches, execution_records)
self._frame_motion_math(generated_frames, execution_records, "EventMath_concatenated_frame_motion")
execution_records.append({
"stage": "EventHorizonCascadeEnd",
"status": "ok",
"segments": requested_cascade_count,
"total_requested_frames": requested_cascade_count * int(frames),
})
self._event_universal_stage_math(
execution_records,
"EventHorizonCascadeEnd",
input_state=segment_batches[0] if segment_batches else None,
output_state=generated_frames,
observed_behavior="multiple cascade frame batches concatenated into one continuous output sequence",
formula_role="FRAME_BATCHES segment outcomes -> FRAMES full cascade outcome",
route_id="route_cascade_concat",
next_requirement="video combine requires one ordered frame batch",
control_mode="REPORT_ONLY",
metadata={"segments": requested_cascade_count, "frames_per_cascade": int(frames)},
)
if save_frames or save_output_image or generation_target == "IMAGE":
try:
self._save_image_attempt(generated_frames, save_prefix, execution_records)
except Exception as e:
execution_records.append({"stage": "EventSaveFrames", "status": "failed", "error": str(e)})
if requested_video and save_video:
try:
saved_video_path = self._video_combine_attempt(generated_frames, fps, save_prefix, execution_records, video_format=video_format, force_vhs=force_vhs_video_combine, output_target=output_target, output_folder_mode=output_folder_mode, output_folder=output_folder, custom_output_folder=custom_output_folder)
result_status = "VIDEO"
except Exception as e:
execution_records.append({"stage": "EventVideoCombine", "status": "failed", "error": str(e)})
if generation_target == "VIDEO":
result_status = "FAILED"
failure_reason = f"VIDEO target requested but video combine failed: {e}"
packet, cid = self._fail_conflict(packet, "EventVideoCombine", failure_reason, {"generation_target": generation_target})
conflict_ids.append(cid)
else:
result_status = "FRAMES"
else:
result_status = "FRAMES" if generated_frames is not None else "NONE"
ui_images = []
if getattr(self, "_pause_flag_triggered", False):
result_status = "PAUSED"
if generated_frames is not None:
ui_images = self._save_pause_frames_to_temp(generated_frames)
video_ui_payload = self._extract_vhs_ui_payload_from_records(execution_records)
generated_frames, _ = self._run_branch_barrier(
phase_name="AFTER_GENERATION",
cleanup_timing=cleanup_timing,
strategy_state=generated_frames,
strategy_label="final_frames_outcome",
records=execution_records,
cleanup_records=cleanup_records,
barrier_records=branch_barrier_records,
route_id="route_cleanup_after_generation",
stage_name="EventCleanupAfterGeneration",
formula_role="OUTPUT frames + cleanup behavior -> saved/previewable output state",
observed_behavior="Smart Branch Barrier after generation preserves final frames while releasing disposable memory.",
next_requirement="report/output finalization continues after cleanup",
use_custom_cleanup_nodes=use_custom_cleanup_nodes,
use_custom_chain=False,
cleanup_label="after_generation",
)
except Exception as e:
failure_reason = str(e)
result_status = "FAILED"
execution_records.append({"stage": "terminal_wan_execution", "status": "failed", "error": failure_reason})
packet, cid = self._fail_conflict(packet, "WanEventWorkflowCore/Execution", failure_reason, {"execution_mode": execution_mode, "generation_target": generation_target})
conflict_ids.append(cid)
elif execution_mode == "REPORT_ONLY":
result_status = "INCOMPLETE"
execution_records.append({"stage": "execution", "status": "skipped_report_only"})
else:
result_status = "INCOMPLETE"
execution_records.append({
"stage": "execution",
"status": "skipped_unsupported_execution_mode",
"execution_mode": execution_mode,
"expected_modes": ["RUN", "TRY_EVENT_HORIZON_CASCADE", "TRY_WAN_SOLO_EXECUTION", "REPORT_ONLY"],
})
# r12 safety: saved_video_path must be a clean string path, never raw VHS payload.
if not isinstance(saved_video_path, str):
raw_saved_video_payload = str(saved_video_path)[:1000]
try:
saved_video_path = self._extract_path_from_video_result(saved_video_path, save_prefix, execution_records)
except Exception as e:
execution_records.append({"stage": "EventVideoPathFinalCoerce", "status": "failed", "error": str(e), "raw_payload": raw_saved_video_payload})
saved_video_path = ""
else:
execution_records.append({"stage": "EventVideoPathFinalCoerce", "status": "ok", "path": saved_video_path, "raw_payload": raw_saved_video_payload})
barrier_phase_counts = {}
barrier_with_strategy = 0
barrier_preserved = 0
for item in branch_barrier_records:
phase = str(item.get("barrier_phase", "unknown"))
barrier_phase_counts[phase] = barrier_phase_counts.get(phase, 0) + 1
if item.get("strategy_state_required"):
barrier_with_strategy += 1
if item.get("strategy_state_preserved"):
barrier_preserved += 1
packet["metadata"]["cleanup_records"] = cleanup_records
packet["metadata"]["branch_barrier_records"] = branch_barrier_records
packet["metadata"]["branch_barrier_summary"] = {
"record_count": len(branch_barrier_records),
"phase_counts": barrier_phase_counts,
"strategy_state_checks": barrier_with_strategy,
"strategy_state_preserved_count": barrier_preserved,
"observer_only": True,
}
packet["metadata"]["execution_records"] = execution_records
packet["metadata"]["result_status"] = {
"generation_target": generation_target,
"result_status": result_status,
"failure_reason": failure_reason,
"saved_video_path": saved_video_path,
"fake_success_prevented": True,
"source_image_was_not_returned_as_successful_video": True,
"source_image_file": str(source_image_file) if source_image_file else "",
}
packet["metadata"]["program_output_policy"] = {
"terminal_mode": bool(terminal_mode),
"normal_result": "saved files + status, no IMAGE outputs",
"continuation_outputs": "removed from main terminal node in r15",
"image_passthrough_policy": "never claim source image as generated VIDEO result",
"video_required_for_success": generation_target == "VIDEO",
}
# update program status with final result
packet["metadata"]["event_program_status"]["result_status"] = result_status
packet["metadata"]["event_program_status"]["saved_video_path"] = saved_video_path
packet["metadata"]["event_program_status"]["failure_reason"] = failure_reason
packet = self._event_core_body_finalize(packet, execution_records, result_status, saved_video_path, failure_reason)
packet, sstate = build_sstate_from_packet(packet, position="S0_wan_terminal", active_relation_ids=relation_ids)
packet = record_stage(
packet,
stage_name="WanEventWorkflowCore",
action="BUILD_TERMINAL_WAN_REPORT",
observed_behavior="Built terminal Wan Event Program report with save-first output policy",
input_signal_ids=signal_ids,
relation_ids=relation_ids,
sstate_ids=[sstate["id"]] if sstate else [],
conflict_ids=conflict_ids,
formula_note="terminal external inputs -> internal Event-Horizon-aware subnodes -> EventSamplerCore high/low -> decoded frames/video -> saved result",
metadata={
"result_status": result_status,
"failure_reason": failure_reason,
"saved_video_path": saved_video_path,
"execution_records": execution_records,
},
)
if save_report:
report = build_markdown_report(packet)
if not str(report or "").strip():
execution_records.append({
"stage": "EventSaveReportBuild",
"status": "empty_report_from_builder",
"runtime_version": EVENT_HORIZON_RUNTIME_VERSION,
})
else:
execution_records.append({
"stage": "EventSaveReportBuild",
"status": "ok",
"chars": len(str(report)),
"runtime_version": EVENT_HORIZON_RUNTIME_VERSION,
})
try:
saved_report_path = self._save_report_file(report, save_prefix, "COMFY_OUTPUT", "DEFAULT", "default", "")
try:
report_size = Path(saved_report_path).stat().st_size
except Exception:
report_size = -1
execution_records.append({
"stage": "EventSaveReport",
"status": "standard_comfy_output_ok",
"path": saved_report_path,
"bytes": report_size,
"nonempty": bool(report_size and report_size > 0),
"runtime_version": EVENT_HORIZON_RUNTIME_VERSION,
})
if report_size <= 0:
raise RuntimeError(f"Report was saved but is empty: {saved_report_path}")
sidecar_info = self._save_runtime_monitor_sidecars(packet, saved_report_path, saved_video_path, save_prefix)
packet.setdefault("metadata", {})["runtime_monitor_sidecars"] = sidecar_info
execution_records.append({
"stage": "EventRuntimeMonitorSidecars",
"status": sidecar_info.get("status", "unknown"),
"json_path": sidecar_info.get("json_path", ""),
"csv_path": sidecar_info.get("csv_path", ""),
"diff_path": sidecar_info.get("diff_path", ""),
"previous_json_path": sidecar_info.get("previous_json_path", ""),
"record_count": sidecar_info.get("record_count", 0),
"motion_stage": sidecar_info.get("motion_stage", ""),
"motion_profile": sidecar_info.get("motion_profile", ""),
"motion_stability_score": sidecar_info.get("motion_stability_score", ""),
"observer_only": True,
"formula": "Runtime Monitor writes machine-readable ObservedBehavior sidecars without changing generation.",
})
report = build_markdown_report(packet)
report_size = self._rewrite_report_file(saved_report_path, report)
except Exception as e:
execution_records.append({"stage": "EventSaveReport", "status": "failed", "error": str(e), "runtime_version": EVENT_HORIZON_RUNTIME_VERSION})
saved_report_path = ""
else:
report = ""
saved_report_path = ""
execution_records.append({"stage": "EventSaveReport", "status": "disabled_by_user", "runtime_version": EVENT_HORIZON_RUNTIME_VERSION})
# Preview model:
# - source_preview always shows the input/source.
# - result_preview shows generated frames only if real generation produced frames.
# - if generation failed, result_preview is a blank placeholder, not the source image.
if enable_continuation_outputs and generated_frames is not None and result_status in ("FRAMES", "VIDEO"):
result_preview = self._representative_preview_frame(generated_frames, width, height, mode="last")
else:
result_preview = self._placeholder_image(width, height)
if not ui_images:
ui_images = self._make_ui_previews(source_preview, result_preview, save_prefix, execution_records, include_result_preview=enable_continuation_outputs)
packet["metadata"]["ui_preview"] = {
"source_preview": "source image or upload",
"result_preview": "disabled; no PreviewImage calls in terminal node",
"continuation_seed_frame": "not emitted by main terminal node in r15; use future extractor/chain node",
"ui_images_count": len(ui_images),
"video_ui_payload_returned": bool(video_ui_payload),
}
status = (
f"EventHorizon v{EVENT_HORIZON_RUNTIME_VERSION} | target={generation_target} | result={result_status} | "
f"terminal={terminal_mode} | continuation={enable_continuation_outputs} | "
f"video_path={saved_video_path or 'none'} | report_path={saved_report_path or 'none'}"
)
if failure_reason:
status += f" | failure={failure_reason}"
continuation_image = self._representative_preview_frame(generated_frames, width, height, mode="last") if (enable_continuation_outputs and generated_frames is not None) else self._placeholder_image(width, height)
continuation_latent = generated_latent if enable_continuation_outputs else None
continuation_packet = packet if enable_continuation_outputs else None
result_tuple = (
status,
saved_video_path,
saved_report_path,
report,
)
if not video_ui_payload:
video_ui_payload = {}
if ui_images:
if getattr(self, "_pause_flag_triggered", False):
video_ui_payload = {} # Wipe VHS video payload to prevent overlap
video_ui_payload["pause_frames"] = ui_images
if video_ui_payload:
return {"ui": video_ui_payload, "result": result_tuple}
return result_tuple
class EventHorizonCascadeSimple(WanEventWorkflowCore):
@classmethod
def INPUT_TYPES(cls):
return {
"required": {
"primary_model": ("MODEL",),
"clip": ("CLIP",),
"vae": ("VAE",),
"source_image_file": (_event_core_list_input_images(), {"image_upload": True}),
"positive_prompt": ("STRING", {"default": "", "multiline": True, "height": 180, "dynamicPrompts": False}),
"negative_prompt": ("STRING", {"default": "", "multiline": True, "height": 180, "dynamicPrompts": False}),
"temporal_texture_lock": ("BOOLEAN", {"default": True}),
"cascade_count": ("INT", {"default": 1, "min": 1, "max": 5}),
"pause_after_cascade_1": ("BOOLEAN", {"default": False}),
"pause_after_cascade_2": ("BOOLEAN", {"default": False}),
"pause_after_cascade_3": ("BOOLEAN", {"default": False}),
"pause_after_cascade_4": ("BOOLEAN", {"default": False}),
"resume_frame_index": ("INT", {"default": -1, "min": -1, "max": 4096}),
"frames_per_cascade": ("INT", {"default": 49, "min": 1, "max": 4096}),
"width": ("INT", {"default": 416, "min": 16, "max": 8192, "step": 8}),
"height": ("INT", {"default": 608, "min": 16, "max": 8192, "step": 8}),
"fps": ("INT", {"default": 16, "min": 1, "max": 240}),
"seed": ("INT", {"default": 359, "min": 0, "max": 0xffffffffffffffff}),
"sampler_name": ("STRING", {"default": "euler"}),
"scheduler": ("STRING", {"default": "simple"}),
"global_steps": ("INT", {"default": 4, "min": 0, "max": 10000}),
"primary_cfg": ("FLOAT", {"default": 1.0, "min": -1000.0, "max": 1000.0, "step": 0.01}),
"secondary_cfg": ("FLOAT", {"default": 1.0, "min": -1000.0, "max": 1000.0, "step": 0.01}),
"primary_start_step": ("INT", {"default": 0, "min": 0, "max": 10000}),
"primary_end_step": ("INT", {"default": 1, "min": 0, "max": 10000}),
"secondary_start_step": ("INT", {"default": 1, "min": 0, "max": 10000}),
"secondary_end_step": ("INT", {"default": 4, "min": 0, "max": 10000}),
# Keep combo UX, but include lowercase legacy tokens so stale workflows do not hard-fail before normalization.
"math_control_mode": (["OBSERVE_ONLY", "LATENT_DELTA_SCALE", "DEEP_STEP_DELTA_CONTROL", "observe_only", "latent_delta_scale", "deep_step_delta_control"], {"default": "LATENT_DELTA_SCALE"}),
"high_delta_strength": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 2.0, "step": 0.0001, "round": 0.0001}),
"low_delta_strength": ("FLOAT", {"default": 1.0, "min": 0.0, "max": 2.0, "step": 0.0001, "round": 0.0001}),
"decode_tile_size": ("INT", {"default": 512, "min": 64, "max": 8192, "step": 8}),
# Allow wider transport range; runtime normalization will clamp to safe decode constraints.
"decode_overlap": ("INT", {"default": 64, "min": 0, "max": 65535, "step": 8}),
"decode_temporal_size": ("INT", {"default": 32, "min": 1, "max": 4096}),
"decode_temporal_overlap": ("INT", {"default": 12, "min": 0, "max": 65535}),
"image_upscale_method": (["nearest-exact", "nearest", "bilinear", "area", "bicubic", "lanczos"], {"default": "nearest-exact"}),
"image_crop": (["disabled", "center"], {"default": "disabled"}),
"cleanup_timing": ([
"NONE", "BEFORE_GENERATION", "BETWEEN_SAMPLERS", "AFTER_GENERATION", "BEFORE_AND_AFTER", "ALL",
"none", "before_generation", "between_samplers", "after_generation", "before_and_after", "all",
], {"default": "ALL"}),
"save_video": ("BOOLEAN", {"default": True}),
"video_format": (["video/h264-mp4", "video/h265-mp4", "image/webp", "image/gif"], {"default": "video/h264-mp4"}),
"save_report": ("BOOLEAN", {"default": True}),
"save_prefix": ("STRING", {"default": "Event Horizon"}),
"sampler_trace_mode": (["OFF", "SHADOW_STEP_TRACE"], {"default": "OFF"}),
"sampler_trace_max_steps": ("INT", {"default": 64, "min": 1, "max": 65535}),
},
"optional": {
"secondary_model": ("MODEL",),
"image": ("IMAGE",),
"mask": ("MASK",),
}
}
RETURN_TYPES = ("STRING", "STRING", "STRING", "STRING")
RETURN_NAMES = ("status", "saved_video_path", "saved_report_path", "report")
FUNCTION = "run"
CATEGORY = "Event Equality/Event Horizon"
OUTPUT_NODE = True
@staticmethod
def _append_input_adjustment(adjustments, field, original, normalized, reason):
if original == normalized:
return
adjustments.append({
"field": str(field),
"original": _event_json_safe(original),
"normalized": _event_json_safe(normalized),
"reason": str(reason),
})
def _get_ksampler_allowed_values(self):
samplers = []
schedulers = []
try:
cls = self._get_node_class("KSamplerAdvanced")
input_types = cls.INPUT_TYPES() if hasattr(cls, "INPUT_TYPES") else {}
required = input_types.get("required", {}) if isinstance(input_types, dict) else {}
sampler_def = required.get("sampler_name")
scheduler_def = required.get("scheduler")
if isinstance(sampler_def, (list, tuple)) and sampler_def:
options = sampler_def[0]
if isinstance(options, (list, tuple)):
samplers = [str(v) for v in options if str(v).strip()]
if isinstance(scheduler_def, (list, tuple)) and scheduler_def:
options = scheduler_def[0]
if isinstance(options, (list, tuple)):
schedulers = [str(v) for v in options if str(v).strip()]
except Exception:
pass
if not samplers:
samplers = ["euler"]
if not schedulers:
schedulers = ["simple"]
return samplers, schedulers
@staticmethod
def _sanitize_save_prefix_text(value):
text = str(value or "").strip()
if not text:
return ""
text = re.sub(r'[\x00-\x1f]', "_", text)
text = re.sub(r'[<>:"/\\|?*]+', "_", text)
text = re.sub(r"\s+", " ", text).strip()
text = text.rstrip(" .")
if len(text) > 120:
text = text[:120].rstrip(" .")
return text
def _event_runtime_aliases(self):
aliases = []
try:
mappings = globals().get("NODE_CLASS_MAPPINGS", {}) or {}
for key, value in mappings.items():
if value is self.__class__:
aliases.append(str(key))
except Exception:
return []
return sorted(set(aliases))
def _normalize_clean_inputs(
self,
*,
secondary_model,
cascade_count,
frames_per_cascade,
width,
height,
fps,
seed,
sampler_name,
scheduler,
global_steps,
primary_cfg,
secondary_cfg,
primary_start_step,
primary_end_step,
secondary_start_step,
secondary_end_step,
math_control_mode,
high_delta_strength,
low_delta_strength,
decode_tile_size,
decode_overlap,
decode_temporal_size,
decode_temporal_overlap,
image_upscale_method,
image_crop,
cleanup_timing,
video_format,
save_prefix,
sampler_trace_mode,
sampler_trace_max_steps,
):
adjustments = []
allowed_samplers, allowed_schedulers = self._get_ksampler_allowed_values()
def clamp_int(field, value, default, min_value, max_value):
original = value
try:
out = int(value)
except Exception:
out = int(default)
self._append_input_adjustment(adjustments, field, original, out, "invalid_int_fallback")
clamped = max(int(min_value), min(int(max_value), int(out)))
if clamped != out:
self._append_input_adjustment(adjustments, field, out, clamped, f"clamped_to_{min_value}_{max_value}")
return clamped
def clamp_float(field, value, default, min_value, max_value, digits=None):
original = value
try:
out = float(value)
except Exception:
out = float(default)
self._append_input_adjustment(adjustments, field, original, out, "invalid_float_fallback")
clamped = max(float(min_value), min(float(max_value), float(out)))
if digits is not None:
rounded = round(clamped, int(digits))
if rounded != clamped:
self._append_input_adjustment(adjustments, field, clamped, rounded, f"rounded_{digits}_digits")
clamped = rounded
if clamped != out:
self._append_input_adjustment(adjustments, field, out, clamped, f"clamped_to_{min_value}_{max_value}")
return clamped
def clamp_enum(field, value, allowed_values, default_value, casefold=False, reason="invalid_enum_fallback"):
original = value
text = str(value if value is not None else default_value)
if casefold:
text_cmp = text.upper()
allowed_cmp = {str(v).upper(): str(v) for v in allowed_values}
if text_cmp in allowed_cmp:
out = allowed_cmp[text_cmp]
else:
out = str(default_value)
self._append_input_adjustment(adjustments, field, original, out, reason)
return out
if text in allowed_values:
return text
out = str(default_value)
self._append_input_adjustment(adjustments, field, original, out, reason)
return out
def align_to_step(field, value, step, min_value):
if step <= 1:
return int(value)
aligned = int(value) - (int(value) % int(step))
if aligned < int(min_value):
aligned = int(min_value)
if aligned != int(value):
self._append_input_adjustment(adjustments, field, value, aligned, f"aligned_to_step_{step}")
return int(aligned)
dual_branch = secondary_model is not None
cascade_count_n = clamp_int("cascade_count", cascade_count, 1, 1, 5)
frames_per_cascade_n = clamp_int("frames_per_cascade", frames_per_cascade, 49, 1, 4096)
if cascade_count_n > 1 and frames_per_cascade_n < 2:
self._append_input_adjustment(
adjustments,
"frames_per_cascade",
frames_per_cascade_n,
2,
"cascade_requires_at_least_two_frames",
)
frames_per_cascade_n = 2
width_n = align_to_step(
"width",
clamp_int("width", width, 416, 16, 8192),
step=8,
min_value=16,
)
height_n = align_to_step(
"height",
clamp_int("height", height, 608, 16, 8192),
step=8,
min_value=16,
)
fps_n = clamp_int("fps", fps, 16, 1, 240)
seed_n = clamp_int("seed", seed, 359, 0, 0xFFFFFFFFFFFFFFFF)
sampler_default = allowed_samplers[0] if allowed_samplers else "euler"
scheduler_default = allowed_schedulers[0] if allowed_schedulers else "simple"
sampler_name_n = clamp_enum(
"sampler_name",
str(sampler_name or "").strip() or sampler_default,
allowed_samplers,
sampler_default,
reason="unsupported_sampler_fallback",
)
scheduler_n = clamp_enum(
"scheduler",
str(scheduler or "").strip() or scheduler_default,
allowed_schedulers,
scheduler_default,
reason="unsupported_scheduler_fallback",
)
global_steps_n = clamp_int("global_steps", global_steps, 4, 1, 10000)
if dual_branch and global_steps_n < 2:
self._append_input_adjustment(adjustments, "global_steps", global_steps_n, 2, "dual_branch_min_steps")
global_steps_n = 2
if dual_branch:
primary_start_max = max(0, global_steps_n - 2)
primary_start_n = clamp_int("primary_start_step", primary_start_step, 0, 0, primary_start_max)
primary_end_min = primary_start_n + 1
primary_end_max = max(primary_end_min, global_steps_n - 1)
primary_end_n = clamp_int("primary_end_step", primary_end_step, primary_end_min, primary_end_min, primary_end_max)
secondary_start_n = primary_end_n
try:
secondary_start_original = int(secondary_start_step)
except Exception:
secondary_start_original = secondary_start_step
if secondary_start_original != secondary_start_n:
self._append_input_adjustment(
adjustments,
"secondary_start_step",
secondary_start_original,
secondary_start_n,
"forced_to_primary_end_for_dual_branch",
)
secondary_end_n = clamp_int(
"secondary_end_step",
secondary_end_step,
min(global_steps_n, secondary_start_n + 1),
secondary_start_n + 1,
global_steps_n,
)
else:
primary_start_n = clamp_int("primary_start_step", primary_start_step, 0, 0, max(0, global_steps_n - 1))
primary_end_n = clamp_int("primary_end_step", primary_end_step, max(1, primary_start_n + 1), primary_start_n + 1, global_steps_n)
secondary_start_n = clamp_int("secondary_start_step", secondary_start_step, primary_end_n, 0, max(0, global_steps_n - 1))
secondary_end_n = clamp_int("secondary_end_step", secondary_end_step, max(1, secondary_start_n + 1), secondary_start_n + 1, global_steps_n)
primary_cfg_n = clamp_float("primary_cfg", primary_cfg, 1.0, -1000.0, 1000.0, digits=4)
secondary_cfg_n = clamp_float("secondary_cfg", secondary_cfg, 1.0, -1000.0, 1000.0, digits=4)
math_control_mode_n = clamp_enum(
"math_control_mode",
math_control_mode,
{"OBSERVE_ONLY", "LATENT_DELTA_SCALE", "DEEP_STEP_DELTA_CONTROL"},
"OBSERVE_ONLY",
casefold=True,
)
high_delta_strength_n = clamp_float("high_delta_strength", high_delta_strength, 1.0, 0.0, 2.0, digits=4)
low_delta_strength_n = clamp_float("low_delta_strength", low_delta_strength, 1.0, 0.0, 2.0, digits=4)
decode_tile_size_n = align_to_step(
"decode_tile_size",
clamp_int("decode_tile_size", decode_tile_size, 512, 64, 8192),
step=8,
min_value=64,
)
decode_overlap_n = align_to_step(
"decode_overlap",
clamp_int("decode_overlap", decode_overlap, 64, 0, 8192),
step=8,
min_value=0,
)
max_overlap = max(0, decode_tile_size_n - 8)
if decode_overlap_n > max_overlap:
self._append_input_adjustment(adjustments, "decode_overlap", decode_overlap_n, max_overlap, "decode_overlap_cannot_exceed_tile_size_minus_step")
decode_overlap_n = max_overlap
decode_temporal_size_n = clamp_int("decode_temporal_size", decode_temporal_size, 32, 1, 4096)
decode_temporal_overlap_n = clamp_int("decode_temporal_overlap", decode_temporal_overlap, 12, 0, 4096)
max_temporal_overlap = max(0, decode_temporal_size_n - 1)
if decode_temporal_overlap_n > max_temporal_overlap:
self._append_input_adjustment(
adjustments,
"decode_temporal_overlap",
decode_temporal_overlap_n,
max_temporal_overlap,
"decode_temporal_overlap_cannot_exceed_temporal_size_minus_one",
)
decode_temporal_overlap_n = max_temporal_overlap
image_upscale_method_n = clamp_enum(
"image_upscale_method",
image_upscale_method,
{"nearest-exact", "nearest", "bilinear", "area", "bicubic", "lanczos"},
"nearest-exact",
)
image_crop_n = clamp_enum("image_crop", image_crop, {"disabled", "center"}, "disabled")
cleanup_timing_n = clamp_enum(
"cleanup_timing",
cleanup_timing,
{"NONE", "BEFORE_GENERATION", "BETWEEN_SAMPLERS", "AFTER_GENERATION", "BEFORE_AND_AFTER", "ALL"},
"ALL",
casefold=True,
)
video_format_n = clamp_enum(
"video_format",
video_format,
{"video/h264-mp4", "video/h265-mp4", "image/webp", "image/gif"},
"video/h264-mp4",
)
sampler_trace_mode_n = clamp_enum(
"sampler_trace_mode",
sampler_trace_mode,
{"OFF", "SHADOW_STEP_TRACE"},
"OFF",
casefold=True,
)
sampler_trace_max_steps_n = clamp_int("sampler_trace_max_steps", sampler_trace_max_steps, 64, 1, 65535)
save_prefix_n = self._sanitize_save_prefix_text(save_prefix)
if save_prefix_n != str(save_prefix or "").strip():
self._append_input_adjustment(adjustments, "save_prefix", save_prefix, save_prefix_n, "sanitized_for_windows_filename")
if not save_prefix_n:
save_prefix_n = "Event Horizon"
self._append_input_adjustment(adjustments, "save_prefix", save_prefix, save_prefix_n, "empty_string_fallback")
normalized = {
"cascade_count": cascade_count_n,
"frames_per_cascade": frames_per_cascade_n,
"width": width_n,
"height": height_n,
"fps": fps_n,
"seed": seed_n,
"sampler_name": sampler_name_n,
"scheduler": scheduler_n,
"global_steps": global_steps_n,
"primary_cfg": primary_cfg_n,
"secondary_cfg": secondary_cfg_n,
"primary_start_step": primary_start_n,
"primary_end_step": primary_end_n,
"secondary_start_step": secondary_start_n,
"secondary_end_step": secondary_end_n,
"math_control_mode": math_control_mode_n,
"high_delta_strength": high_delta_strength_n,
"low_delta_strength": low_delta_strength_n,
"decode_tile_size": decode_tile_size_n,
"decode_overlap": decode_overlap_n,
"decode_temporal_size": decode_temporal_size_n,
"decode_temporal_overlap": decode_temporal_overlap_n,
"image_upscale_method": image_upscale_method_n,
"image_crop": image_crop_n,
"cleanup_timing": cleanup_timing_n,
"video_format": video_format_n,
"save_prefix": save_prefix_n,
"sampler_trace_mode": sampler_trace_mode_n,
"sampler_trace_max_steps": sampler_trace_max_steps_n,
"branch_mode": "DUAL_HIGH_LOW" if dual_branch else "SINGLE",
"cascade_mode": "SOLO_1" if cascade_count_n <= 1 else f"CASCADE_{cascade_count_n}",
}
signature_source = {
"normalized": normalized,
"adjustment_count": len(adjustments),
"adjustment_reasons": sorted(
str(adj.get("reason", "unknown") or "unknown")
for adj in adjustments
if isinstance(adj, dict)
),
}
encoded = json.dumps(_event_json_safe(signature_source), sort_keys=True, ensure_ascii=True)
normalized_signature = hashlib.sha256(encoded.encode("utf-8")).hexdigest()[:24]
return {
"normalized": normalized,
"adjustments": adjustments,
"normalized_signature": normalized_signature,
"normalized_signature_source": signature_source,
}
def run(
self,
primary_model,
clip,
vae,
source_image_file,
positive_prompt,
negative_prompt,
temporal_texture_lock,
cascade_count,
frames_per_cascade,
width,
height,
fps,
seed,
sampler_name,
scheduler,
global_steps,
primary_cfg,
secondary_cfg,
primary_start_step,
primary_end_step,
secondary_start_step,
secondary_end_step,
math_control_mode,
high_delta_strength,
low_delta_strength,
decode_tile_size,
decode_overlap,
decode_temporal_size,
decode_temporal_overlap,
image_upscale_method,
image_crop,
cleanup_timing,
save_video,
video_format,
save_report,
save_prefix,
sampler_trace_mode,
sampler_trace_max_steps,
pause_after_cascade_1=False,
pause_after_cascade_2=False,
pause_after_cascade_3=False,
pause_after_cascade_4=False,
resume_frame_index=-1,
secondary_model=None,
image=None,
mask=None,
):
self._event_strategy_coupling = {"low_strength_multiplier": 1.0}
normalization = self._normalize_clean_inputs(
secondary_model=secondary_model,
cascade_count=cascade_count,
frames_per_cascade=frames_per_cascade,
width=width,
height=height,
fps=fps,
seed=seed,
sampler_name=sampler_name,
scheduler=scheduler,
global_steps=global_steps,
primary_cfg=primary_cfg,
secondary_cfg=secondary_cfg,
primary_start_step=primary_start_step,
primary_end_step=primary_end_step,
secondary_start_step=secondary_start_step,
secondary_end_step=secondary_end_step,
math_control_mode=math_control_mode,
high_delta_strength=high_delta_strength,
low_delta_strength=low_delta_strength,
decode_tile_size=decode_tile_size,
decode_overlap=decode_overlap,
decode_temporal_size=decode_temporal_size,
decode_temporal_overlap=decode_temporal_overlap,
image_upscale_method=image_upscale_method,
image_crop=image_crop,
cleanup_timing=cleanup_timing,
video_format=video_format,
save_prefix=save_prefix,
sampler_trace_mode=sampler_trace_mode,
sampler_trace_max_steps=sampler_trace_max_steps,
)
normalized = normalization.get("normalized", {})
self._event_input_normalization = normalization
cascade_count = int(normalized.get("cascade_count", 1))
frames_per_cascade = int(normalized.get("frames_per_cascade", 49))
width = int(normalized.get("width", 416))
height = int(normalized.get("height", 608))
fps = int(normalized.get("fps", 16))
seed = int(normalized.get("seed", 359))
sampler_name = str(normalized.get("sampler_name", "euler"))
scheduler = str(normalized.get("scheduler", "simple"))
global_steps = int(normalized.get("global_steps", 4))
primary_cfg = float(normalized.get("primary_cfg", 1.0))
secondary_cfg = float(normalized.get("secondary_cfg", 1.0))
primary_start_step = int(normalized.get("primary_start_step", 0))
primary_end_step = int(normalized.get("primary_end_step", 1))
secondary_start_step = int(normalized.get("secondary_start_step", 1))
secondary_end_step = int(normalized.get("secondary_end_step", 4))
math_control_mode = str(normalized.get("math_control_mode", "OBSERVE_ONLY"))
high_delta_strength = float(normalized.get("high_delta_strength", 1.0))
low_delta_strength = float(normalized.get("low_delta_strength", 1.0))
decode_tile_size = int(normalized.get("decode_tile_size", 512))
decode_overlap = int(normalized.get("decode_overlap", 64))
decode_temporal_size = int(normalized.get("decode_temporal_size", 32))
decode_temporal_overlap = int(normalized.get("decode_temporal_overlap", 12))
image_upscale_method = str(normalized.get("image_upscale_method", "nearest-exact"))
image_crop = str(normalized.get("image_crop", "disabled"))
cleanup_timing = str(normalized.get("cleanup_timing", "ALL"))
video_format = str(normalized.get("video_format", "video/h264-mp4"))
save_prefix = str(normalized.get("save_prefix", "Event Horizon"))
sampler_trace_mode = str(normalized.get("sampler_trace_mode", "OFF"))
sampler_trace_max_steps = int(normalized.get("sampler_trace_max_steps", 64))
branch_mode = str(normalized.get("branch_mode", "SINGLE"))
cascade_mode = str(normalized.get("cascade_mode", "SOLO_1"))
self._event_requested_runtime_controls = {
"cascade_count": int(cascade_count),
"frames_per_cascade": int(frames_per_cascade or 49),
"width": int(width),
"height": int(height),
"fps": int(fps),
"seed": int(seed),
"branch_mode": str(branch_mode),
}
# Shift is intentionally disabled in the clean node.
# If a model needs SD3 shift, apply it outside this node before connecting MODEL.
if bool(temporal_texture_lock):
positive_prompt = (str(positive_prompt or "").rstrip() + "\n"
"Small liquid droplets, sweat beads, reflections, and tiny highlights keep a stable temporal identity across frames; "
"any visible liquid motion follows one continuous gravity-consistent direction with no sudden reversal or reset.")
negative_prompt = (str(negative_prompt or "").rstrip() + "\n"
"upward moving droplets, reversing sweat, liquid moving against gravity, flickering specular noise, "
"jumping highlights, temporal reset, frame snap, duplicated motion boundary, jittering droplets, crawling texture")
self._event_math_control_mode = str(math_control_mode or "OBSERVE_ONLY")
self._event_delta_strengths = {
"high": float(high_delta_strength),
"low": float(low_delta_strength),
}
self._event_requested_math_controls = {
"math_control_mode": str(math_control_mode or "OBSERVE_ONLY"),
"high_delta_strength": float(high_delta_strength),
"low_delta_strength": float(low_delta_strength),
"active_math_sampler_path": "semantic_overlay_native_sampler_when_latent_delta_scale",
"high_low_strategy_carrier_coupling": True,
"precision_step": 0.0001,
"precision_round": 0.0001,
"sampler_trace_mode": str(sampler_trace_mode or "OFF").upper(),
"sampler_trace_max_steps": int(sampler_trace_max_steps or 64),
}
self._event_sampler_trace = {
"mode": str(sampler_trace_mode or "OFF").upper(),
"max_steps": int(sampler_trace_max_steps or 64),
}
return super().run(
primary_model=primary_model,
clip=clip,
vae=vae,
source_image_file=source_image_file,
positive_prompt=positive_prompt,
negative_prompt=negative_prompt,
event_strategy="r44: one external Event Horizon node with internal Event Core Body, RouteMemory, S-Wire, and universal stage math",
generation_target="VIDEO",
terminal_mode=True,
enable_continuation_outputs=False,
execution_mode="RUN",
branch_mode=branch_mode,
cascade_count=cascade_count,
cascade_mode=cascade_mode,
frames_per_cascade=int(frames_per_cascade or 49),
width=int(width),
height=int(height),
frames=int(frames_per_cascade or 49),
batch_size=1,
fps=int(fps),
seed=int(seed),
sampler_name=sampler_name,
scheduler=scheduler,
global_steps=int(global_steps),
primary_cfg=float(primary_cfg),
secondary_cfg=float(secondary_cfg),
primary_start_step=int(primary_start_step),
primary_end_step=int(primary_end_step),
secondary_start_step=int(secondary_start_step),
secondary_end_step=int(secondary_end_step),
primary_sd3_shift=0.0,
secondary_sd3_shift=0.0,
decode_tile_size=int(decode_tile_size),
decode_overlap=int(decode_overlap),
decode_temporal_size=int(decode_temporal_size),
decode_temporal_overlap=int(decode_temporal_overlap),
image_upscale_method=image_upscale_method,
image_crop=image_crop,
cleanup_timing=cleanup_timing,
stage_delay_seconds=0.0,
use_custom_cleanup_nodes=True,
save_video=bool(save_video),
video_format=video_format,
force_vhs_video_combine=True,
save_frames=False,
save_report=bool(save_report),
output_target="COMFY_OUTPUT",
save_output_image=False,
save_prefix=save_prefix,
pause_after_cascade_1=pause_after_cascade_1,
pause_after_cascade_2=pause_after_cascade_2,
pause_after_cascade_3=pause_after_cascade_3,
pause_after_cascade_4=pause_after_cascade_4,
resume_frame_index=resume_frame_index,
output_folder_mode="DEFAULT",
output_folder="default",
custom_output_folder="",
report_detail="STANDARD",
secondary_model=secondary_model,
image=image,
mask=mask,
)
NODE_CLASS_MAPPINGS = {
# Public clean release: one visible ComfyUI node.
# Development/debug classes remain internal implementation details.
"EventHorizon": EventHorizonCascadeSimple,
}
NODE_DISPLAY_NAME_MAPPINGS = {
"EventHorizon": "Event Horizon",
}