#!/usr/bin/env python3
"""skill_birth — topic-agnostic skill-birth engine (Builder A).

Implements the frozen `interfaces.py` API. One universal decision path for every family:
detect registry miss -> name the gap from I/O shape -> write a plain-English empirical
standard -> enumerate generic operators over the OBSERVED shape (+ compositions over born
skills) -> keep train-consistent candidates -> negative / hold-out / adversarial /
injected-contradiction gates -> adopt iff exactly ONE distinct surviving behavior on a
discriminating hold-out, else a constructive refusal naming the exact missing evidence.

The engine reads example DATA and I/O SHAPE and calls the family checker. It never reads a
fixture's topic identity (G1): parameters are enumerated from observed values, never from a
topic word. Zero RNG, zero wall-clock, zero `hash()`; every enumeration canonically sorted
before use (G10). All failures are typed and loud.
"""
from __future__ import annotations

import hashlib
import json
from typing import Any, Callable, Dict, List, Mapping, Optional, Tuple

from interfaces import (
    AccountingLedger,
    BirthCounts,
    BirthOutcome,
    BirthStatus,
    BornSkill,
    CandidateSpec,
    EmpiricalStandard,
    GENERIC_OPERATORS,
    MalformedFamilyError,
    NonDeterminismError,
    RefusalKind,
    RefusalRequest,
    SkillBirthError,
    SkillHypothesis,
    TaskFamily,
)
from interfaces import SkillBirthEngine as _EngineAPI


class CandidateRefusal(SkillBirthError):
    """A candidate program's CONTROLLED refusal on an input outside its guard. Counts as
    'refuse' (never a confident wrong answer) in adversarial evaluation. Any OTHER exception
    from a program is a crash and eliminates the candidate."""


_REFUSE: Tuple[str, ...] = ("__refuse__",)
_CRASH: Tuple[str, ...] = ("__crash__",)

_SCHEMES: Tuple[str, ...] = ("casefold", "casefold+strip", "dedup", "sort", "sort+dedup", "strip")
_SEQ_SCHEMES: Tuple[str, ...] = ("dedup", "sort", "sort+dedup")
_AGGS: Tuple[str, ...] = ("count", "first", "join", "last", "max", "min", "sum")


def canon(value: Any) -> str:
    """Canonical deterministic serialization used for equality, ordering and hashing keys.
    Never `hash()` (G10)."""
    return json.dumps(value, sort_keys=True, default=repr)


def canonical_eq(a: Any, b: Any) -> bool:
    return canon(a) == canon(b)


def _is_num(v: Any) -> bool:
    return isinstance(v, (int, float)) and not isinstance(v, bool)


def _kind(value: Any) -> str:
    """Coarse structural kind used for guards/branching. Shape inspection is allowed (G1);
    these short tags are never topic words."""
    if value is None:
        return "nul"
    if isinstance(value, bool) or _is_num(value):
        return "sc"
    if isinstance(value, str):
        return "txt"
    if isinstance(value, (list, tuple)):
        return "sq"
    if isinstance(value, dict):
        return "rec"
    raise MalformedFamilyError("unsupported value kind: %r" % type(value).__name__)


def shape(value: Any) -> str:
    """Structural, topic-blind shape tag (DESIGN section 1)."""
    k = _kind(value)
    if k == "nul":
        return "null"
    if k == "sc":
        return "scalar"
    if k == "txt":
        return "text"
    if k == "sq":
        inner = sorted({shape(e) for e in value})
        return "seq[" + "|".join(inner) + "]"
    return "record<n=%d>" % len(value)


class _SplitView:
    """ENGINE-AUTHORED train/hold-out split (integration fix, 2026-07-05): the fixture ships
    one example POOL; the ENGINE decides the split, so standard-authorship is demonstrably
    the engine's, not the fixture author's. Deterministic: pool ordered by sha256 of each
    canonical example pair; every 3rd position (hash order) becomes hold-out (>=1 guaranteed).
    Same attribute surface as TaskFamily minus the harness-only name/expect fields — the
    engine never reads those (G1)."""
    __slots__ = ("examples", "holdout", "adversarial_seeds", "negatives", "probe_pool",
                 "checker", "available_atoms", "_engine_split")

    def __init__(self, family: TaskFamily):
        pool = list(family.examples) + list(family.holdout)
        order = sorted(range(len(pool)),
                       key=lambda i: hashlib.sha256(canon(pool[i]).encode()).hexdigest())
        hold = {i for pos, i in enumerate(order) if pos % 3 == 2}
        if not hold and pool:
            hold = {order[-1]}
        self.examples = [pool[i] for i in range(len(pool)) if i not in hold]
        self.holdout = [pool[i] for i in range(len(pool)) if i in hold]
        self.adversarial_seeds = list(family.adversarial_seeds)
        self.negatives = list(family.negatives)
        self.probe_pool = list(family.probe_pool)
        self.checker = family.checker
        self.available_atoms = tuple(getattr(family, "available_atoms", ()) or ())
        self._engine_split = True


def engine_split(family: TaskFamily) -> Any:
    """Idempotent: name_skill/write_standard/birth all funnel through this, so the split
    happens exactly once regardless of call order."""
    return family if getattr(family, "_engine_split", False) else _SplitView(family)


