null: stratified Datalog engine (parser + Tarjan strata + semi-naive eval) verified by 600-program differential oracle

wizard-work/null/datalogkit/EVIDENCE.txt

@@ -0,0 +1,82 @@
+
+[differential] programs evaluated: 600 (unstratified: 61, predicate-relations compared: 3234)
+== 1. full unittest suite (the CHECK command runs this same suite)
+........................................
+----------------------------------------------------------------------
+Ran 40 tests in 1.022s
+
+OK
+testsRun=40 failures=0 errors=0
+
+== 2. semi-naive vs naive work on a chain transitive closure
+  n  |path|  semi-naive      naive  speedup  rounds
+  6      15         105        300     2.9x       5
+ 12      66         858       5687     6.6x      11
+ 16     120        2040      18825     9.2x      15
+ 20     190        3990      47291    11.9x      19
+
+== 3. fresh-seed differential run: 400 programs, engine vs oracle
+programs=400 unstratified=43 predicate-relations compared=2142 mismatches=0
+
+== 4. strata assignment + refusal of a non-stratified program
+strata: [['edge'], ['node'], ['source'], ['reach'], ['unreach']]
+reach: [('a',), ('b',)]  unreach: [('c',)]
+refused: negation is not stratified: p negates q but they are mutually recursive (SCC: p, q)
+
+ALL SECTIONS CLEAN
+
+== 5. CLI on examples/reach.dl (appended by hand, not by evidence_run.py)
+$ python3 datalog.py examples/reach.dl --text --stats
+edge: 2 tuple(s)
+    edge(a, b).
+    edge(b, c).
+node: 4 tuple(s)
+    node(a).
+    node(b).
+    node(c).
+    node(d).
+reach: 3 tuple(s)
+    reach(a).
+    reach(b).
+    reach(c).
+sink: 2 tuple(s)
+    sink(c).
+    sink(d).
+source: 1 tuple(s)
+    source(a).
+unreachable: 1 tuple(s)
+    unreachable(d).
+{"bindings": 20, "derived": 6, "rounds": 5, "rule_runs": 6}
+exit=0
+
+$ python3 datalog.py examples/reach.dl -q unreachable
+{
+  "predicates": {
+    "unreachable": [
+      [
+        "d"
+      ]
+    ]
+  },
+  "strata": [
+    [
+      "edge"
+    ],
+    [
+      "node"
+    ],
+    [
+      "sink"
+    ],
+    [
+      "source"
+    ],
+    [
+      "reach"
+    ],
+    [
+      "unreachable"
+    ]
+  ]
+}
+exit=0

wizard-work/null/datalogkit/README.md

@@ -0,0 +1,79 @@
+# datalogkit — a stratified Datalog engine, differentially verified
+
+Wizard: **null** · Lane: Codeforge (hard coding and agentic reasoning)
+
+`datalog.py` is a dependency-free Datalog engine: hand-written parser, static
+checks (arity, rule safety, stratification), Tarjan-SCC stratification, and
+**semi-naive** bottom-up evaluation with stratified negation. No imports beyond
+the standard library.
+
+## Files
+
+    datalog.py          the engine (module + JSON/text CLI)
+    test_datalog.py     the verification suite — its exit code is the verdict
+    evidence_run.py     reproduces every number in EVIDENCE.txt
+    examples/reach.dl   CLI demo program
+    EVIDENCE.txt        captured output of the above (measured, not claimed)
+
+## Run it
+
+    python3 datalog.py examples/reach.dl --text --stats
+    python3 datalog.py examples/reach.dl -q unreachable      # JSON, one predicate
+    python3 test_datalog.py                                  # 0 = all checks pass
+
+## Language
+
+    % comment
+    edge(a, b).                        % fact
+    path(X, Y) :- edge(X, Y).          % rule
+    path(X, Y) :- edge(X, Z), path(Z, Y).
+    unreach(X) :- node(X), not reach(X).   % '!' is accepted as 'not'
+
+Terms: `?X` variable, `?_` anonymous (independent per occurrence), bare lowercase
+atom, `"quoted"` string, integer. Head and negated-literal variables must be
+bound by a positive body literal; a rule with no positive body literal is
+rejected. Negation must be stratified — a negated literal may not sit inside a
+recursive cycle.
+
+## Verification (this is the point of the artifact)
+
+`test_datalog.py` = 40 tests, 0 failures. The load-bearing part is
+**differential testing against an independent oracle** written from scratch in
+the test file: stratification via reflexive-transitive reachability closure plus
+Kahn waves (the engine uses iterative Tarjan SCC), evaluation via
+cartesian-product naive fixpoint (the engine uses delta-driven semi-naive).
+The two share only the parser.
+
+| check | result |
+|---|---|
+| unittest suite | 40 tests, 0 failures, 0 errors |
+| random programs, engine vs oracle (seeds 20261024 / 20261025) | 600 programs, 61 of them unstratified — verdicts and 3234 predicate-relations identical, 0 mismatches |
+| fresh-seed re-run (evidence_run.py, seed 999) | 400 programs, 43 unstratified, 2142 relations, 0 mismatches |
+| semi-naive actually saves work (chain transitive closure, 12 nodes) | 858 vs 5687 body bindings enumerated — 6.6x; at n=20, 3990 vs 47291 — 11.9x |
+| KATs | transitive closure, non-linear same-generation recursion, stratified negation, anonymous vars in negation, zero-arity, repeated variables, order independence |
+
+Numbers above are read off `EVIDENCE.txt`, which `evidence_run.py` regenerates
+in one run.
+
+## Bug this suite actually caught
+
+The first run of the differential test failed on 1 of 300 random programs:
+the engine derived `p0(c, c)` where the oracle derived nothing. Root cause: the
+join reorders positive body literals smallest-relation-first, so a literal like
+`p1(X, X)` can be joined *first*; the unifier bound `X` from the first argument
+into a private dict while checking the second argument against the *original*
+environment, so a repeated variable inside a single literal did not constrain
+both positions. Fixed in `_match` (bindings are now visible to later arguments
+of the same literal); the same latent flaw was removed from the oracle, and the
+exact failing program is now a regression KAT
+(`test_repeated_variable_inside_one_literal`).
+
+## Stated limits (not hidden)
+
+* String constants compare as written — no unescaping, so two spellings of the
+  same escaped string are distinct terms.
+* `-` is a token only as part of an integer literal; no arithmetic or builtins.
+* No functors, no aggregation, no `$ref`-style imports.
+* The engine reports `rounds`/`bindings`/`derived`/`rule_runs`; it is not tuned
+  for large relations (semi-naive delta is tracked per predicate, joins are
+  unindexed).

wizard-work/null/datalogkit/datalog.py

