From 7cf828a78ccc6dce92842b3cfd35e68a96837f9f Mon Sep 17 00:00:00 2001 From: null Date: Sat, 24 Oct 2026 22:01:36 +0000 Subject: [PATCH] null: stratified Datalog engine (parser + Tarjan strata + semi-naive eval) verified by 600-program differential oracle --- wizard-work/null/datalogkit/EVIDENCE.txt | 82 ++ wizard-work/null/datalogkit/README.md | 79 ++ wizard-work/null/datalogkit/datalog.py | 691 +++++++++++++++++ wizard-work/null/datalogkit/evidence_run.py | 83 ++ wizard-work/null/datalogkit/examples/reach.dl | 10 + wizard-work/null/datalogkit/test_datalog.py | 706 ++++++++++++++++++ 6 files changed, 1651 insertions(+) create mode 100644 wizard-work/null/datalogkit/EVIDENCE.txt create mode 100644 wizard-work/null/datalogkit/README.md create mode 100644 wizard-work/null/datalogkit/datalog.py create mode 100644 wizard-work/null/datalogkit/evidence_run.py create mode 100644 wizard-work/null/datalogkit/examples/reach.dl create mode 100644 wizard-work/null/datalogkit/test_datalog.py diff --git a/wizard-work/null/datalogkit/EVIDENCE.txt b/wizard-work/null/datalogkit/EVIDENCE.txt new file mode 100644 index 0000000..ae6f371 --- /dev/null +++ b/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 diff --git a/wizard-work/null/datalogkit/README.md b/wizard-work/null/datalogkit/README.md new file mode 100644 index 0000000..c4d7265 --- /dev/null +++ b/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). diff --git a/wizard-work/null/datalogkit/datalog.py b/wizard-work/null/datalogkit/datalog.py new file mode 100644 index 0000000..f5fea1d --- /dev/null +++ b/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[ \t\r\n]+)" + r"|(?P%[^\n]*)" + r"|(?P\"(?:[^\"\\]|\\.)*\")" + r"|(?P-?\d+)" + r"|(?P[A-Za-z_][A-Za-z0-9_]*)" + r"|(?P:-|[!(),.])" +) + + +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()) diff --git a/wizard-work/null/datalogkit/evidence_run.py b/wizard-work/null/datalogkit/evidence_run.py new file mode 100644 index 0000000..fc93f6c --- /dev/null +++ b/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()) diff --git a/wizard-work/null/datalogkit/examples/reach.dl b/wizard-work/null/datalogkit/examples/reach.dl new file mode 100644 index 0000000..da9ce63 --- /dev/null +++ b/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, _). diff --git a/wizard-work/null/datalogkit/test_datalog.py b/wizard-work/null/datalogkit/test_datalog.py new file mode 100644 index 0000000..c7120ef --- /dev/null +++ b/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)