def _digest(family: TaskFamily) -> Dict[str, Any]:
    """Observed, sorted, data-driven parameter sources. Reads DATA and SHAPE only."""
    if not isinstance(family.examples, list) or not family.examples:
        raise MalformedFamilyError("family has no worked examples")
    if not callable(family.checker):
        raise MalformedFamilyError("family checker is not callable")
    xs = [x for x, _ in family.examples]
    ys = [y for _, y in family.examples]
    in_shape, out_shape = shape(xs[0]), shape(ys[0])
    for x, y in family.examples:
        if shape(x) != in_shape or shape(y) != out_shape:
            raise MalformedFamilyError(
                "unstable example I/O shape: %s->%s vs %s->%s"
                % (in_shape, out_shape, shape(x), shape(y)))
    for x, _ in family.holdout:
        if _kind(x) != _kind(xs[0]):
            raise MalformedFamilyError("held-out input kind differs from worked examples")
    ik = _kind(xs[0])
    keys: List[str] = []
    num_keys: List[str] = []
    if ik == "rec":
        common = set(xs[0].keys())
        for x in xs[1:]:
            common &= set(x.keys())
        keys = sorted(common)
        num_keys = [k for k in keys if all(_is_num(x[k]) for x in xs)]
    idxs: List[int] = []
    elem_nums: List[Any] = []
    if ik == "sq":
        m = min(len(x) for x in xs)
        idxs = list(range(m)) + ([-1] if m else [])
        elem_nums = sorted({e for x in xs for e in x if _is_num(e)})
    delims: List[str] = []
    split_idx: Dict[str, List[int]] = {}
    if ik == "txt":
        seen = sorted({ch for x in xs for ch in x if not ch.isalnum()})
        delims = [d for d in seen if all(d in x for x in xs)]
        for d in delims:
            m = min(len(x.split(d)) for x in xs)
            split_idx[d] = list(range(m)) + ([-1] if m else [])
    answers: List[Any] = []
    seen_a = set()
    for y in sorted(ys, key=canon):
        c = canon(y)
        if c not in seen_a:
            seen_a.add(c)
            answers.append(y)
    return {
        "xs": xs, "ys": ys, "in_kind": ik, "in_shape": in_shape, "out_shape": out_shape,
        "keys": keys, "num_keys": num_keys, "idxs": idxs, "elem_nums": elem_nums,
        "delims": delims, "split_idx": split_idx, "answers": answers,
    }


def generate_adversarial(inputs: List[Any], seeds: List[Any]) -> List[Any]:
    """Engine-generated adversarial inputs from OBSERVED inputs + shape only (DESIGN section 3):
    shape mutations (reorder / duplicate / drop / field swap), boundary (empty / singleton /
    longest-seen), garbage (type mismatch, None), plus family-supplied seeds. Deterministic
    construction order, canon-deduplicated."""
    if not inputs:
        raise MalformedFamilyError("cannot generate adversarial inputs from zero examples")
    base = inputs[0]
    longest = max(inputs, key=lambda v: (len(v) if hasattr(v, "__len__") else 0, canon(v)))
    k = _kind(base)
    out: List[Any] = []
    if k == "txt":
        out += ["".join(reversed(base)), base + base, base[1:], "", base[:1], longest]
    elif k == "sq":
        b = list(base)
        out += [list(reversed(b)), b + b[:1], b[1:], [], b[:1], list(longest)]
        if len(b) >= 2:
            out.append([b[1], b[0]] + b[2:])
    elif k == "rec":
        ks = sorted(base.keys())
        if len(ks) >= 2:
            sw = dict(base)
            sw[ks[0]], sw[ks[1]] = base[ks[1]], base[ks[0]]
            out.append(sw)
        if ks:
            dr = {kk: base[kk] for kk in ks[1:]}
            out.append(dr)
            nl = dict(base)
            nl[ks[0]] = None
            out.append(nl)
        out.append({})
    else:
        out += [""]
    out += [0, None]
    out += list(seeds)
    dedup: List[Any] = []
    seen = set()
    for v in out:
        c = canon(v)
        if c not in seen:
            seen.add(c)
            dedup.append(v)
    return dedup

# ─────────────────────────────────────────────────────────────────────────────────────────
# Program construction — every candidate program is rebuilt from its JSON-able spec.
# ─────────────────────────────────────────────────────────────────────────────────────────
def _dedup_seq(v: List[Any]) -> List[Any]:
    out: List[Any] = []
    seen = set()
    for e in v:
        c = canon(e)
        if c not in seen:
            seen.add(c)
            out.append(e)
    return out


def _nat_sorted(v: List[Any]) -> List[Any]:
    try:
        return sorted(v)
    except TypeError:
        raise CandidateRefusal("elements not mutually orderable")


def _apply_scheme(scheme: str, v: Any) -> Any:
    if isinstance(v, str):
        if scheme == "casefold":
            return v.casefold()
        if scheme == "strip":
            return v.strip()
        if scheme == "casefold+strip":
            return v.casefold().strip()
        if scheme == "sort":
            return "".join(sorted(v))
        if scheme == "dedup":
            return "".join(_dedup_seq(list(v)))
        if scheme == "sort+dedup":
            return "".join(sorted(_dedup_seq(list(v))))
    elif isinstance(v, (list, tuple)):
        if scheme == "sort":
            return _nat_sorted(list(v))
        if scheme == "dedup":
            return _dedup_seq(list(v))
        if scheme == "sort+dedup":
            return _nat_sorted(_dedup_seq(list(v)))
        if scheme in ("casefold", "strip", "casefold+strip"):
            return [_apply_scheme(scheme, e) for e in v]
    raise CandidateRefusal("scheme %r inapplicable to this input" % scheme)


def _stage_fn(stage: List[Any], reg: Mapping[str, CandidateSpec]) -> Callable[[Any], Any]:
    """One pipeline stage: ['key',k] | ['idx',i] | ['split',d,i] | ['norm',s] | ['skill',n]."""
    tag = stage[0]
    if tag == "key":
        k = stage[1]

        def f_key(v: Any) -> Any:
            if not isinstance(v, dict) or k not in v:
                raise CandidateRefusal("no such field")
            return v[k]
        return f_key
    if tag == "idx":
        i = stage[1]

        def f_idx(v: Any) -> Any:
            if not isinstance(v, (list, tuple)) or not (-len(v) <= i < len(v)):
                raise CandidateRefusal("index out of range")
            return v[i]
        return f_idx
    if tag == "split":
        d, i = stage[1], stage[2]

        def f_split(v: Any) -> Any:
            if not isinstance(v, str):
                raise CandidateRefusal("split needs a str input")
            parts = v.split(d)
            if not (-len(parts) <= i < len(parts)):
                raise CandidateRefusal("split index out of range")
            return parts[i]
        return f_split
    if tag == "norm":
        s = stage[1]
        return lambda v: _apply_scheme(s, v)
    if tag == "skill":
        n = stage[1]
        if n not in reg:
            raise SkillBirthError("unresolved prerequisite skill %r" % n)
        return materialize(reg[n], reg)
    raise SkillBirthError("unknown pipeline stage %r" % stage)