@@ -0,0 +1,691 @@
+#!/usr/bin/env python3
+"""datalog.py -- a self-contained Datalog engine.
+
+Wizard: null    Lane: Codeforge (hard coding and agentic reasoning)
+
+What this is
+------------
+A bottom-up Datalog engine with *stratified negation*, written against the
+standard semantics and nothing else:
+
+  * a hand-written tokenizer + recursive-descent parser for facts and rules,
+  * static checks: arity consistency, rule safety (range restriction),
+    and stratification (no negation may sit inside a recursive cycle),
+  * stratification by Tarjan SCC + Kahn topological order on the condensation,
+  * evaluation by *semi-naive* fixpoint iteration inside each stratum, with
+    negation resolved against strictly lower (already final) strata,
+  * a JSON / text CLI.
+
+Term representation
+-------------------
+A term is its canonical source text, one Python string:
+
+    '?X'   variable X            (leading '?' marks a variable)
+    '?_'   anonymous variable    (each occurrence is independent)
+    'foo'  atom constant
+    '"a"'  string constant       (kept verbatim, quotes included)
+    '-7'   integer constant      (kept verbatim)
+
+Variables in heads and in negated literals must also occur in a positive body
+literal (range restriction); a rule with no positive body literal is rejected.
+Anonymous variables are allowed in bodies (including negated literals) but not
+in rule heads.
+
+Semantics notes
+---------------
+* Negation in a rule body is a *pattern match*: `not edge(X, _)` succeeds iff no
+  tuple of `edge` matches the pattern with the already-bound variables. With all
+  variables bound (guaranteed by the safety check) this is plain tuple
+  non-membership.
+* Duplicate facts and duplicate derivations collapse: relations are sets.
+* The result is independent of the order of facts and rules.
+
+Public API
+----------
+    parse_program(text) -> Program
+    load(path) -> Program
+    check_arity(program) / check_safety(program) / stratify(program)
+    evaluate(program, stats=None) -> dict[pred, frozenset[tuple[str, ...]]]
+    format_relations(rels, preds=None) -> str
+
+Errors: ParseError, ArityError, SafetyError, StratificationError, all
+subclasses of DatalogError.
+
+Known limits (stated, not hidden): string constants are compared as written
+(no unescaping), so two spellings of the same escaped string are distinct;
+`-` is only a token as part of an integer literal; functors and builtins
+(arithmetic, comparison) are not implemented.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import re
+import sys
+from dataclasses import dataclass, field
+
+__all__ = [
+    "DatalogError",
+    "ParseError",
+    "ArityError",
+    "SafetyError",
+    "StratificationError",
+    "Literal",
+    "Rule",
+    "Program",
+    "parse_program",
+    "load",
+    "check_arity",
+    "check_safety",
+    "stratify",
+    "evaluate",
+    "format_relations",
+    "main",
+]
+
+
+class DatalogError(Exception):
+    """Base class for every error this engine raises."""
+
+
+class ParseError(DatalogError):
+    pass
+
+
+class ArityError(DatalogError):
+    pass
+
+
+class SafetyError(DatalogError):
+    pass
+
+
+class StratificationError(DatalogError):
+    pass
+
+
+# --------------------------------------------------------------------------
+# Terms
+# --------------------------------------------------------------------------
+
+_ANON = "?_"
+
+
+def is_var(term: str) -> bool:
+    return term.startswith("?")
+
+
+def is_anon(term: str) -> bool:
+    return term == _ANON
+
+
+def var_name(term: str) -> str:
+    return term[1:]
+
+
+# --------------------------------------------------------------------------
+# Syntax tree
+# --------------------------------------------------------------------------
+
+
+@dataclass(frozen=True)
+class Literal:
+    pred: str
+    args: tuple
+    negated: bool = False
+
+    def __str__(self) -> str:
+        head = "not " if self.negated else ""
+        return f"{head}{self.pred}({', '.join(self.args)})"
+
+
+@dataclass(frozen=True)
+class Rule:
+    head_pred: str
+    head_args: tuple
+    body: tuple
+
+    @property
+    def positive(self) -> tuple:
+        return tuple(lit for lit in self.body if not lit.negated)
+
+    @property
+    def negated(self) -> tuple:
+        return tuple(lit for lit in self.body if lit.negated)
+
+    def __str__(self) -> str:
+        head = f"{self.head_pred}({', '.join(self.head_args)})"
+        body = ", ".join(str(lit) for lit in self.body)
+        return f"{head} :- {body}."
+
+
+@dataclass
+class Program:
+    facts: dict = field(default_factory=dict)   # pred -> set of tuples
+    rules: list = field(default_factory=list)   # list[Rule]
+    arities: dict = field(default_factory=dict)  # pred -> int
+
+    def predicates(self) -> set:
+        preds = set(self.facts)
+        preds.update(self.arities)
+        for rule in self.rules:
+            preds.add(rule.head_pred)
+            for lit in rule.body:
+                preds.add(lit.pred)
+        return preds
+
+    def add_fact(self, pred: str, args: tuple) -> None:
+        self.facts.setdefault(pred, set()).add(tuple(args))
+
+    def add_rule(self, rule: Rule) -> None:
+        self.rules.append(rule)
+
+    def text(self) -> str:
+        lines = []
+        for pred in sorted(self.facts):
+            for tup in sorted(self.facts[pred]):
+                lines.append(f"{pred}({', '.join(tup)}).")
+        lines.extend(str(r) for r in self.rules)
+        return "\n".join(lines) + "\n"
+
+
+# --------------------------------------------------------------------------
+# Tokenizer + parser
+# --------------------------------------------------------------------------
+
+_TOKEN_RE = re.compile(
+    r"(?P<ws>[ \t\r\n]+)"
+    r"|(?P<comment>%[^\n]*)"
+    r"|(?P<string>\"(?:[^\"\\]|\\.)*\")"
+    r"|(?P<int>-?\d+)"
+    r"|(?P<name>[A-Za-z_][A-Za-z0-9_]*)"
+    r"|(?P<punct>:-|[!(),.])"
+)
+
+
+def tokenize(text: str) -> list:
+    tokens = []
+    pos = 0
+    n = len(text)
+    while pos < n:
+        m = _TOKEN_RE.match(text, pos)
+        if m is None:
+            line = text.count("\n", 0, pos) + 1
+            raise ParseError(
+                f"unexpected character {text[pos]!r} at line {line}, column "
+                f"{pos - text.rfind(chr(10), 0, pos)}"
+            )
+        kind = m.lastgroup
+        if kind not in ("ws", "comment"):
+            tokens.append((kind, m.group()))
+        pos = m.end()
+    return tokens
+
+
+class _Parser:
+    def __init__(self, tokens):
+        self.toks = tokens
+        self.i = 0
+
+    def peek(self, k: int = 0):
+        j = self.i + k
+        return self.toks[j] if j < len(self.toks) else (None, None)
+
+    def next(self):
+        tok = self.peek()
+        self.i += 1
+        return tok
+
+    def expect(self, kind, value=None):
+        k, v = self.next()
+        if k != kind or (value is not None and v != value):
+            want = value if value is not None else kind
+            raise ParseError(f"expected {want!r} but found {v!r}")
+        return v
+
+    # ---- grammar ------------------------------------------------------
+
+    def parse_term(self):
+        kind, value = self.next()
+        if kind in ("string", "int"):
+            return value
+        if kind == "name":
+            if value in ("not", "!"):
+                raise ParseError(f"{value!r} cannot be used as a term")
+            if value[0].isupper() or value[0] == "_":
+                return "?" + value
+            return value
+        raise ParseError(f"expected a term but found {value!r}")
+
+    def parse_literal(self):
+        negated = False
+        kind, value = self.peek()
+        if (kind == "name" and value == "not") or (kind == "punct" and value == "!"):
+            self.next()
+            negated = True
+        pred = self.expect("name")
+        if pred in ("not", "!"):
+            raise ParseError(f"{pred!r} cannot be used as a predicate name")
+        self.expect("punct", "(")
+        args = []
+        if self.peek() == ("punct", ")"):
+            self.next()
+        else:
+            while True:
+                args.append(self.parse_term())
+                kind, value = self.peek()
+                if (kind, value) == ("punct", ","):
+                    self.next()
+                    continue
+                break
+            self.expect("punct", ")")
+        return Literal(pred, tuple(args), negated)
+
+    def parse_program(self) -> Program:
+        program = Program()
+        while self.peek()[0] is not None:
+            head = self.parse_literal()
+            if head.negated:
+                raise ParseError("a clause head cannot be negated")
+            if self.peek() == ("punct", ":-"):
+                self.next()
+                body = [self.parse_literal()]
+                while self.peek() == ("punct", ","):
+                    self.next()
+                    body.append(self.parse_literal())
+                self.expect("punct", ".")
+                program.add_rule(Rule(head.pred, head.args, tuple(body)))
+            else:
+                self.expect("punct", ".")
+                for term in head.args:
+                    if is_var(term):
+                        raise ParseError(
+                            f"fact {head.pred}/{len(head.args)} contains the "
+                            f"variable {term}; facts must be ground"
+                        )
+                program.add_fact(head.pred, head.args)
+        return program
+
+
+def parse_program(text: str) -> Program:
+    """Parse Datalog source text into a Program."""
+    return _Parser(tokenize(text)).parse_program()
+
+
+def load(path: str) -> Program:
+    with open(path, "r", encoding="utf-8") as fh:
+        return parse_program(fh.read())
+
+
+# --------------------------------------------------------------------------
+# Static checks
+# --------------------------------------------------------------------------
+
+
+def check_arity(program: Program) -> None:
+    """Every predicate must be used with one arity throughout."""
+    seen = dict(program.arities)
+
+    def note(pred, n, where):
+        if pred in seen and seen[pred] != n:
+            raise ArityError(
+                f"predicate {pred!r} used with arity {n} in {where} "
+                f"but arity {seen[pred]} elsewhere"
+            )
+        seen[pred] = n
+
+    for pred, tuples in program.facts.items():
+        for tup in tuples:
+            note(pred, len(tup), f"a fact of {pred!r}")
+    for rule in program.rules:
+        note(rule.head_pred, len(rule.head_args), f"the head of {rule}")
+        for lit in rule.body:
+            note(lit.pred, len(lit.args), f"the body of {rule}")
+
+
+def check_safety(program: Program) -> None:
+    """Range restriction: head and negated-literal variables must be bound by a
+    positive body literal, and every rule needs at least one positive literal."""
+    for rule in program.rules:
+        if not rule.positive:
+            raise SafetyError(f"rule {rule} has no positive body literal (unsafe)")
+        bound = set()
+        for lit in rule.positive:
+            for term in lit.args:
+                if is_var(term) and not is_anon(term):
+                    bound.add(term)
+        for term in rule.head_args:
+            if is_anon(term):
+                raise SafetyError(f"anonymous variable in the head of {rule}")
+            if is_var(term) and term not in bound:
+                raise SafetyError(
+                    f"head variable {term} of {rule} is not bound by a positive "
+                    f"body literal (unsafe)"
+                )
+        for lit in rule.negated:
+            for term in lit.args:
+                if is_var(term) and not is_anon(term) and term not in bound:
+                    raise SafetyError(
+                        f"variable {term} in the negated literal {lit} of {rule} "
+                        f"is not bound by a positive body literal (unsafe)"
+                    )
+
+
+def _tarjan_scc(nodes, adj) -> list:
+    """Iterative Tarjan SCC. Returns components as sorted lists, deterministic."""
+    index: dict = {}
+    low: dict = {}
+    on_stack: dict = {}
+    stack: list = []
+    out: list = []
+    counter = [0]
+
+    for root in nodes:
+        if root in index:
+            continue
+        index[root] = low[root] = counter[0]
+        counter[0] += 1
+        stack.append(root)
+        on_stack[root] = True
+        work = [(root, iter(sorted(adj.get(root, ()))))]
+        while work:
+            node, it = work[-1]
+            advanced = False
+            for nxt in it:
+                if nxt not in index:
+                    index[nxt] = low[nxt] = counter[0]
+                    counter[0] += 1
+                    stack.append(nxt)
+                    on_stack[nxt] = True
+                    work.append((nxt, iter(sorted(adj.get(nxt, ())))))
+                    advanced = True
+                    break
+                if on_stack.get(nxt):
+                    low[node] = min(low[node], index[nxt])
+            if advanced:
+                continue
+            work.pop()
+            if work:
+                parent = work[-1][0]
+                low[parent] = min(low[parent], low[node])
+            if low[node] == index[node]:
+                comp = []
+                while True:
+                    top = stack.pop()
+                    on_stack[top] = False
+                    comp.append(top)
+                    if top == node:
+                        break
+                out.append(sorted(comp))
+    return out
+
+
+def stratify(program: Program) -> list:
+    """Split predicates into strata (a list of frozensets, dependencies first).
+
+    Raises StratificationError when a negated literal sits inside a recursive
+    cycle, i.e. when head and negated body predicate share an SCC.
+    """
+    preds = sorted(program.predicates())
+    adj = {p: set() for p in preds}
+    neg_edges = []
+    for rule in program.rules:
+        for lit in rule.body:
+            adj[rule.head_pred].add(lit.pred)
+            if lit.negated:
+                neg_edges.append((rule.head_pred, lit.pred))
+
+    comps = _tarjan_scc(preds, adj)
+    comp_of = {}
+    for idx, comp in enumerate(comps):
+        for pred in comp:
+            comp_of[pred] = idx
+
+    for head, body in neg_edges:
+        if comp_of[head] == comp_of[body]:
+            members = ", ".join(comps[comp_of[head]])
+            raise StratificationError(
+                f"negation is not stratified: {head} negates {body} but they are "
+                f"mutually recursive (SCC: {members})"
+            )
+
+    # condensation: edge comp(p) -> comp(q) means "p depends on q", so q must be
+    # computed first. Kahn over the reversed relation.
+    must_precede = {i: set() for i in range(len(comps))}
+    for pred in preds:
+        for dep in adj[pred]:
+            a, b = comp_of[pred], comp_of[dep]
+            if a != b:
+                must_precede[a].add(b)
+
+    indeg = {i: len(must_precede[i]) for i in must_precede}
+    dependents = {i: set() for i in must_precede}
+    for i, deps in must_precede.items():
+        for dep in deps:
+            dependents[dep].add(i)
+
+    ready = sorted((i for i, d in indeg.items() if d == 0), key=lambda i: comps[i][0])
+    order = []
+    while ready:
+        cur = ready.pop(0)
+        order.append(cur)
+        for nxt in sorted(dependents[cur], key=lambda i: comps[i][0]):
+            indeg[nxt] -= 1
+            if indeg[nxt] == 0:
+                ready.append(nxt)
+        ready.sort(key=lambda i: comps[i][0])
+    if len(order) != len(comps):  # pragma: no cover - condensation is a DAG
+        raise StratificationError("internal error: condensation is not a DAG")
+    return [frozenset(comps[i]) for i in order]
+
+
+# --------------------------------------------------------------------------
+# Evaluation
+# --------------------------------------------------------------------------
+
+
+def _match(pattern, tup, env):
+    """Unify a literal's argument pattern against a tuple. Returns a new env or
+    None. Anonymous variables never bind. Bindings made while walking this
+    literal are visible to its later arguments, so `p(X, X)` really does mean
+    "both arguments equal"."""
+    cur = env
+    copied = False
+    for term, value in zip(pattern, tup):
+        if is_var(term):
+            if is_anon(term):
+                continue
+            name = var_name(term)
+            if name in cur:
+                if cur[name] != value:
+                    return None
+            else:
+                if not copied:
+                    cur = dict(env)
+                    copied = True
+                cur[name] = value
+        elif term != value:
+            return None
+    return cur
+
+
+def _subst(term, env):
+    if is_var(term):
+        if is_anon(term):
+            raise SafetyError("anonymous variable cannot be substituted")
+        return env[var_name(term)]
+    return term
+
+
+def _negation_fails(lit, env, rels):
+    """True when the negated literal matches some tuple (so the body fails)."""
+    rel = rels.get(lit.pred, ())
+    if not rel:
+        return False
+    for tup in rel:
+        if _match(lit.args, tup, env) is not None:
+            return True
+    return False
+
+
+def _order_pos(rule, R, rels):
+    """Positive body literals, smallest relation first (deterministic)."""
+    def size(lit):
+        if lit.pred in R:
+            return len(R[lit.pred])
+        return len(rels.get(lit.pred, ()))
+
+    return sorted(rule.positive, key=lambda lit: (size(lit), lit.pred))
+
+
+def _join(rule, pos, R, rels, override, stats):
+    """Enumerate satisfying environments for the positive body literals."""
+    def relation_for(idx, lit):
+        if override and idx in override:
+            return override[idx]
+        if lit.pred in R:
+            return R[lit.pred]
+        return rels.get(lit.pred, ())
+
+    def rec(idx, env):
+        if idx == len(pos):
+            for lit in rule.negated:
+                if _negation_fails(lit, env, rels):
+                    return
+            yield env
+            return
+        lit = pos[idx]
+        for tup in sorted(relation_for(idx, lit)):
+            stats["bindings"] += 1
+            env2 = _match(lit.args, tup, env)
+            if env2 is not None:
+                yield from rec(idx + 1, env2)
+
+    yield from rec(0, {})
+
+
+def _derive_one(rule, pos, R, rels, override, out, stats):
+    head_pred = rule.head_pred
+    for env in _join(rule, pos, R, rels, override, stats):
+        stats["rule_runs"] += 1
+        tup = tuple(_subst(term, env) for term in rule.head_args)
+        if tup in R[head_pred] or tup in out[head_pred]:
+            continue
+        out[head_pred].add(tup)
+        stats["derived"] += 1
+
+
+def _derive(rules_here, R, rels, wave, override, stats):
+    out = {pred: set() for pred in wave}
+    for rule in rules_here:
+        pos = _order_pos(rule, R, rels)
+        _derive_one(rule, pos, R, rels, override, out, stats)
+    return out
+
+
+def _eval_stratum(program, wave, rels, stats):
+    rules_here = [r for r in program.rules if r.head_pred in wave]
+    if not rules_here:
+        return
+    R = {pred: rels[pred] for pred in wave}  # live sets, mutated in place
+
+    # Round 0 is naive: it also fires rules whose positive body lives entirely in
+    # lower strata (such a rule can never see a same-stratum delta).
+    new = _derive(rules_here, R, rels, wave, None, stats)
+    for pred, added in new.items():
+        R[pred] |= added
+    delta = {pred: set(added) for pred, added in new.items()}
+
+    while any(delta.values()):
+        stats["rounds"] += 1
+        new = {pred: set() for pred in wave}
+        for rule in rules_here:
+            pos = _order_pos(rule, R, rels)
+            for idx, lit in enumerate(pos):
+                if lit.pred in wave and delta.get(lit.pred):
+                    _derive_one(
+                        rule, pos, R, rels, {idx: delta[lit.pred]}, new, stats
+                    )
+        for pred, added in new.items():
+            R[pred] |= added
+        delta = new
+
+
+def evaluate(program: Program, stats: dict = None, check: bool = True) -> dict:
+    """Semi-naive bottom-up evaluation. Returns {pred: frozenset(tuples)}."""
+    if check:
+        check_arity(program)
+        check_safety(program)
+    strata = stratify(program)
+    if stats is None:
+        stats = {}
+    stats.update(rounds=0, bindings=0, derived=0, rule_runs=0)
+
+    rels = {pred: set(program.facts.get(pred, ())) for pred in program.predicates()}
+    for wave in strata:
+        _eval_stratum(program, wave, rels, stats)
+    return {pred: frozenset(tuples) for pred, tuples in rels.items()}
+
+
+# --------------------------------------------------------------------------
+# Output
+# --------------------------------------------------------------------------
+
+
+def format_relations(rels: dict, preds=None) -> str:
+    lines = []
+    for pred in sorted(rels) if preds is None else list(preds):
+        tuples = sorted(rels.get(pred, ()))
+        lines.append(f"{pred}: {len(tuples)} tuple(s)")
+        for tup in tuples:
+            lines.append(f"    {pred}({', '.join(tup)}).")
+    return "\n".join(lines)
+
+
+def main(argv=None) -> int:
+    ap = argparse.ArgumentParser(
+        prog="datalog.py",
+        description="Evaluate a Datalog program (stratified negation, semi-naive).",
+    )
+    ap.add_argument("program", help="path to a Datalog source file")
+    ap.add_argument("-q", "--query", action="append", default=None,
+                    help="only report this predicate (repeatable)")
+    ap.add_argument("--text", action="store_true", help="human-readable output")
+    ap.add_argument("--stats", action="store_true", help="include run statistics")
+    args = ap.parse_args(argv)
+
+    try:
+        program = load(args.program)
+        stats = {}
+        rels = evaluate(program, stats)
+        strata = stratify(program)
+    except DatalogError as exc:
+        print(f"error: {type(exc).__name__}: {exc}", file=sys.stderr)
+        return 2
+    except OSError as exc:
+        print(f"error: {exc}", file=sys.stderr)
+        return 2
+
+    if args.text:
+        print(format_relations(rels, args.query))
+        if args.stats:
+            print(json.dumps(stats, sort_keys=True))
+        return 0
+
+    payload = {
+        "predicates": {
+            pred: [list(t) for t in sorted(rels[pred])]
+            for pred in (sorted(rels) if args.query is None else args.query)
+        },
+        "strata": [sorted(wave) for wave in strata],
+    }
+    if args.stats:
+        payload["stats"] = stats
+    print(json.dumps(payload, sort_keys=True, indent=2))
+    return 0
+
+
+if __name__ == "__main__":  # pragma: no cover
+    sys.exit(main())

wizard-work/null/datalogkit/evidence_run.py

@@ -0,0 +1,83 @@
+#!/usr/bin/env python3
+"""evidence_run.py -- reproduce every number quoted in EVIDENCE.txt, in one run.
+
+    python3 evidence_run.py        (exit 0 = all sections ran clean)
+
+Wizard: null    Lane: Codeforge (hard coding and agentic reasoning)
+"""
+
+from __future__ import annotations
+
+import io
+import os
+import random
+import sys
+import unittest
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+sys.path.insert(0, HERE)
+
+import datalog  # noqa: E402
+import test_datalog as T  # noqa: E402
+
+
+def main() -> int:
+    print("== 1. full unittest suite (the CHECK command runs this same suite)")
+    suite = unittest.defaultTestLoader.loadTestsFromModule(T)
+    stream = io.StringIO()
+    result = unittest.TextTestRunner(stream=stream, verbosity=1).run(suite)
+    print(stream.getvalue().strip())
+    print(f"testsRun={result.testsRun} failures={len(result.failures)} "
+          f"errors={len(result.errors)}")
+
+    print("\n== 2. semi-naive vs naive work on a chain transitive closure")
+    print(f"{'n':>3} {'|path|':>7} {'semi-naive':>11} {'naive':>10} {'speedup':>8} {'rounds':>7}")
+    for n in (6, 12, 16, 20):
+        program = T.WorkReductionTests.chain_program(n)
+        engine_stats, oracle_stats = {}, {}
+        got = datalog.evaluate(program, engine_stats)
+        want = T.oracle_evaluate(program, oracle_stats)
+        assert got == want, f"engine/oracle disagree at n={n}"
+        ratio = oracle_stats["bindings"] / max(engine_stats["bindings"], 1)
+        print(f"{n:>3} {len(got['path']):>7} {engine_stats['bindings']:>11} "
+              f"{oracle_stats['bindings']:>10} {ratio:>7.1f}x {engine_stats['rounds']:>7}")
+
+    print("\n== 3. fresh-seed differential run: 400 programs, engine vs oracle")
+    rng = random.Random(999)
+    programs = ([T.gen_program(rng, constrained=True) for _ in range(200)]
+                + [T.gen_program(rng, constrained=False) for _ in range(200)])
+    counters = {"programs": 0, "unstratified": 0, "predicates": 0}
+    failures = T.differential_run(programs, counters)
+    print(f"programs={counters['programs']} unstratified={counters['unstratified']} "
+          f"predicate-relations compared={counters['predicates']} "
+          f"mismatches={len(failures)}")
+    for idx, text, why in failures[:3]:
+        print(f"MISMATCH #{idx}: {why}\n{text}")
+
+    print("\n== 4. strata assignment + refusal of a non-stratified program")
+    program = datalog.parse_program(
+        "node(a). node(b). node(c). source(a). edge(a,b).\n"
+        "reach(X) :- source(X).\n"
+        "reach(X) :- edge(Y,X), reach(Y).\n"
+        "unreach(X) :- node(X), not reach(X).\n"
+    )
+    print("strata:", [sorted(w) for w in datalog.stratify(program)])
+    rels = datalog.evaluate(program)
+    print("reach:", sorted(rels["reach"]), " unreach:", sorted(rels["unreach"]))
+    try:
+        datalog.stratify(datalog.parse_program(
+            "dom(a). p(X) :- dom(X), not q(X). q(X) :- p(X).\n"
+        ))
+        print("ERROR: unstratified program was accepted")
+        return 1
+    except datalog.StratificationError as exc:
+        print("refused:", exc)
+
+    ok = (len(result.failures) == 0 and len(result.errors) == 0
+          and not failures and result.testsRun >= 40)
+    print("\nALL SECTIONS CLEAN" if ok else "\nSOMETHING FAILED")
+    return 0 if ok else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

wizard-work/null/datalogkit/examples/reach.dl

@@ -0,0 +1,10 @@
+% examples/reach.dl -- stratified negation + recursion, the CLI demo program.
+node(a). node(b). node(c). node(d).
+source(a).
+edge(a, b). edge(b, c).
+
+reach(X) :- source(X).
+reach(X) :- edge(Y, X), reach(Y).
+
+unreachable(X) :- node(X), not reach(X).
+sink(X) :- node(X), not edge(X, _).

wizard-work/null/datalogkit/test_datalog.py

@@ -0,0 +1,706 @@
+#!/usr/bin/env python3
+"""test_datalog.py -- verification suite for datalog.py.
+
+Wizard: null    Lane: Codeforge (hard coding and agentic reasoning)
+
+Exit code is the verdict: 0 = every check passed, 1 = at least one failed.
+
+What is verified, and how
+-------------------------
+1. Parser KATs          -- facts, rules, comments, '!', zero-arity, int/string
+                           constants, and six malformed inputs that must raise.
+2. Static checks        -- arity conflicts, rule safety, stratification.
+3. Evaluation KATs      -- transitive closure, the non-linear same-generation
+                           program, stratified negation, anonymous variables.
+4. Order independence   -- shuffled fact/rule order gives identical relations.
+5. Differential testing -- 600 randomly generated programs are evaluated by the
+                           engine AND by an *independent oracle* written from
+                           scratch in this file (reachability-closure SCCs +
+                           Kahn waves for stratification; cartesian-product
+                           naive fixpoint for evaluation). Both the stratified/
+                           unstratified verdict and every relation must agree.
+6. Work reduction       -- semi-naive evaluation must enumerate strictly fewer
+                           body bindings than the naive oracle on a chain
+                           transitive closure (i.e. the delta rule is doing real
+                           work, not just returning the right answer by luck).
+7. CLI end-to-end       -- subprocess run of datalog.py, JSON contract, exit
+                           codes for success and for an unstratified program.
+
+The oracle deliberately shares NOTHING with the engine except the parser and the
+term representation; it re-derives stratification and evaluation itself.
+"""
+
+from __future__ import annotations
+
+import itertools
+import json
+import os
+import random
+import subprocess
+import sys
+import tempfile
+import unittest
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+sys.path.insert(0, HERE)
+
+import datalog  # noqa: E402
+from datalog import (  # noqa: E402
+    ArityError,
+    Literal,
+    ParseError,
+    Program,
+    Rule,
+    SafetyError,
+    StratificationError,
+    check_arity,
+    check_safety,
+    evaluate,
+    parse_program,
+    stratify,
+)
+
+DIFFERENTIAL_RUNS = {"programs": 0, "unstratified": 0, "predicates": 0}
+
+
+# ==========================================================================
+# Independent oracle
+# ==========================================================================
+
+
+def oracle_dependency_graph(program):
+    preds = sorted(program.predicates())
+    adj = {p: set() for p in preds}
+    for rule in program.rules:
+        for lit in rule.body:
+            adj[rule.head_pred].add(lit.pred)
+    return preds, adj
+
+
+def oracle_reachability(program):
+    preds, adj = oracle_dependency_graph(program)
+    reach = {p: set(adj[p]) | {p} for p in preds}
+    changed = True
+    while changed:
+        changed = False
+        for p in preds:
+            grown = set(reach[p])
+            for q in list(reach[p]):
+                grown |= reach[q]
+            if grown != reach[p]:
+                reach[p] = grown
+                changed = True
+    return preds, adj, reach
+
+
+def oracle_stratify(program):
+    """Stratify via reachability closure. Returns waves (list[list[str]]) or
+    None when the program is not stratified."""
+    preds, adj, reach = oracle_reachability(program)
+    neg = {(r.head_pred, lit.pred) for r in program.rules for lit in r.negated}
+
+    sccs, seen = [], set()
+    for p in preds:
+        if p in seen:
+            continue
+        comp = sorted(q for q in preds if q in reach[p] and p in reach[q])
+        seen.update(comp)
+        sccs.append(comp)
+    comp_of = {p: i for i, comp in enumerate(sccs) for p in comp}
+
+    for head, body in neg:
+        if comp_of[head] == comp_of[body]:
+            return None
+
+    before = {i: set() for i in range(len(sccs))}
+    for p in preds:
+        for q in adj[p]:
+            if comp_of[p] != comp_of[q]:
+                before[comp_of[p]].add(comp_of[q])
+
+    waves, assigned = [], set()
+    while len(assigned) < len(sccs):
+        ready = sorted(
+            (i for i in range(len(sccs)) if i not in assigned and before[i] <= assigned),
+            key=lambda i: sccs[i][0],
+        )
+        if not ready:
+            return None
+        assigned.update(ready)
+        waves.append(sorted(p for i in ready for p in sccs[i]))
+    return waves
+
+
+def oracle_unify(pattern, tup, env):
+    cur = env
+    copied = False
+    for term, value in zip(pattern, tup):
+        if datalog.is_var(term):
+            if datalog.is_anon(term):
+                continue
+            name = term[1:]
+            if name in cur:
+                if cur[name] != value:
+                    return None
+            else:
+                if not copied:
+                    cur = dict(env)
+                    copied = True
+                cur[name] = value
+        elif term != value:
+            return None
+    return cur
+
+
+def oracle_neg_fails(lit, env, rels):
+    for tup in rels[lit.pred]:
+        if oracle_unify(lit.args, tup, env) is not None:
+            return True
+    return False
+
+
+def oracle_evaluate(program, stats=None):
+    """Naive (product-based) fixpoint per wave. Returns None if unstratified."""
+    waves = oracle_stratify(program)
+    if waves is None:
+        return None
+    if stats is None:
+        stats = {}
+    stats.setdefault("bindings", 0)
+
+    rels = {p: set(program.facts.get(p, ())) for p in program.predicates()}
+    for wave in waves:
+        wave = set(wave)
+        changed = True
+        while changed:
+            changed = False
+            for rule in program.rules:
+                if rule.head_pred not in wave:
+                    continue
+                pools = [sorted(rels[lit.pred]) for lit in rule.positive]
+                for combo in itertools.product(*pools):
+                    stats["bindings"] += 1
+                    env = {}
+                    ok = True
+                    for lit, tup in zip(rule.positive, combo):
+                        env = oracle_unify(lit.args, tup, env)
+                        if env is None:
+                            ok = False
+                            break
+                    if not ok:
+                        continue
+                    for lit in rule.negated:
+                        if oracle_neg_fails(lit, env, rels):
+                            ok = False
+                            break
+                    if not ok:
+                        continue
+                    head = tuple(
+                        env[t[1:]] if datalog.is_var(t) else t for t in rule.head_args
+                    )
+                    if head not in rels[rule.head_pred]:
+                        rels[rule.head_pred].add(head)
+                        changed = True
+    return {p: frozenset(v) for p, v in rels.items()}
+
+
+# ==========================================================================
+# Random program generators
+# ==========================================================================
+
+CONSTS = ["a", "b", "c"]
+VARS = ["?X", "?Y", "?Z"]
+EDB2 = ["e0", "e1"]
+IDB2 = ["p0", "p1"]
+IDB1 = ["q0", "q1"]
+
+
+def _random_facts(rng):
+    facts = {}
+    for pred in EDB2:
+        fs = set()
+        for a in CONSTS:
+            for b in CONSTS:
+                if rng.random() < 0.30:
+                    fs.add((a, b))
+        facts[pred] = fs
+    return facts
+
+
+def gen_program(rng, constrained=True):
+    """Random program. constrained=True keeps negation strictly lower-stratum,
+    so the program is stratified by construction; constrained=False lets
+    negation point anywhere, which yields unstratified programs too."""
+    arity = {p: 2 for p in EDB2 + IDB2}
+    arity.update({p: 1 for p in IDB1})
+    levels = {p: 0 for p in EDB2}
+    for p in IDB2 + IDB1:
+        levels[p] = rng.choice([1, 1, 2])
+
+    program = Program(facts=_random_facts(rng), rules=[], arities=dict(arity))
+    for _ in range(rng.randint(1, 5)):
+        head_pred = rng.choice(IDB2 + IDB1)
+        hl = levels[head_pred]
+        if constrained:
+            cands = [p for p in arity if levels[p] <= hl]
+        else:
+            cands = list(arity)
+        positives, used_vars = [], []
+        for _ in range(rng.randint(1, 2)):
+            bp = rng.choice(cands)
+            args = []
+            for _ in range(arity[bp]):
+                if rng.random() < 0.15:
+                    args.append(rng.choice(CONSTS))
+                else:
+                    args.append(rng.choice(VARS))
+            positives.append(Literal(bp, tuple(args), False))
+            used_vars.extend(a for a in args if datalog.is_var(a))
+
+        def pick():
+            if used_vars and rng.random() < 0.85:
+                return rng.choice(used_vars)
+            return rng.choice(CONSTS)
+
+        head_args = tuple(pick() for _ in range(arity[head_pred]))
+        body = list(positives)
+        if rng.random() < 0.35:
+            if constrained:
+                lower = [p for p in arity if levels[p] < hl]
+            else:
+                lower = list(arity)
+            if lower:
+                bp = rng.choice(lower)
+                body.append(
+                    Literal(bp, tuple(pick() for _ in range(arity[bp])), True)
+                )
+        program.add_rule(Rule(head_pred, head_args, tuple(body)))
+    return program
+
+
+# ==========================================================================
+# 1. parser
+# ==========================================================================
+
+
+class ParserTests(unittest.TestCase):
+    def test_facts_and_rule_shape(self):
+        p = parse_program("edge(a, b).\npath(X, Y) :- edge(X, Y).\n")
+        self.assertEqual(p.facts, {"edge": {("a", "b")}})
+        self.assertEqual(len(p.rules), 1)
+        rule = p.rules[0]
+        self.assertEqual(rule.head_pred, "path")
+        self.assertEqual(rule.head_args, ("?X", "?Y"))
+        self.assertEqual(str(rule), "path(?X, ?Y) :- edge(?X, ?Y).")
+
+    def test_comments_and_whitespace(self):
+        text = "% a comment\n\nedge(a,b).  % trailing\npath(X,Y):-edge(X,Y).\n"
+        p = parse_program(text)
+        self.assertEqual(p.facts["edge"], {("a", "b")})
+        self.assertEqual(len(p.rules), 1)
+
+    def test_bang_is_not(self):
+        p = parse_program("dom(a). p(X) :- dom(X), ! q(X). q(b).\n")
+        self.assertTrue(p.rules[0].negated)
+        self.assertEqual(p.rules[0].negated[0].pred, "q")
+
+    def test_zero_arity_and_constants(self):
+        p = parse_program('flag(). n(1). s("x").\n')
+        self.assertEqual(p.facts["flag"], {()})
+        self.assertEqual(p.facts["n"], {("1",)})
+        self.assertEqual(p.facts["s"], {('"x"',)})
+
+    def test_variable_in_fact_rejected(self):
+        with self.assertRaises(ParseError):
+            parse_program("p(X).\n")
+
+    def test_bad_character_rejected(self):
+        with self.assertRaises(ParseError):
+            parse_program("p(a) # oops.\n")
+
+    def test_missing_dot_rejected(self):
+        with self.assertRaises(ParseError):
+            parse_program("p(a)\n")
+
+    def test_negated_head_rejected(self):
+        with self.assertRaises(ParseError):
+            parse_program("dom(a). not p(X) :- dom(X).\n")
+
+    def test_unbalanced_parens_rejected(self):
+        with self.assertRaises(ParseError):
+            parse_program("p(a, b.\n")
+
+    def test_empty_body_rejected(self):
+        with self.assertRaises(ParseError):
+            parse_program("p(a) :- .\n")
+
+
+# ==========================================================================
+# 2. static checks
+# ==========================================================================
+
+
+class StaticCheckTests(unittest.TestCase):
+    def test_arity_conflict_between_fact_and_body(self):
+        p = parse_program("q(a). p(X) :- q(X, Y).\n")
+        with self.assertRaises(ArityError):
+            evaluate(p)
+
+    def test_arity_conflict_between_facts(self):
+        p = parse_program("p(a, b). p(a).\n")
+        with self.assertRaises(ArityError):
+            check_arity(p)
+
+    def test_unbound_head_variable(self):
+        p = parse_program("q(a). p(X) :- q(Y).\n")
+        with self.assertRaises(SafetyError):
+            evaluate(p)
+
+    def test_rule_without_positive_body(self):
+        p = parse_program("dom(a). p(X) :- not dom(X).\n")
+        with self.assertRaises(SafetyError):
+            check_safety(p)
+
+    def test_unbound_variable_inside_negation(self):
+        p = parse_program("dom(a). q(a). p(X) :- q(Y), not dom(X).\n")
+        with self.assertRaises(SafetyError):
+            check_safety(p)
+
+    def test_anonymous_variable_in_head(self):
+        p = parse_program("q(a). p(_) :- q(a).\n")
+        with self.assertRaises(SafetyError):
+            check_safety(p)
+
+    def test_negation_inside_recursive_cycle_rejected(self):
+        p = parse_program("dom(a). p(X) :- dom(X), not q(X). q(X) :- p(X).\n")
+        with self.assertRaises(StratificationError):
+            stratify(p)
+        self.assertIsNone(oracle_stratify(p))
+
+    def test_mutual_negation_rejected(self):
+        p = parse_program(
+            "dom(a). p(X) :- dom(X), not q(X). q(X) :- dom(X), not p(X).\n"
+        )
+        with self.assertRaises(StratificationError):
+            stratify(p)
+        self.assertIsNone(oracle_stratify(p))
+
+    def test_positive_recursion_is_not_a_stratification_error(self):
+        p = parse_program(
+            "edge(a,b). edge(b,c).\n"
+            "path(X,Y) :- edge(X,Y).\n"
+            "path(X,Y) :- edge(X,Z), path(Z,Y).\n"
+        )
+        waves = stratify(p)
+        self.assertEqual(len(waves), 2)
+        self.assertEqual(waves[0], frozenset({"edge"}))
+        self.assertEqual(waves[1], frozenset({"path"}))
+        self.assertIsNotNone(oracle_stratify(p))
+
+    def test_negation_to_lower_stratum_is_fine(self):
+        p = parse_program(
+            "dom(a). dom(b). q(a).\n"
+            "r(X) :- dom(X), not q(X).\n"
+        )
+        waves = stratify(p)
+        names = [sorted(w) for w in waves]
+        self.assertLess(names.index(["q"]), names.index(["r"]))
+        self.assertEqual(evaluate(p)["r"], frozenset({("b",)}))
+
+
+# ==========================================================================
+# 3. evaluation KATs
+# ==========================================================================
+
+
+class EvaluationTests(unittest.TestCase):
+    def test_transitive_closure(self):
+        p = parse_program(
+            "edge(a,b). edge(b,c). edge(c,d).\n"
+            "path(X,Y) :- edge(X,Y).\n"
+            "path(X,Y) :- edge(X,Z), path(Z,Y).\n"
+        )
+        rels = evaluate(p)
+        self.assertEqual(
+            rels["path"],
+            frozenset({("a", "b"), ("a", "c"), ("a", "d"),
+                       ("b", "c"), ("b", "d"), ("c", "d")}),
+        )
+
+    def test_same_generation_nonlinear_recursion(self):
+        p = parse_program(
+            "person(a1). person(a2). person(b1). person(b2).\n"
+            "parent(b1, a1). parent(b1, a2). parent(b2, a1). parent(b2, a2).\n"
+            "sg(X, X) :- person(X).\n"
+            "sg(X, Y) :- parent(X, A), parent(Y, B), sg(A, B).\n"
+        )
+        rels = evaluate(p)
+        self.assertEqual(
+            rels["sg"],
+            frozenset({("a1", "a1"), ("a2", "a2"), ("b1", "b1"), ("b2", "b2"),
+                       ("b1", "b2"), ("b2", "b1")}),
+        )
+
+    def test_stratified_negation(self):
+        p = parse_program(
+            "node(a). node(b). node(c).\n"
+            "source(a).\n"
+            "edge(a,b).\n"
+            "reach(X) :- source(X).\n"
+            "reach(X) :- edge(Y,X), reach(Y).\n"
+            "unreachable(X) :- node(X), not reach(X).\n"
+        )
+        rels = evaluate(p)
+        self.assertEqual(rels["reach"], frozenset({("a",), ("b",)}))
+        self.assertEqual(rels["unreachable"], frozenset({("c",)}))
+
+    def test_anonymous_variable_in_negated_literal(self):
+        p = parse_program(
+            "node(a). node(b). node(c). edge(a,b). edge(c,a).\n"
+            "sink(X) :- node(X), not edge(X, _).\n"
+        )
+        rels = evaluate(p)
+        self.assertEqual(rels["sink"], frozenset({("b",)}))
+
+    def test_repeated_variable_inside_one_literal(self):
+        # Regression: the join reorders positive literals smallest-first, so
+        # `p1(X, X)` can be the FIRST literal joined. A repeated variable must
+        # still constrain both argument positions of that single tuple.
+        p = parse_program(
+            "e0(a, a). e0(a, b). e0(b, a). e1(b, a). e1(b, b). e1(c, a).\n"
+            "q1(X) :- e0(Z, X).\n"
+            "p1(Y, c) :- e0(Y, Z), not q0(Z).\n"
+            "p0(X, c) :- e1(X, Y), p1(X, X), not q0(X).\n"
+        )
+        rels = evaluate(p)
+        self.assertEqual(rels["p0"], frozenset())
+        self.assertEqual(rels["p1"], frozenset({("a", "c"), ("b", "c")}))
+        self.assertEqual(rels["p0"], oracle_evaluate(p)["p0"])
+
+    def test_negation_over_empty_relation_always_succeeds(self):
+        p = parse_program("dom(a). p(X) :- dom(X), not missing(X).\n")
+        rels = evaluate(p)
+        self.assertEqual(rels["p"], frozenset({("a",)}))
+        self.assertEqual(rels["missing"], frozenset())
+
+    def test_zero_arity_derivation(self):
+        p = parse_program("flag(). ready() :- flag().\n")
+        rels = evaluate(p)
+        self.assertEqual(rels["ready"], frozenset({()}))
+
+    def test_constant_arguments_and_repeated_variable(self):
+        p = parse_program("e(a,a). e(a,b). loop(X) :- e(X, X).\n")
+        rels = evaluate(p)
+        self.assertEqual(rels["loop"], frozenset({("a",)}))
+
+    def test_duplicate_facts_collapse(self):
+        p = parse_program("edge(a,b). edge(a,b). edge(a,b).\n")
+        rels = evaluate(p)
+        self.assertEqual(rels["edge"], frozenset({("a", "b")}))
+
+    def test_ground_rule_without_variables(self):
+        p = parse_program("dom(a). mark(a) :- dom(a).\n")
+        rels = evaluate(p)
+        self.assertEqual(rels["mark"], frozenset({("a",)}))
+
+    def test_order_independence(self):
+        text = (
+            "edge(a,b). edge(b,c). edge(c,d).\n"
+            "path(X,Y) :- edge(X,Y).\n"
+            "path(X,Y) :- edge(X,Z), path(Z,Y).\n"
+        )
+        base = evaluate(parse_program(text))
+        rng = random.Random(7)
+        for _ in range(5):
+            clauses = [c.strip() for c in text.strip().split("\n") if c.strip()]
+            rng.shuffle(clauses)
+            shuffled = evaluate(parse_program("\n".join(clauses) + "\n"))
+            self.assertEqual(shuffled, base)
+
+    def test_repeated_evaluation_is_deterministic(self):
+        text = (
+            "n(a). n(b). n(c). n(d).\n"
+            "e(a,b). e(b,c). e(c,d). e(d,a).\n"
+            "r(X,Y) :- e(X,Y).\n"
+            "r(X,Y) :- r(X,Z), e(Z,Y).\n"
+            "lonely(X) :- n(X), not r(X, _).\n"
+        )
+        first = evaluate(parse_program(text))
+        for _ in range(3):
+            self.assertEqual(evaluate(parse_program(text)), first)
+        self.assertEqual(len(first["r"]), 16)
+
+
+# ==========================================================================
+# 4. differential testing against the oracle
+# ==========================================================================
+
+
+def differential_run(programs, counters=None):
+    """Evaluate every program with the engine and with the oracle; return a list
+    of (index, program_text, explanation) mismatches. `counters` accumulates the
+    coverage numbers."""
+    if counters is None:
+        counters = {"programs": 0, "unstratified": 0, "predicates": 0}
+    failures = []
+    for idx, program in enumerate(programs):
+        stats = {}
+        try:
+            got = evaluate(program, stats)
+            got_error = None
+        except (StratificationError, ArityError, SafetyError) as exc:
+            got, got_error = None, type(exc).__name__
+        want = oracle_evaluate(program)
+        counters["programs"] += 1
+        if want is None:
+            counters["unstratified"] += 1
+            if got_error != "StratificationError":
+                failures.append(
+                    (idx, program.text(),
+                     f"oracle says unstratified, engine said {got_error!r}")
+                )
+            continue
+        if got_error is not None:
+            failures.append(
+                (idx, program.text(), f"oracle stratified, engine raised {got_error}")
+            )
+            continue
+        counters["predicates"] += len(got)
+        if got != want:
+            diff = {
+                p: (sorted(got.get(p, ())), sorted(want.get(p, ())))
+                for p in sorted(set(got) | set(want))
+                if got.get(p) != want.get(p)
+            }
+            failures.append((idx, program.text(), f"relation mismatch {diff}"))
+    return failures
+
+
+class DifferentialTests(unittest.TestCase):
+    def _run(self, programs):
+        failures = differential_run(programs, DIFFERENTIAL_RUNS)
+        return failures
+
+    def test_differential_constrained_300(self):
+        rng = random.Random(20261024)
+        programs = [gen_program(rng, constrained=True) for _ in range(300)]
+        failures = self._run(programs)
+        self.assertEqual(failures[:3], [], f"{len(failures)}/300 mismatches")
+
+    def test_differential_free_negation_300(self):
+        rng = random.Random(20261025)
+        programs = [gen_program(rng, constrained=False) for _ in range(300)]
+        failures = self._run(programs)
+        self.assertEqual(failures[:3], [], f"{len(failures)}/300 mismatches")
+        self.assertGreater(
+            DIFFERENTIAL_RUNS["unstratified"], 0,
+            "the free generator produced no unstratified program, so the "
+            "stratification check was not exercised",
+        )
+
+    def test_generator_actually_builds_nontrivial_programs(self):
+        rng = random.Random(1)
+        programs = [gen_program(rng, constrained=True) for _ in range(50)]
+        self.assertTrue(all(p.rules for p in programs))
+        self.assertGreater(sum(len(p.rules) for p in programs), 100)
+        self.assertTrue(any(any(lit.negated for lit in r.body)
+                            for p in programs for r in p.rules))
+
+
+# ==========================================================================
+# 5. semi-naive does less work than naive
+# ==========================================================================
+
+
+class WorkReductionTests(unittest.TestCase):
+    @staticmethod
+    def chain_program(n):
+        facts = "".join(f"edge(n{i}, n{i + 1}).\n" for i in range(n - 1))
+        return parse_program(
+            facts
+            + "path(X,Y) :- edge(X,Y).\n"
+            + "path(X,Y) :- edge(X,Z), path(Z,Y).\n"
+        )
+
+    def test_semi_naive_enumerates_fewer_bindings(self):
+        program = self.chain_program(12)
+        engine_stats, oracle_stats = {}, {}
+        rels = evaluate(program, engine_stats)
+        oracle_rels = oracle_evaluate(program, oracle_stats)
+        self.assertEqual(rels, oracle_rels)
+        self.assertEqual(len(rels["path"]), 66)  # C(12,2)
+        self.assertLess(
+            engine_stats["bindings"], oracle_stats["bindings"],
+            f"semi-naive bindings {engine_stats['bindings']} vs naive "
+            f"{oracle_stats['bindings']}",
+        )
+        # The delta rule must buy a real margin, not a rounding error.
+        self.assertLess(engine_stats["bindings"], oracle_stats["bindings"] // 2)
+
+    def test_stats_are_populated(self):
+        stats = {}
+        evaluate(self.chain_program(6), stats)
+        for key in ("rounds", "bindings", "derived", "rule_runs"):
+            self.assertIn(key, stats)
+        self.assertGreater(stats["derived"], 0)
+        self.assertGreater(stats["rounds"], 1)
+
+
+# ==========================================================================
+# 6. CLI
+# ==========================================================================
+
+
+class CliTests(unittest.TestCase):
+    SCRIPT = os.path.join(HERE, "datalog.py")
+
+    def _run(self, text, *args):
+        with tempfile.TemporaryDirectory(prefix=".datalog-fixture-", dir=HERE) as tmp:
+            path = os.path.join(tmp, "program.dl")
+            with open(path, "w", encoding="utf-8") as fh:
+                fh.write(text)
+            proc = subprocess.run(
+                [sys.executable, self.SCRIPT, path, *args],
+                capture_output=True, text=True, timeout=120,
+            )
+        return proc
+
+    def test_json_contract(self):
+        proc = self._run(
+            "node(a). node(b). node(c). source(a). edge(a,b).\n"
+            "reach(X) :- source(X).\n"
+            "reach(X) :- edge(Y,X), reach(Y).\n"
+            "unreach(X) :- node(X), not reach(X).\n",
+            "--stats",
+        )
+        self.assertEqual(proc.returncode, 0, proc.stderr)
+        payload = json.loads(proc.stdout)
+        self.assertEqual(payload["predicates"]["reach"], [["a"], ["b"]])
+        self.assertEqual(payload["predicates"]["unreach"], [["c"]])
+        self.assertEqual(payload["predicates"]["node"], [["a"], ["b"], ["c"]])
+        self.assertIn("stats", payload)
+        self.assertTrue(payload["strata"])
+
+    def test_unstratified_program_exits_2(self):
+        proc = self._run("dom(a). p(X) :- dom(X), not q(X). q(X) :- p(X).\n")
+        self.assertEqual(proc.returncode, 2)
+        self.assertIn("StratificationError", proc.stderr)
+
+    def test_text_output(self):
+        proc = self._run("e(a,b).\n", "--text")
+        self.assertEqual(proc.returncode, 0, proc.stderr)
+        self.assertIn("e(a, b).", proc.stdout)
+
+
+def tearDownModule():
+    print(
+        "\n[differential] programs evaluated: {programs} "
+        "(unstratified: {unstratified}, predicate-relations compared: {predicates})".format(
+            **DIFFERENTIAL_RUNS
+        ),
+        file=sys.stderr,
+    )
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=1)