diff --git a/docs/benchmarks.md b/docs/benchmarks.md
index b005652..e03fa90 100644
--- a/docs/benchmarks.md
+++ b/docs/benchmarks.md
@@ -117,6 +117,69 @@ correctness. An incomplete preparation is not published; existing prepared
 datasets are verified rather than overwritten. Do not reuse another dataset's
 interned constants or compare old-generation facts under the same application
 name without recording that change.
+Prepared directories are immutable snapshots: their manifests retain the exact
+adapter hash used to create them. Reuse verifies those stored tuples, not whether
+the adapter source is unchanged; use a new data root when applying normalization
+changes. Execution conservatively requires the recorded canonical query source
+hash to match the current query, even for source-only edits.
+
+## Running the DOOP correctness and timing suite
+
+`examples/run_doop_suite.py` runs the canonical Python query, with each dataset's
+own metadata. The CPU backend translates that instantiated logical program to
+Soufflé; it does not substitute a separately maintained query. It requires
+Soufflé and its development headers, a C++17/OpenMP compiler, and zlib/SQLite
+development libraries. `SOUFFLE`, `SOUFFLE_INCLUDE_DIR`, `CXX`, `CPPFLAGS`,
+`CXXFLAGS`, and `LDFLAGS` support nonstandard installations.
+The GPU backend requires the normal [CUDA build setup](getting_started).
+
+```bash
+# Prepare first; each run needs a new external output directory.
+CXX=g++ python examples/run_doop_suite.py --all \
+    --root /path/to/doop-data --output /path/to/results/cpu \
+    --backend cpu --threads 12
+
+python examples/run_doop_suite.py --all \
+    --root /path/to/doop-data --output /path/to/results/gpu-baseline \
+    --backend gpu --plan baseline --jobs 2 \
+    --reference /path/to/results/cpu/suite.json
+
+python examples/run_doop_suite.py --all \
+    --root /path/to/doop-data --output /path/to/results/gpu-bitmap \
+    --backend gpu --plan bitmap --jobs 2 \
+    --reference /path/to/results/cpu/suite.json
+```
+
+Use `--dataset NAME ...` or `--tier TIER` instead of `--all` for smaller runs.
+The bitmap variant applies the opt-in plan only to `VPT_Assign`'s two recursive
+variants; the baseline and logical query remain unchanged.
+Defaults are one warmup and three measured repetitions, each in a fresh process.
+`--timeout` bounds each build/execution process; `--warmups 0 --repeats 1` is
+useful for validation but is not a stable performance measurement.
+
+The suite records build, load, execution, and export separately. GPU execution
+includes host-to-device initialization and synchronizes before the timer stops;
+CPU execution times the Soufflé query after loading. These scopes are recorded
+in the reports and must not be presented as interchangeable kernel-only times.
+Every run reaches an unlimited fixedpoint and must exit normally.
+Failures and timeouts retain logs and appear explicitly in `suite.json`;
+the command exits unsuccessfully if any selected dataset fails.
+
+The final measured run records all 74 relation cardinalities and exports all
+37 derived relation sets. With `--reference`, comparison requires matching
+query/input identities, complete relation coverage, equal cardinalities, and
+exact integer tuple sets after external sorting—not just matching VPT counts.
+Without a reference, correctness is `not_compared`, even when execution passes.
+Reports distinguish `input_rows`/`input_bytes` for the 37 consumed inputs from
+`prepared_input_rows`/`prepared_input_bytes` for all 39 prepared files.
+The runner rejects a prepared benchmark with no selected main method rather than
+accepting a vacuous empty-analysis match. The per-process timeout also covers
+loading and final tuple export; increase it for large exports.
+
+Reserve substantial disk space for derived results and sort scratch, especially
+Jython; small compressed inputs do not imply small fixedpoints. Run performance
+measurements without competing CPU/GPU workloads. Compilation and a small GPU
+smoke test do not establish that a complete dataset fits in available VRAM.
 
 ## Regenerating from Nim
 