_RELS: Dict[str, Callable[[Any, Any], bool]] = {
    "eq": lambda a, b: canonical_eq(a, b),
    "ge": lambda a, b: a >= b,
    "gt": lambda a, b: a > b,
    "le": lambda a, b: a <= b,
    "lt": lambda a, b: a < b,
}


def _build_op(op: str, params: Dict[str, Any], reg: Mapping[str, CandidateSpec]) -> Callable[[Any], Any]:
    if op == "normalize":
        s = params["scheme"]
        return lambda v: _apply_scheme(s, v)
    if op == "fieldmap":
        # record<n=K> -> record<n=K>: apply a normalize scheme to the string value at ONE
        # observed key, identity on every other field. Generic record->record constructor
        # (integration fix, 2026-07-05 — the algebra had no record-output op at all).
        k, s = params["key"], params["scheme"]

        def f_fieldmap(v: Any) -> Any:
            if not isinstance(v, dict) or k not in v:
                raise CandidateRefusal("fieldmap needs a record containing the observed key")
            u = v[k]
            if not isinstance(u, str):
                raise CandidateRefusal("fieldmap scheme needs a str value at the key")
            w = dict(v)
            w[k] = _apply_scheme(s, u)
            return w
        return f_fieldmap
    if op == "map":
        s = params["scheme"]

        def f_map(v: Any) -> Any:
            if not isinstance(v, (list, tuple)):
                raise CandidateRefusal("map needs a seq input")
            return [_apply_scheme(s, e) for e in v]
        return f_map
    if op == "extract":
        return _stage_fn(params["sel"], reg)
    if op == "lookup":
        fns = [_stage_fn(st, reg) for st in params["pre"]]
        tbl = params["tbl"]

        def f_lookup(v: Any) -> Any:
            for fn in fns:
                v = fn(v)
            c = canon(v)
            if c not in tbl:
                raise CandidateRefusal("no induced entry for this key")
            return tbl[c]
        return f_lookup
    if op == "classify":
        src, rule = params["src"], params["rule"]
        get = (lambda v: v) if src[0] == "self" else _stage_fn(src, reg)
        if rule[0] == "ge":
            _, t, hi, lo = rule

            def f_cge(v: Any) -> Any:
                u = get(v)
                if not _is_num(u):
                    raise CandidateRefusal("threshold needs a numeric value")
                return hi if u >= t else lo
            return f_cge
        if rule[0] == "in":
            member = frozenset(rule[1])
            a, b = rule[2], rule[3]
            return lambda v: a if canon(get(v)) in member else b
        raise SkillBirthError("unknown classify rule %r" % rule[0])
    if op == "reduce":
        agg = params["agg"]

        def f_reduce(v: Any) -> Any:
            if not isinstance(v, (list, tuple)):
                raise CandidateRefusal("reduce needs a seq input")
            if agg == "count":
                return len(v)
            if agg in ("first", "last"):
                if not v:
                    raise CandidateRefusal("empty seq")
                return v[0] if agg == "first" else v[-1]
            if agg == "join":
                if not all(isinstance(e, str) for e in v):
                    raise CandidateRefusal("join needs str elements")
                return "".join(v)
            if not all(_is_num(e) for e in v):
                raise CandidateRefusal("numeric agg needs numeric elements")
            if agg == "sum":
                return sum(v)
            if not v:
                raise CandidateRefusal("empty seq")
            return min(v) if agg == "min" else max(v)
        return f_reduce
    if op == "filter":
        pred = params["pred"]

        def f_filter(v: Any) -> Any:
            if not isinstance(v, (list, tuple)):
                raise CandidateRefusal("filter needs a seq input")
            if pred[0] == "truthy":
                return [e for e in v if e]
            if pred[0] == "nonnull":
                return [e for e in v if e is not None]
            t = pred[1]
            if not all(_is_num(e) for e in v):
                raise CandidateRefusal("threshold filter needs numeric elements")
            if pred[0] == "ge":
                return [e for e in v if e >= t]
            return [e for e in v if e <= t]
        return f_filter
    if op == "compare":
        rel = _RELS[params["rel"]]
        fa, fb = _stage_fn(params["a"], reg), _stage_fn(params["b"], reg)
        strict = params["rel"] != "eq"

        def f_cmp(v: Any) -> Any:
            a, b = fa(v), fb(v)
            if strict and not (_is_num(a) and _is_num(b)):
                raise CandidateRefusal("relational compare needs numeric fields")
            return bool(rel(a, b))
        return f_cmp
    if op == "compose":
        for nm in (params["g"], params["h"]):
            if nm not in reg:
                raise SkillBirthError("unresolved prerequisite skill %r" % nm)
        g = materialize(reg[params["g"]], reg)
        h = materialize(reg[params["h"]], reg)
        return lambda v: h(g(v))
    if op == "sequence":
        fns = []
        for st in params["steps"]:
            if st[0] == "skill":
                fns.append(_stage_fn(st, reg))
            elif st[0] == "op":
                fns.append(_build_op(st[1]["op"], st[1]["params"], reg))
            else:
                raise SkillBirthError("unknown sequence step %r" % st[0])

        def f_seq(v: Any) -> Any:
            for fn in fns:
                v = fn(v)
            return v
        return f_seq
    raise SkillBirthError("unknown operator %r" % op)


def materialize(spec: CandidateSpec, registry: Optional[Mapping[str, CandidateSpec]] = None) -> Callable[[Any], Any]:
    """PUBLIC: rebuild the executable program from a JSON-able spec (never pickled). `registry`
    maps born-skill names to their specs; a spec whose deps are unresolved fails loud."""
    if spec.op not in GENERIC_OPERATORS:
        raise SkillBirthError("operator %r is outside GENERIC_OPERATORS" % spec.op)
    return _build_op(spec.op, spec.params, dict(registry or {}))


