"""Combine module resolver — syntax-aware template fill. Reads payload.template (or instance.template_override when set), resolves all $var / {a|b|c} / {N$$sep$$...} constructs against the runtime ctx, binds the result to payload.output_var. Its own negative is payload.negative, or instance.negative_override when that is a string. Surface-gated: combine resolves $var (it's the consumer) but not @{uuid} (refs only resolve from wildcard option values — RefOutOfSurfaceError). Seed lock: instance.locked_seed pins {a|b|c} resolution per instance; without it, derives from chain seed via the shared engine.modules._seed.derive_module_rng helper. """ from __future__ import annotations import re from typing import Any from engine import negatives from engine.modules import build_resolve_ctx from engine.modules._detail import module_detail from engine.modules._seed import derive_module_rng, instance_key, stream_key from engine.modules.dispatcher import ModuleHandler from engine.syntax import resolve_text #: `$name` (not `$$`) in a template, for WP Debug's list of what it read. _TEMPLATE_VAR_RE = re.compile(r"(? None: if not isinstance(payload, dict): raise ValueError("combine payload must be an object") template = payload.get("template") if not isinstance(template, str): raise ValueError("combine payload.template must be a string") # Empty template means the module emits an empty string for its # output_var — almost always a config mistake (the user removed # the template content but forgot the module is still in the # chain). Reject so the broken state surfaces at save time. if not template: raise ValueError("combine payload.template must not be empty") if len(template) > _MAX_TEMPLATE_LEN: raise ValueError( f"combine payload.template must be at most {_MAX_TEMPLATE_LEN} " f"chars (got {len(template)})" ) output_var = payload.get("output_var") if not isinstance(output_var, str) or not output_var: raise ValueError("combine payload.output_var must be a non-empty string") if len(output_var) > _MAX_IDENT_LEN: raise ValueError( f"combine payload.output_var must be at most {_MAX_IDENT_LEN} " f"chars (got {len(output_var)})" ) if output_var.startswith("__"): raise ValueError( "combine payload.output_var must not start with '__' " "(reserved for engine-internal keys)" ) if not _IDENT_RE.match(output_var): raise ValueError( f"combine payload.output_var {output_var!r} is not a valid identifier" ) input_vars = payload.get("input_vars", []) if not isinstance(input_vars, list): raise ValueError("combine payload.input_vars must be a list") for v in input_vars: if not isinstance(v, str): raise ValueError("combine payload.input_vars entries must be strings") neg = payload.get("negative") if neg is not None and not isinstance(neg, str): raise ValueError("combine payload.negative must be a string") @classmethod def resolve( cls, payload: dict[str, Any], instance: dict[str, Any], ctx: Any, ) -> dict[str, str]: cls.validate_payload(payload) # instance.template_override wins over payload.template — same # precedence pattern wildcard's option_weights / fixed_values' # values_overrides use. override = instance.get("template_override") template: str = override if isinstance(override, str) else payload["template"] # instance.variable_binding rebinds the produced var (mirrors # wildcard precedence). Empty string / non-string falls back to # payload.output_var. Strip leading `$` so callers can write # either `$foo` or `foo` in the modal. binding_override = instance.get("variable_binding") if isinstance(binding_override, str) and binding_override.strip(): output_var: str = binding_override.lstrip("$").strip() else: output_var = payload["output_var"] # Effective seed selection (mirrors WildcardHandler): # - locked_seed when present → reproducible per-instance # - chain seed otherwise → re-rolls each queue # Both feed derive_module_rng(seed, instance key) so locking with # locked_seed = chain_seed reproduces the unlocked roll exactly. if instance.get("seed_scope") == "hold": # Held: reuse the frame-0 resolved value verbatim (from the # pipeline's base pass) so a held combine's alternation stays # frozen across the whole run, nested refs included. _hb = ctx.get("__wp_hold_base_ctx__") if isinstance(_hb, dict) and output_var in _hb: held_detail = module_detail(ctx) if held_detail is not None: held_detail["held"] = True negatives.copy_from(ctx, _hb, [output_var]) return {output_var: _hb[output_var]} chain_seed = int(ctx.get("__wp_node_seed_hold__", ctx.get("__wp_node_seed__", 0)) or 0) else: chain_seed = int(ctx.get("__wp_node_seed__", 0) or 0) locked_seed = instance.get("locked_seed") if isinstance(locked_seed, (int, float)): effective_seed = int(locked_seed) else: effective_seed = chain_seed ikey = instance_key(ctx, output_var) rng = derive_module_rng(effective_seed, ikey) # Patch ctx for the scope of this resolve so build_resolve_ctx # picks up our derived RNG instead of the chain RNG. ctx_local = {**ctx, "__wp_rng__": rng} resolve_ctx = build_resolve_ctx( ctx_local, surface="combine", stream=stream_key(effective_seed, ikey), ) with negatives.collecting(resolve_ctx) as col: result = resolve_text(template, resolve_ctx) # Send-to-negative: the phrase carries the negatives of every variable # its template rendered, plus its own (resolved quietly on the # combine's negative stream so its {a|b} never shifts the value's). # instance.negative_override wins like template_override ("" clears). neg_override = instance.get("negative_override") own = negatives.clean_negative( neg_override if isinstance(neg_override, str) else payload.get("negative")) entries = negatives.entries_for_reads(ctx, col.reads) entries += negatives.own_entries( [own] if own else [], resolve_ctx, negatives.neg_rng(effective_seed, ikey), pick=None, source=output_var, ) negatives.set_entries(ctx, output_var, entries) detail = module_detail(ctx) if detail is not None: # WP Debug: the template it filled and what each `$var` read. detail["template"] = template[:400] detail["template_source"] = "override" if isinstance(override, str) else "library" reads: dict[str, str | None] = {} names = [name for name, _index in col.reads] # A `$var` nothing set records no read; list it from the template. names += _TEMPLATE_VAR_RE.findall(template) for name in names: if name in reads or name.startswith("__"): continue val = ctx.get(name) reads[name] = None if val is None else str(val)[:160] if reads: detail["reads"] = reads if instance.get("locked_seed") is not None: detail["locked_seed"] = True return {output_var: result}