DumiFlex
4ff45d2492
A toggle with a Value writes it to its variable while on and nothing while off; its Negative goes with the variable. $name.bool reads the switch (true/false) across nodes, in text and in conditions, through __wp_toggle_bools__. Payloads reading $name.bool stamp schema 12; bool is reserved as a tag group.
783 lines
33 KiB
Python
783 lines
33 KiB
Python
"""Derivation module resolver — IF/ELIF/ELSE rules over the runtime context.
|
|
|
|
Each rule has an ordered list of branches (index 0 == IF, the rest == ELIF)
|
|
and an optional ``else`` clause. A branch ``condition`` is either one test
|
|
(``{var, op, value}``) or a group (``{match: "all"|"any", conditions: [...]}``)
|
|
whose members are tests or further groups, so a branch can AND / OR several
|
|
tests together (schema v7). Rules evaluate independently top-to-bottom:
|
|
the first branch whose condition matches wins; if none match, the ``else``
|
|
action (if any) fires; otherwise the rule is a no-op.
|
|
|
|
A branch (and the ``else`` clause) runs its ``action`` and then, in order,
|
|
each entry of its optional ``extra_actions`` list (schema v8): "THEN ... AND
|
|
...". Actions mutate the runtime context directly. Multiple rules can target the
|
|
same variable; later rules see earlier mutations. Action ``value`` strings
|
|
are resolved through resolve_text with surface="derivation" so $var tokens
|
|
and {a|b|c} picks work inside derivation action values.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from typing import Any
|
|
|
|
from engine import negatives
|
|
from engine.modules import build_resolve_ctx, restream
|
|
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
|
|
from engine.syntax.resolve import BOOL_ACCESSOR, NEG_ACCESSOR
|
|
from engine.syntax.types import (
|
|
deref_var_value,
|
|
parse_var_reference,
|
|
split_var_accessor,
|
|
)
|
|
|
|
_VALID_OPS = {
|
|
"equals", "not_equals", "contains", "matches",
|
|
# Presence-check ops added in 2026-05-09 cycle. `exists`/`not_exists`
|
|
# check key presence in ctx regardless of value (empty string still
|
|
# counts as present); `is_set`/`is_unset` additionally require the
|
|
# value to be non-empty. Both pairs ship so users can pick the
|
|
# semantics they need — wildcards can pick options with empty
|
|
# values, leaving keys present-but-empty.
|
|
"exists", "not_exists", "is_set", "is_unset",
|
|
# Emptiness refinements the editor's "exists" switch has always offered.
|
|
# `is_empty` is "present and empty" (a wildcard that rolled its null
|
|
# option); `is_not_empty` is "resolved to something non-empty".
|
|
"is_empty", "is_not_empty",
|
|
# The editor's NOT toggle (schema v10): the negations of `contains` and
|
|
# `matches`, the only tests that had none.
|
|
"not_contains", "not_matches",
|
|
}
|
|
_VALID_MATCHES = {"all", "any"}
|
|
# Groups nest; the editor stops at three levels, the cap here only guards a
|
|
# hand-built payload against runaway recursion.
|
|
_MAX_CONDITION_DEPTH = 8
|
|
# `negative` (schema v8, send-to-negative): "Add to negative" — the value is
|
|
# added to the target variable's negatives; the variable itself is untouched.
|
|
# `negative_replace` (also v8): "Replace negative" — the value becomes the
|
|
# variable's only negative, dropping what it carried.
|
|
NEGATIVE_MODES = {"negative", "negative_replace"}
|
|
# The separator an append value starts with (", red") or a prepend value
|
|
# ends with ("red, "), dropped when the target is still empty.
|
|
_JOIN_EDGE = re.compile(r"^[\s,;]+")
|
|
_JOIN_TAIL = re.compile(r"[\s,;]+$")
|
|
_VALID_MODES = {"replace", "append", "prepend", *NEGATIVE_MODES}
|
|
|
|
|
|
def _ctx_get_raw(ctx: Any, name: str) -> Any:
|
|
"""Read the RAW stored value for `name` (may be a ListVar) through the
|
|
callable-getter -> dict-lookup indirection. Missing / error -> None."""
|
|
if ctx is None:
|
|
return None
|
|
getter = getattr(ctx, "get", None)
|
|
if callable(getter):
|
|
try:
|
|
return getter(name, None)
|
|
except TypeError:
|
|
try:
|
|
return getter(name)
|
|
except Exception:
|
|
return None
|
|
try:
|
|
if name in ctx: # type: ignore[operator]
|
|
return ctx[name] # type: ignore[index]
|
|
except Exception:
|
|
return None
|
|
return None
|
|
|
|
|
|
def _ctx_axis(ctx: Any, base: str, axis: str, index: int | None) -> str:
|
|
"""Read a rolled axis choice out of `ctx["__wp_axes__"]`.
|
|
|
|
Mirrors the resolver's `_resolve_axis` shaping so a condition and a
|
|
template agree on what `$outfit.SHOES` means. Missing yields "" rather than
|
|
raising: derivations are control flow, so an unanswerable condition is
|
|
simply false.
|
|
"""
|
|
if not isinstance(ctx, dict):
|
|
return ""
|
|
entry = (ctx.get("__wp_axes__") or {}).get(base)
|
|
if isinstance(entry, list):
|
|
# Positional, like the resolver: a pick with no tag on this axis keeps
|
|
# its slot so `.K` lines up with `$outfit.K`.
|
|
vals = [
|
|
(e.get(axis) or "") if isinstance(e, dict) else "" for e in entry
|
|
]
|
|
if index is None:
|
|
return ", ".join(v for v in vals if v)
|
|
return vals[index] if 0 <= index < len(vals) else ""
|
|
if isinstance(entry, dict):
|
|
got = entry.get(axis, "")
|
|
# A single pick behaves as a one-element list, as everywhere else.
|
|
if index is not None:
|
|
return got if index == 0 else ""
|
|
return got
|
|
return ""
|
|
|
|
|
|
def _ctx_get(ctx: Any, name: str) -> str:
|
|
"""Read `name` as a string, honoring a `.K` list accessor, a `.AXIS` read
|
|
(2026-08 tag axes), and folding a ListVar to its joined value (SP2a):
|
|
`$mood.0` splits to base `mood` + index 0. The fold/accessor contract lives
|
|
in deref_var_value; the axis contract in `_ctx_axis`."""
|
|
base, index, axis = parse_var_reference(name)
|
|
if axis == NEG_ACCESSOR:
|
|
return _ctx_negative(ctx, base, index)
|
|
if axis == BOOL_ACCESSOR:
|
|
return _ctx_bool(ctx, base)
|
|
if axis:
|
|
return _ctx_axis(ctx, base, axis, index)
|
|
return deref_var_value(_ctx_get_raw(ctx, base), index)
|
|
|
|
|
|
def _ctx_negative(ctx: Any, base: str, index: int | None) -> str:
|
|
"""`$name.neg` in a condition: the variable's negatives as one line,
|
|
read the same way the resolver renders them."""
|
|
texts = [
|
|
str(e.get("text", "")) for e in negatives.get_entries(ctx, base)
|
|
if index is None or e.get("pick") is None or e.get("pick") == index
|
|
]
|
|
return negatives.join_unique(texts)
|
|
|
|
|
|
def _ctx_bool(ctx: Any, base: str) -> str:
|
|
"""`$name.bool` in a condition: WP Toggles' on/off for the variable, else
|
|
its own value when that already is "true" / "false", else ""."""
|
|
table = ctx.get("__wp_toggle_bools__") if isinstance(ctx, dict) else None
|
|
got = table.get(base) if isinstance(table, dict) else None
|
|
if isinstance(got, bool):
|
|
return "true" if got else "false"
|
|
own = _ctx_get_raw(ctx, base)
|
|
word = own.strip().lower() if isinstance(own, str) else ""
|
|
return word if word in ("true", "false") else ""
|
|
|
|
|
|
def _ctx_has(ctx: Any, name: str) -> bool:
|
|
"""Whether `name` is a known key in `ctx`, regardless of value.
|
|
|
|
Mirrors `_ctx_get`'s indirection chain (callable getter →
|
|
`__contains__` / dict lookup) but returns bool. Required so the
|
|
`exists` / `not_exists` ops can distinguish "key absent" from
|
|
"key present but value is empty string" — `_ctx_get` collapses
|
|
both into `""`. Used only by presence-check ops; non-presence
|
|
ops continue to read through `_ctx_get`.
|
|
"""
|
|
if ctx is None:
|
|
return False
|
|
base, _index, axis = parse_var_reference(name)
|
|
if axis == NEG_ACCESSOR:
|
|
return bool(_ctx_negative(ctx, base, _index))
|
|
if axis == BOOL_ACCESSOR:
|
|
return bool(_ctx_bool(ctx, base))
|
|
if axis:
|
|
# An axis is present when the binding actually rolled it. Checked
|
|
# against the axes bucket, not ctx: `$outfit.SHOES` can exist while
|
|
# `outfit` is unbound, and vice versa.
|
|
entry = (
|
|
(ctx.get("__wp_axes__") or {}).get(base)
|
|
if isinstance(ctx, dict) else None
|
|
)
|
|
if isinstance(entry, dict):
|
|
return axis in entry
|
|
if isinstance(entry, list):
|
|
return any(isinstance(e, dict) and axis in e for e in entry)
|
|
return False
|
|
name = base # SP2a: presence checks the base var
|
|
# Engine ctx is a dict in tests + at runtime; check `__contains__`
|
|
# before falling back to a getter probe so we get the cheap path.
|
|
try:
|
|
return name in ctx # type: ignore[operator]
|
|
except Exception:
|
|
pass
|
|
getter = getattr(ctx, "get", None)
|
|
if callable(getter):
|
|
# Use a sentinel default so a "missing" key can be told apart
|
|
# from a key whose value is `None` or `""`.
|
|
sentinel = object()
|
|
try:
|
|
value = getter(name, sentinel)
|
|
except TypeError:
|
|
try:
|
|
value = getter(name)
|
|
except Exception:
|
|
return False
|
|
return value is not sentinel
|
|
return False
|
|
|
|
|
|
def _ctx_set(ctx: Any, name: str, value: str) -> None:
|
|
if ctx is None:
|
|
return
|
|
name = split_var_accessor(name)[0] # SP2a: writing $x.K targets the base var
|
|
setter = getattr(ctx, "set", None)
|
|
if callable(setter):
|
|
setter(name, value)
|
|
return
|
|
try:
|
|
ctx[name] = value # type: ignore[index]
|
|
except Exception:
|
|
return
|
|
|
|
|
|
def _match_condition(condition: dict[str, Any], ctx: Any) -> bool:
|
|
var = condition.get("var", "")
|
|
op = condition.get("op", "")
|
|
value = condition.get("value", "")
|
|
# Presence-check ops short-circuit before reading the value — they
|
|
# only care about key presence (and, for is_set/is_unset, whether
|
|
# the stored value is non-empty). The `value` field is ignored by
|
|
# these ops; the SPA disables the value input when one is selected.
|
|
if op == "exists":
|
|
return _ctx_has(ctx, var)
|
|
if op == "not_exists":
|
|
return not _ctx_has(ctx, var)
|
|
if op == "is_set":
|
|
return _ctx_has(ctx, var) and _ctx_get(ctx, var) != ""
|
|
if op == "is_unset":
|
|
return not _ctx_has(ctx, var) or _ctx_get(ctx, var) == ""
|
|
if op == "is_empty":
|
|
return _ctx_has(ctx, var) and _ctx_get(ctx, var) == ""
|
|
if op == "is_not_empty":
|
|
return _ctx_get(ctx, var) != ""
|
|
actual = _ctx_get(ctx, var)
|
|
if op == "equals":
|
|
return actual == value
|
|
if op == "not_equals":
|
|
return actual != value
|
|
if op == "contains":
|
|
return value in actual
|
|
if op == "not_contains":
|
|
return value not in actual
|
|
if op in ("matches", "not_matches"):
|
|
try:
|
|
hit = re.search(value, actual) is not None
|
|
except re.error:
|
|
# A broken pattern matches nothing, so its negation holds nothing
|
|
# either: an invalid test never fires a branch.
|
|
return False
|
|
return hit if op == "matches" else not hit
|
|
return False
|
|
|
|
|
|
def is_condition_group(condition: Any) -> bool:
|
|
"""A group carries ``conditions``; a single test carries ``var``."""
|
|
return isinstance(condition, dict) and "conditions" in condition
|
|
|
|
|
|
def condition_leaves(condition: Any) -> list[dict[str, Any]]:
|
|
"""Every single test in ``condition``, depth-first. The position in this
|
|
list is the test's override index (see ``condition_override_key``)."""
|
|
if is_condition_group(condition):
|
|
out: list[dict[str, Any]] = []
|
|
for child in condition.get("conditions") or []:
|
|
out.extend(condition_leaves(child))
|
|
return out
|
|
return [condition] if isinstance(condition, dict) else []
|
|
|
|
|
|
def condition_override_key(branch_index: int, leaf_index: int) -> str:
|
|
"""Key into ``condition_value_overrides[rule_id]`` for one test of a
|
|
branch. The first test keeps the bare branch index it always had, so a
|
|
single-test branch and its saved overrides are unchanged; later tests of a
|
|
grouped branch append ``.K`` (their depth-first position).
|
|
|
|
MUST stay byte-identical to the TS twin ``conditionOverrideKey`` in
|
|
``src/extension/derivation-conditions.ts``.
|
|
"""
|
|
return str(branch_index) if leaf_index == 0 else f"{branch_index}.{leaf_index}"
|
|
|
|
|
|
def _with_value_overrides(
|
|
condition: Any, overrides: dict[str, Any], branch_index: int,
|
|
) -> Any:
|
|
"""Copy of ``condition`` with each test's ``value`` swapped for its
|
|
instance override, when one is set."""
|
|
counter = [0]
|
|
|
|
def walk(node: Any) -> Any:
|
|
if is_condition_group(node):
|
|
return {**node, "conditions": [walk(c) for c in node.get("conditions") or []]}
|
|
if not isinstance(node, dict):
|
|
return node
|
|
key = condition_override_key(branch_index, counter[0])
|
|
counter[0] += 1
|
|
override = overrides.get(key)
|
|
return {**node, "value": override} if isinstance(override, str) else node
|
|
|
|
return walk(condition)
|
|
|
|
|
|
def _eval_condition(condition: Any, ctx: Any) -> bool:
|
|
if is_condition_group(condition):
|
|
children = condition.get("conditions") or []
|
|
if condition.get("match") == "any":
|
|
return any(_eval_condition(c, ctx) for c in children)
|
|
return all(_eval_condition(c, ctx) for c in children)
|
|
if not isinstance(condition, dict):
|
|
return False
|
|
return _match_condition(condition, ctx)
|
|
|
|
|
|
def _explain_condition(condition: Any, ctx: Any) -> dict[str, Any]:
|
|
"""The condition tree with each test's outcome and the value it read, for
|
|
the WP Debug node. Evaluates every test, unlike ``_eval_condition``, so a
|
|
group that short-circuited still shows why each member did or didn't
|
|
match. Reads only; nothing here writes to ``ctx``."""
|
|
if is_condition_group(condition):
|
|
children = [_explain_condition(c, ctx) for c in condition.get("conditions") or []]
|
|
match = "any" if condition.get("match") == "any" else "all"
|
|
results = [c["result"] for c in children]
|
|
return {
|
|
"match": match,
|
|
"result": any(results) if match == "any" else all(results),
|
|
"conditions": children,
|
|
}
|
|
if not isinstance(condition, dict):
|
|
return {"var": "", "op": "", "value": "", "actual": None, "result": False}
|
|
var = str(condition.get("var", ""))
|
|
return {
|
|
"var": var,
|
|
"op": str(condition.get("op", "")),
|
|
"value": condition.get("value", ""),
|
|
"actual": _ctx_get(ctx, var) if _ctx_has(ctx, var) else None,
|
|
"result": _match_condition(condition, ctx),
|
|
}
|
|
|
|
|
|
def branch_carrier_key(rule_id: str, branch: int | str) -> str:
|
|
"""Carrier key for a derivation branch occurrence -- the `option_id` the
|
|
SP3 nested-occurrence model matches a constraint `pick` against. `branch`
|
|
is the 0-based branch index (0 = IF) or the literal ``"else"``.
|
|
|
|
MUST stay byte-identical to the TS twin ``branchKey`` in
|
|
``src/extension/constraint-pairs.ts``; both are corpus-locked against
|
|
``tests/fixtures/constraint-corpus.json`` ``branch_key_cases``.
|
|
"""
|
|
return f"{rule_id}:{branch}"
|
|
|
|
|
|
def _apply_action(
|
|
action: dict[str, Any],
|
|
ctx: Any,
|
|
resolve_ctx: Any,
|
|
carrier_key: str | None = None,
|
|
neg_seed: int | None = None,
|
|
neg_key: str | None = None,
|
|
) -> tuple[str, str | None] | None:
|
|
"""Run one action. Returns (target, new value), with value None for an
|
|
"Add to negative" / "Replace negative" action (it writes no value), or None when it did
|
|
nothing. `neg_seed` + `neg_key` key the action's negative stream
|
|
(`neg_key` defaults to the carrier key)."""
|
|
target = action.get("target_var", "")
|
|
if not target:
|
|
return None
|
|
mode = action.get("mode", "replace")
|
|
raw_value = str(action.get("value", ""))
|
|
n_rng = negatives.neg_rng(
|
|
neg_seed or 0, f"derivation::{neg_key or carrier_key or target}",
|
|
)
|
|
if mode in NEGATIVE_MODES:
|
|
text = negatives.quiet_resolve(raw_value, resolve_ctx, n_rng)
|
|
negatives.set_entries(
|
|
ctx, target,
|
|
[{"text": text, "pick": None, "source": carrier_key or "derivation"}],
|
|
add=mode == "negative",
|
|
)
|
|
return target, None
|
|
col = negatives.Collector()
|
|
# Stamp this derivation instance as the CARRIER of any `@{}` in the action
|
|
# value, keyed by the branch key (`${rule_id}:${bi}` / `${rule_id}:else`) —
|
|
# the `option_id` the SP3 nested-occurrence model matches a `pick` against.
|
|
# Save/restore around the resolve, mirroring the resolver's own carrier
|
|
# save/restore for nested recursion.
|
|
set_carrier: Any = getattr(resolve_ctx, "set_carrier", None)
|
|
get_carrier: Any = getattr(resolve_ctx, "get_carrier", None)
|
|
if carrier_key is not None and callable(set_carrier) and callable(get_carrier):
|
|
module_uid = None
|
|
if isinstance(ctx, dict):
|
|
module_uid = ctx.get("__wp_current_module_uid__") or ctx.get("__wp_current_module_id__")
|
|
prev: Any = get_carrier()
|
|
set_carrier(module_uid, carrier_key)
|
|
try:
|
|
with negatives.collecting(resolve_ctx) as col:
|
|
new_value = resolve_text(raw_value, resolve_ctx)
|
|
finally:
|
|
set_carrier(*prev)
|
|
else:
|
|
with negatives.collecting(resolve_ctx) as col:
|
|
new_value = resolve_text(raw_value, resolve_ctx)
|
|
if mode not in ("replace", "append", "prepend"):
|
|
return None
|
|
# The written text carries the negatives of the variables it read and of
|
|
# the options its refs picked. Replace swaps the target's negatives for
|
|
# those; append / prepend keep the old ones and add.
|
|
entries = negatives.entries_for_reads(ctx, col.reads)
|
|
entries += negatives.own_entries(
|
|
col.ref_negatives, resolve_ctx, n_rng,
|
|
pick=None, source=carrier_key or "derivation",
|
|
)
|
|
negatives.set_entries(ctx, target, entries, add=mode != "replace")
|
|
current = _ctx_get(ctx, target)
|
|
if mode == "replace":
|
|
result = new_value
|
|
elif not current.strip():
|
|
# Adding to nothing: the joining comma or space the value brought
|
|
# for a non-empty target would otherwise start the text.
|
|
edge = _JOIN_EDGE if mode == "append" else _JOIN_TAIL
|
|
result = edge.sub("", new_value)
|
|
elif mode == "append":
|
|
result = current + new_value
|
|
else:
|
|
result = new_value + current
|
|
_ctx_set(ctx, target, result)
|
|
return target, result
|
|
|
|
|
|
def clause_actions(clause: Any) -> list[dict[str, Any]]:
|
|
"""Every action a branch or ``else`` clause runs, in order: its ``action``
|
|
then its ``extra_actions``. Non-dict entries are skipped."""
|
|
if not isinstance(clause, dict):
|
|
return []
|
|
out = [clause.get("action")]
|
|
extra = clause.get("extra_actions")
|
|
if isinstance(extra, list):
|
|
out.extend(extra)
|
|
return [a for a in out if isinstance(a, dict)]
|
|
|
|
|
|
def action_override_key(branch: int | str, action_index: int) -> str:
|
|
"""Key into ``action_value_overrides[rule_id]`` for one action of a branch
|
|
(``branch`` is the index or ``"else"``). The first action keeps the bare
|
|
key it always had; later ones append ``.K``.
|
|
|
|
MUST stay byte-identical to the TS twin ``actionOverrideKey`` in
|
|
``src/extension/derivation-conditions.ts``.
|
|
"""
|
|
return str(branch) if action_index == 0 else f"{branch}.{action_index}"
|
|
|
|
|
|
def _explain_action(
|
|
action: dict[str, Any], pair: tuple[str, str | None] | None,
|
|
) -> dict[str, Any]:
|
|
"""What a fired branch did: the variable, how it wrote, and the result."""
|
|
return {
|
|
"target": str(action.get("target_var", "")),
|
|
"mode": str(action.get("mode", "replace")),
|
|
"value": str(action.get("value", "")),
|
|
"result": pair[1] if pair is not None else None,
|
|
}
|
|
|
|
|
|
def _validate_condition(condition: Any, where: str, depth: int = 0) -> None:
|
|
if not isinstance(condition, dict):
|
|
raise ValueError(f"derivation {where}.condition must be an object")
|
|
if is_condition_group(condition):
|
|
if depth >= _MAX_CONDITION_DEPTH:
|
|
raise ValueError(
|
|
f"derivation {where}.condition nests deeper than "
|
|
f"{_MAX_CONDITION_DEPTH} groups"
|
|
)
|
|
if condition.get("match") not in _VALID_MATCHES:
|
|
raise ValueError(
|
|
f"derivation {where}.condition.match must be one of "
|
|
f"{sorted(_VALID_MATCHES)}"
|
|
)
|
|
children = condition.get("conditions")
|
|
if not isinstance(children, list) or not children:
|
|
raise ValueError(
|
|
f"derivation {where}.condition.conditions must be a non-empty list"
|
|
)
|
|
for ci, child in enumerate(children):
|
|
_validate_condition(child, f"{where}.condition.conditions[{ci}]", depth + 1)
|
|
return
|
|
var = condition.get("var")
|
|
if not isinstance(var, str) or not var:
|
|
raise ValueError(f"derivation {where}.condition.var must be a non-empty string")
|
|
op = condition.get("op")
|
|
if op not in _VALID_OPS:
|
|
raise ValueError(
|
|
f"derivation {where}.condition.op must be one of {sorted(_VALID_OPS)}"
|
|
)
|
|
if not isinstance(condition.get("value", ""), str):
|
|
raise ValueError(f"derivation {where}.condition.value must be a string")
|
|
|
|
|
|
def _validate_action(action: Any, where: str) -> None:
|
|
if not isinstance(action, dict):
|
|
raise ValueError(f"derivation {where}.action must be an object")
|
|
target = action.get("target_var")
|
|
if not isinstance(target, str) or not target:
|
|
raise ValueError(
|
|
f"derivation {where}.action.target_var must be a non-empty string"
|
|
)
|
|
mode = action.get("mode", "replace")
|
|
if mode not in _VALID_MODES:
|
|
raise ValueError(
|
|
f"derivation {where}.action.mode must be one of {sorted(_VALID_MODES)}"
|
|
)
|
|
if not isinstance(action.get("value", ""), str):
|
|
raise ValueError(f"derivation {where}.action.value must be a string")
|
|
|
|
|
|
def _validate_extra_actions(clause: dict[str, Any], where: str) -> None:
|
|
extra = clause.get("extra_actions")
|
|
if extra is None:
|
|
return
|
|
if not isinstance(extra, list):
|
|
raise ValueError(f"derivation {where}.extra_actions must be a list")
|
|
for ai, action in enumerate(extra):
|
|
_validate_action(action, f"{where}.extra_actions[{ai}]")
|
|
|
|
|
|
class DerivationHandler(ModuleHandler):
|
|
"""IF / ELIF / ELSE rule evaluator that mutates ctx."""
|
|
|
|
type_id = "derivation"
|
|
|
|
@classmethod
|
|
def validate_payload(cls, payload: dict[str, Any]) -> None:
|
|
if not isinstance(payload, dict):
|
|
raise ValueError("derivation payload must be an object")
|
|
rules = payload.get("rules")
|
|
if not isinstance(rules, list):
|
|
raise ValueError("derivation payload.rules must be a list")
|
|
seen_rule_ids: set[str] = set()
|
|
for ri, rule in enumerate(rules):
|
|
if not isinstance(rule, dict):
|
|
raise ValueError(f"derivation payload.rules[{ri}] must be an object")
|
|
rule_id = rule.get("id")
|
|
if not isinstance(rule_id, str) or not rule_id:
|
|
raise ValueError(
|
|
f"derivation payload.rules[{ri}].id must be a non-empty string"
|
|
)
|
|
# Rule ids index `instance.disabled_rule_ids` and
|
|
# `rule_order_override` — dupes make both ambiguous.
|
|
if rule_id in seen_rule_ids:
|
|
raise ValueError(
|
|
f"derivation payload.rules[{ri}].id {rule_id!r} duplicates "
|
|
f"an earlier rule id"
|
|
)
|
|
seen_rule_ids.add(rule_id)
|
|
branches = rule.get("branches")
|
|
if not isinstance(branches, list) or not branches:
|
|
raise ValueError(
|
|
f"derivation payload.rules[{ri}].branches must be a non-empty list"
|
|
)
|
|
for bi, branch in enumerate(branches):
|
|
if not isinstance(branch, dict):
|
|
raise ValueError(
|
|
f"derivation payload.rules[{ri}].branches[{bi}] must be an object"
|
|
)
|
|
_validate_condition(
|
|
branch.get("condition"), f"rules[{ri}].branches[{bi}]"
|
|
)
|
|
_validate_action(branch.get("action"), f"rules[{ri}].branches[{bi}]")
|
|
_validate_extra_actions(branch, f"rules[{ri}].branches[{bi}]")
|
|
else_clause = rule.get("else")
|
|
if else_clause is not None:
|
|
if not isinstance(else_clause, dict):
|
|
raise ValueError(
|
|
f"derivation payload.rules[{ri}].else must be an object"
|
|
)
|
|
_validate_action(else_clause.get("action"), f"rules[{ri}].else")
|
|
_validate_extra_actions(else_clause, f"rules[{ri}].else")
|
|
|
|
@classmethod
|
|
def resolve(
|
|
cls,
|
|
payload: dict[str, Any],
|
|
instance: dict[str, Any],
|
|
ctx: Any,
|
|
) -> dict[str, str]:
|
|
cls.validate_payload(payload)
|
|
|
|
# Effective seed selection — same pattern wildcard / combine /
|
|
# fixed_values use. Locked seed pins {a|b|c} resolution in
|
|
# action.value across runs; without lock the chain seed
|
|
# propagates per-queue. Falls back to the host ctx's RNG when
|
|
# the ctx wasn't built with `__wp_rng__` (legacy callers).
|
|
# seed_scope == "hold" freezes the DICE per branch, NOT which branch
|
|
# fires. The conditions still pick a branch live each iteration, so a
|
|
# varying input can switch IF→ELIF and we use THAT branch — but every
|
|
# branch's inline {a|b} / @{} roll is frozen to its frame-0 value via
|
|
# the constant hold seed, keyed PER BRANCH (see _branch_resolve_ctx).
|
|
# So each branch's value is identical every iteration it fires and
|
|
# equals the value it WOULD have had at frame 0, independent of which
|
|
# other branches fired. (Contrast wildcard/combine/fixed_values hold,
|
|
# which freeze the whole resolved VALUE because a constraint can
|
|
# reshape their pool; a derivation branch's pool is fixed, so a frozen
|
|
# seed alone is enough — and freezing the branch SELECTION would be
|
|
# wrong: the user wants the chosen branch to follow the inputs.)
|
|
_hold = False
|
|
effective_seed = 0
|
|
module_id = "derivation"
|
|
ctx_local: dict[str, Any] = {}
|
|
if isinstance(ctx, dict) and "__wp_rng__" in ctx:
|
|
_hold = instance.get("seed_scope") == "hold"
|
|
if _hold:
|
|
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")
|
|
effective_seed = (
|
|
int(locked_seed) if isinstance(locked_seed, (int, float))
|
|
else chain_seed
|
|
)
|
|
module_id = instance_key(ctx, "derivation")
|
|
rng = derive_module_rng(effective_seed, module_id)
|
|
ctx_local = {**ctx, "__wp_rng__": rng}
|
|
resolve_ctx = build_resolve_ctx(ctx_local, surface="derivation")
|
|
else:
|
|
resolve_ctx = build_resolve_ctx(ctx, surface="derivation")
|
|
|
|
def _branch_resolve_ctx(branch_key: str) -> Any:
|
|
# An INDEPENDENT stream per branch (keyed by rule:branch), and
|
|
# per action below, so a value rolls the same whichever other
|
|
# branches fired before it. Under hold the seed is the constant
|
|
# hold seed, so the value is also frozen to its frame-0 roll.
|
|
if not ctx_local:
|
|
return resolve_ctx
|
|
b_key = f"{module_id}::{branch_key}"
|
|
return build_resolve_ctx(
|
|
{**ctx_local, "__wp_rng__": derive_module_rng(effective_seed, b_key)},
|
|
surface="derivation", stream=stream_key(effective_seed, b_key),
|
|
)
|
|
|
|
# Tier-D instance overrides (2026-05-10 cycle).
|
|
disabled_rule_ids = set(instance.get("disabled_rule_ids") or [])
|
|
disabled_branch_keys = set(instance.get("disabled_branch_keys") or [])
|
|
action_overrides = instance.get("action_value_overrides") or {}
|
|
cond_overrides = instance.get("condition_value_overrides") or {}
|
|
|
|
# Per-instance rule reorder — sort by rule_order_override list.
|
|
# IDs not in the override fall through in their original library
|
|
# order at the end (defensive against partial reorders).
|
|
rules = list(payload.get("rules", []))
|
|
order = instance.get("rule_order_override")
|
|
if isinstance(order, list) and order:
|
|
ordering = {rid: i for i, rid in enumerate(order)}
|
|
tail = len(order)
|
|
rules.sort(
|
|
key=lambda r: ordering.get(r.get("id"), tail + 1),
|
|
)
|
|
|
|
out: dict[str, str] = {}
|
|
# Values this module has written so far, by base variable. Every
|
|
# action resolves against them, so a later action (in the same branch
|
|
# or a later rule) reads the earlier write, as conditions already do
|
|
# through the live ctx.
|
|
written: dict[str, Any] = {}
|
|
# WP Debug: which branch of each rule fired and why (explain runs only).
|
|
detail = module_detail(ctx)
|
|
explained: list[dict[str, Any]] | None = [] if detail is not None else None
|
|
|
|
def run_clause(
|
|
rule_id: str, key: int | str, clause: dict[str, Any],
|
|
rd: dict[str, Any] | None,
|
|
) -> None:
|
|
"""Run a fired branch's (or the else clause's) actions in order.
|
|
Each action sees the writes of the ones before it. Action-value
|
|
overrides are keyed per action (``action_override_key``)."""
|
|
carrier = branch_carrier_key(rule_id, key)
|
|
rule_overrides = action_overrides.get(rule_id)
|
|
b_ctx = _branch_resolve_ctx(carrier)
|
|
vars_view = getattr(b_ctx, "_vars", None)
|
|
if isinstance(vars_view, dict):
|
|
vars_view.update(written)
|
|
shown: list[dict[str, Any]] = []
|
|
for ai, action in enumerate(clause_actions(clause)):
|
|
override = (
|
|
rule_overrides.get(action_override_key(key, ai))
|
|
if isinstance(rule_overrides, dict) else None
|
|
)
|
|
if isinstance(override, str):
|
|
action = {**action, "value": override}
|
|
# Later actions key their negative stream apart so two
|
|
# "Add to negative" actions on one branch never share rolls.
|
|
pair = _apply_action(
|
|
action, ctx,
|
|
restream(b_ctx, f"{b_ctx.stream_key}|a{ai}")
|
|
if getattr(b_ctx, "stream_key", None) else b_ctx,
|
|
carrier_key=carrier,
|
|
neg_seed=effective_seed,
|
|
neg_key=(
|
|
f"{module_id}::{carrier}" if ai == 0
|
|
else f"{module_id}::{carrier}.{ai}"
|
|
),
|
|
)
|
|
if pair is not None and pair[1] is not None:
|
|
out[pair[0]] = pair[1]
|
|
# The next action (this branch's or a later rule's)
|
|
# reads what this one wrote.
|
|
base = split_var_accessor(pair[0])[0]
|
|
written[base] = pair[1]
|
|
if isinstance(vars_view, dict):
|
|
vars_view[base] = pair[1]
|
|
shown.append(_explain_action(action, pair))
|
|
if rd is not None:
|
|
rd["fired"] = key
|
|
if shown:
|
|
rd["action"] = shown[0]
|
|
rd["actions"] = shown
|
|
|
|
for rule in rules:
|
|
rule_id = rule.get("id", "")
|
|
rd: dict[str, Any] | None = None
|
|
if explained is not None:
|
|
rd = {
|
|
"id": rule_id,
|
|
"fired": None,
|
|
"branches": [],
|
|
"has_else": isinstance(rule.get("else"), dict),
|
|
}
|
|
explained.append(rd)
|
|
if rule_id in disabled_rule_ids:
|
|
if rd is not None:
|
|
rd["disabled"] = True
|
|
continue
|
|
|
|
applied = False
|
|
for bi, branch in enumerate(rule.get("branches", [])):
|
|
# Branch-level disable — `r1:1` etc. IF (bi=0) ignored
|
|
# even when listed because disabling IF == disabling rule
|
|
# (the per-rule toggle handles that case cleanly).
|
|
if bi != 0 and branch_carrier_key(rule_id, bi) in disabled_branch_keys:
|
|
if rd is not None:
|
|
rd["branches"].append({"index": bi, "disabled": True})
|
|
continue
|
|
|
|
# Condition-value overrides, one per test of the branch.
|
|
cond = branch.get("condition", {})
|
|
rule_cond_overrides = cond_overrides.get(rule_id)
|
|
if isinstance(rule_cond_overrides, dict) and rule_cond_overrides:
|
|
cond = _with_value_overrides(cond, rule_cond_overrides, bi)
|
|
|
|
matched = _eval_condition(cond, ctx)
|
|
if rd is not None:
|
|
rd["branches"].append({
|
|
"index": bi,
|
|
"matched": matched,
|
|
"condition": _explain_condition(cond, ctx),
|
|
})
|
|
if matched:
|
|
run_clause(rule_id, bi, branch, rd)
|
|
applied = True
|
|
break
|
|
|
|
if not applied:
|
|
# ELSE skip when listed in disabled_branch_keys.
|
|
if branch_carrier_key(rule_id, "else") in disabled_branch_keys:
|
|
if rd is not None:
|
|
rd["else_disabled"] = True
|
|
continue
|
|
else_clause = rule.get("else")
|
|
if isinstance(else_clause, dict):
|
|
run_clause(rule_id, "else", else_clause, rd)
|
|
if detail is not None:
|
|
detail["rules"] = explained
|
|
return out
|