def _run(prog: Callable[[Any], Any], x: Any) -> Tuple[Any, ...]:
    """Execute guarded: ('ok', value) | _REFUSE (controlled) | _CRASH (eliminates)."""
    try:
        return ("ok", prog(x))
    except CandidateRefusal:
        return _REFUSE
    except SkillBirthError:
        raise
    except Exception:
        return _CRASH


def _beh(r: Tuple[Any, ...]) -> Tuple[str, ...]:
    """Behavior element: canonical outcome of one execution (refusals are first-class)."""
    return ("ok", canon(r[1])) if r[0] == "ok" else r  # type: ignore[return-value]


def _check(checker: Callable[[Any, Any], bool], x: Any, v: Any) -> bool:
    """Call the family oracle. An oracle crash on a candidate's value counts as NOT approved
    (a candidate may never profit from breaking the oracle); deterministic either way."""
    try:
        return bool(checker(x, v))
    except Exception:
        return False


def _check_adv(checker: Callable[[Any, Any], bool], x: Any, v: Any) -> str:
    """Adversarial-stage oracle call -> 'ok' | 'wrong' | 'not_adjudicable'.
    Integration fix (2026-07-05): engine-GENERATED mutations can fall outside the oracle's
    own input domain (e.g. a record the mutator stripped a field from) — the oracle then
    crashes on x no matter what the candidate answered. Eliminating the candidate for the
    ORACLE's crash is a false 'confidently wrong' (Law 1: elimination must mean the oracle
    judged the answer wrong). Discrimination: if the oracle also crashes on a sentinel value
    for the same x, the INPUT is outside its domain -> not adjudicable; if the sentinel is
    judged fine, the candidate's VALUE broke the oracle -> wrong (original protection kept)."""
    try:
        return "ok" if bool(checker(x, v)) else "wrong"
    except Exception:
        try:
            checker(x, None)
        except Exception:
            return "not_adjudicable"
        return "wrong"


def _spec_key(s: CandidateSpec) -> str:
    return canon({"deps": list(s.deps), "op": s.op, "params": s.params})


def _render(s: CandidateSpec) -> str:
    """Compact human-readable rendering for refusal messages (big induced maps abbreviated)."""
    p = dict(s.params)
    if "tbl" in p:
        p["tbl"] = "<induced:%d entries>" % len(s.params["tbl"])
    return canon({"deps": list(s.deps), "op": s.op, "params": p})

# ─────────────────────────────────────────────────────────────────────────────────────────
# Candidate enumeration — generic operators instantiated over OBSERVED shape (params are
# data-driven, never a topic branch) ++ compositions over the born-skill registry.
# ─────────────────────────────────────────────────────────────────────────────────────────
def _induce_tbl(chain: List[List[Any]], xs: List[Any], ys: List[Any],
                reg: Mapping[str, CandidateSpec]) -> Optional[Dict[str, Any]]:
    """Induce a key->answer map from TRAIN via the pre-chain. None if the chain fails on any
    worked input or two inputs collide on the same key with different answers."""
    try:
        fns = [_stage_fn(st, reg) for st in chain]
    except SkillBirthError:
        return None
    tbl: Dict[str, Any] = {}
    for x, y in zip(xs, ys):
        v = x
        try:
            for fn in fns:
                v = fn(v)
        except CandidateRefusal:
            return None
        except SkillBirthError:
            raise
        except Exception:
            return None
        c = canon(v)
        if c in tbl and not canonical_eq(tbl[c], y):
            return None
        tbl[c] = y
    return dict(sorted(tbl.items()))