diff --git a/examples/doop_suite/compare.py b/examples/doop_suite/compare.py
new file mode 100644
index 0000000..af5d231
--- /dev/null
+++ b/examples/doop_suite/compare.py
@@ -0,0 +1,84 @@
+"""Compare complete logical tuple sets, independent of engine export row order."""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import tempfile
+from pathlib import Path
+
+
+def compare_results(left: dict, right: dict, output: Path) -> dict:
+  if left['status'] != 'passed' or right['status'] != 'passed':
+    raise ValueError('Only successful complete fixedpoints can be compared')
+  for key in ('dataset', 'source_sha256', 'metadata_sha256', 'input_manifest_sha256'):
+    if not left.get(key) or left[key] != right.get(key):
+      raise ValueError(f'Cannot compare different {key}')
+  expected = set(left['expected_relations'])
+  inputs = set(left['input_relations'])
+  outputs = expected - inputs
+  if (
+    not outputs
+    or expected != set(right['expected_relations'])
+    or inputs != set(right['input_relations'])
+  ):
+    raise ValueError('Reference and candidate have different relation schemas')
+  for result in (left, right):
+    if set(result['relation_counts']) != expected or set(result['outputs']) != outputs:
+      raise ValueError('Result does not cover every relation in the query')
+  output.parent.mkdir(parents=True, exist_ok=True)
+  if output.exists():
+    raise FileExistsError(f'Refusing to overwrite comparison: {output}')
+  checks = []
+  with tempfile.TemporaryDirectory(prefix='.compare-', dir=output.parent) as temporary:
+    scratch = Path(temporary)
+    for name in sorted(expected):
+      check = {
+        'relation': name,
+        'left_rows': left['relation_counts'][name],
+        'right_rows': right['relation_counts'][name],
+      }
+      check['passed'] = check['left_rows'] == check['right_rows']
+      if name in outputs:
+        for label, result in (('left', left), ('right', right)):
+          subprocess.run(
+            [
+              'sort',
+              '-u',
+              '-T',
+              str(scratch),
+              '-o',
+              str(scratch / label),
+              str(Path(result['outputs'][name]).resolve(strict=True)),
+            ],
+            env=dict(os.environ, LC_ALL='C'),
+            check=True,
+          )
+        equal = True
+        rows = [0, 0]
+        with (scratch / 'left').open('rb') as lhs, (scratch / 'right').open('rb') as rhs:
+          while True:
+            a, b = lhs.read(8 * 1024 * 1024), rhs.read(8 * 1024 * 1024)
+            if not a and not b:
+              break
+            equal = equal and a == b
+            rows[0] += a.count(b'\n')
+            rows[1] += b.count(b'\n')
+        check['export_rows'] = rows
+        check['equal_tuple_sets'] = equal
+        check['passed'] = (
+          check['passed'] and equal and rows == [check['left_rows'], check['right_rows']]
+        )
+      checks.append(check)
+  result = {
+    'passed': all(row['passed'] for row in checks),
+    'dataset': left['dataset'],
+    'method': 'Complete lexicographically sorted integer TSV tuple sets; no sampling',
+    'input_manifest_sha256': left['input_manifest_sha256'],
+    'relations': checks,
+  }
+  with output.open('x') as stream:
+    json.dump(result, stream, indent=2)
+    stream.write('\n')
+  return result
diff --git a/examples/doop_suite/cpu.py b/examples/doop_suite/cpu.py
new file mode 100644
index 0000000..4d59e6f
--- /dev/null
+++ b/examples/doop_suite/cpu.py
@@ -0,0 +1,484 @@
+"""Compile the instantiated canonical DOOP program and run fresh Souffle fixedpoints.
+
+Requires ``souffle``, its development headers, and a C++17/OpenMP compiler with
+zlib and SQLite development libraries. SOUFFLE and CXX select executables;
+SOUFFLE_INCLUDE_DIR, CPPFLAGS, CXXFLAGS and LDFLAGS support nonstandard installs.
+No binary or generated source is reused from outside the caller's new output.
+"""
+
+from __future__ import annotations
+
+import argparse
+import contextlib
+import hashlib
+import importlib.util
+import json
+import os
+import re
+import shlex
+import shutil
+import signal
+import subprocess
+import time
+import traceback
+from pathlib import Path
+
+from srdatalog.dsl import ArgKind, Atom, Filter, Negation, Program, Split
+
+_IDENTIFIER = re.compile(r"[A-Za-z][A-Za-z0-9_]*\Z")
+_VARIABLE = re.compile(r"[A-Za-z_][A-Za-z0-9_]*\Z")
+_INTEGER = r"-?(?:0|[1-9][0-9]*)"
+_OPERAND = rf"(?:[A-Za-z_][A-Za-z0-9_]*|{_INTEGER})"
+_COMPARISON = re.compile(rf"\s*({_OPERAND})\s*(!=|==)\s*({_OPERAND})\s*\Z")
+
+
+def _digest(path: Path) -> str:
+  digest = hashlib.sha256()
+  with path.open("rb") as stream:
+    for block in iter(lambda: stream.read(1024 * 1024), b""):
+      digest.update(block)
+  return digest.hexdigest()
+
+
+def _json(path: Path, value: dict) -> None:
+  path.write_text(json.dumps(value, indent=2) + "\n", encoding="utf-8")
+
+
+def _integer(value: int) -> str:
+  if type(value) is not int or not -(2**31) <= value < 2**31:
+    raise ValueError(f"Expected a signed int32 literal, got {value!r}")
+  return str(value)
+
+
+def _variable(name: str) -> str:
+  if name == "_":
+    return name
+  if not isinstance(name, str) or not _VARIABLE.fullmatch(name):
+    raise ValueError(f"Unsupported variable: {name!r}")
+  return "v_" + name
+
+
+def _filter(clause: Filter) -> list[str]:
+  match = re.fullmatch(r"\s*return\s+(.+);\s*", clause.code)
+  if not match:
+    raise ValueError(f"Unsupported C++ filter: {clause.code}")
+  translated = []
+  for part in match.group(1).split("&&"):
+    comparison = _COMPARISON.fullmatch(part)
+    if not comparison:
+      raise ValueError(f"Unsupported C++ filter: {clause.code}")
+    left, operator, right = comparison.groups()
+    operands = []
+    for operand in (left, right):
+      if re.fullmatch(_INTEGER, operand):
+        operands.append(_integer(int(operand)))
+      elif operand != "_" and operand in clause.vars:
+        operands.append(_variable(operand))
+      else:
+        raise ValueError(f"Undeclared filter variable: {operand!r}")
+    translated.append(f'{operands[0]} {"=" if operator == "==" else operator} {operands[1]}')
+  return translated
+
+
+def translate_program(program: Program) -> tuple[str, dict]:
+  """Export integer set rules, rejecting unsupported semantics rather than guessing.
+
+  SPLIT and GPU plans affect execution only. Multiheads become independent
+  rules with the same body; anonymous variables and negation stay intact.
+  All IDBs are observable outputs, including intermediates needed for parity.
+  """
+  relations = {relation.name: relation for relation in program.relations}
+  if len(relations) != len(program.relations):
+    raise ValueError("Duplicate relation names")
+  declarations = []
+  schema = []
+  for relation in program.relations:
+    if not _IDENTIFIER.fullmatch(relation.name):
+      raise ValueError(f"Unsupported relation name: {relation.name!r}")
+    if (
+      relation.arity < 1
+      or len(relation.column_types) != relation.arity
+      or any(kind is not int for kind in relation.column_types)
+      or relation.semiring != "NoProvenance"
+    ):
+      raise ValueError(f"Unsupported relation type/semiring: {relation.name}")
+    columns = ", ".join(f"c{index}:number" for index in range(relation.arity))
+    declarations.append(f".decl {relation.name}({columns})")
+    if relation.input_file:
+      filename = Path(relation.input_file)
+      if filename.is_absolute() or ".." in filename.parts:
+        raise ValueError(f"Nonrelative input filename: {filename}")
+      declarations.append(
+        f'.input {relation.name}(IO="file", filename={json.dumps(str(filename))}, delimiter="\\t")'
+      )
+    else:
+      declarations.append(
+        f'.output {relation.name}(IO="file", filename="{relation.name}.tsv", delimiter="\\t")'
+      )
+    schema.append(
+      {"name": relation.name, "arity": relation.arity, "input_file": relation.input_file or None}
+    )
+
+  def argument(arg) -> str:
+    if arg.kind is ArgKind.LVAR:
+      return _variable(arg.var_name)
+    if arg.kind is ArgKind.CONST:
+      literal = _integer(arg.const_value)
+      if arg.const_cpp_expr not in (None, literal):
+        raise ValueError(f"Nonliteral C++ constant: {arg.const_cpp_expr}")
+      return literal
+    raise ValueError(f"Unsupported argument: {arg}")
+
+  def atom(value: Atom) -> str:
+    if value.rel not in relations or len(value.args) != relations[value.rel].arity:
+      raise ValueError(f"Undeclared or malformed atom: {value}")
+    return value.rel + "(" + ", ".join(argument(arg) for arg in value.args) + ")"
+
+  rules = []
+  rule_map = []
+  filters = []
+  for index, rule in enumerate(program.rules):
+    if rule.count or rule.debug_code or not rule.heads:
+      raise ValueError(f"Non-relational rule behavior: {rule.name}")
+    body = []
+    for clause in rule.body:
+      if isinstance(clause, Atom):
+        body.append(atom(clause))
+      elif isinstance(clause, Negation):
+        body.append("!" + atom(clause.atom))
+      elif isinstance(clause, Filter):
+        translated = _filter(clause)
+        body.extend(translated)
+        filters.append({"rule": rule.name, "cpp": clause.code, "datalog": translated})
+      elif isinstance(clause, Split):
+        continue
+      else:
+        raise ValueError(f"Unsupported clause in {rule.name}: {clause!r}")
+    emitted = [atom(head) + (" :- " + ", ".join(body) if body else "") + "." for head in rule.heads]
+    rules.extend(emitted)
+    rule_map.append(
+      {
+        "source_index": index,
+        "name": rule.name,
+        "execution_plans_omitted": len(rule.plans),
+        "emitted": emitted,
+      }
+    )
+  text = "\n".join(
+    [
+      "// Instantiated canonical Python Program; integer set semantics, logical column order.",
+      *declarations,
+      "",
+      *rules,
+      "",
+    ]
+  )
+  return text, {
+    "relations": schema,
+    "source_rule_count": len(program.rules),
+    "emitted_rule_count": len(rules),
+    "rule_map": rule_map,
+    "filters": filters,
+  }
+
+
+def _command(
+  argv: list[str], directory: Path, label: str, env: dict[str, str], timeout: int
+) -> dict:
+  """Keep command/log evidence even if a compiler or run fails or times out."""
+  started = time.perf_counter()
+  record = {
+    "argv": argv,
+    "cwd": str(directory),
+    "timeout_seconds": timeout,
+    "stdout": str(directory / f"{label}.stdout.log"),
+    "stderr": str(directory / f"{label}.stderr.log"),
+  }
+  process = None
+  try:
+    with Path(record["stdout"]).open("w") as stdout, Path(record["stderr"]).open("w") as stderr:
+      process = subprocess.Popen(
+        argv,
+        cwd=directory,
+        env=env,
+        stdout=stdout,
+        stderr=stderr,
+        start_new_session=os.name == "posix",
+      )
+      try:
+        record["returncode"] = process.wait(timeout=timeout)
+      except BaseException:
+        if os.name == "posix":
+          with contextlib.suppress(ProcessLookupError):
+            os.killpg(process.pid, signal.SIGKILL)
+        else:
+          process.kill()
+        process.wait()
+        raise
+  except subprocess.TimeoutExpired:
+    record["returncode"] = "timeout"
+    raise TimeoutError(f"CPU {label} exceeded {timeout}s; inspect {directory}") from None
+  except BaseException as error:
+    record["error"] = f"{type(error).__name__}: {error}"
+    if process is not None:
+      record["returncode"] = process.returncode
+    raise
+  finally:
+    record["process_wall_seconds"] = time.perf_counter() - started
+    _json(directory / f"{label}.process.json", record)
+  if record["returncode"] != 0:
+    raise RuntimeError(f"CPU {label} exited {record['returncode']}; inspect {directory}")
+  return record
+
+
+def _build(source: Path, directory: Path, threads: int, env: dict[str, str], timeout: int) -> dict:
+  souffle = shutil.which(env.get("SOUFFLE", "souffle"))
+  compiler = shlex.split(env.get("CXX", "c++"))
+  if not souffle:
+    raise FileNotFoundError("Souffle not found; install souffle or set SOUFFLE")
+  if not compiler or not shutil.which(compiler[0]):
+    raise FileNotFoundError("C++ compiler not found; install a C++17/OpenMP compiler or set CXX")
+  include = env.get("SOUFFLE_INCLUDE_DIR")
+  if include:
+    includes = ["-I" + str(Path(include).expanduser().resolve())]
+  else:
+    # A prefix install keeps bin/ and include/ beside each other. Standard
+    # system headers also work through the compiler's normal search paths.
+    prefix = Path(souffle).resolve().parent.parent / "include"
+    includes = ["-I" + str(prefix)] if (prefix / "souffle/SouffleInterface.h").is_file() else []
+  driver = Path(__file__).resolve().with_name("souffle_driver.cpp")
+  generated = directory / "doop_reference.cpp"
+  executable = directory / "souffle_reference"
+  generation = _command(
+    [souffle, "-j", str(threads), "-g", str(generated), str(source)],
+    directory,
+    "generate",
+    env,
+    timeout,
+  )
+  command = [
+    *compiler,
+    "-O3",
+    "-DNDEBUG",
+    "-std=c++17",
+    "-fopenmp",
+    "-pthread",
+    *shlex.split(env.get("CPPFLAGS", "")),
+    *shlex.split(env.get("CXXFLAGS", "")),
+    "-D__EMBEDDED_SOUFFLE__",
+    *includes,
+    str(generated),
+    str(driver),
+    "-o",
+    str(executable),
+    *shlex.split(env.get("LDFLAGS", "")),
+    "-lz",
+    "-lsqlite3",
+    "-ldl",
+  ]
+  compilation = _command(command, directory, "compile", env, timeout)
+  result = {
+    "status": "built",
+    "binary": str(executable),
+    "binary_sha256": _digest(executable),
+    "factory_name": "doop_reference",
+    "driver_sha256": _digest(driver),
+    "souffle_program_sha256": _digest(source),
+    "generation": generation,
+    "compilation": compilation,
+  }
+  _json(directory / "build.json", result)
+  return result
+
+
+def run_cpu(
+  facts: Path,
+  output: Path,
+  *,
+  threads: int = 12,
+  timeout: int = 900,
+  warmups: int = 1,
+  repeats: int = 3,
+) -> dict:
+  """Run exact CPU fixedpoints; ``output`` must not exist, even if empty.
+
+  Each warmup/repetition starts a fresh process with empty IDBs. Only run()
+  is timed as the fixedpoint; load, compilation, counting and final all-IDB
+  TSV export are separate. Timeout applies to each build or execution process.
+  """
+  for name, value, minimum in (
+    ("threads", threads, 1),
+    ("timeout", timeout, 1),
+    ("warmups", warmups, 0),
+    ("repeats", repeats, 1),
+  ):
+    if type(value) is not int or value < minimum:
+      raise ValueError(f"{name} must be an integer >= {minimum}")
+  if Path(output).is_symlink():
+    raise FileExistsError(f"Output already exists: {output}")
+  facts, output = Path(facts).resolve(), Path(output).resolve()
+  output.mkdir(parents=True, exist_ok=False)
+  result = {
+    "schema_version": 1,
+    "status": "running",
+    "backend": "cpu",
+    "plan": "baseline",
+    "engine": "souffle-compiled",
+    "dataset": facts.name,
+    "facts": str(facts),
+    "threads": threads,
+    "warmups": warmups,
+    "repeats": repeats,
+    "timings_seconds": [],
+    "runs": [],
+  }
+  try:
+    manifest_path, metadata_path = facts / "manifest.json", facts / "meta.json"
+    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+    metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
+    if not isinstance(manifest, dict) or not isinstance(metadata, dict):
+      raise ValueError("Prepared manifest and metadata must be JSON objects")
+    source = Path(__file__).resolve().parents[1] / "doop.py"
+    result.update(
+      source_sha256=_digest(source),
+      metadata_sha256=_digest(metadata_path),
+      input_source_sha256=manifest["source_sha256"],
+      input_manifest_sha256=_digest(manifest_path),
+    )
+    if manifest["program"]["sha256"] != result["source_sha256"]:
+      raise ValueError("Canonical program differs from the prepared manifest; prepare again")
+    if (
+      manifest["metadata"]["path"] != "meta.json"
+      or manifest["metadata"]["sha256"] != result["metadata_sha256"]
+    ):
+      raise ValueError("Prepared metadata differs from manifest")
+    spec = importlib.util.spec_from_file_location("_doop_suite_cpu_canonical", source)
+    if spec is None or spec.loader is None:
+      raise ImportError(f"Cannot load canonical program: {source}")
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    text, exported = translate_program(module.build_doopdb_program(metadata))
+    schema = {relation["name"]: relation for relation in exported["relations"]}
+    inputs = {name: relation for name, relation in schema.items() if relation["input_file"]}
+    idbs = set(schema) - inputs.keys()
+    result.update(expected_relations=sorted(schema), input_relations=sorted(inputs))
+    from .prepare import SCHEMA
+
+    # The source declares two EDBs unused by the instantiated Program.
+    # Keep the complete preparation contract, but execute only active relations.
+    if set(manifest["relations"]) != set(SCHEMA) or not inputs.keys() <= SCHEMA.keys():
+      raise ValueError("Prepared manifest does not contain the declared input contract")
+    input_counts = {}
+    for name, relation in inputs.items():
+      entry = manifest["relations"][name]
+      path = facts / relation["input_file"]
+      if entry["path"] != relation["input_file"] or entry["arity"] != relation["arity"]:
+        raise ValueError(f"Prepared schema mismatch: {name}")
+      if not path.is_file() or path.stat().st_size != entry["bytes"]:
+        raise ValueError(f"Prepared input missing or size differs from manifest: {path}")
+      if type(entry["rows"]) is not int or entry["rows"] < 0:
+        raise ValueError(f"Invalid prepared row count: {name}")
+      input_counts[name] = entry["rows"]
+    # Do not scan multi-gigabyte inputs here: preparation records hashes and
+    # validates int32 set data. Scanning immediately before load warms cache.
+    build_dir = output / "build"
+    build_dir.mkdir()
+    dl = build_dir / "doop_reference.dl"
+    dl.write_text(text, encoding="utf-8")
+    exported.update(
+      source_sha256=result["source_sha256"], metadata_sha256=result["metadata_sha256"]
+    )
+    _json(build_dir / "export.json", exported)
+    env = os.environ.copy()
+    env["OMP_NUM_THREADS"] = str(threads)
+    temp = build_dir / "tmp"
+    temp.mkdir()
+    env["TMPDIR"] = str(temp)
+    build = _build(dl, build_dir, threads, env, timeout)
+    result["build"] = build
+    result["generation_seconds"] = build["generation"]["process_wall_seconds"]
+    result["compile_seconds"] = build["compilation"]["process_wall_seconds"]
+    counts = None
+    for index in range(warmups + repeats):
+      warmup = index < warmups
+      run_dir = output / (f"warmup-{index:03d}" if warmup else f"run-{index - warmups:03d}")
+      run_dir.mkdir()
+      export = index == warmups + repeats - 1
+      timing_path = run_dir / "timing.json"
+      command = [
+        build["binary"],
+        build["factory_name"],
+        str(facts),
+        str(run_dir / "tuples"),
+        str(threads),
+        str(timing_path),
+        "tsv" if export else "none",
+      ]
+      process = _command(command, run_dir, "run", env, timeout)
+      timing = json.loads(timing_path.read_text(encoding="utf-8"))
+      observed = timing["relation_counts"]
+      if set(observed) != schema.keys() or any(
+        type(n) is not int or n < 0 for n in observed.values()
+      ):
+        raise ValueError(f"CPU did not retain all canonical relations: {run_dir}")
+      if any(observed[name] != count for name, count in input_counts.items()):
+        raise ValueError(f"Loaded input counts differ from prepared set counts: {run_dir}")
+      if counts is not None and observed != counts:
+        raise ValueError(f"Fresh CPU fixedpoints disagree on cardinalities: {run_dir}")
+      counts = observed
+      if set(timing["exported_relations"]) != (idbs if export else set()):
+        raise ValueError(f"CPU export omitted or added a relation: {run_dir}")
+      result["runs"].append({"warmup": warmup, "process": process, "timing": timing})
+      if not warmup:
+        result["timings_seconds"].append(timing["run_seconds"])
+      if export:
+        result["outputs"] = {name: str(run_dir / "tuples" / f"{name}.tsv") for name in sorted(idbs)}
+        for path in result["outputs"].values():
+          if not Path(path).is_file():
+            raise FileNotFoundError(path)
+    result.update(
+      status="passed",
+      relation_counts=counts,
+      load_seconds=[run["timing"]["load_seconds"] for run in result["runs"] if not run["warmup"]],
+      export_seconds=result["runs"][-1]["timing"]["export_seconds"],
+      page_cache_policy="Uncontrolled; fresh processes, no automatic cache flush; explicit warmups only.",
+      input_verification="Prepared manifest sizes/schema and loaded set counts; input hashes recorded at preparation.",
+    )
+    _json(output / "result.json", result)
+    return result
+  except BaseException as error:
+    result.update(
+      status="failed", error=f"{type(error).__name__}: {error}", traceback=traceback.format_exc()
+    )
+    # Preserve the original failure if the filesystem itself failed.
+    with contextlib.suppress(OSError):
+      _json(output / "failure.json", result)
+    raise
+
+
+def main() -> None:
+  parser = argparse.ArgumentParser(description=__doc__)
+  parser.add_argument("--facts", required=True, type=Path)
+  parser.add_argument("--output", required=True, type=Path)
+  parser.add_argument("--threads", type=int, default=12)
+  parser.add_argument("--timeout", type=int, default=900)
+  parser.add_argument("--warmups", type=int, default=1)
+  parser.add_argument("--repeats", type=int, default=3)
+  args = parser.parse_args()
+  print(
+    json.dumps(
+      run_cpu(
+        args.facts,
+        args.output,
+        threads=args.threads,
+        timeout=args.timeout,
+        warmups=args.warmups,
+        repeats=args.repeats,
+      ),
+      indent=2,
+    )
+  )
+
+
+if __name__ == "__main__":
+  main()
diff --git a/examples/doop_suite/gpu.py b/examples/doop_suite/gpu.py
new file mode 100644
index 0000000..87a3723
--- /dev/null
+++ b/examples/doop_suite/gpu.py
@@ -0,0 +1,400 @@
+"""Build and measure the canonical DOOP program through the generated CUDA ABI.
+
+Every warmup/repetition runs in a fresh process and must exit normally after
+checked shutdown. Build, CSV loading and exact IDB exports are not timed as
+fixedpoint work. A successful report does not assert cross-engine equality.
+"""
+
+from __future__ import annotations
+
+import argparse
+import contextlib
+import ctypes
+import hashlib
+import json
+import os
+import signal
+import subprocess
+import sys
+import time
+from pathlib import Path
+
+_ROOT = Path(__file__).resolve().parents[2]
+_SOURCE = _ROOT / "examples" / "doop.py"
+
+
+def _sha256(path: Path) -> str:
+  digest = hashlib.sha256()
+  with path.open("rb") as source:
+    for chunk in iter(lambda: source.read(1024 * 1024), b""):
+      digest.update(chunk)
+  return digest.hexdigest()
+
+
+def _save(path: Path, value: dict) -> None:
+  temporary = path.with_suffix(path.suffix + ".tmp")
+  temporary.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+  temporary.replace(path)
+
+
+def _program(meta: dict, plan: str):
+  # Resolve both compiler and logical program from this checkout, not an
+  # installed historical version or a workstation-specific module directory.
+  sys.path[:0] = [str(_ROOT / "src"), str(_ROOT / "examples")]
+  from doop import build_doopdb_program
+
+  from srdatalog import Program
+
+  program = build_doopdb_program(meta)
+  if plan == "baseline":
+    return program
+  if plan != "bitmap":
+    raise ValueError(f"Unknown GPU plan: {plan}")
+  if sum(rule.name == "VPT_Assign" for rule in program.rules) != 1:
+    raise ValueError("Expected exactly one canonical VPT_Assign rule")
+  return Program(
+    rules=[
+      rule.with_plan(delta=0, dedup_bitmap=True).with_plan(delta=1, dedup_bitmap=True)
+      if rule.name == "VPT_Assign"
+      else rule
+      for rule in program.rules
+    ]
+  )
+
+
+def _check_prepared(facts: Path, program) -> dict:
+  from doop_suite.prepare import SCHEMA
+
+  manifest = json.loads((facts / "manifest.json").read_text(encoding="utf-8"))
+  if manifest["program"]["sha256"] != _sha256(_SOURCE):
+    raise ValueError("Prepared facts target a different logical DOOP program")
+  for key, filename in (("metadata", "meta.json"), ("symbols", "str2num.json")):
+    if manifest[key]["sha256"] != _sha256(facts / filename):
+      raise ValueError(f"Prepared {filename} checksum mismatch")
+  inputs = {relation.name: relation for relation in program.relations if relation.input_file}
+  if set(manifest["relations"]) != set(SCHEMA) or not inputs.keys() <= SCHEMA.keys():
+    raise ValueError("Prepared input relation schema differs from the declared input contract")
+  for name, relation in inputs.items():
+    entry = manifest["relations"][name]
+    if entry["path"] != relation.input_file or entry["arity"] != relation.arity:
+      raise ValueError(f"Prepared input schema mismatch: {name}")
+    if entry["sha256"] != _sha256(facts / relation.input_file):
+      raise ValueError(f"Prepared input checksum mismatch: {name}")
+  return manifest
+
+
+def _run_process(command: list[str], log: Path, timeout: int) -> float:
+  """Gate on actual process exit, killing its entire process group on timeout."""
+  started = time.perf_counter()
+  with log.open("xb") as stream:
+    child = subprocess.Popen(
+      command,
+      stdout=stream,
+      stderr=subprocess.STDOUT,
+      start_new_session=True,
+      cwd=log.parent.resolve(),
+    )
+    try:
+      code = child.wait(timeout=timeout)
+    except BaseException:
+      # The group is ours (start_new_session=True), including compiler children.
+      # Do not leave a compiler/GPU worker alive after its caller gives up.
+      with contextlib.suppress(ProcessLookupError):
+        os.killpg(child.pid, signal.SIGKILL)
+      child.wait()
+      raise
+  if code != 0:
+    raise RuntimeError(f"GPU worker exited {code}; see {log}")
+  return time.perf_counter() - started
+
+
+def _build(facts: Path, output: Path, plan: str, jobs: int) -> None:
+  started = time.perf_counter()
+  program = _program(json.loads((facts / "meta.json").read_text(encoding="utf-8")), plan)
+  manifest = _check_prepared(facts, program)
+  preparation_seconds = time.perf_counter() - started
+  from srdatalog import CompilerConfig, build_project, compile_jit_project
+  from srdatalog.runtime import (
+    cuda_compile_flags,
+    cuda_include_paths,
+    cuda_libs,
+    cuda_link_flags,
+    runtime_defines,
+    runtime_include_paths,
+  )
+
+  started = time.perf_counter()
+  project = build_project(program, "DoopSuite", cache_base=str(output / "build-cache"))
+  emit_seconds = time.perf_counter() - started
+  config = CompilerConfig(
+    include_paths=runtime_include_paths() + cuda_include_paths(),
+    defines=runtime_defines(),
+    cxx_flags=cuda_compile_flags() + ["-fPIC"],
+    link_flags=cuda_link_flags(),
+    libs=cuda_libs() + ["boost_container"],
+    shared=True,
+    jobs=jobs,
+  )
+  started = time.perf_counter()
+  build = compile_jit_project(project, config)
+  compile_seconds = time.perf_counter() - started
+  for result in [*build.compile_results, *([build.link_result] if build.link_result else [])]:
+    print(json.dumps({"command": result.command, "returncode": result.returncode}), flush=True)
+    if result.stdout:
+      print(result.stdout, flush=True)
+    if result.stderr:
+      print(result.stderr, file=sys.stderr, flush=True)
+  if not build.ok() or not build.artifact or not Path(build.artifact).is_file():
+    raise RuntimeError("DOOP GPU compilation/linking failed")
+  _save(
+    output / "build.json",
+    {
+      "status": "built",
+      "library": str(Path(build.artifact).resolve()),
+      "library_sha256": _sha256(Path(build.artifact)),
+      "source_sha256": _sha256(_SOURCE),
+      "metadata_sha256": _sha256(facts / "meta.json"),
+      "input_source_sha256": manifest["source_sha256"],
+      "input_manifest_sha256": _sha256(facts / "manifest.json"),
+      "preparation_seconds": preparation_seconds,
+      "emit_seconds": emit_seconds,
+      "compile_seconds": compile_seconds,
+      "relations": [
+        {"name": relation.name, "arity": relation.arity, "input_file": relation.input_file}
+        for relation in program.relations
+      ],
+      "input_rows": {
+        relation.name: manifest["relations"][relation.name]["rows"]
+        for relation in program.relations
+        if relation.input_file
+      },
+    },
+  )
+
+
+def _execute(build: dict, facts: Path, report_path: Path, export: Path | None) -> None:
+  # The library remains loaded until normal process exit: do not dlclose while
+  # generated function-local GPU scratch or CUDA stream pools still exist.
+  library = ctypes.CDLL(build["library"], mode=ctypes.RTLD_GLOBAL)
+  signatures = {
+    "init": [],
+    "load_all": [ctypes.c_char_p],
+    "run": [ctypes.c_ulonglong],
+    "synchronize": [],
+    "shutdown": [],
+    "get_size": [ctypes.c_char_p, ctypes.POINTER(ctypes.c_ulonglong)],
+    "export_tsv": [ctypes.c_char_p, ctypes.c_char_p],
+  }
+  for name, args in signatures.items():
+    function = getattr(library, "srdatalog_" + name)
+    function.argtypes = args
+    function.restype = ctypes.c_int
+
+  def checked(name: str, *arguments) -> None:
+    code = getattr(library, "srdatalog_" + name)(*arguments)
+    if code != 0:
+      raise RuntimeError(f"srdatalog_{name} returned {code}")
+
+  report = {"status": "running", "stages_seconds": {}, "outputs": {}}
+  initialized = False
+  _save(report_path, report)
+  try:
+    started = time.perf_counter()
+    checked("init")
+    initialized = True
+    checked("synchronize")
+    report["stages_seconds"]["init"] = time.perf_counter() - started
+    started = time.perf_counter()
+    checked("load_all", os.fsencode(facts))
+    checked("synchronize")
+    report["stages_seconds"]["load"] = time.perf_counter() - started
+    # Includes fresh host-to-device DB construction and *every* fixedpoint step.
+    # Zero means no iteration cap, not a single-iteration smoke benchmark.
+    started = time.perf_counter()
+    checked("run", 0)
+    checked("synchronize")
+    report["fixedpoint_seconds"] = time.perf_counter() - started
+    started = time.perf_counter()
+    counts = {}
+    for relation in build["relations"]:
+      count = ctypes.c_ulonglong()
+      checked("get_size", relation["name"].encode(), ctypes.byref(count))
+      counts[relation["name"]] = count.value
+    report["relation_counts"] = counts
+    report["stages_seconds"]["counts"] = time.perf_counter() - started
+    for name, rows in build["input_rows"].items():
+      if counts[name] != rows:
+        raise RuntimeError(
+          f"Loaded input cardinality mismatch for {name}: {counts[name]} != {rows}"
+        )
+    if export is not None:
+      export.mkdir()
+      started = time.perf_counter()
+      for relation in build["relations"]:
+        if relation["input_file"]:
+          continue
+        name = relation["name"]
+        path = export / f"{name}.tsv"
+        checked("export_tsv", name.encode(), os.fsencode(path))
+        if not path.is_file():
+          raise RuntimeError(f"Native export did not create {path}")
+        report["outputs"][name] = str(path)
+      checked("synchronize")
+      report["stages_seconds"]["export"] = time.perf_counter() - started
+  except BaseException as error:
+    report["status"] = "failed"
+    report["error"] = repr(error)
+    raise
+  finally:
+    try:
+      if initialized:
+        started = time.perf_counter()
+        checked("shutdown")
+        report["stages_seconds"]["shutdown"] = time.perf_counter() - started
+    except BaseException as error:
+      report["status"] = "failed"
+      report["shutdown_error"] = repr(error)
+      raise
+    finally:
+      _save(report_path, report)
+  report["status"] = "native_completed"
+  _save(report_path, report)
+
+
+def run_gpu(
+  facts: Path,
+  output: Path,
+  *,
+  plan: str = "baseline",
+  jobs: int = 2,
+  timeout: int = 900,
+  warmups: int = 1,
+  repeats: int = 3,
+) -> dict:
+  """Return a process-gated report; timeout applies to build and each fresh run.
+
+  Requires the checkout's CUDA compiler/runtime dependencies and an available
+  GPU. Export borrows canonical index columns and uses bounded host buffers.
+  Neither a child-written report nor native return alone is success:
+  normal process teardown must also finish within the timeout.
+  """
+  if plan not in ("baseline", "bitmap"):
+    raise ValueError("plan must be 'baseline' or 'bitmap'")
+  if jobs < 1 or timeout < 1 or warmups < 0 or repeats < 1:
+    raise ValueError("jobs/timeout/repeats must be positive; warmups must be nonnegative")
+  facts, output = Path(facts).resolve(), Path(output).resolve()
+  if not facts.is_dir():
+    raise NotADirectoryError(facts)
+  if output.is_relative_to(_ROOT) or output.is_relative_to(facts):
+    raise ValueError("GPU output must be outside the repository and prepared facts")
+  output.mkdir(parents=True, exist_ok=False)
+  report = {
+    "status": "running",
+    "backend": "gpu",
+    "dataset": facts.name,
+    "plan": plan,
+    "facts": str(facts),
+    "source_sha256": _sha256(_SOURCE),
+    "timings_seconds": [],
+    "relation_counts": {},
+    "outputs": {},
+    "runs": [],
+    "jobs": jobs,
+    "timeout_seconds_per_process": timeout,
+    "warmups": warmups,
+    "repeats": repeats,
+    "timing_boundary": "fresh H2D database + entire unlimited fixedpoint + checked device synchronization",
+    "process_isolation": "one fresh process per warmup/measured run; normal exit required",
+  }
+  report_path = output / "report.json"
+  _save(report_path, report)
+  worker = [sys.executable, "-B", str(Path(__file__).resolve())]
+  try:
+    report["build_process_seconds"] = _run_process(
+      [*worker, "_build", str(facts), str(output), plan, str(jobs)],
+      output / "build.log",
+      timeout,
+    )
+    build = json.loads((output / "build.json").read_text(encoding="utf-8"))
+    if build["status"] != "built" or build["source_sha256"] != report["source_sha256"]:
+      raise RuntimeError("Build report incomplete or logical program changed during build")
+    for field in (
+      "metadata_sha256",
+      "input_source_sha256",
+      "input_manifest_sha256",
+      "library",
+      "library_sha256",
+      "preparation_seconds",
+      "emit_seconds",
+      "compile_seconds",
+    ):
+      report[field] = build[field]
+    reference_counts = None
+    for index in range(-warmups, repeats):
+      label = f"warmup-{index + warmups:03d}" if index < 0 else f"run-{index:03d}"
+      run_path = output / f"{label}.json"
+      export = output / "relations" if index == repeats - 1 else None
+      command = [*worker, "_run", str(output / "build.json"), str(facts), str(run_path)]
+      if export is not None:
+        command.append(str(export))
+      elapsed = _run_process(command, output / f"{label}.log", timeout)
+      run = json.loads(run_path.read_text(encoding="utf-8"))
+      if run["status"] != "native_completed":
+        raise RuntimeError(f"{label} exited without completing every native stage")
+      counts = run["relation_counts"]
+      if set(counts) != {relation["name"] for relation in build["relations"]}:
+        raise RuntimeError(f"{label} did not report the complete relation schema")
+      if reference_counts is not None and counts != reference_counts:
+        raise RuntimeError(f"Relation cardinalities changed in fresh repetition {label}")
+      reference_counts = counts
+      run.update({"index": index, "warmup": index < 0, "process_seconds": elapsed})
+      report["runs"].append(run)
+      if index >= 0:
+        report["timings_seconds"].append(run["fixedpoint_seconds"])
+      report["relation_counts"] = counts
+      if export is not None:
+        expected = {
+          relation["name"] for relation in build["relations"] if not relation["input_file"]
+        }
+        if set(run["outputs"]) != expected:
+          raise RuntimeError("Final repetition did not export every IDB relation")
+        report["outputs"] = run["outputs"]
+        report["export_seconds"] = run["stages_seconds"]["export"]
+      _save(report_path, report)
+    report["load_seconds"] = [
+      run["stages_seconds"]["load"] for run in report["runs"] if not run["warmup"]
+    ]
+    report["status"] = "passed"
+    _save(report_path, report)
+    return report
+  except BaseException as error:
+    report["status"] = "failed"
+    report["error"] = repr(error)
+    _save(report_path, report)
+    raise
+
+
+def _main() -> None:
+  if len(sys.argv) > 1 and sys.argv[1] == "_build":
+    _build(Path(sys.argv[2]), Path(sys.argv[3]), sys.argv[4], int(sys.argv[5]))
+    return
+  if len(sys.argv) > 1 and sys.argv[1] == "_run":
+    build = json.loads(Path(sys.argv[2]).read_text(encoding="utf-8"))
+    _execute(
+      build, Path(sys.argv[3]), Path(sys.argv[4]), Path(sys.argv[5]) if len(sys.argv) > 5 else None
+    )
+    return
+  parser = argparse.ArgumentParser(description=__doc__)
+  parser.add_argument("facts", type=Path)
+  parser.add_argument("output", type=Path)
+  parser.add_argument("--plan", choices=("baseline", "bitmap"), default="baseline")
+  parser.add_argument("--jobs", type=int, default=2)
+  parser.add_argument("--timeout", type=int, default=900)
+  parser.add_argument("--warmups", type=int, default=1)
+  parser.add_argument("--repeats", type=int, default=3)
+  args = parser.parse_args()
+  print(json.dumps(run_gpu(**vars(args)), indent=2))
+
+
+if __name__ == "__main__":
+  _main()
diff --git a/examples/doop_suite/souffle_driver.cpp b/examples/doop_suite/souffle_driver.cpp
new file mode 100644
index 0000000..708fe37
--- /dev/null
+++ b/examples/doop_suite/souffle_driver.cpp
@@ -0,0 +1,122 @@
+// Link generated Souffle C++ with -D__EMBEDDED_SOUFFLE__.
+#include <souffle/SouffleInterface.h>
+#include <chrono>
+#include <cstdint>
+#include <filesystem>
+#include <fstream>
+#include <iomanip>
+#include <iostream>
+#include <limits>
+#include <memory>
+#include <stdexcept>
+#include <string>
+#include <utility>
+#include <vector>
+
+using Clock = std::chrono::steady_clock;
+
+static double seconds(Clock::time_point begin, Clock::time_point end) {
+    return std::chrono::duration<double>(end - begin).count();
+}
+
+int main(int argc, char** argv) {
+    try {
+        if (argc != 7) {
+            std::cerr << "usage: souffle_reference PROGRAM FACTS OUTPUT THREADS REPORT none|tsv\n";
+            return 2;
+        }
+        const std::string program_name = argv[1];
+        const std::filesystem::path facts = argv[2], output = argv[3], report_path = argv[5];
+        const std::string format = argv[6];
+        if (format != "none" && format != "tsv") {
+            throw std::runtime_error("Output format must be none or tsv");
+        }
+        std::size_t consumed = 0;
+        const int threads = std::stoi(argv[4], &consumed);
+        if (threads < 1 || consumed != std::string(argv[4]).size()) {
+            throw std::runtime_error("threads must be a positive integer");
+        }
+        if (std::filesystem::exists(report_path)) {
+            throw std::runtime_error("Refusing to replace an existing timing report");
+        }
+        if (format == "tsv" && !std::filesystem::create_directory(output)) {
+            throw std::runtime_error("Refusing to replace an existing tuple directory");
+        }
+
+        const auto begin = Clock::now();
+        std::unique_ptr<souffle::SouffleProgram> program(souffle::ProgramFactory::newInstance(program_name));
+        if (!program) {
+            throw std::runtime_error("Unknown generated program factory: " + program_name);
+        }
+        program->setNumThreads(static_cast<std::size_t>(threads));
+        const auto instantiated = Clock::now();
+        program->loadAll(facts.string());
+        const auto loaded = Clock::now();
+        // No input/output or intermediate pruning inside the measured fixedpoint.
+        // Completion of this synchronous call includes all OpenMP work.
+        program->runAll("", "", false, false);
+        const auto fixedpoint = Clock::now();
+
+        std::vector<std::pair<std::string, std::size_t>> counts;
+        for (auto* relation : program->getAllRelations()) {
+            counts.emplace_back(relation->getName(), relation->size());
+        }
+        const auto counted = Clock::now();
+        std::vector<std::string> exported;
+        if (format == "tsv") {
+            // Every canonical IDB is marked as an output in the translated source.
+            // tuple[column] uses logical declaration order, not physical index order.
+            for (auto* relation : program->getOutputRelations()) {
+                const auto name = relation->getName();
+                std::ofstream stream;
+                stream.exceptions(std::ios::badbit | std::ios::failbit);
+                stream.open(output / (name + ".tsv"));
+                for (const auto& tuple : *relation) {
+                    for (std::size_t column = 0; column < relation->getArity(); ++column) {
+                        const auto value = tuple[column];
+                        if (value < std::numeric_limits<std::int32_t>::min()
+                                || value > std::numeric_limits<std::int32_t>::max()) {
+                            throw std::runtime_error("Tuple outside the common signed int32 domain: " + name);
+                        }
+                        if (column) stream << '\t';
+                        stream << value;
+                    }
+                    stream << '\n';
+                }
+                stream.close();
+                exported.push_back(name);
+            }
+        }
+        const auto exported_at = Clock::now();
+        std::ofstream report;
+        report.exceptions(std::ios::badbit | std::ios::failbit);
+        report.open(report_path);
+        report << std::setprecision(12)
+               << "{\"schema_version\":1,\"threads\":" << threads
+               << ",\"instantiate_seconds\":" << seconds(begin, instantiated)
+               << ",\"load_seconds\":" << seconds(instantiated, loaded)
+               << ",\"run_seconds\":" << seconds(loaded, fixedpoint)
+               << ",\"count_seconds\":" << seconds(fixedpoint, counted)
+               << ",\"export_seconds\":" << seconds(counted, exported_at)
+               << ",\"relation_counts\":{";
+        bool first = true;
+        for (const auto& item : counts) {
+            if (!first) report << ',';
+            first = false;
+            report << '"' << item.first << "\":" << item.second;
+        }
+        report << "},\"exported_relations\":[";
+        first = true;
+        for (const auto& name : exported) {
+            if (!first) report << ',';
+            first = false;
+            report << '"' << name << '"';
+        }
+        report << "]}\n";
+        report.close();
+        return 0;
+    } catch (const std::exception& error) {
+        std::cerr << "CPU reference error: " << error.what() << '\n';
+        return 1;
+    }
+}
diff --git a/examples/doop_suite/suite.py b/examples/doop_suite/suite.py
new file mode 100644
index 0000000..04a8f8a
--- /dev/null
+++ b/examples/doop_suite/suite.py
@@ -0,0 +1,160 @@
+"""Sequential, failure-accounted execution of explicitly selected DOOP datasets."""
+
+from __future__ import annotations
+
+import importlib.util
+import json
+import sys
+from pathlib import Path
+
+from .catalog import digest, load_catalog, verify_prepared
+from .compare import compare_results
+
+REPO = Path(__file__).resolve().parents[2]
+
+
+def logical_program(metadata: Path):
+  sys.path.insert(0, str(REPO / 'src'))
+  source = REPO / 'examples' / 'doop.py'
+  spec = importlib.util.spec_from_file_location('doop_suite_model', source)
+  module = importlib.util.module_from_spec(spec)
+  spec.loader.exec_module(module)
+  return module.build_doopdb_program(json.loads(metadata.read_text()))
+
+
+def run_matrix(
+  datasets: list[dict],
+  root: Path,
+  output: Path,
+  *,
+  backend: str,
+  plan: str = 'baseline',
+  threads: int = 12,
+  jobs: int = 2,
+  timeout: int = 900,
+  warmups: int = 1,
+  repeats: int = 3,
+  reference: Path | None = None,
+) -> dict:
+  if backend not in ('cpu', 'gpu'):
+    raise ValueError(f'Unknown backend: {backend}')
+  if plan not in ('baseline', 'bitmap') or (backend == 'cpu' and plan != 'baseline'):
+    raise ValueError('CPU executes logical baseline; bitmap is an opt-in GPU plan')
+  if min(threads, jobs, timeout, repeats) < 1 or warmups < 0:
+    raise ValueError('Positive threads/jobs/timeout/repeats and nonnegative warmups required')
+  root, output = root.resolve(), output.resolve()
+  references = {}
+  if reference is not None:
+    document = json.loads(reference.read_text())
+    references = {item['dataset']: item for item in document['results']}
+    if any(item['name'] not in references for item in datasets):
+      raise ValueError('Reference does not cover every selected dataset')
+  output.mkdir(parents=True, exist_ok=False)
+  report = {
+    'schema_version': 1,
+    'suite': load_catalog()['suite'],
+    'status': 'running',
+    'backend': backend,
+    'plan': plan,
+    'selected_datasets': [d['name'] for d in datasets],
+    'tier_metric': load_catalog()['tier_metric'],
+    'results': [],
+  }
+
+  def save():
+    (output / 'suite.json').write_text(json.dumps(report, indent=2) + '\n')
+
+  save()
+  for dataset in datasets:
+    name = dataset['name']
+    facts = root / 'prepared' / name
+    print(f'[run] {name} backend={backend} plan={plan}', flush=True)
+    try:
+      manifest = verify_prepared(facts, dataset)
+      if not manifest['entrypoints']['selected_main_methods']:
+        raise ValueError('Refusing an empty-root DOOP benchmark: no main method was selected')
+      program = logical_program(facts / 'meta.json')
+      expected = {relation.name for relation in program.relations}
+      inputs = {relation.name for relation in program.relations if relation.input_file}
+      from .prepare import SCHEMA
+
+      if set(manifest['relations']) != set(SCHEMA) or not inputs <= SCHEMA.keys():
+        raise ValueError('Prepared relation contract differs from the declared inputs')
+      if backend == 'cpu':
+        from .cpu import run_cpu
+
+        result = run_cpu(
+          facts, output / name, threads=threads, timeout=timeout, warmups=warmups, repeats=repeats
+        )
+      else:
+        from .gpu import run_gpu
+
+        result = run_gpu(
+          facts,
+          output / name,
+          plan=plan,
+          jobs=jobs,
+          timeout=timeout,
+          warmups=warmups,
+          repeats=repeats,
+        )
+      if result['status'] != 'passed':
+        raise RuntimeError(f'Backend did not complete: {result["status"]}')
+      if set(result['relation_counts']) != expected or set(result['outputs']) != expected - inputs:
+        raise ValueError('Backend did not report/export every query relation')
+      if result['source_sha256'] != digest(REPO / 'examples' / 'doop.py') or result[
+        'metadata_sha256'
+      ] != digest(facts / 'meta.json'):
+        raise ValueError('Backend used a different query or dataset metadata')
+      result.update(
+        tier=dataset['tier'],
+        reference_vpt_rows=dataset['reference_vpt_rows'],
+        expected_relations=sorted(expected),
+        input_relations=sorted(inputs),
+        input_manifest_sha256=digest(facts / 'manifest.json'),
+        input_rows=sum(manifest['relations'][name]['rows'] for name in inputs),
+        input_bytes=sum(manifest['relations'][name]['bytes'] for name in inputs),
+        prepared_input_rows=sum(r['rows'] for r in manifest['relations'].values()),
+        prepared_input_bytes=sum(r['bytes'] for r in manifest['relations'].values()),
+        correctness='not_compared',
+      )
+      if reference is not None:
+        try:
+          comparison = compare_results(references[name], result, output / name / 'comparison.json')
+        except Exception as error:
+          result['status'] = 'failed'
+          result['error'] = f'Reference comparison failed: {error!r}'
+        else:
+          result['correctness'] = 'exact_match' if comparison['passed'] else 'mismatch'
+          if not comparison['passed']:
+            result['status'] = 'failed'
+      report['results'].append(result)
+      print(
+        json.dumps(
+          {
+            'dataset': name,
+            'status': result['status'],
+            'correctness': result['correctness'],
+            'timings_seconds': result['timings_seconds'],
+          }
+        ),
+        flush=True,
+      )
+    except Exception as error:
+      report['results'].append(
+        {
+          'dataset': name,
+          'tier': dataset['tier'],
+          'backend': backend,
+          'plan': plan,
+          'status': 'failed',
+          'error': repr(error),
+        }
+      )
+      print(f'[failed] {name}: {error}', file=sys.stderr, flush=True)
+    save()
+  report['status'] = (
+    'passed' if all(r['status'] == 'passed' for r in report['results']) else 'failed'
+  )
+  save()
+  return report
diff --git a/examples/run_doop_suite.py b/examples/run_doop_suite.py
new file mode 100755
index 0000000..49b4716
--- /dev/null
+++ b/examples/run_doop_suite.py
@@ -0,0 +1,64 @@
+#!/usr/bin/env python3
+"""Run selected prepared DOOP datasets to complete fixedpoints and export exact results."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from pathlib import Path
+
+from doop_suite.catalog import load_catalog, select_datasets
+from doop_suite.suite import run_matrix
+
+
+def main(argv: list[str] | None = None) -> int:
+  parser = argparse.ArgumentParser(description=__doc__)
+  selection = parser.add_mutually_exclusive_group(required=True)
+  selection.add_argument('--all', action='store_true')
+  selection.add_argument('--dataset', nargs='+', metavar='NAME')
+  selection.add_argument('--tier', choices=list(load_catalog()['tiers']))
+  parser.add_argument(
+    '--root', type=Path, required=True, help='Data root used by doop_benchmark.py prepare'
+  )
+  parser.add_argument('--output', type=Path, required=True, help='New external result directory')
+  parser.add_argument('--backend', choices=['cpu', 'gpu'], required=True)
+  parser.add_argument('--plan', choices=['baseline', 'bitmap'], default='baseline')
+  parser.add_argument('--threads', type=int, default=12, help='CPU execution threads')
+  parser.add_argument('--jobs', type=int, default=2, help='GPU compile jobs')
+  parser.add_argument(
+    '--timeout', type=int, default=900, help='Per build/execution process timeout in seconds'
+  )
+  parser.add_argument('--warmups', type=int, default=1)
+  parser.add_argument('--repeats', type=int, default=3)
+  parser.add_argument(
+    '--reference',
+    type=Path,
+    help='Reference suite.json; require every selected relation set to match',
+  )
+  args = parser.parse_args(argv)
+  try:
+    report = run_matrix(
+      select_datasets(args.dataset, args.tier),
+      args.root,
+      args.output,
+      backend=args.backend,
+      plan=args.plan,
+      threads=args.threads,
+      jobs=args.jobs,
+      timeout=args.timeout,
+      warmups=args.warmups,
+      repeats=args.repeats,
+      reference=args.reference,
+    )
+  except (OSError, ValueError, RuntimeError) as error:
+    print(f'[error] {error}', file=sys.stderr)
+    return 1
+  print(
+    json.dumps({'status': report['status'], 'report': str(args.output.resolve() / 'suite.json')})
+  )
+  return 0 if report['status'] == 'passed' else 1
+
+
+if __name__ == '__main__':
+  raise SystemExit(main())
diff --git a/src/srdatalog/build.py b/src/srdatalog/build.py
index 621abeb..b8e8c30 100644
--- a/src/srdatalog/build.py
+++ b/src/srdatalog/build.py
@@ -113,6 +113,7 @@ def build_project(
       project_name,
       cr.hir.relation_decls,
       _extract_count_result_relations(cr.mir),
+      canonical_indices=cr.canonical_indices,
     )
 
   # In unity mode we want no jit_batch_*.cpp files — they'd be
diff --git a/src/srdatalog/ir/codegen/cuda/main_file.py b/src/srdatalog/ir/codegen/cuda/main_file.py
index add93d7..52df225 100644
--- a/src/srdatalog/ir/codegen/cuda/main_file.py
+++ b/src/srdatalog/ir/codegen/cuda/main_file.py
@@ -736,10 +736,102 @@ def gen_unity_main_file_content(
   return out
 
 
+def _gen_relation_export_helper() -> str:
+  '''Bounded host-side TSV export without rebuilding an index from stale columns.'''
+  return r'''
+#include <array>
+#include <fstream>
+#include <locale>
+#include <type_traits>
+
+namespace {
+inline void srdatalog_check_gpu(GPU_ERROR_T status) {
+  if (status != GPU_SUCCESS)
+    throw std::runtime_error(GPU_GET_ERROR_STRING(status));
+}
+
+template<class Schema, class DB>
+void srdatalog_write_tsv(DB& db, const char* path, const SRDatalog::IndexSpec& canonical) {
+  using Attrs = typename Schema::attr_ts_type;
+  constexpr std::size_t arity = std::tuple_size_v<Attrs>;
+  auto& relation = get_relation_by_schema<Schema, FULL_VER>(db);
+  using Rel = std::remove_reference_t<decltype(relation)>;
+  using Value = typename Rel::value_type;
+  constexpr bool integer_columns = []<std::size_t... I>(std::index_sequence<I...>) {
+    return ((std::is_integral_v<std::tuple_element_t<I, Attrs>> &&
+             sizeof(std::tuple_element_t<I, Attrs>) <= sizeof(Value)) && ...);
+  }(std::make_index_sequence<arity>{});
+  constexpr bool column_index = requires(typename Rel::IndexTypeInst& index) {
+    index.data().view().column_ptr(0);
+  };
+  if constexpr (!integer_columns || !column_index ||
+                Rel::Layout != SRDatalog::GPU::StorageLayout::SoA) {
+    throw std::runtime_error("TSV export requires integer-valued SoA column indexes");
+  } else {
+    // Index maintenance can leave intern columns stale. Borrow authoritative
+    // index storage and undo its column permutation without a second GPU copy.
+    std::array<const Value*, arity> device_columns{};
+    std::size_t rows = relation.size();
+    if (!canonical.cols.empty()) {
+      if (canonical.cols.size() != arity || !relation.has_index(canonical))
+        throw std::runtime_error("TSV export: missing canonical index");
+      auto& index = relation.get_index(canonical);
+      // Two-level indexes expose only FULL through data(); consolidate HEAD
+      // before borrowing it. Fixedpoint-exit reconstruction normally did this.
+      if constexpr (requires { index.compact(); }) index.compact();
+      srdatalog_check_gpu(GPU_DEVICE_SYNCHRONIZE());
+      rows = index.size();
+      if (rows) {
+        const auto view = index.data().view();
+        std::array<bool, arity> seen{};
+        for (std::size_t position = 0; position < arity; ++position) {
+          const auto logical = canonical.cols[position];
+          if (logical < 0 || logical >= arity || seen[logical])
+            throw std::runtime_error("TSV export: invalid canonical permutation");
+          seen[logical] = true;
+          device_columns[logical] = view.column_ptr(position);
+        }
+      }
+    } else if (rows) {
+      const auto view = relation.unsafe_interned_columns().view();
+      for (std::size_t column = 0; column < arity; ++column)
+        device_columns[column] = view.column_ptr(column);
+    }
+    std::ofstream output;
+    output.exceptions(std::ios::failbit | std::ios::badbit);
+    output.imbue(std::locale::classic());
+    output.open(path, std::ios::out | std::ios::trunc);
+    constexpr std::size_t chunk_rows = 65536;
+    std::array<std::vector<Value>, arity> host;
+    for (auto& column : host) column.resize(std::min(rows, chunk_rows));
+    for (std::size_t first = 0; first < rows; first += chunk_rows) {
+      const auto count = std::min(chunk_rows, rows - first);
+      for (std::size_t column = 0; column < arity; ++column) {
+        srdatalog_check_gpu(GPU_MEMCPY(
+            host[column].data(), device_columns[column] + first,
+            count * sizeof(Value), GPU_DEVICE_TO_HOST));
+      }
+      for (std::size_t row = 0; row < count; ++row) {
+        [&]<std::size_t... I>(std::index_sequence<I...>) {
+          ((output << (I ? "\t" : "")
+                   << +static_cast<std::tuple_element_t<I, Attrs>>(host[I][row])), ...);
+        }(std::make_index_sequence<arity>{});
+        output << '\n';
+      }
+    }
+    output.close();
+  }
+}
+}  // namespace
+'''
+
+
 def gen_extern_c_shim(
   ruleset_name: str,
   decls: list[RelationDecl],
   count_relations: list[str] | None = None,
+  *,
+  canonical_indices: dict[str, list[int]] | None = None,
 ) -> str:
   '''Emit an `extern "C"` shim the Python ctypes loader can call.
 
@@ -749,13 +841,20 @@ def gen_extern_c_shim(
     - `srdatalog_run(max_iters)`             — copy-to-device + _Runner::run
     - `srdatalog_shutdown()`                 — free host + device DB
     - `srdatalog_size(rel_name)`             — count result or device relation size
+    - `srdatalog_synchronize()`              — checked device synchronization
+    - `srdatalog_get_size(rel_name, out)`    — checked authoritative cardinality
+    - `srdatalog_export_tsv(rel_name, path)` — integer tuples in logical column order
 
   The shim uses a file-scope `HostDB*` holding the live SemiNaiveDatabase
   so Python can stage data via multiple `load_csv` calls before `run`. It
   retains the post-run device DB because computed results are not copied
   back to the host DB.
+  The checked cardinality/export APIs use the compiler's canonical index map.
+  Export supports integer-valued SoA relations, rejects count-only results, and
+  requires a completed run. The caller owns the destination file/directory.
   '''
   count_relations = count_relations or []
+  canonical = canonical_indices or {}
   ext_db = f"{ruleset_name}_DB"
   blueprint = f"{ext_db}_Blueprint"
   host_db = f"{blueprint}_HostDB"
@@ -764,6 +863,7 @@ def gen_extern_c_shim(
   out = [
     "// ======== Python ctypes shim (extern \"C\") ========",
     '#include "gpu/init.h"  // SRDatalog::GPU::init_cuda',
+    _gen_relation_export_helper(),
     "",
     f"using {host_db} = SRDatalog::AST::SemiNaiveDatabase<{blueprint}>;",
     f"static {host_db}* g_host_db = nullptr;",
@@ -781,10 +881,11 @@ def gen_extern_c_shim(
     "",
     "int srdatalog_load_csv(const char* rel_name, const char* path) {",
     "  if (!rel_name || !path) return 1;",
-    f"  if (!g_host_db) g_host_db = new {host_db}();",
     "  try {",
+    f"    if (!g_host_db) g_host_db = new {host_db}();",
     "    std::string rn(rel_name);",
   ]
+  # Allocate inside the exception boundary so allocation failure cannot cross C.
   # Only emit load_from_file dispatch for relations marked with
   # `input_file=...`. Non-input relations (pure IDB, or relations
   # with non-default index types like Device2LevelIndex) fail to
@@ -819,8 +920,8 @@ def gen_extern_c_shim(
     "// Matches the Nim driver pattern: single call, relies on _Runner::load_data.",
     "int srdatalog_load_all(const char* data_dir) {",
     "  if (!data_dir) return 1;",
-    f"  if (!g_host_db) g_host_db = new {host_db}();",
     "  try {",
+    f"    if (!g_host_db) g_host_db = new {host_db}();",
     f"    {ruleset_name}_Runner::load_data(*g_host_db, std::string(data_dir));",
     "    return 0;",
     "  } catch (const std::exception& e) {",
@@ -836,7 +937,7 @@ def gen_extern_c_shim(
   ]
   out += [
     f"    g_device_db = new {device_db}(SRDatalog::GPU::copy_host_to_device(*g_host_db));",
-    f"    {ruleset_name}_Runner::run(*g_device_db, max_iters ? (std::size_t)max_iters : std::numeric_limits<int>::max());",
+    f"    {ruleset_name}_Runner::run(*g_device_db, max_iters ? (std::size_t)max_iters : std::numeric_limits<std::size_t>::max());",
     "    return 0;",
     "  } catch (const std::exception& e) {",
     '    std::cerr << "srdatalog_run: " << e.what() << std::endl;',
@@ -873,10 +974,73 @@ def gen_extern_c_shim(
     "  return 0;",
     "}",
     "",
+    "int srdatalog_synchronize() {",
+    "  return (int)GPU_DEVICE_SYNCHRONIZE();",
+    "}",
+    "",
+    "int srdatalog_get_size(const char* rel_name, unsigned long long* result) {",
+    "  if (!rel_name || !result || !g_device_db) return 1;",
+    "  try {",
+    "    std::string rn(rel_name);",
+  ]
+  if count_relations:
+    out.append(f"    if ({ruleset_name}_Runner::get_count_result(rn, *result)) return 0;")
+  for d in decls:
+    out.append(f'    if (rn == "{d.rel_name}") {{')
+    out.append(f'      auto& rel = get_relation_by_schema<{d.rel_name}, FULL_VER>(*g_device_db);')
+    if d.rel_name in canonical:
+      cols = ", ".join(str(c) for c in canonical[d.rel_name])
+      out.append(f"      SRDatalog::IndexSpec spec{{{cols}}};")
+      out.append('      if (!rel.has_index(spec)) return 2;')
+      out.append('      *result = (unsigned long long)rel.get_index(spec).size();')
+    else:
+      out.append('      *result = (unsigned long long)rel.size();')
+    out += ["      return 0;", "    }"]
+  out += [
+    "    return 2;",
+    "  } catch (const std::exception& e) {",
+    '    std::cerr << "srdatalog_get_size: " << e.what() << std::endl;',
+    "    return 3;",
+    "  } catch (...) { return 4; }",
+    "}",
+    "",
+    "int srdatalog_export_tsv(const char* rel_name, const char* path) {",
+    "  if (!rel_name || !path || !g_device_db) return 1;",
+    "  try {",
+    "    srdatalog_check_gpu(GPU_DEVICE_SYNCHRONIZE());",
+    "    std::string rn(rel_name);",
+  ]
+  for d in decls:
+    out.append(f'    if (rn == "{d.rel_name}") {{')
+    if d.rel_name in count_relations or d.count_only:
+      out.append("      return 2; // A count-only result has no materialized tuples.")
+    else:
+      cols = ", ".join(str(c) for c in canonical.get(d.rel_name, []))
+      out.append(
+        f"      srdatalog_write_tsv<{d.rel_name}>(*g_device_db, path, "
+        f"SRDatalog::IndexSpec{{{cols}}});"
+      )
+      out.append("      return 0;")
+    out.append("    }")
+  out += [
+    "    return 2;",
+    "  } catch (const std::exception& e) {",
+    '    std::cerr << "srdatalog_export_tsv: " << e.what() << std::endl;',
+    "    return 3;",
+    "  } catch (...) { return 4; }",
+    "}",
+    "",
     "int srdatalog_shutdown() {",
-    "  if (g_device_db) { delete g_device_db; g_device_db = nullptr; }",
-    "  if (g_host_db) { delete g_host_db; g_host_db = nullptr; }",
-    "  return 0;",
+    "  try {",
+    "    srdatalog_check_gpu(GPU_DEVICE_SYNCHRONIZE());",
+    "    if (g_device_db) { delete g_device_db; g_device_db = nullptr; }",
+    "    if (g_host_db) { delete g_host_db; g_host_db = nullptr; }",
+    "    srdatalog_check_gpu(GPU_DEVICE_SYNCHRONIZE());",
+    "    return 0;",
+    "  } catch (const std::exception& e) {",
+    '    std::cerr << "srdatalog_shutdown: " << e.what() << std::endl;',
+    "    return 1;",
+    "  } catch (...) { return 2; }",
     "}",
     "",
     "}  // extern \"C\"",
diff --git a/tests/test_doop_compare.py b/tests/test_doop_compare.py
new file mode 100644
index 0000000..f05dbdb
--- /dev/null
+++ b/tests/test_doop_compare.py
@@ -0,0 +1,45 @@
+"""Equal cardinalities do not establish relation equality; row order is immaterial."""
+
+import sys
+from pathlib import Path
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'examples'))
+from doop_suite.compare import compare_results
+
+
+def report(path):
+  return {
+    'status': 'passed',
+    'dataset': 'sample',
+    'source_sha256': 'same-source',
+    'metadata_sha256': 'same-meta',
+    'input_manifest_sha256': 'same-inputs',
+    'expected_relations': ['Input', 'Output'],
+    'input_relations': ['Input'],
+    'relation_counts': {'Input': 1, 'Output': 2},
+    'outputs': {'Output': str(path)},
+  }
+
+
+def test_comparison_checks_tuples_not_cardinality_or_export_order(tmp_path):
+  left, right = tmp_path / 'left.tsv', tmp_path / 'right.tsv'
+  left.write_text('1\t2\n3\t4\n')
+  right.write_text('3\t4\n1\t2\n')
+  assert compare_results(report(left), report(right), tmp_path / 'equal.json')['passed']
+  right.write_text('3\t4\n1\t9\n')
+  assert not compare_results(report(left), report(right), tmp_path / 'different.json')['passed']
+
+
+def test_comparison_rejects_missing_exports_and_different_inputs(tmp_path):
+  path = tmp_path / 'out.tsv'
+  path.write_text('1\t2\n3\t4\n')
+  incomplete = report(path)
+  incomplete['outputs'] = {}
+  with pytest.raises(ValueError, match='every relation'):
+    compare_results(report(path), incomplete, tmp_path / 'missing.json')
+  changed = report(path)
+  changed['input_manifest_sha256'] = 'different-input'
+  with pytest.raises(ValueError, match='input_manifest_sha256'):
+    compare_results(report(path), changed, tmp_path / 'changed.json')
diff --git a/tests/test_doop_gpu_process.py b/tests/test_doop_gpu_process.py
new file mode 100644
index 0000000..b95432f
--- /dev/null
+++ b/tests/test_doop_gpu_process.py
@@ -0,0 +1,60 @@
+"""A native report is not success until the fresh GPU worker exits cleanly."""
+
+import importlib.util
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+_SPEC = importlib.util.spec_from_file_location(
+  "doop_gpu_runner", Path(__file__).parents[1] / "examples" / "doop_suite" / "gpu.py"
+)
+assert _SPEC is not None and _SPEC.loader is not None
+gpu = importlib.util.module_from_spec(_SPEC)
+_SPEC.loader.exec_module(gpu)
+
+
+def test_worker_failure_after_writing_report_is_not_success(tmp_path):
+  report = tmp_path / "native.json"
+  command = [
+    sys.executable,
+    "-c",
+    "import pathlib,sys; pathlib.Path(sys.argv[1]).write_text("
+    "'{\"status\":\"native_completed\"}'); print('teardown failed', flush=True); sys.exit(17)",
+    str(report),
+  ]
+  log = tmp_path / "worker.log"
+  with pytest.raises(RuntimeError, match="exited 17"):
+    gpu._run_process(command, log, timeout=10)
+  assert report.read_text() == '{"status":"native_completed"}'
+  assert "teardown failed" in log.read_text()
+
+
+def test_worker_timeout_is_an_error_with_preserved_diagnostics(tmp_path):
+  log = tmp_path / "timeout.log"
+  command = [
+    sys.executable,
+    "-c",
+    "import time; print('entered native fixedpoint', flush=True); time.sleep(60)",
+  ]
+  with pytest.raises(subprocess.TimeoutExpired):
+    gpu._run_process(command, log, timeout=2)
+  assert "entered native fixedpoint" in log.read_text()
+
+
+def test_failed_worker_keeps_relative_artifacts_outside_checkout(tmp_path, monkeypatch):
+  checkout, output = tmp_path / "checkout", tmp_path / "output"
+  checkout.mkdir()
+  output.mkdir()
+  monkeypatch.setattr(gpu, "_ROOT", checkout)
+  command = [
+    sys.executable,
+    "-c",
+    "from pathlib import Path; Path('allocator_failure.log').write_text('pool exhausted'); "
+    "raise SystemExit(17)",
+  ]
+  with pytest.raises(RuntimeError, match="exited 17"):
+    gpu._run_process(command, output / "worker.log", timeout=10)
+  assert (output / "allocator_failure.log").read_text() == "pool exhausted"
+  assert not (checkout / "allocator_failure.log").exists()
diff --git a/tests/test_doop_oracle.py b/tests/test_doop_oracle.py
new file mode 100644
index 0000000..40b3ab7
--- /dev/null
+++ b/tests/test_doop_oracle.py
@@ -0,0 +1,86 @@
+"""Semantic checks for the conservative canonical-Program CPU translation."""
+
+import shutil
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+from srdatalog.dsl import SPLIT, Const, Filter, Program, Relation, Var
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "examples"))
+from doop_suite.cpu import translate_program
+
+
+def test_translation_preserves_recursive_multihead_filter_negation_and_split(tmp_path):
+  souffle = shutil.which("souffle")
+  if souffle is None:
+    pytest.skip("Souffle is required to execute the translated logical program")
+  x, y, z = Var("x"), Var("y"), Var("z")
+  seed = Relation("Seed", 2, input_file="Seed.csv")
+  blocked = Relation("Blocked", 2, input_file="Blocked.csv")
+  forward = Relation("Forward", 2)
+  reverse = Relation("Reverse", 2)
+  from_one = Relation("FromOne", 1)
+  program = Program(
+    rules=[
+      (
+        (forward(x, y) | reverse(y, x))
+        <= seed(x, y)
+        & SPLIT
+        & ~blocked(x, Var("_"))
+        & Filter(("x", "y"), "return x != -1 && y != 8;")
+      ).with_plan(var_order=["y", "x"]),
+      forward(x, z) <= forward(x, y) & forward(y, z),
+      from_one(y) <= forward(x, y) & Filter(("x",), "return x == 1;"),
+    ]
+  )
+  text, _ = translate_program(program)
+  source = tmp_path / "semantic.dl"
+  source.write_text(text, encoding="utf-8")
+  (tmp_path / "Seed.csv").write_text("1\t2\n2\t3\n3\t4\n5\t6\n1\t2\n7\t8\n-1\t0\n")
+  (tmp_path / "Blocked.csv").write_text("3\t99\n5\t1\n")
+  outputs = tmp_path / "outputs"
+  outputs.mkdir()
+  subprocess.run(
+    [souffle, "-F", str(tmp_path), "-D", str(outputs), str(source)],
+    check=True,
+    capture_output=True,
+    text=True,
+    timeout=30,
+  )
+
+  def tuples(name):
+    rows = (outputs / f"{name}.tsv").read_text().splitlines()
+    parsed = [tuple(map(int, row.split("\t"))) for row in rows]
+    assert len(parsed) == len(set(parsed)), "The oracle must export set, not bag, results"
+    return set(parsed)
+
+  assert tuples("Forward") == {(1, 2), (2, 3), (1, 3)}
+  assert tuples("Reverse") == {(2, 1), (3, 2)}
+  assert tuples("FromOne") == {(2,), (3,)}
+
+
+@pytest.mark.parametrize(
+  "code",
+  [
+    "return x != 1 || x != 2;",  # Disjunction must not be silently treated as conjunction.
+    "return x == 010;",  # C++ octal is not a decimal identifier.
+    "return x == 2147483648;",  # GPU's integer domain is signed int32.
+  ],
+)
+def test_unsupported_filter_semantics_are_rejected(code):
+  x = Var("x")
+  seed = Relation("Seed", 1, input_file="Seed.csv")
+  result = Relation("Result", 1)
+  with pytest.raises(ValueError):
+    translate_program(Program(rules=[result(x) <= seed(x) & Filter(("x",), code)]))
+
+
+def test_constant_cpp_expression_cannot_override_metadata_literal():
+  x = Var("x")
+  seed = Relation("Seed", 1, input_file="Seed.csv")
+  result = Relation("Result", 2)
+  with pytest.raises(ValueError):
+    translate_program(Program(rules=[result(x, Const(1, cpp_expr="2")) <= seed(x)]))