"""Weight-Coupled Invariant Organ Atlas, Funktor Equivalence, M-Rail & Graveyard (`organs.py`)."""
from __future__ import annotations
import hashlib
import json
import math
import re
from dataclasses import asdict, dataclass, field
from typing import Any, Dict, List, Optional, Sequence, Tuple, Union
import torch
from .hypercodec import ZeroUNKCodec
from .vocab import (
CANONICAL_DE_NUM,
CANONICAL_EN_NUM,
COUNTABLE_NOUN_SET,
DE_NUMBER_LEXICON,
EN_NUMBER_LEXICON,
)
_BINARY_SET = frozenset({"add", "sub", "mul", "div", "pow", "moebius", "casimir_sqrt", "pade11"})
_UNARY_SET = frozenset({"abs", "cos", "exp", "log", "neg", "sin", "sqrt", "lorentz_gamma", "chebyshev_t2"})
_ATOM_SET = frozenset({"x", "y", "z", "c0", "c1", "pi", "e", "1", "2", "0.5"})
_COMMUTATIVE_OPS = frozenset({"add", "mul", "moebius"})
def eval_symbolic_prefix_f64(
expr_prefix: str,
x: float,
y: float = 0.0,
z: float = 0.0,
c0: float = 1.0,
c1: float = 1.0,
) -> Optional[float]:
"""IEEE-754 float64 evaluator for Polish-prefix symbolic organ expressions."""
toks = [t for t in expr_prefix.strip().split() if t != "NOP"]
if not toks:
return None
stack: List[float] = []
atom_vals = {
"x": float(x), "y": float(y), "z": float(z),
"c0": float(c0), "c1": float(c1),
"pi": math.pi, "e": math.e, "1": 1.0, "2": 2.0, "0.5": 0.5,
}
try:
for t in reversed(toks):
if t in atom_vals:
stack.append(atom_vals[t])
elif t in _UNARY_SET:
if not stack:
return None
a1 = stack.pop()
if t == "abs":
r = abs(a1)
elif t == "cos":
r = math.cos(a1)
elif t == "exp":
if abs(a1) > 20.0:
return None
r = math.exp(a1)
elif t == "log":
if a1 <= 1e-9:
return None
r = math.log(a1)
elif t == "neg":
r = -a1
elif t == "sin":
r = math.sin(a1)
elif t == "sqrt":
if a1 < -1e-12:
return None
r = math.sqrt(max(a1, 0.0))
elif t == "lorentz_gamma":
ins = 1.0 - a1 * a1
if ins <= 1e-6:
return None
r = 1.0 / math.sqrt(ins)
elif t == "chebyshev_t2":
r = 2.0 * a1 * a1 - 1.0
else:
return None
if not math.isfinite(r) or abs(r) >= 60000.0:
return None
stack.append(r)
elif t in _BINARY_SET:
if len(stack) < 2:
return None
a1 = stack.pop()
a2 = stack.pop()
if t == "add":
r = a1 + a2
elif t == "sub":
r = a1 - a2
elif t == "mul":
r = a1 * a2
elif t == "div":
if abs(a2) <= 1e-7:
return None
r = a1 / a2
elif t == "pow":
if a1 < 0.0 and abs(a2 - round(a2)) > 1e-9:
return None
if abs(a2) > 8.0 or (abs(a1) < 1e-9 and a2 <= 0.0):
return None
r = float(a1 ** a2)
elif t == "moebius":
den = 1.0 + a1 * a2
if abs(den) <= 1e-7:
return None
r = (a1 + a2) / den
elif t == "casimir_sqrt":
ins = a2 * a2 + a1 * a1 * (1.0 - a2 * a2)
if ins < -1e-12:
return None
r = math.sqrt(max(ins, 0.0))
elif t == "pade11":
den = 1.0 + a2
if abs(den) <= 1e-7:
return None
r = a1 / den
else:
return None
if not math.isfinite(r) or abs(r) >= 60000.0:
return None
stack.append(r)
else:
return None
except (OverflowError, ValueError, ZeroDivisionError):
return None
return stack[-1] if len(stack) == 1 else None
def canonical_symbolic_skeleton(expr_prefix: str) -> Tuple[str, str]:
"""Compute canonical structural skeleton and its 16-hex SHA-256 `skeleton_hash`."""
toks = [t for t in expr_prefix.strip().split() if t != "NOP"]
def _parse(idx: int) -> Tuple[str, int]:
if idx >= len(toks):
return "?", idx
t = toks[idx]
if t in ("c0", "c1"):
return "CONST", idx + 1
if t in _ATOM_SET:
return t, idx + 1
if t in _UNARY_SET:
child, nxt = _parse(idx + 1)
return f"{t}({child})", nxt
if t in _BINARY_SET:
left, mid = _parse(idx + 1)
right, nxt = _parse(mid)
if t in _COMMUTATIVE_OPS and right < left:
left, right = right, left
return f"{t}({left},{right})", nxt
return t, idx + 1
skel, _ = _parse(0)
skel_hash = hashlib.sha256(skel.encode("utf-8")).hexdigest()[:16]
return skel, skel_hash
# Backwards-compatible aliases used by organism.py
EN_NUM_WORDS = EN_NUMBER_LEXICON
DE_NUM_WORDS = DE_NUMBER_LEXICON
INV_EN_NUM = CANONICAL_EN_NUM
INV_DE_NUM = CANONICAL_DE_NUM
COUNTABLE_NOUNS = COUNTABLE_NOUN_SET
TRIGGER_FINALS = {"is", "ist", "=", "->"}
def _default_prefix_for_family(family: str, params: Dict[str, float]) -> str:
if family == "add":
return "add x y"
if family == "sub":
return "sub x y"
if family == "mul":
return "mul x y"
if family == "div":
return "div x y"
if family == "diff":
return "abs sub x y"
if family == "affine2":
ca = float(params.get("ca", 1.0))
cb = float(params.get("cb", 1.0))
c0 = float(params.get("c0", 0.0))
if (ca, cb, c0) == (1.0, 1.0, 0.0):
return "add x y"
if (ca, cb, c0) == (1.0, -1.0, 0.0):
return "sub x y"
if abs(c0) < 1e-9:
t_x = "x" if abs(ca - 1.0) < 1e-9 else ("mul 2 x" if abs(ca - 2.0) < 1e-9 else "mul c0 x")
t_y = "y" if abs(cb - 1.0) < 1e-9 else ("mul 2 y" if abs(cb - 2.0) < 1e-9 else "mul c1 y")
return f"add {t_x} {t_y}"
return "add add mul c0 x mul c1 y c0"
return ""
@dataclass
class OrganSpec:
"""Serializable sub-kilobyte invariant organ crystal."""
organ_id: str
family: str # max, min, median, closer, count, diff, add, sub, mul, div, mod, parity, affine2, symbolic_hcgm
triggers: Tuple[str, ...]
accepted: bool = True
support_agreement: float = 1.0
params: Dict[str, float] = field(default_factory=dict)
source: str = "seed_k7"
muted: bool = False
expr_prefix: str = ""
skeleton: str = ""
skeleton_hash: str = ""
weltbuch_receipt: str = ""
def __post_init__(self) -> None:
if not self.expr_prefix:
self.expr_prefix = _default_prefix_for_family(self.family, self.params)
if self.expr_prefix and not self.skeleton_hash:
self.skeleton, self.skeleton_hash = canonical_symbolic_skeleton(self.expr_prefix)
def byte_size(self) -> int:
return len(
json.dumps(asdict(self), separators=(",", ":"), ensure_ascii=False).encode("utf-8")
)
def compute_invariant(self, values: Sequence[float], count_len: int = 0) -> Optional[float]:
if not self.accepted or self.muted:
return None
fam = self.family
if fam == "symbolic_hcgm" and len(values) >= 1 and self.expr_prefix:
a = float(values[0])
b = float(values[1]) if len(values) >= 2 else 0.0
z = float(values[2]) if len(values) >= 3 else 0.0
c0 = float(self.params.get("c0", 1.0))
c1 = float(self.params.get("c1", 1.0))
return eval_symbolic_prefix_f64(self.expr_prefix, a, b, z, c0, c1)
if fam == "max" and len(values) >= 2:
if len(values) == 2:
a, b = values[0], values[1]
sigma = float(self.params.get("sigma", 1.0))
return 0.5 * (a + b) + sigma * 0.5 * abs(a - b)
return float(max(values))
if fam == "min" and len(values) >= 2:
if len(values) == 2:
a, b = values[0], values[1]
sigma = float(self.params.get("sigma", -1.0))
return 0.5 * (a + b) + sigma * 0.5 * abs(a - b)
return float(min(values))
if fam == "median" and len(values) >= 3 and len(values) % 2 == 1:
s_vals = sorted(float(v) for v in values)
return s_vals[len(s_vals) // 2]
if fam == "closer" and len(values) == 3:
a, b, target = values[0], values[1], values[2]
if abs(a - target) == abs(b - target):
return None
return a if abs(a - target) < abs(b - target) else b
if fam == "count":
if count_len >= 1:
inc = float(self.params.get("increment", 1.0))
off = float(self.params.get("offset", 0.0))
return inc * count_len + off
if len(values) >= 1:
return float(len(values))
if fam == "diff" and len(values) == 2:
return float(abs(values[0] - values[1]))
if fam == "add" and len(values) >= 2:
return float(sum(values))
if fam == "sub" and len(values) >= 2:
acc = values[0]
for v in values[1:]:
acc -= v
return float(acc)
if fam == "mul" and len(values) >= 2:
if all(v > 0 and abs(v - round(v)) < 1e-9 for v in values):
return float(round(math.exp(sum(math.log(v) for v in values))))
prod = 1.0
for v in values:
prod *= v
return float(prod)
if fam == "div" and len(values) == 2:
denom = int(round(values[1]))
if denom == 0:
return None
return float(int(round(values[0])) // denom)
if fam == "mod":
if "modulus" in self.params and len(values) >= 1:
modulus = max(1, int(round(self.params["modulus"])))
return float(int(round(sum(values))) % modulus)
if len(values) == 2:
denom = int(round(values[1]))
if denom == 0:
return None
return float(int(round(values[0])) % denom)
if fam == "parity" and len(values) == 1:
return float(int(round(values[0])) % 2)
if fam == "affine2" and len(values) == 2:
ca = float(self.params.get("ca", 1.0))
cb = float(self.params.get("cb", 1.0))
c0 = float(self.params.get("c0", 0.0))
return ca * values[0] + cb * values[1] + c0
return None
def default_k6_k7_organs() -> List[OrganSpec]:
"""Complete 12-Organ Atlas uniting arithmetic, selection & Z2 parity."""
return [
OrganSpec(
organ_id="k7.max",
family="max",
triggers=("larger", "größer", "groesser", "max", "maximum"),
support_agreement=1.0,
params={"sigma": 1.0},
source="o1_invariant_atlas",
),
OrganSpec(
organ_id="k7.min",
family="min",
triggers=("smaller", "kleiner", "min", "minimum"),
support_agreement=1.0,
params={"sigma": -1.0},
source="o1_invariant_atlas",
),
OrganSpec(
organ_id="k7.median",
family="median",
triggers=("middle", "mitte", "median"),
support_agreement=0.9792,
params={},
source="o1_invariant_atlas",
),
OrganSpec(
organ_id="k7.closer",
family="closer",
triggers=("near", "closer", "näher", "naeher"),
support_agreement=0.9792,
params={},
source="o1_invariant_atlas",
),
OrganSpec(
organ_id="k7.count",
family="count",
triggers=("__noun_list__", "count", "anzahl"),
support_agreement=1.0,
params={"increment": 1.0, "offset": 0.0},
source="o1_invariant_atlas",
),
OrganSpec(
organ_id="k7.diff",
family="diff",
triggers=("distance", "abstand", "diff", "unterschied"),
support_agreement=1.0,
params={},
source="o1_invariant_atlas",
),
OrganSpec(
organ_id="k6.add",
family="add",
triggers=("plus", "+", "add"),
support_agreement=0.9938,
params={"sign": 1.0},
source="o1_invariant_atlas",
),
OrganSpec(
organ_id="k6.sub",
family="sub",
triggers=("minus", "subtract", "less", "-"),
support_agreement=0.9889,
params={"sign": -1.0},
source="o1_invariant_atlas",
),
OrganSpec(
organ_id="k6.mul",
family="mul",
triggers=("times", "mal", "*"),
support_agreement=1.0,
params={"lattice": 1.0},
source="o1_invariant_atlas",
),
OrganSpec(
organ_id="k6.div",
family="div",
triggers=("divided", "durch", "div", "//"),
support_agreement=0.991,
params={"lattice": -1.0},
source="o1_invariant_atlas",
),
OrganSpec(
organ_id="k6.mod",
family="mod",
triggers=("mod", "modulo", "rest", "%"),
support_agreement=0.995,
params={},
source="o1_invariant_atlas",
),
OrganSpec(
organ_id="z2.parity",
family="parity",
triggers=("parity", "parität", "paritaet"),
support_agreement=1.0,
params={},
source="o1_invariant_atlas",
),
]
class OrganBank:
"""Registry & weight-subspace-coupled router for grafted Organ Shells."""
def __init__(self, dose_s0: float = 10.0, seed_defaults: bool = True) -> None:
self.dose_s0 = float(dose_s0)
self.organs: Dict[str, OrganSpec] = {}
if seed_defaults:
for spec in default_k6_k7_organs():
self.register(spec)
def register(self, spec: OrganSpec) -> None:
self.organs[spec.organ_id] = spec
def total_bytes(self) -> int:
return sum(o.byte_size() for o in self.organs.values())
def funktor_equivalences(self) -> List[Dict[str, Any]]:
"""Group active organs by canonical `skeleton_hash` (`Funktor` cross-domain isomorphism)."""
by_hash: Dict[str, List[OrganSpec]] = {}
for spec in self.organs.values():
if spec.accepted and not spec.muted and spec.skeleton_hash:
by_hash.setdefault(spec.skeleton_hash, []).append(spec)
groups: List[Dict[str, Any]] = []
for skel_hash, members in sorted(by_hash.items()):
if len(members) >= 2:
groups.append(
{
"skeleton_hash": skel_hash,
"skeleton": members[0].skeleton,
"organ_ids": [m.organ_id for m in members],
"triggers": [t for m in members for t in m.triggers],
"count": len(members),
}
)
return groups
def export_json(self) -> Dict[str, Any]:
return {
"dose_s0": self.dose_s0,
"total_bytes": self.total_bytes(),
"funktor_equivalences": self.funktor_equivalences(),
"organs": {oid: asdict(spec) for oid, spec in self.organs.items()},
}
@staticmethod
def _parse_number_token(tok: str) -> Tuple[Optional[float], str]:
low = tok.lower()
if re.fullmatch(r"-?\d+", low):
return float(int(low)), "digit"
if re.fullmatch(r"-?\d+\.\d+", low):
return float(low), "float"
if low in EN_NUM_WORDS:
return float(EN_NUM_WORDS[low]), "en"
if low in DE_NUM_WORDS:
return float(DE_NUM_WORDS[low]), "de"
return None, ""
@staticmethod
def _extract_active_clause_bytes(raw_bytes: bytes) -> bytes:
last_gt = raw_bytes.rfind(b">")
while last_gt != -1 and last_gt > 0 and raw_bytes[last_gt - 1 : last_gt + 1] == b"->":
last_gt = raw_bytes.rfind(b">", 0, last_gt - 1)
if last_gt != -1:
raw_bytes = raw_bytes[last_gt + 1 :]
for sep in (b"\n", b";", b"?", b"!", b":"):
pos = raw_bytes.rfind(sep)
if pos != -1:
raw_bytes = raw_bytes[pos + 1 :]
dot_space = raw_bytes.rfind(b". ")
if dot_space != -1:
raw_bytes = raw_bytes[dot_space + 2 :]
return raw_bytes
def _find_organ_for_op(self, op_word: str) -> Optional[OrganSpec]:
for spec in self.organs.values():
if op_word in spec.triggers and spec.accepted and not spec.muted:
return spec
return None
def _eval_chained_ops(
self, values: List[float], ops: List[str]
) -> Tuple[Optional[float], Optional[str], Optional[str]]:
"""Evaluate alternating `v0 op0 v1 op1 ...` with multiplicative (`*`, `//`, `%`) precedence."""
if len(values) != len(ops) + 1 or not ops:
return None, None, None
organ_seq: List[OrganSpec] = []
for op in ops:
spec = self._find_organ_for_op(op)
if spec is None:
return None, None, None
organ_seq.append(spec)
# Pass 1: high-precedence multiplicative/quotient organs (`mul`, `div`, `mod`)
vals = list(values)
rem_organs: List[OrganSpec] = []
for spec in organ_seq:
if spec.family in ("mul", "div", "mod") and "modulus" not in spec.params:
res = spec.compute_invariant([vals[len(rem_organs)], vals[len(rem_organs) + 1]])
if res is None:
return None, None, None
vals[len(rem_organs) : len(rem_organs) + 2] = [res]
else:
rem_organs.append(spec)
# Pass 2: left-to-right binary reduction
acc = vals[0]
for step_idx, spec in enumerate(rem_organs):
nxt = vals[step_idx + 1]
res = spec.compute_invariant([acc, nxt])
if res is None:
return None, None, None
acc = res
lead_id = (
organ_seq[-1].organ_id
if len(organ_seq) == 1
else "+".join(dict.fromkeys(s.organ_id for s in organ_seq))
)
route_name = organ_seq[-1].family.upper() if len(organ_seq) == 1 else "CHAIN"
return acc, lead_id, route_name
def _eval_numeric_subbody(
self, low_body: List[str]
) -> Tuple[Optional[float], Optional[str], Optional[str], List[str]]:
values: List[float] = []
formats: List[str] = []
op_tokens: List[str] = []
alternating_valid = len(low_body) >= 3 and (len(low_body) % 2 == 1)
for idx, tok in enumerate(low_body):
val, fmt = self._parse_number_token(tok)
if val is not None:
values.append(float(val))
formats.append(fmt)
if idx % 2 == 1:
alternating_valid = False
else:
op_tokens.append(tok)
if idx % 2 == 0:
alternating_valid = False
if len(low_body) == 1 and len(values) == 1:
return values[0], "literal", "LITERAL", formats
if not values or not op_tokens:
return None, None, None, formats
unique_ops = set(op_tokens)
if len(unique_ops) == 1:
op_word = next(iter(unique_ops))
matched_organ = self._find_organ_for_op(op_word)
if matched_organ is not None:
u_val = matched_organ.compute_invariant(values)
if u_val is not None:
return (
u_val,
matched_organ.organ_id,
matched_organ.family.upper(),
formats,
)
if alternating_valid and len(op_tokens) >= 2:
u_val, oid, rname = self._eval_chained_ops(values, op_tokens)
return u_val, oid, rname, formats
return None, None, None, formats
def route_context(
self,
text_prefix: str = "",
*,
prefix_bytes: Optional[bytes] = None,
) -> Dict[str, Any]:
"""Structural router supporting all 12 organs, chains, postfix parity & 17-Op HCGM organs."""
raw = prefix_bytes if prefix_bytes is not None else text_prefix.encode("utf-8")
clause = self._extract_active_clause_bytes(raw).lstrip(b" \t\r")
if not clause:
return {"matched": False, "route": "TEXT"}
m = re.match(
rb"^(.*?(?:\b(?:ist|is)\b|->|=))([ \t]*)([A-Za-z0-9\.\-\x80-\xff]*)$",
clause,
flags=re.IGNORECASE,
)
if not m:
return {"matched": False, "route": "TEXT"}
head_bytes = m.group(1).strip()
space_bytes = m.group(2)
partial_bytes = m.group(3)
try:
head_str = head_bytes.decode("utf-8")
except UnicodeDecodeError:
return {"matched": False, "route": "TEXT"}
words = head_str.split()
if len(words) < 2:
return {"matched": False, "route": "TEXT"}
final_tok = words[-1].lower()
if final_tok not in TRIGGER_FINALS:
return {"matched": False, "route": "TEXT"}
body = words[:-1]
low_body = [w.lower() for w in body]
has_space_after_trigger = len(space_bytes) > 0
# 1. K7 Counting route (pure countable noun list OR trailing `count`/`anzahl`)
explicit_count = low_body[-1] in ("count", "anzahl") and len(low_body) >= 2
noun_slice = low_body[:-1] if explicit_count else low_body
if len(noun_slice) >= 2 and all(w in COUNTABLE_NOUNS for w in noun_slice):
count_organ = self.organs.get("k7.count")
if count_organ and count_organ.accepted and not count_organ.muted:
u_val = count_organ.compute_invariant([], count_len=len(noun_slice))
is_de = (
final_tok == "ist"
or low_body[-1] == "anzahl"
or any(
w
in {
"katze", "hund", "vogel", "pferd", "fisch", "baum", "haus",
"auto", "schiff", "könig", "koenig", "fluss", "stadt", "buch",
"lied", "spiel", "stein", "bär", "baer", "kind", "apfel",
"stern", "knoten", "welle", "zelle",
}
for w in noun_slice
)
)
return {
"matched": u_val is not None,
"route": "COUNT",
"organ_id": count_organ.organ_id,
"u": u_val,
"format": "de" if is_de else "en",
"partial_bytes": partial_bytes,
"has_space_after_trigger": has_space_after_trigger,
}
# 2. Postfix Z2 Parity route (`<expr> parity is` / `<expr> parität ist`)
parity_organ = self._find_organ_for_op(low_body[-1])
if parity_organ is not None and parity_organ.family == "parity" and len(low_body) >= 2:
sub_u, sub_oid, _sub_r, formats = self._eval_numeric_subbody(low_body[:-1])
if sub_u is not None:
par_bit = parity_organ.compute_invariant([sub_u])
if par_bit is not None:
is_de_par = (
final_tok == "ist"
or low_body[-1] in ("parität", "paritaet")
or "de" in formats
)
if final_tok in ("=", "->") and "en" not in formats and "de" not in formats:
p_fmt = "digit"
else:
p_fmt = "parity_de" if is_de_par else "parity_en"
comb_id = (
parity_organ.organ_id
if sub_oid in (None, "literal")
else f"{sub_oid}+{parity_organ.organ_id}"
)
return {
"matched": True,
"route": "PARITY",
"organ_id": comb_id,
"u": par_bit,
"format": p_fmt,
"partial_bytes": partial_bytes,
"has_space_after_trigger": has_space_after_trigger,
}
# 3. General Single-Organ or Multi-Organ Chained Evaluation
u_val, organ_id_out, route_name, formats = self._eval_numeric_subbody(low_body)
if u_val is None or organ_id_out in (None, "literal") or route_name is None:
return {"matched": False, "route": "TEXT"}
is_non_integer = abs(u_val - round(u_val)) > 1e-4
if is_non_integer:
out_fmt = "float"
elif final_tok in ("=", "->") or "digit" in formats or "float" in formats:
out_fmt = "digit"
elif final_tok == "ist" or "de" in formats:
out_fmt = "de"
else:
out_fmt = "en"
if not is_non_integer:
u_int = int(round(u_val))
if out_fmt == "en" and u_int not in INV_EN_NUM:
out_fmt = "digit"
elif out_fmt == "de" and u_int not in INV_DE_NUM:
out_fmt = "digit"
return {
"matched": True,
"route": route_name,
"organ_id": organ_id_out,
"u": u_val,
"format": out_fmt,
"partial_bytes": partial_bytes,
"has_space_after_trigger": has_space_after_trigger,
}
@staticmethod
def _candidate_table(out_fmt: str, u_center: float) -> List[Tuple[Union[int, float], str]]:
if out_fmt == "float":
exact_str = f"{round(u_center, 6):.6g}"
if exact_str == "-0":
exact_str = "0"
return [
(u_center, exact_str),
(u_center + 1.0, f"{round(u_center + 1.0, 6):.6g}"),
]
if out_fmt == "parity_en":
return [(0, "even"), (1, "odd")]
if out_fmt == "parity_de":
return [(0, "gerade"), (1, "ungerade")]
if out_fmt == "en":
return [(val, word) for val, word in sorted(INV_EN_NUM.items())]
if out_fmt == "de":
return [(val, word) for val, word in sorted(INV_DE_NUM.items())]
center = int(round(u_center))
lo = min(-20, center - 16)
hi = max(100, center + 16)
return [(val, str(val)) for val in range(lo, hi + 1)]
def _try_weight_subspace_shell(
self,
logits_1d: torch.Tensor,
embed_inv: torch.Tensor,
u_val: float,
out_fmt: str,
need_leading_space: bool,
s0: float,
) -> Optional[Tuple[torch.Tensor, str]]:
"""Direct rank-2 dot product against `model.embed.weight[:, -16:]` invariant slice!
Uses exact algebraic identity on imprinted coordinates `inv[:, 0] = c/20`,
`inv[:, 1] = -c^2/800`:
40*S0*u * inv[:, 0] + 800*S0 * inv[:, 1] - S0*u^2 == -S0 * (c - u)^2
Executes natively on GPU/MPS/NPU tensor memory in 1 single crystal step.
"""
if out_fmt == "float" or abs(u_val - round(u_val)) > 1e-4:
return None
if embed_inv.ndim != 2 or embed_inv.shape[1] < 16:
return None
space_flag = 1.0 if need_leading_space else 0.0
space_mask = (embed_inv[:, 14] - space_flag).abs() < 0.1
if out_fmt in ("parity_en", "parity_de"):
lang_col = 9 if out_fmt == "parity_en" else 10
cand_mask = (
space_mask
& (embed_inv[:, 11] > 0.5)
& (embed_inv[:, lang_col] > 0.5)
)
if int(cand_mask.sum().item()) < 2:
return None
target_z2 = 1.0 if int(round(u_val)) % 2 == 0 else -1.0
margin = min(max(s0 * 2.5, 8.0), 40.0)
sculpted = logits_1d.clone()
potentials = margin * target_z2 * embed_inv[cand_mask, 4]
sculpted[cand_mask] = sculpted[cand_mask] + potentials
target_word = (
("even" if target_z2 > 0 else "odd")
if out_fmt == "parity_en"
else ("gerade" if target_z2 > 0 else "ungerade")
)
return sculpted, target_word
fmt_col = {"digit": 8, "en": 9, "de": 10}.get(out_fmt)
if fmt_col is None:
return None
cand_mask = (
space_mask
& (embed_inv[:, 7] > 0.5)
& (embed_inv[:, fmt_col] > 0.5)
)
if int(cand_mask.sum().item()) == 0:
return None
c_slice = embed_inv[cand_mask, 0] * 20.0
min_dist = float(torch.min(torch.abs(c_slice - float(u_val))).item())
if min_dist > 1e-2:
return None
q_vec = torch.tensor(
[40.0 * s0 * float(u_val), 800.0 * s0],
device=embed_inv.device,
dtype=embed_inv.dtype,
)
raw_pot = embed_inv[cand_mask, :2] @ q_vec - (s0 * float(u_val) * float(u_val))
margin_cap = min(max(s0 * 2.5, 8.0), 40.0)
deltas = torch.clamp(raw_pot + margin_cap, min=-margin_cap, max=margin_cap)
sculpted = logits_1d.clone()
sculpted[cand_mask] = sculpted[cand_mask] + deltas.to(sculpted.dtype)
u_int = int(round(u_val))
if out_fmt == "en":
target_str = INV_EN_NUM.get(u_int, str(u_int))
elif out_fmt == "de":
target_str = INV_DE_NUM.get(u_int, str(u_int))
else:
target_str = str(u_int)
return sculpted, target_str
def apply_logit_shell(
self,
logits_1d: torch.Tensor,
text_prefix: str = "",
codec: Optional[ZeroUNKCodec] = None,
*,
prefix_bytes: Optional[bytes] = None,
dose_override: Optional[float] = None,
embed_inv: Optional[torch.Tensor] = None,
) -> Tuple[torch.Tensor, Dict[str, Any]]:
"""Sculpt next-token logits via weight-native `embed_inv` slice or lossless byte fallback."""
s0 = self.dose_s0 if dose_override is None else float(dose_override)
route_info = self.route_context(text_prefix, prefix_bytes=prefix_bytes)
if not route_info.get("matched") or s0 == 0.0:
return logits_1d, route_info
u_val = float(route_info["u"])
out_fmt = str(route_info["format"])
partial_bytes: bytes = route_info["partial_bytes"]
has_space = bool(route_info["has_space_after_trigger"])
need_leading_space = len(partial_bytes) == 0 and not has_space
# Fast-path 1: Direct tensor product on model's imprinted invariant embedding slice!
if len(partial_bytes) == 0 and embed_inv is not None:
ws_hit = self._try_weight_subspace_shell(
logits_1d, embed_inv, u_val, out_fmt, need_leading_space, s0
)
if ws_hit is not None:
sculpted_ws, target_ws = ws_hit
route_info["partial"] = ""
route_info.pop("partial_bytes", None)
route_info["target_surface"] = target_ws
route_info["dose_s0"] = s0
route_info["execution_mode"] = "weight_invariant_subspace"
return sculpted_ws, route_info
# Fast-path 2: Exact UTF-8 byte / crystal prefix continuation (handles floats & arbitrary big ints)
candidates = self._candidate_table(out_fmt, u_val)
sculpted = logits_1d.clone()
best_target_str = None
best_shell = -1e30
next_tok_best_shell: Dict[int, float] = {}
for c_val, word in candidates:
word_bytes = word.encode("utf-8")
if not word_bytes.startswith(partial_bytes):
continue
shell_potential = -s0 * ((float(c_val) - u_val) ** 2)
if shell_potential > best_shell:
best_shell = shell_potential
best_target_str = word
rem_bytes = word_bytes[len(partial_bytes) :]
if len(rem_bytes) == 0:
delims = [ord(" "), ord("\n")]
if codec is not None:
delims.append(codec.EOS)
for delim_tok in delims:
prev = next_tok_best_shell.get(delim_tok, -1e30)
if shell_potential > prev:
next_tok_best_shell[delim_tok] = shell_potential
else:
emit_bytes = (b" " + rem_bytes) if need_leading_space else rem_bytes
first_tok: Optional[int] = None
if codec is not None:
try:
emit_str = emit_bytes.decode("utf-8")
tok_ids = codec.encode(emit_str, add_bos=False, add_eos=False)
if tok_ids:
first_tok = tok_ids[0]
except UnicodeDecodeError:
first_tok = int(emit_bytes[0])
else:
first_tok = int(emit_bytes[0])
if first_tok is not None:
prev = next_tok_best_shell.get(first_tok, -1e30)
if shell_potential > prev:
next_tok_best_shell[first_tok] = shell_potential
if next_tok_best_shell:
peak = max(next_tok_best_shell.values())
margin_cap = min(max(s0 * 2.5, 8.0), 40.0)
for tok_id, pot in next_tok_best_shell.items():
delta = max(pot - peak + margin_cap, -margin_cap)
sculpted[tok_id] = sculpted[tok_id] + delta
route_info["partial"] = partial_bytes.decode("utf-8", errors="replace")
route_info.pop("partial_bytes", None)
route_info["target_surface"] = best_target_str
route_info["dose_s0"] = s0
route_info["execution_mode"] = "byte_prefix_shell"
return sculpted, route_info
class MRail:
"""K1 Context-Keyed Trajectory Cache (`key_width` tokens -> verified rail)."""
def __init__(self, key_width: int = 6, min_hits_to_lock: int = 2) -> None:
self.key_width = key_width
self.min_hits_to_lock = min_hits_to_lock
self._tracks: Dict[Tuple[int, ...], Dict[int, int]] = {}
self.total_queries = 0
self.total_hits = 0
def observe_sequence(self, token_ids: Sequence[int]) -> None:
w = self.key_width
n = len(token_ids)
for i in range(w, n):
key = tuple(int(t) for t in token_ids[i - w : i])
nxt = int(token_ids[i])
bucket = self._tracks.setdefault(key, {})
bucket[nxt] = bucket.get(nxt, 0) + 1
def lookup(self, recent_ids: Sequence[int]) -> Optional[Tuple[int, float, int]]:
self.total_queries += 1
if len(recent_ids) < self.key_width:
return None
key = tuple(int(t) for t in recent_ids[-self.key_width :])
bucket = self._tracks.get(key)
if not bucket:
return None
best_tok, best_cnt = max(bucket.items(), key=lambda kv: kv[1])
total = sum(bucket.values())
conf = best_cnt / float(total)
if best_cnt >= self.min_hits_to_lock and conf >= 0.85:
self.total_hits += 1
return best_tok, conf, best_cnt
return None
def stats(self) -> Dict[str, Any]:
return {
"rail_keys": len(self._tracks),
"queries": self.total_queries,
"hits": self.total_hits,
"hit_rate": round(self.total_hits / max(1, self.total_queries), 4),
}
@dataclass
class GraveyardRegion:
region_id: str
domain: str
signature: str
reason: str
residual: float
samples_seen: int
class GraveyardStore:
"""Wirkung Negative-Knowledge Store: records falsified hypotheses so they never re-run."""
def __init__(self) -> None:
self.regions: Dict[str, GraveyardRegion] = {}
@staticmethod
def make_signature(
domain: str, op_token: str, pairs: Sequence[Tuple[float, float, float]]
) -> str:
canonical = json.dumps(
{"domain": domain, "op": op_token, "n": len(pairs), "head": pairs[:6]},
sort_keys=True,
)
return hashlib.sha256(canonical.encode("utf-8")).hexdigest()[:16]
def is_covered(self, signature: str) -> bool:
return signature in self.regions
def bury(
self,
domain: str,
signature: str,
reason: str,
residual: float,
samples_seen: int,
) -> GraveyardRegion:
reg = GraveyardRegion(
region_id=f"grave.{domain}.{signature[:8]}",
domain=domain,
signature=signature,
reason=reason,
residual=float(residual),
samples_seen=int(samples_seen),
)
self.regions[signature] = reg
return reg
"""Weight-Coupled Invariant Organ Atlas, Funktor Equivalence, M-Rail & Graveyard (`organs.py`)."""
from __future__ import annotations
import hashlib
import json
import math
import re
from dataclasses import asdict, dataclass, field
from typing import Any, Dict, List, Optional, Sequence, Tuple, Union
import torch
from .hypercodec import ZeroUNKCodec
from .vocab import (
CANONICAL_DE_NUM,
CANONICAL_EN_NUM,
COUNTABLE_NOUN_SET,
DE_NUMBER_LEXICON,
EN_NUMBER_LEXICON,
)
_BINARY_SET = frozenset({"add", "sub", "mul", "div", "pow", "moebius", "casimir_sqrt", "pade11"})
_UNARY_SET = frozenset({"abs", "cos", "exp", "log", "neg", "sin", "sqrt", "lorentz_gamma", "chebyshev_t2"})
_ATOM_SET = frozenset({"x", "y", "z", "c0", "c1", "pi", "e", "1", "2", "0.5"})
_COMMUTATIVE_OPS = frozenset({"add", "mul", "moebius"})
def eval_symbolic_prefix_f64(
expr_prefix: str,
x: float,
y: float = 0.0,
z: float = 0.0,
c0: float = 1.0,
c1: float = 1.0,
) -> Optional[float]:
"""IEEE-754 float64 evaluator for Polish-prefix symbolic organ expressions."""
toks = [t for t in expr_prefix.strip().split() if t != "NOP"]
if not toks:
return None
stack: List[float] = []
atom_vals = {
"x": float(x), "y": float(y), "z": float(z),
"c0": float(c0), "c1": float(c1),
"pi": math.pi, "e": math.e, "1": 1.0, "2": 2.0, "0.5": 0.5,
}
try:
for t in reversed(toks):
if t in atom_vals:
stack.append(atom_vals[t])
elif t in _UNARY_SET:
if not stack:
return None
a1 = stack.pop()
if t == "abs":
r = abs(a1)
elif t == "cos":
r = math.cos(a1)
elif t == "exp":
if abs(a1) > 20.0:
return None
r = math.exp(a1)
elif t == "log":
if a1 <= 1e-9:
return None
r = math.log(a1)
elif t == "neg":
r = -a1
elif t == "sin":
r = math.sin(a1)
elif t == "sqrt":
if a1 < -1e-12:
return None
r = math.sqrt(max(a1, 0.0))
elif t == "lorentz_gamma":
ins = 1.0 - a1 * a1
if ins <= 1e-6:
return None
r = 1.0 / math.sqrt(ins)
elif t == "chebyshev_t2":
r = 2.0 * a1 * a1 - 1.0
else:
return None
if not math.isfinite(r) or abs(r) >= 60000.0:
return None
stack.append(r)
elif t in _BINARY_SET:
if len(stack) < 2:
return None
a1 = stack.pop()
a2 = stack.pop()
if t == "add":
r = a1 + a2
elif t == "sub":
r = a1 - a2
elif t == "mul":
r = a1 * a2
elif t == "div":
if abs(a2) <= 1e-7:
return None