def _enumerate(dg: Dict[str, Any], reg: Dict[str, CandidateSpec]) -> List[CandidateSpec]:
    """FULL enumeration (F001: len(result) is the honest `proposals` count), canonically sorted."""
    out: List[Tuple[str, Dict[str, Any], Tuple[str, ...]]] = []
    ik, xs, ys = dg["in_kind"], dg["xs"], dg["ys"]
    born = sorted(reg)
    sels: List[List[Any]] = []
    if ik == "txt":
        for s in _SCHEMES:
            out.append(("normalize", {"scheme": s}, ()))
        for d in dg["delims"]:
            for i in dg["split_idx"][d]:
                sels.append(["split", d, i])
    elif ik == "sq":
        for s in _SEQ_SCHEMES:
            out.append(("normalize", {"scheme": s}, ()))
        for s in _SCHEMES:
            out.append(("map", {"scheme": s}, ()))
        for i in dg["idxs"]:
            sels.append(["idx", i])
        for a in _AGGS:
            out.append(("reduce", {"agg": a}, ()))
        preds: List[List[Any]] = [["nonnull"], ["truthy"]]
        for o in ("ge", "le"):
            for t in dg["elem_nums"]:
                preds.append([o, t])
        for p in preds:
            out.append(("filter", {"pred": p}, ()))
    elif ik == "rec":
        for k in dg["keys"]:
            sels.append(["key", k])
        for k1 in dg["num_keys"]:
            for k2 in dg["num_keys"]:
                if k1 == k2:
                    continue
                for r in sorted(_RELS):
                    out.append(("compare", {"a": ["key", k1], "b": ["key", k2], "rel": r}, ()))
    for sel in sels:
        out.append(("extract", {"sel": sel}, ()))
    # fieldmap — record -> record of the same arity: ONE observed key's str value through a
    # normalize scheme, identity on the rest (the generic record-output constructor; keys are
    # observed data, schemes are the fixed generic set — no topic word anywhere)
    if ik == "rec" and all(isinstance(y, dict) for y in ys):
        str_keys = [k for k in dg["keys"] if all(isinstance(x.get(k), str) for x in xs)]
        for k in str_keys:
            for s in _SCHEMES:
                out.append(("fieldmap", {"key": k, "scheme": s}, ()))
    # classify — exactly-two-class rules, thresholds/membership drawn from observed values
    ans = dg["answers"]
    if len(ans) == 2:
        a0, a1 = ans
        srcs = [["self"]] if ik in ("sc", "txt") else [["key", k] for k in dg["keys"]]
        for src in srcs:
            get_vals = [x if src[0] == "self" else x[src[1]] for x in xs]
            if all(_is_num(u) for u in get_vals):
                # thresholds: every observed value PLUS boundary-adjacent candidates between
                # adjacent observed values (upper neighbor + integral midpoint). Data-derived
                # and topic-blind; without these, a true boundary that the engine's own split
                # placed in the hold-out is never proposable (integration fix, 2026-07-05).
                cand_t = set(get_vals)
                sv = sorted(set(get_vals))
                for lo, hi in zip(sv, sv[1:]):
                    mid = (lo + hi) / 2
                    if isinstance(lo, int) and isinstance(hi, int) and (lo + hi) % 2 == 0:
                        mid = (lo + hi) // 2
                    cand_t.add(mid)
                    cand_t.add(hi)
                for t in sorted(cand_t):
                    out.append(("classify", {"src": src, "rule": ["ge", t, a0, a1]}, ()))
                    out.append(("classify", {"src": src, "rule": ["ge", t, a1, a0]}, ()))
            m0 = sorted({canon(u) for u, y in zip(get_vals, ys) if canonical_eq(y, a0)})
            m1 = sorted({canon(u) for u, y in zip(get_vals, ys) if canonical_eq(y, a1)})
            out.append(("classify", {"src": src, "rule": ["in", m0, a0, a1]}, ()))
            out.append(("classify", {"src": src, "rule": ["in", m1, a1, a0]}, ()))
    # lookup — induced map, key optionally through ONE selector / ONE scheme / a born skill /
    # selector-then-born-skill (the compounding seam: reachable ONLY once the prereq exists)
    chains: List[List[List[Any]]] = [[]]
    chains += [[sel] for sel in sels]
    chains += [[["norm", s]] for s in _SCHEMES]
    chains += [[["skill", n]] for n in born]
    chains += [[sel, ["skill", n]] for sel in sels for n in born]
    for ch in chains:
        tbl = _induce_tbl(ch, xs, ys, reg)
        if tbl is None:
            continue
        deps = tuple(sorted({st[1] for st in ch if st[0] == "skill"}))
        out.append(("lookup", {"pre": ch, "tbl": tbl}, deps))
    # compose / sequence — born skills as ATOMS (empty until adoptions exist)
    for g in born:
        for h in born:
            out.append(("compose", {"g": g, "h": h}, tuple(sorted({g, h}))))
    for n in born:
        for s in _SCHEMES:
            nrm = {"op": "normalize", "params": {"scheme": s}}
            out.append(("sequence", {"steps": [["skill", n], ["op", nrm]]}, (n,)))
            out.append(("sequence", {"steps": [["op", nrm], ["skill", n]]}, (n,)))
        for a in _AGGS:
            out.append(("sequence", {"steps": [["skill", n], ["op", {"op": "reduce", "params": {"agg": a}}]]}, (n,)))
    specs = [CandidateSpec(op=o, params=p, deps=d) for o, p, d in out]
    specs.sort(key=_spec_key)
    return specs

# ─────────────────────────────────────────────────────────────────────────────────────────
# The engine — one universal decision path over every family (steps 2-9 + compounding).
# ─────────────────────────────────────────────────────────────────────────────────────────
class SkillBirthEngine(_EngineAPI):
    """See interfaces.SkillBirthEngine for the frozen contract."""

    def __init__(self) -> None:
        self._registry: Dict[str, BornSkill] = {}
        self._specs: Dict[str, CandidateSpec] = {}
        self._ledger = AccountingLedger()
        self._misses: List[str] = []

    def registry_names(self) -> List[str]:
        return sorted(self._registry)

    def detect_unknown(self, family: TaskFamily) -> bool:
        dg = _digest(family)
        for n in sorted(self._specs):
            prog = materialize(self._specs[n], self._specs)
            hit = True
            for x, y in family.examples:
                r = _run(prog, x)
                if r[0] != "ok" or not canonical_eq(r[1], y):
                    hit = False
                    break
            if hit:
                return False
        self._misses.append("%s->%s" % (dg["in_shape"], dg["out_shape"]))
        return True

    def name_skill(self, family: TaskFamily) -> SkillHypothesis:
        family = engine_split(family)
        dg = _digest(family)
        pairs = sorted(([x, y] for x, y in family.examples), key=canon)
        h8 = hashlib.sha256(canon(pairs).encode("utf-8")).hexdigest()[:8]

        def tag(s: str) -> str:
            return "".join(ch for ch in s if ch.isalnum())

        return SkillHypothesis(
            skill_name="skill_%s_to_%s_%s" % (tag(dg["in_shape"]), tag(dg["out_shape"]), h8),
            in_shape=dg["in_shape"], out_shape=dg["out_shape"], n_train=len(family.examples),
            derived_from="shape tags + sha256(canonical sorted example pairs)[:8]; no topic word")

    def write_standard(self, hypothesis: SkillHypothesis, family: TaskFamily) -> EmpiricalStandard:
        family = engine_split(family)
        dg = _digest(family)
        n_adv = len(generate_adversarial(dg["xs"], list(family.adversarial_seeds)))
        n_h, n_neg = len(family.holdout), len(family.negatives)
        n_pool = hypothesis.n_train + n_h
        plain = (
            "Skill %s: from a pool of %d worked examples, the engine split the pool itself "
            "(canonical-hash order) into %d training and %d held-out cases; each maps a %s to "
            "a %s. A candidate is correct iff it (a) reproduces every training answer, (b) "
            "reproduces the family oracle on the %d engine-held-out inputs, (c) never returns "
            "a wrong confident answer on %d adversarial inputs the engine generated from the "
            "observed input shape (mutation/boundary/garbage) plus supplied seeds, (d) matches "
            "none of the %d negative rules, and (e) is the unique surviving behavior on a "
            "discriminating hold-out. Adopt iff exactly one candidate meets (a)-(e); otherwise "
            "refuse and say what is missing."
            % (hypothesis.skill_name, n_pool, hypothesis.n_train, n_h, hypothesis.in_shape,
               hypothesis.out_shape, n_h, n_adv, n_neg))
        return EmpiricalStandard(
            skill_name=hypothesis.skill_name, plain_english=plain,
            success_rule=("adopt iff exactly one distinct candidate behavior survives train + "
                          "hold-out + adversarial + negative + contradiction gates on a "
                          "discriminating hold-out; otherwise refuse"),
            n_train=hypothesis.n_train, n_holdout=n_h, n_adversarial=n_adv, n_negative=n_neg,
            contradiction_check=("the engine injects one poisoned copy of a worked example "
                                 "(answer swapped to a different observed answer) and re-runs "
                                 "train consistency; the loop must refuse, never adopt, under "
                                 "the engine-injected inconsistency"),
            refusal_conditions=(
                "reach_limit: no enumerated candidate fits the worked examples and survives validation",
                "underdetermined: >1 distinct surviving behavior; the exact separating input is named",
                "indistinguishable: >1 surviving rule and no available input separates them, or the hold-out does not discriminate the candidate set",
                "contradiction: two worked examples give the same input different answers",
            ))

    def enumerate_candidates(self, family: TaskFamily) -> List[CandidateSpec]:
        return _enumerate(_digest(engine_split(family)), self._specs)

    def ledger(self) -> AccountingLedger:
        return self._ledger

    # ── internal helpers (all topic-blind) ──────────────────────────────────────────────
    def _approved(self, family: TaskFamily, x: Any, prog: Callable[[Any], Any]) -> bool:
        r = _run(prog, x)
        return r[0] == "ok" and _check(family.checker, x, r[1])

    @staticmethod
    def _example_contradiction(family: TaskFamily) -> Optional[RefusalRequest]:
        seen: Dict[str, Any] = {}
        for x, y in family.examples:
            cx = canon(x)
            if cx in seen and not canonical_eq(seen[cx], y):
                vals = tuple(sorted((seen[cx], y), key=canon))
                return RefusalRequest(
                    kind=RefusalKind.CONTRADICTION,
                    plain_english=("Your worked examples disagree: the input %s is recorded "
                                   "with two different answers (%s). Remove or correct one of "
                                   "them and I can learn." % (canon(x), canon(list(vals)))),
                    missing_input=x, disagreeing_values=vals)
            seen[cx] = y
        return None

    @staticmethod
    def _poison(family: TaskFamily, dg: Dict[str, Any]) -> Tuple[Any, Any]:
        """Poisoned pair for the soundness check: first example's input with a DIFFERENT
        observed answer (examples first, then held-out answers; deterministic fallback)."""
        x0, y0 = family.examples[0]
        for y in dg["answers"]:
            if not canonical_eq(y, y0):
                return x0, y
        for y in sorted((yy for _, yy in family.holdout), key=canon):
            if not canonical_eq(y, y0):
                return x0, y
        return x0, (None if not canonical_eq(None, y0) else 0)

    def _underdetermined(self, family: TaskFamily,
                         groups: Dict[Tuple[Any, ...], List[Tuple[CandidateSpec, Callable[[Any], Any]]]],
                         n_holdout: int) -> RefusalRequest:
        """>1 distinct surviving behavior: name the single probe input that separates them
        (S047 pick_discriminating_unit), or refuse INDISTINGUISHABLE if none does."""
        keys = sorted(groups)
        reps = [min((sp for sp, _ in groups[k]), key=_spec_key) for k in keys]
        rendered = tuple(sorted(_render(sp) for sp in reps))
        pool = list(family.probe_pool)
        best_j, best_spread = -1, 1
        for j in range(len(pool)):
            spread = len({k[n_holdout + j] for k in keys})
            if spread > best_spread:
                best_j, best_spread = j, spread
        if best_j < 0:
            return RefusalRequest(
                kind=RefusalKind.INDISTINGUISHABLE,
                plain_english=("I can't tell yet — %d surviving rules give the SAME answer on "
                               "every available input: %s. I can't separate them without "
                               "different data." % (len(keys), "; ".join(rendered))),
                candidates=rendered)
        u_star = pool[best_j]
        raw: List[Any] = []
        seen: set = set()
        for k in keys:
            r = _run(groups[k][0][1], u_star)
            v = r[1] if r[0] == "ok" else "<refuses>"
            c = canon(v)
            if c not in seen:
                seen.add(c)
                raw.append(v)
        vals = tuple(sorted(raw, key=canon))
        return RefusalRequest(
            kind=RefusalKind.UNDERDETERMINED,
            plain_english=("I can't tell yet — your examples fit these %d rules: %s. To decide, "
                           "give me the answer for %s; those rules disagree there (they would "
                           "give %s)." % (len(keys), "; ".join(rendered), canon(u_star),
                                          canon(list(vals)))),
            candidates=rendered, missing_input=u_star, disagreeing_values=vals)

    # ── steps 2-9 for one family ──────────────────────────────────────────────────────────
    def birth(self, family: TaskFamily, label: str) -> BirthOutcome:
        family = engine_split(family)                   # ENGINE authors the train/hold-out split
        detected = self.detect_unknown(family)
        hyp = self.name_skill(family)
        std = self.write_standard(hyp, family)          # G3: standard BEFORE any decision
        dg = _digest(family)
        specs = self.enumerate_candidates(family)
        progs = [materialize(s, self._specs) for s in specs]
        n = dict(proposals=len(specs), train_checks=0, holdout_checks=0, adversarial_checks=0,
                 negative_checks=0, contradiction_checks=0, survivors=0, adoptions=0,
                 refusals=0, reuse_events=0)
        adopted: Optional[BornSkill] = None
        refusal = self._example_contradiction(family)

        if refusal is None:
            # 1) train fit — FULL enumeration, one counted check per candidate (F001/G5:
            #    candidates are proposed and fitted on the worked examples ONLY)
            fit: List[Tuple[CandidateSpec, Callable[[Any], Any]]] = []
            for sp, pr in zip(specs, progs):
                n["train_checks"] += 1
                if all(self._approved(family, x, pr) for x, _ in family.examples):
                    fit.append((sp, pr))
            # 2) negative rules — drop any candidate whose probe-pool behavior equals a
            #    known-wrong rule's behavior (kills coincidental train-fit)
            pool = list(family.probe_pool)
            if pool and family.negatives:
                negbeh = [tuple(_beh(_run(ng.program, u)) for u in pool)
                          for ng in family.negatives]
                kept = []
                for sp, pr in fit:
                    beh = tuple(_beh(_run(pr, u)) for u in pool)
                    bad = False
                    for nb in negbeh:
                        n["negative_checks"] += 1
                        if beh == nb:
                            bad = True
                            break
                    if not bad:
                        kept.append((sp, pr))
                fit = kept
            # 3) held-out + engine-generated adversarial (+ family seeds): approved or refuse;
            #    a confident wrong answer (or a crash) eliminates the candidate
            adv = generate_adversarial(dg["xs"], list(family.adversarial_seeds))
            surv: List[Tuple[CandidateSpec, Callable[[Any], Any]]] = []
            for sp, pr in fit:
                ok = True
                for x, _ in family.holdout:
                    n["holdout_checks"] += 1
                    if not self._approved(family, x, pr):
                        ok = False
                        break
                if ok:
                    for u in adv:
                        n["adversarial_checks"] += 1
                        r = _run(pr, u)
                        if r == _REFUSE:
                            continue
                        if r == _CRASH or _check_adv(family.checker, u, r[1]) == "wrong":
                            ok = False
                            break
                if ok:
                    surv.append((sp, pr))
            n["survivors"] = len(surv)
            # 4) injected-contradiction soundness: on examples + one poisoned copy the loop
            #    must NOT uniquely determine a behavior (recorded, never adopted from)
            px, py = self._poison(family, dg)
            poisoned = list(family.examples) + [(px, py)]
            pfit: List[Callable[[Any], Any]] = []
            for sp, pr in zip(specs, progs):
                n["contradiction_checks"] += 1
                good = True
                for x, y in poisoned:
                    r = _run(pr, x)
                    if r[0] != "ok" or not canonical_eq(r[1], y):
                        good = False
                        break
                if good:
                    pfit.append(pr)
            if pfit:
                dpool0 = [x for x, _ in family.holdout] + pool
                pbeh = {tuple(_beh(_run(pr, u)) for u in dpool0) for pr in pfit}
                if len(pbeh) == 1:
                    raise SkillBirthError(
                        "soundness violation: a self-contradictory example set still uniquely "
                        "determined 1 behavior (%d candidates)" % len(pfit))
            # 5) unique determination on the discriminating pool (S035/S036 sealed rule)
            dpool = [x for x, _ in family.holdout] + pool
            groups: Dict[Tuple[Any, ...], List[Tuple[CandidateSpec, Callable[[Any], Any]]]] = {}
            for sp, pr in surv:
                key = tuple(_beh(_run(pr, u)) for u in dpool)
                groups.setdefault(key, []).append((sp, pr))
            if not surv:
                refusal = RefusalRequest(
                    kind=RefusalKind.REACH_LIMIT,
                    plain_english=("No enumerated candidate (%d proposed: generic operators over "
                                   "the observed %s -> %s shape plus compositions over %d born "
                                   "skills) fits all %d worked examples and survives hold-out/"
                                   "adversarial/negative validation. This needs an operation "
                                   "outside my current algebra — supply a prerequisite skill "
                                   "that provides it, or corrected examples."
                                   % (len(specs), hyp.in_shape, hyp.out_shape,
                                      len(self._specs), len(family.examples))))
            elif len(groups) > 1:
                refusal = self._underdetermined(family, groups, len(family.holdout))
            else:
                hx = [x for x, _ in family.holdout]
                sigs = {tuple(_beh(_run(pr, u)) for u in hx) for pr in progs}
                if len(sigs) < 2:
                    refusal = RefusalRequest(
                        kind=RefusalKind.INDISTINGUISHABLE,
                        plain_english=("The hold-out is uninformative: all %d enumerated "
                                       "candidates behave identically on it, so adoption cannot "
                                       "be discriminating. Supply held-out inputs on which "
                                       "candidate rules differ." % len(specs)))
                else:
                    members = groups[sorted(groups)[0]]
                    rep = min((sp for sp, _ in members), key=_spec_key)
                    name = hyp.skill_name
                    if name in self._specs:
                        raise SkillBirthError("duplicate skill name %r" % name)
                    adopted = BornSkill(
                        name=name, spec=rep, deps=rep.deps,
                        in_shape=hyp.in_shape, out_shape=hyp.out_shape,
                        provenance=("uniquely determined: 1 distinct behavior among %d survivors "
                                    "of %d proposals; fit on %d worked examples only; passed %d "
                                    "held-out, %d adversarial and %d negative-rule checks; "
                                    "injected-contradiction check refused as required; deps=%s"
                                    % (len(surv), len(specs), len(family.examples),
                                       n["holdout_checks"], n["adversarial_checks"],
                                       n["negative_checks"], canon(list(rep.deps)))))
                    self._registry[name] = adopted   # step 10: born skill becomes an atom
                    self._specs[name] = rep

        # F001 ledger + outcome
        if adopted is not None:
            n["adoptions"], n["reuse_events"] = 1, len(adopted.deps)
        else:
            if refusal is None:
                raise SkillBirthError("decision fell through with neither adoption nor refusal")
            n["refusals"] = 1
        counts = BirthCounts(**n)
        self._ledger.per_family.append(counts)
        if adopted is not None:
            self._ledger.skills_adopted.append(hyp.skill_name)
            self._ledger.families_with_adoption.append(label)
            for dp in sorted(set(adopted.deps)):
                self._ledger.reuse_edges.append((dp, label))
        else:
            self._ledger.refusals.append(label)
        return BirthOutcome(
            family=label, status=BirthStatus.ADOPTED if adopted else BirthStatus.REFUSED,
            hypothesis=hyp, standard=std, detected_unknown=detected, adopted=adopted,
            refusal=refusal, reused_deps=(adopted.deps if adopted else ()), counts=counts)

# ─────────────────────────────────────────────────────────────────────────────────────────
# Standalone smoke — synthetic mini-families, inline only (never part of the real run).
# ─────────────────────────────────────────────────────────────────────────────────────────
def _smoke() -> None:
    import dataclasses
    from interfaces import ExpectTag, NegativeRef

    def chk1(x: Any, a: Any) -> bool:
        return isinstance(x, str) and a == x.casefold().strip()

    f1 = TaskFamily(
        name="f1", expect=ExpectTag.ADOPT,
        examples=[(" Ab", "ab"), ("cD ", "cd"), (" eF ", "ef")],
        holdout=[("Gh", "gh"), (" iJ", "ij")],
        adversarial_seeds=["  Q  "],
        negatives=[NegativeRef("cf-only", lambda x: x.casefold())],
        probe_pool=[" Kl ", "Mn"], checker=chk1)

    def chk2(x: Any, a: Any) -> bool:
        return isinstance(x, (list, tuple)) and all(_is_num(e) for e in x) and a == sum(x)

    f2 = TaskFamily(
        name="f2", expect=ExpectTag.ADOPT,
        examples=[([1, 2], 3), ([2, 3], 5), ([4], 4)],
        holdout=[([5, 6], 11), ([7], 7)],
        adversarial_seeds=[], negatives=[], probe_pool=[[8, 9]], checker=chk2)

    truth = {"ab": "u", "cd": "v", "ef": "w"}

    def chk3(x: Any, a: Any) -> bool:
        return (isinstance(x, dict) and isinstance(x.get("k"), str)
                and truth.get(x["k"].casefold().strip()) == a)

    f3 = TaskFamily(
        name="f3", expect=ExpectTag.REUSE,
        examples=[({"k": " Ab", "z": 1}, "u"), ({"k": "cD ", "z": 2}, "v"),
                  ({"k": " eF ", "z": 3}, "w")],
        holdout=[({"k": "AB ", "z": 9}, "u"), ({"k": " Cd", "z": 8}, "v")],
        adversarial_seeds=[], negatives=[], probe_pool=[], checker=chk3)

    def chk4(x: Any, a: Any) -> bool:
        if not isinstance(x, str):
            return False
        ps = x.split(" ")
        return a in (ps[1], ps[-1]) if len(ps) > 1 else a == ps[0]

    f4 = TaskFamily(
        name="f4", expect=ExpectTag.REFUSE,
        examples=[("a b", "b"), ("c d", "d"), ("e f", "f")],
        holdout=[("g h", "h")],
        adversarial_seeds=[],
        negatives=[NegativeRef("head", lambda x: x.split(" ")[0])],
        probe_pool=["p q r"], checker=chk4)

    def drive() -> Tuple[List[BirthOutcome], SkillBirthEngine]:
        e = SkillBirthEngine()
        outs = [e.birth(f, lb) for f, lb in ((f1, "L1"), (f2, "L2"), (f3, "L3"), (f4, "L4"))]
        return outs, e

    outs, eng = drive()
    o1, o2, o3, o4 = outs
    assert o1.status is BirthStatus.ADOPTED and o1.adopted.spec.op == "normalize" \
        and o1.adopted.spec.params["scheme"] == "casefold+strip", o1
    assert o2.status is BirthStatus.ADOPTED and o2.adopted.spec.op == "reduce" \
        and o2.adopted.spec.params["agg"] == "sum", o2
    s1 = o1.hypothesis.skill_name
    assert o3.status is BirthStatus.ADOPTED and o3.reused_deps == (s1,) \
        and o3.counts.reuse_events == 1, o3
    assert (s1, "L3") in eng.ledger().reuse_edges
    assert o4.status is BirthStatus.REFUSED and o4.refusal.kind is RefusalKind.UNDERDETERMINED \
        and o4.refusal.missing_input == "p q r" \
        and o4.refusal.disagreeing_values == ("q", "r"), o4.refusal
    assert all(o.detected_unknown for o in outs)
    for o in outs:
        assert o.counts.proposals >= o.counts.survivors > -1
    # reach-before: the reuse family REFUSES (reach limit) on a fresh, empty engine (G7)
    fresh = SkillBirthEngine().birth(f3, "L3-fresh")
    assert fresh.status is BirthStatus.REFUSED \
        and fresh.refusal.kind is RefusalKind.REACH_LIMIT, fresh.refusal
    # materialize: independent re-execution of the reuse survivor from specs alone (G8 seam)
    reg = {o.hypothesis.skill_name: o.adopted.spec for o in outs if o.adopted is not None}
    prog = materialize(o3.adopted.spec, reg)
    for x, y in f3.holdout:
        assert canonical_eq(prog(x), y)
    # determinism: a second fresh full run is canonically identical (G10)
    outs2, _ = drive()
    a = canon([dataclasses.asdict(o) for o in outs])
    b = canon([dataclasses.asdict(o) for o in outs2])
    if a != b:
        raise NonDeterminismError("smoke double-run diverged")
    t = eng.ledger().totals()
    print("smoke OK: adopt/adopt/reuse-adopt/underdetermined-refuse + reach-before + "
          "materialize + double-run identity")
    print("totals: proposals=%d train=%d holdout=%d adversarial=%d negative=%d "
          "contradiction=%d survivors=%d adoptions=%d refusals=%d reuse=%d"
          % (t.proposals, t.train_checks, t.holdout_checks, t.adversarial_checks,
             t.negative_checks, t.contradiction_checks, t.survivors, t.adoptions,
             t.refusals, t.reuse_events))


if __name__ == "__main__":
    _smoke()
