diff --git a/.claude/board/STATUS_BOARD.md b/.claude/board/STATUS_BOARD.md
index 92a317e2f..c8fc132dd 100644
--- a/.claude/board/STATUS_BOARD.md
+++ b/.claude/board/STATUS_BOARD.md
@@ -6,7 +6,8 @@ Plans in #1667 (`d-rft-relalg-thought-git-frontier-v1.md`, `d-qsp-classview-quac
 |---|---|---|---|
 | **D-RFT-0 / D-QSP-0** | K3 normalization safety: rewrites keep SQL three-valued TRUE rows | Shipped (#1668; EXO-50 claim `session_01G22yT6htkcdyXsihxxXdrg/031`; `entries/2026-10-21-d-rft-0-kleene-normalization.md`): `normalize_kleene`; `SetNormalize` admits it under `ThreeValued`. MEASURED: PostgreSQL 16.14 = Kleene oracle = `normalize_kleene` on 2000/2000; the two-valued form is wrong on 474 | Kleene evaluator over random expressions + PostgreSQL differential; 5 disables red. Remainder open: NULL literals, `COALESCE`, nullable arithmetic, empty `SUM`, `COUNT(expr)` |
 | **D-QSP-0 remainder, slice 1** | scalar aggregates under SQL NULL: empty / all-NULL `SUM` is NULL, `COUNT(x)` vs `COUNT(*)`, MIN/MAX/AVG | In PR (EXO-50 claim `session_013z2e1YXkm6o6HNrDn5BPWA/026`; `entries/2026-10-21-d-qsp-0-scalar-aggregates-null.md`): Quack `lower_sum` + `sum_finish` over existing terminals. MEASURED: PostgreSQL 16.15 = Quack = Kleene oracle on 3600/3600 queries | `tests/sql_aggregate_null.rs` (4 disables red); `examples/qsp_agg_pg_parity.rs` (SKIP without a server) |
-| **D-RFT-1..5, D-LFA-0** | composition, substitution, recursion, frontier, replay; flow census | PROPOSAL (`entries/2026-10-21-d-rft-d-lfa-0-assessment.md`) | per-row falsifiers in the entry; unclaimed |
+| **D-RFT-1** | typed composition: membership × role carriers over one slot coordinate; Distinct/Bag from the cardinality witness; NULL key matches nothing; XOR, seal mismatch, stale binding, bag overflow refused | In PR, PROBE (EXO-50 claim `session_01G22yT6htkcdyXsihxxXdrg/036`; `crates/dir-rmap-probe/src/compose.rs`; `entries/2026-10-21-d-rft-1-typed-composition.md`). r2il polyfill: `IntXor`, `ANY`, `ALL` + procedures. MEASURED: PostgreSQL 16.14 lane equal over 600 users × 64 roles, bag 1421 = `count(*)`, distinct 1374 = `count(DISTINCT)`; composition 310 µs vs the same plan in direct SIMD 14 µs | IAM diamond (Bag 2, Distinct 1); independent DFS oracle lane-equal; 7 compose + 5 r2il disables red |
+| **D-RFT-2..5, D-LFA-0** | substitution, recursion, frontier, replay; flow census | PROPOSAL (`entries/2026-10-21-d-rft-d-lfa-0-assessment.md`) | per-row falsifiers in the entry; unclaimed |
 | **D-LFA-1 / D-LFA-2** (amended 2026-10-21) | guarded-control verification of a Rubicon flow; integration with the existing drivers | PROPOSAL (plan amendment, `d-lfa-…-v1.md` § Amendment 2026-10-21; map `entries/2026-10-21-coresearch-plans-mapping-foresight.md`) | GKAT equivalence + separate no-commit-without-revision-and-MUL trace check; 3 disables; `GoTo`/`GoBack` refused; unclaimed |
 | **D-GKAT-0** | GKAT fixture probe: G1 factoring, G2 nondeterminism refusal, G5 declared atom bound | In PR, PROBE (EXO-50 claim `session_01VncKoyG6JkpsoqhqTGTv3f/026`; `crates/gkat-probe`; `entries/2026-10-21-d-gkat-0-probe.md`). TEST-PINNED: Rubicon reference ≡ optimized, both safe; CHECK-authorized flows equivalent and unsafe; ≤ 8 tests (256 atoms) else refused | 3 flow disables + 11 checker disables red; independent bounded oracle; not a production verifier |
 
diff --git a/.claude/board/entries/2026-10-21-d-rft-1-typed-composition.md b/.claude/board/entries/2026-10-21-d-rft-1-typed-composition.md
new file mode 100644
index 000000000..b0b38b711
--- /dev/null
+++ b/.claude/board/entries/2026-10-21-d-rft-1-typed-composition.md
@@ -0,0 +1,101 @@
+# 2026-10-21 — D-RFT-1-COMPOSE: typed composition over the DR256 slot coordinate (D-RFT-1)
+
+**Status:** PROBE (lab, `crates/dir-rmap-probe`). TEST-PINNED and MEASURED as stated below; no production ABI changes.
+**Claim:** EXO-50 `session_01G22yT6htkcdyXsihxxXdrg/036`, status in `/038`.
+**Plan:** `.claude/plans/d-rft-relalg-thought-git-frontier-v1.md` (D-RFT-1); assessment `entries/2026-10-21-d-rft-d-lfa-0-assessment.md`.
+
+## What landed
+
+**`dir-rmap-probe::compose`.** It composes user → group membership (`Page.member`) with the role carriers (`Qr.role_slots`, and `Qr.role` for the slot-major direction).
+- Both operands share one coordinate, the 256 slots, and one `Witness`.
+- Per coordinate (user `i`, role `r`), the path mask is `member_i ∧ valid_i ∧ carriers_r`.
+- Bag is the path count and Distinct is its existence. They are two readings of one computation, never two computations.
+
+**Laws, each TEST-PINNED with a disable that goes red:**
+
+| law | how |
+|---|---|
+| Stratum comes from a witness, not the caller | Distinct when the left fan (present keys per user) or the right fan (carrier slots per role) is ≤ 1, else Bag |
+| A NULL key matches nothing | a key whose payload bit is set but whose presence bit is clear contributes no path (the same model as Quack's validity plane) |
+| XOR is refused | an even number of parallel paths would read 0, an answer neither stratum gives |
+| Seal mismatch is refused | the two operands were bound to different registers |
+| A stale binding is refused | the caller's current `Witness` differs from the binding's |
+| A page answers only for its own binding | `Page` carries the `Witness` it was resolved against; an old page beside a newly bound register is a seal mismatch |
+| The bag sum is checked | `add_bag` refuses a wrap |
+| The IAM diamond | two direct groups reaching one role: Bag 2, Distinct 1 |
+
+**Execution.** Every count, existence and mask equality is an r2il procedure. Each is compiled once by `FoldDialect` and replayed through mask-risc `execute_into` over zero-copy row slices (a prepared statement):
+
+| step | procedure | rows |
+|---|---|---|
+| present keys | `keep(intersect(member_i, valid_i))` | 256 slots |
+| support (the user's roles) | `keep(union(role[s] for s in present_i))` | 64 roles |
+| path count, support coordinates only | `count_and(member_i, valid_i, carriers_r)` | 256 slots |
+| distinct | `count` over the support plane | `users × 64` |
+
+Coordinates outside the support have no path by construction and are never evaluated. The probe has no counting loop of its own. The two pointer tables, of present slots and of supported roles, are the inversion from a plane to its local positions.
+
+**r2il polyfill** (`lance-graph-r2il`):
+- `IntXor` (R2IL ordinal 22) lowers to `MaskOp::Xor`, or to `BoolExpr::Xor` when fused.
+- `ANY` and `ALL` lower to `Terminal::Any`/`All` and push 1 or 0.
+- New procedures: `exists`, `all`, `differ` (`IntXor` then `ANY`, mask equality as a fold), `intersect`, `union`, `keep`, `count_and`.
+- 5 r2il disables go red.
+
+## Evidence
+
+- **Independent oracle.** A depth-first walk over the `Prepared` CSRs (nesting restricted to security groups), not `Qr`'s Warshall closure. It is compared lane to lane over every page of `Spec::small()`, with supports compared by `differ`. Anti-vacuity: coordinates with count 1 and with count ≥ 2 both occur.
+- **PostgreSQL 16.14** (`examples/rft1_pg_parity.rs`, with a recursive CTE): the per-coordinate lane is equal over 600 users × 64 roles; bag 1421 = `count(*)` and distinct 1374 = `count(DISTINCT (u, role))`. Every 7th user also has a membership row with a NULL group, which matches nothing in PostgreSQL either.
+- **Disables.** 7 on compose (valid plane ignored, stratum forced to Bag, seal check, stale check, unchecked bag, support loop truncated, union dropping keys) plus 5 on r2il. All red.
+
+## Measured (`examples/rft1_bench.rs`, release, median of 101 runs, 600 users / 3 pages)
+
+| step / arm | time | allocations per run |
+|---|---|---|
+| bind (`Qr::bind`) | 46 µs | |
+| pages | 39 µs | |
+| program prep (`Composer::new`) | 1.4 µs | |
+| A — composition (r2il procedures through mask-risc) | 310 µs | 39 |
+| C — every coordinate, direct `ndarray::simd` | 131 µs | 0 |
+| C+ — A's support-first plan, direct `ndarray::simd` | 14 µs | 0 |
+| B — tuple baseline, LAB ONLY, not a candidate | 134–146 µs | 10 |
+
+All arms agree on (bag, distinct) before anything is timed.
+
+**History:** the first wiring ran one r2il fold per (user, role) coordinate and took 17 ms with 78 210 allocations. The fold count was the cause, not the algebra.
+
+## Open
+
+- **Per-execution cost.** A − C+ is about 296 µs for about 2 600 executions, roughly 115 ns of fixed cost per mask-risc `execute_into` on 4-word planes. The plan and the answers are the same, so the gap is execution overhead. UNKNOWN which part dominates (validation, scratch carving, tile loop); not profiled.
+- **No count-product terminal.** mask-risc has no terminal that produces many coordinates' counts in one execution (a bit-matrix count product, or a masked positional popcount). The support-first plan is what keeps the execution count down. ndarray has no positional popcount either.
+- **Fused wide chains.** `union`/`intersect` reuse two scratch slots in `BoolMode::InPlace` (`call_owned`), so a user with any number of present groups composes. In `BoolMode::Fused` a 24-leaf chain is refused (`ScratchBufferTooSmall`): mask-risc's `fuse` gives each ternlog group its own slot. Pinned as a refusal; not changed here.
+- **Fixture scale.** `Spec::iam(8000)` exceeds 256 security slots (`Refusal::TooLarge`), so the bench runs `Spec::small()`.
+- **Not done:** recursion over composed strata (D-RFT-3), a second composition law beyond Count/Exists, and a PostgreSQL arm in CI (the example prints SKIP without a server).
+
+## Update — PostgreSQL primitive polyfill (bingo card, same PR)
+
+Each PostgreSQL primitive is either an r2il opcode the dialect now lowers, or a procedure (a call sequence that adds no opcode). WHERE-shaped primitives leave a mask. Kept with `KEEP`, that mask is a Lance 13 read mask: `with_row_addr_prefilter(RowAddrMask)`, which `crepe-plan-probe` (#1633) measured as scanning exactly the masked rows (P2). Aggregates leave a scalar.
+
+| PostgreSQL | r2il | status |
+|---|---|---|
+| `=`, `<`, `<=`, `>`, `>=` (int) | `IntEqual`, `IntSLess`, `and_cmp_i32` | existing |
+| `IS NULL` / `IS NOT NULL` | `is_null` / `is_not_null` over the presence plane | NEW (proc) |
+| `BETWEEN` (NULL dropped, empty range keeps nothing) | `and_between_i32` | NEW (proc) |
+| `IN (…)`, `IN ()` | `in_u32` | NEW (proc) |
+| `COUNT(*)` / `COUNT(x)` | `count` / `count_valid` | NEW: `count_valid` |
+| `SUM`, `MIN`, `MAX` | `sum`, `min`, `max` | existing |
+| `AVG` | `avg_parts` (SUM and COUNT; `avg(int)` is `numeric`, so the caller divides) | NEW (proc) |
+| `COUNT(DISTINCT k)` on a key-ordered lane | `KEY_RUNS` → `CountKeyRunsU32`; `count_distinct_sorted` | NEW (opcode + proc); an unordered lane is refused |
+| `bool_or` / `bool_and` (`every`) | `bool_or` / `bool_and` | NEW (proc); an empty selection reads 0 / 1 where PostgreSQL reads NULL |
+| `EXISTS` / `NOT EXISTS` | `exists` (`ANY`) / `not_exists` | NEW |
+| `INTERSECT` / `UNION` / `EXCEPT` (same rows) | `intersect` / `union` / `except` | NEW |
+| `UNION ALL` counts | `count_union_all` (`IntAdd`) | NEW |
+| bigint `+`, `*`, `/` on aggregates | `IntAdd`, `IntMult`, `IntSDiv`, checked | NEW (opcodes); overflow and division by zero refused |
+
+Tests: one row-rule reference per primitive (`postgresql_primitives_match_row_rule_references`), the unordered-lane refusal, and the arithmetic refusals. 6 disables, all red: no division-by-zero guard, wrapping add, `COUNT(x)` ignoring presence, exclusive `BETWEEN` upper bound, `EXCEPT` without its NOT, and `KEY_RUNS` counting rows.
+
+**Not on the card** (open):
+- K3 `NOT IN (…, NULL)` and `IS DISTINCT FROM`: Quack `sql_where` holds the (T, F) pair, and r2il procedures are two-valued.
+- `CASE` (`BLEND`) and grouped `COUNT` (`GroupReduce`): not wired.
+- `COUNT(DISTINCT)` on an unordered lane: no seen-set by design.
+- `ORDER BY` / `LIMIT`, string predicates, nullable arithmetic, `COALESCE`.
+- A PostgreSQL differential for the new procedures: the references are row rules, not a server.
diff --git a/.claude/board/entries/README.md b/.claude/board/entries/README.md
index a0d0036e9..cf658f3d7 100644
--- a/.claude/board/entries/README.md
+++ b/.claude/board/entries/README.md
@@ -25,7 +25,7 @@ index row, (3) no duplicate entry id. Checks 1 and 2 are deliberately
 opposite directions; the stranding this convention prevents shows up in
 exactly one of them, never both.
 
-423 entries, 2026-08-06 .. 2026-10-21.
+424 entries, 2026-08-06 .. 2026-10-21.
 
 | date | entry id | finding | file |
 |---|---|---|---|
@@ -42,6 +42,7 @@ exactly one of them, never both.
 | 2026-10-21 | `D-RU-6-LAB-2` |  | [2026-10-21-d-ru-6-sparse-compute.md](2026-10-21-d-ru-6-sparse-compute.md) |
 | 2026-10-21 | `D-RU-6-LAB-3` |  | [2026-10-21-d-ru-6-dirty-frontier.md](2026-10-21-d-ru-6-dirty-frontier.md) |
 | 2026-10-21 | `D-RFT-1` |  | [2026-10-21-d-rft-d-lfa-0-assessment.md](2026-10-21-d-rft-d-lfa-0-assessment.md) |
+| 2026-10-21 | `D-RFT-1-COMPOSE` |  | [2026-10-21-d-rft-1-typed-composition.md](2026-10-21-d-rft-1-typed-composition.md) |
 | 2026-10-21 | `D-RFT-0` |  | [2026-10-21-d-rft-0-kleene-normalization.md](2026-10-21-d-rft-0-kleene-normalization.md) |
 | 2026-10-21 | `D-RA-2` |  | [2026-10-21-d-ra-2-relalg-symbolic-repoint.md](2026-10-21-d-ra-2-relalg-symbolic-repoint.md) |
 | 2026-10-21 | `D-QSP` |  | [2026-10-21-d-qsp-d-rft-d-lfa-classview-quack-plan.md](2026-10-21-d-qsp-d-rft-d-lfa-classview-quack-plan.md) |
diff --git a/.claude/plans/metta-datalog-assimilation-v1-council-v3.md b/.claude/plans/metta-datalog-assimilation-v1-council-v3.md
index d3135c4b3..2374841d0 100644
--- a/.claude/plans/metta-datalog-assimilation-v1-council-v3.md
+++ b/.claude/plans/metta-datalog-assimilation-v1-council-v3.md
@@ -64,7 +64,7 @@ F12. No model identifier in any committed artifact; status words are `MEASURED/V
 | `crates/lance-graph-relalg/Cargo.toml` deps: mask-risc, r2il, ogar-loco, ogar-r2il (git main) | no quack, no contract direct | dependency decision below (R-2) |
 | `crates/lance-graph-quack/src/lib.rs:340` `pub enum Filter` (`EqU32Via` :358, `Semijoin { fk, foreign }` :387; `pub fn lower` :1383; `pub struct Query` :1301); deps mask-risc + contract only | the SHIPPED hop lowering (D-MTP-1 used it: `crates/lance-graph-metta/examples/join_probe.rs:1-12,27,75-79`) | parity oracle for D-ASM-2's `Program` bytes |
 | `crates/lance-graph-mask-risc/src/ir.rs:127-205` `Pred` (incl. `EqU32Via{fk,key,v} :158`, `Range :180`, `EqU32Strided :184`, `MatchFacet16Strided :200`); `:210-287` `MaskOp` (incl. `Gather :277`); `:288-470` `Terminal` (incl. `Count :290`, `Keep :321`, `ScatterOrU32 :338`, `ScatterCountU32 :364`, `CountKeyRunsU32 :389`, `GroupSumI32 :399`, `GroupReduce :442`, `GroupPowerSumsI32 :465`) | the whole word set D-ASM-2/3 may use | F2: nothing added without proof |
-| `crates/lance-graph-r2il/src/lib.rs:260` `pub struct FoldDialect` (`new(planes, foreign)` :327, `with_keep` :366, `last_program` :447, `emit_pred` :589) | the only path relalg takes to a `Program` | gate G4 (byte-identical `Program`) reads `last_program()` |
+| `crates/lance-graph-r2il/src/lib.rs` `pub struct FoldDialect` (`new(planes, foreign)`, `with_keep`, `last_program`, `emit_pred`) | the only path relalg takes to a `Program` | gate G4 (byte-identical `Program`) reads `last_program()` |
 | `crates/lance-graph-metta/Cargo.toml` deps: mask-risc, quack; dev datafrog (fork master); `src/lib.rs` 7 lines (stub); `examples/common/mod.rs:23 Relation2`, `:46 ingest`, `:139 pairs`, `:180 preimage`, `:200 range_pairs`, `:236 oracle`; `examples/join_probe.rs`, `join_range_probe.rs` | D-MTP-1/1b probes, MEASURED; no `Space` impl yet; no hyperon dependency yet | D-ASM-1/4 land here |
 | `crates/lance-graph-nal/src/lib.rs:94` `pub enum Copula` (Inheritance=1, Similarity=2, Implication=3, Equivalence=4; `Statement` :117, `Belief` :130, `guid` :157-170 = `FacetCascade::from_semantic_tiles([class>>16, class, s, copula, p, id, 0, 0])`, `WEIGHT_SCALE/HORIZON/MAX_CONFIDENCE_CODE/evidence classes` :212-244, `NalError` :494-530, 10 variants counted 2026-10-16) | arity-2 fact already laid out as 6 u16 tiles + 2 spare | D-ASM-1 reuses the tile layout for `(f a₁…aₖ)`, k ≤ 5 |
 | `crates/lance-graph-contract/src/nars.rs:12` `pub enum InferenceType` (Deduction, Induction, Abduction, Revision, Synthesis; doc drift: described as CAM search modes) | out of tranche; inventory §6 item | non-goal here |
diff --git a/.github/workflows/dir-rmap-probe.yml b/.github/workflows/dir-rmap-probe.yml
index 7dbdcbdfa..183223b98 100644
--- a/.github/workflows/dir-rmap-probe.yml
+++ b/.github/workflows/dir-rmap-probe.yml
@@ -11,6 +11,8 @@ on:
       - crates/lance-graph-dir-sim/**
       - crates/lance-graph-quack/**
       - crates/lance-graph-mask-risc/**
+      - crates/lance-graph-r2il/**
+      - crates/lance-graph-contract/**
       - .github/workflows/dir-rmap-probe.yml
       - rust-toolchain.toml
   push:
@@ -21,6 +23,8 @@ on:
       - crates/lance-graph-dir-sim/**
       - crates/lance-graph-quack/**
       - crates/lance-graph-mask-risc/**
+      - crates/lance-graph-r2il/**
+      - crates/lance-graph-contract/**
       - .github/workflows/dir-rmap-probe.yml
       - rust-toolchain.toml
   workflow_dispatch:
diff --git a/crates/dir-rmap-probe/Cargo.toml b/crates/dir-rmap-probe/Cargo.toml
index 268a5910b..1dfd28932 100644
--- a/crates/dir-rmap-probe/Cargo.toml
+++ b/crates/dir-rmap-probe/Cargo.toml
@@ -31,6 +31,12 @@ ndarray = { git = "https://github.com/AdaWorldAPI/ndarray", branch = "master", d
 # fixture's own integer ids (never dir-sim ordinals). AdaWorldAPI fork, by
 # branch (no pins), as in crepe-plan-probe.
 crepe = { git = "https://github.com/AdaWorldAPI/crepe", branch = "ccr-d3ce1263-ied9yh" }
+# D-RFT-1 (src/compose.rs): every count, existence and mask equality of the
+# composition is an r2il procedure (`count_and`, `count`, `differ`) folded by
+# mask-risc. The probe holds no counting loop of its own.
+lance-graph-r2il = { path = "../lance-graph-r2il" }
+# The planes a FoldDialect folds over (`Planes`, `Foreign`).
+lance-graph-mask-risc = { path = "../lance-graph-mask-risc" }
 
 # Same patches as lance-graph-dir-sim: one ndarray identity, and the OGAR
 # sibling checkout so a paired OGAR change is tested before it merges.
@@ -43,7 +49,6 @@ lance-graph-contract = { path = "../lance-graph-contract" }
 # a population is lent as a `Rel::Plane` and folded by relalg's existing
 # mask-risc lowering. No evaluator of the probe's own.
 lance-graph-relalg = { path = "../lance-graph-relalg" }
-lance-graph-mask-risc = { path = "../lance-graph-mask-risc" }
 
 [patch."https://github.com/AdaWorldAPI/ndarray"]
 ndarray = { path = "../../../ndarray" }
@@ -53,3 +58,11 @@ ogar-dir-core = { path = "../../../OGAR/crates/ogar-dir-core" }
 ogar-dir-sim = { path = "../../../OGAR/crates/ogar-dir-sim" }
 ogar-ad = { path = "../../../OGAR/crates/ogar-ad" }
 ogar-az = { path = "../../../OGAR/crates/ogar-az" }
+ogar-loco = { path = "../../../OGAR/crates/ogar-loco" }
+ogar-r2il = { path = "../../../OGAR/crates/ogar-r2il" }
+
+# ogar-loco / ogar-r2il take the contract by git `main`; map it onto the
+# in-tree crate so the graph holds one contract (the lance-graph root does
+# the same).
+[patch."https://github.com/AdaWorldAPI/lance-graph"]
+lance-graph-contract = { path = "../lance-graph-contract" }
diff --git a/crates/dir-rmap-probe/examples/rft1_bench.rs b/crates/dir-rmap-probe/examples/rft1_bench.rs
new file mode 100644
index 000000000..af1fdd22f
--- /dev/null
+++ b/crates/dir-rmap-probe/examples/rft1_bench.rs
@@ -0,0 +1,243 @@
+//! D-RFT-1 benchmark: the admitted query (every user's role multiplicities
+//! through direct security groups and role carriers) three ways, over all
+//! pages of one fixture.
+//!
+//! * A: `Composer::compose` (the typed composition: r2il procedures,
+//!   compiled once, replayed through mask-risc; support first, so only
+//!   coordinates that can hold a path are counted);
+//! * C: every coordinate's AND + popcount with `ndarray::simd` primitives on
+//!   the 4-word rows, no support pass (the brute-force SIMD reference);
+//! * C+: arm A's plan (support first) written directly with
+//!   `ndarray::simd`: the gap A − C+ is mask-risc's per-execution cost;
+//! * B: LAB-ONLY tuple baseline: every path materialized as a
+//!   `(user, role, group)` tuple, sorted, runs counted. It exists only to
+//!   show what leaving the fold costs; it is not a candidate.
+//!
+//! Bind, page preparation, program preparation (`Composer::new`) and
+//! execution are timed separately (median of 101 runs, in ns). All three arms must agree on bag and distinct before any timing.
+//! Allocations are counted by a wrapping global allocator.
+//!
+//! ```text
+//! cargo run --release --example rft1_bench
+//! ```
+
+use dir_rmap_probe::World;
+use dir_rmap_probe::compose::{Carriers, Composer, Law, Members, add_bag};
+use dir_rmap_probe::fixture::{Fixture, Spec};
+use dir_rmap_probe::focus::SLOTS;
+use dir_rmap_probe::prep::{Prepared, Refusal, ones};
+use dir_rmap_probe::qr::{Page, Qr};
+use ndarray::simd::{mask_and_assign, mask_or_assign, popcount_batch_u64};
+use std::alloc::{GlobalAlloc, Layout, System};
+use std::hint::black_box;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::time::Instant;
+
+struct Counting;
+static ALLOCS: AtomicUsize = AtomicUsize::new(0);
+static BYTES: AtomicUsize = AtomicUsize::new(0);
+
+// SAFETY: every method forwards unchanged to `System`; the counters are
+// atomics and do not affect the allocation.
+unsafe impl GlobalAlloc for Counting {
+    unsafe fn alloc(&self, l: Layout) -> *mut u8 {
+        ALLOCS.fetch_add(1, Ordering::Relaxed);
+        BYTES.fetch_add(l.size(), Ordering::Relaxed);
+        // SAFETY: same contract as the caller's.
+        unsafe { System.alloc(l) }
+    }
+    unsafe fn dealloc(&self, p: *mut u8, l: Layout) {
+        // SAFETY: `p` came from `System.alloc` with `l`.
+        unsafe { System.dealloc(p, l) }
+    }
+}
+
+#[global_allocator]
+static GLOBAL: Counting = Counting;
+
+fn counters() -> (usize, usize) {
+    (
+        ALLOCS.load(Ordering::Relaxed),
+        BYTES.load(Ordering::Relaxed),
+    )
+}
+
+/// Median wall time in ns of `reps` runs of `f`.
+fn median_ns<T>(reps: usize, mut f: impl FnMut() -> T) -> u128 {
+    let mut t = Vec::with_capacity(reps);
+    for _ in 0..reps {
+        let s = Instant::now();
+        black_box(f());
+        t.push(s.elapsed().as_nanos());
+    }
+    t.sort_unstable();
+    t[t.len() / 2]
+}
+
+/// Allocations and bytes of one run of `f`.
+fn allocs_of<T>(mut f: impl FnMut() -> T) -> (usize, usize) {
+    let (a0, b0) = counters();
+    black_box(f());
+    let (a1, b1) = counters();
+    (a1 - a0, b1 - b0)
+}
+
+fn build(spec: Spec) -> Result<(World, Prepared, Qr), Refusal> {
+    let w = World::new(Fixture::generate(spec));
+    let b = w.prepare();
+    let qr = Qr::bind(&b, &w.witness())?;
+    Ok((w, b, qr))
+}
+
+fn main() {
+    let (w, b, qr) = match build(Spec::iam(8000)) {
+        Ok(x) => {
+            println!("fixture: Spec::iam(8000)");
+            x
+        }
+        Err(Refusal::TooLarge) => {
+            println!(
+                "fixture: Spec::iam(8000) exceeds 256 security slots (Refusal::TooLarge); \
+                 falling back to Spec::small()"
+            );
+            build(Spec::small()).expect("Spec::small fits 256 slots")
+        }
+        Err(e) => panic!("bind refused: {e:?}"),
+    };
+    let users = b.users;
+    let pages: Vec<Page> = (0..qr.pages())
+        .map(|p| qr.page(&b, p).expect("page"))
+        .collect();
+    let counts: Vec<usize> = (0..pages.len())
+        .map(|p| users.saturating_sub(p * SLOTS).min(SLOTS))
+        .collect();
+    let members: Vec<Members<'_>> = pages
+        .iter()
+        .zip(&counts)
+        .map(|(pg, &n)| Members::of(&qr, pg, n))
+        .collect();
+    let carriers = Carriers::of(&qr);
+    let current = w.witness();
+
+    // Arm A: one prepared composer, reused.
+    let mut composer = Composer::new();
+    let mut arm_a = || {
+        let (mut bag, mut distinct) = (0u64, 0u64);
+        for m in &members {
+            let out = composer
+                .compose(m, &carriers, &current, Law::Count)
+                .expect("compose");
+            bag = add_bag(bag, out.bag).expect("no overflow");
+            distinct = add_bag(distinct, out.distinct).expect("no overflow");
+        }
+        (bag, distinct)
+    };
+    // Arm C: direct ndarray::simd on the 4-word rows.
+    let arm_c = || {
+        let (mut bag, mut distinct) = (0u64, 0u64);
+        for (m, &n) in members.iter().zip(&counts) {
+            for i in 0..n {
+                for r in 0..64 {
+                    let mut buf = m.member[i];
+                    mask_and_assign(&mut buf, &m.valid[i]);
+                    mask_and_assign(&mut buf, &carriers.role_slots[r]);
+                    let c = popcount_batch_u64(&buf);
+                    bag += c;
+                    distinct += u64::from(c > 0);
+                }
+            }
+        }
+        (bag, distinct)
+    };
+    // Arm C+: support first, direct ndarray::simd.
+    let arm_cp = || {
+        let (mut bag, mut distinct) = (0u64, 0u64);
+        for (m, &n) in members.iter().zip(&counts) {
+            for i in 0..n {
+                let mut present = m.member[i];
+                mask_and_assign(&mut present, &m.valid[i]);
+                let mut support = [0u64];
+                for s in ones(&present) {
+                    mask_or_assign(&mut support, std::slice::from_ref(&qr.role[s]));
+                }
+                distinct += popcount_batch_u64(&support);
+                for r in ones(&support) {
+                    let mut buf = present;
+                    mask_and_assign(&mut buf, &carriers.role_slots[r]);
+                    bag += popcount_batch_u64(&buf);
+                }
+            }
+        }
+        (bag, distinct)
+    };
+    // Arm B: LAB-ONLY tuple baseline. A heap row per path, then sort.
+    let arm_b = || {
+        let mut paths: Vec<(u32, u8, u16)> = Vec::new();
+        for (p, (m, &n)) in members.iter().zip(&counts).enumerate() {
+            for i in 0..n {
+                let mut present = m.member[i];
+                mask_and_assign(&mut present, &m.valid[i]);
+                for s in ones(&present) {
+                    for r in 0..64u8 {
+                        if qr.role[s] >> r & 1 == 1 {
+                            paths.push(((p * SLOTS + i) as u32, r, s as u16));
+                        }
+                    }
+                }
+            }
+        }
+        paths.sort_unstable();
+        let bag = paths.len() as u64;
+        let mut distinct = 0u64;
+        let mut last = None;
+        for &(u, r, _) in &paths {
+            if last != Some((u, r)) {
+                distinct += 1;
+                last = Some((u, r));
+            }
+        }
+        (bag, distinct)
+    };
+
+    let (a, c, t) = (arm_a(), arm_c(), arm_b());
+    assert_eq!(a, c, "arm A vs arm C: (bag, distinct)");
+    assert_eq!(a, arm_cp(), "arm A vs arm C+: (bag, distinct)");
+    assert_eq!(a, t, "arm A vs arm B: (bag, distinct)");
+    println!(
+        "users {users}, pages {}, bag {}, distinct {} (all three arms agree)",
+        pages.len(),
+        a.0,
+        a.1
+    );
+
+    const REPS: usize = 101;
+    let bind_ns = median_ns(REPS, || Qr::bind(&b, &current).expect("bind"));
+    let prep_ns = median_ns(REPS, || {
+        (0..qr.pages())
+            .map(|p| qr.page(&b, p).expect("page"))
+            .collect::<Vec<_>>()
+    });
+    let compile_ns = median_ns(REPS, Composer::new);
+    println!("bind     {bind_ns} ns");
+    println!("pages    {prep_ns} ns ({} pages)", pages.len());
+    println!("programs {compile_ns} ns (Composer::new)");
+
+    let ta = median_ns(REPS, &mut arm_a);
+    let tc = median_ns(REPS, arm_c);
+    let tcp = median_ns(REPS, arm_cp);
+    let tb = median_ns(REPS, arm_b);
+    let (aa, ab) = allocs_of(&mut arm_a);
+    let (ca, cb) = allocs_of(arm_c);
+    let (cpa, cpb) = allocs_of(arm_cp);
+    let (ba, bb) = allocs_of(arm_b);
+    println!("A compose (mask-risc)      {ta} ns  allocs {aa} ({ab} B)");
+    println!("C ndarray::simd direct     {tc} ns  allocs {ca} ({cb} B)");
+    println!("C+ support-first simd      {tcp} ns  allocs {cpa} ({cpb} B)");
+    println!("B tuple baseline (lab-only) {tb} ns  allocs {ba} ({bb} B)");
+    println!(
+        "ratios: B/A {:.2}  A/C {:.2}  A/C+ {:.2}",
+        tb as f64 / ta as f64,
+        ta as f64 / tc as f64,
+        ta as f64 / tcp as f64
+    );
+}
diff --git a/crates/dir-rmap-probe/examples/rft1_pg_parity.rs b/crates/dir-rmap-probe/examples/rft1_pg_parity.rs
new file mode 100644
index 000000000..2a3595d6e
--- /dev/null
+++ b/crates/dir-rmap-probe/examples/rft1_pg_parity.rs
@@ -0,0 +1,183 @@
+//! D-RFT-1: the typed composition against PostgreSQL as the semantic
+//! reference.
+//!
+//! The fixture's membership, security set, nesting and grants are emitted in
+//! `Prepared` ordinals. PostgreSQL computes the reflexive nesting closure
+//! over security groups with a recursive CTE, joins membership to the role
+//! carriers, and returns count(*) (Bag), count(DISTINCT (user, role))
+//! (Distinct) and the per-coordinate multiplicities. Every seventh user also
+//! gets a membership row with a NULL group, a NULL join key that must match
+//! nothing in both engines.
+//!
+//! Needs `psql` and a server; connection from the usual `PG*` variables.
+//! Without one it prints `SKIP` and exits 0.
+//!
+//! ```text
+//! PGHOST=/var/tmp/k3pg PGPORT=54329 PGUSER=k3 PGDATABASE=postgres \
+//!   cargo run --release --example rft1_pg_parity
+//! ```
+//! (from this crate's directory)
+
+use dir_rmap_probe::World;
+use dir_rmap_probe::compose::{Carriers, Law, Members, add_bag, compose};
+use dir_rmap_probe::fixture::{Fixture, Spec};
+use dir_rmap_probe::prep::{bit, ones};
+use dir_rmap_probe::qr::Qr;
+use std::fmt::Write as _;
+use std::io::Write as _;
+use std::process::{Command, Stdio};
+
+const CTE: &str = "WITH RECURSIVE reach(g, a) AS (\
+    SELECT g, g FROM sec \
+    UNION \
+    SELECT r.g, n.p FROM reach r JOIN nest n ON n.c = r.a JOIN sec s ON s.g = n.p), \
+  carries AS (\
+    SELECT DISTINCT r.g, gr.role FROM reach r JOIN grant_ gr ON gr.g = r.a \
+    JOIN sec s ON s.g = r.g) ";
+
+fn psql(script: String) -> Result<String, String> {
+    let out = Command::new("psql")
+        .args(["-X", "-q", "-tA", "-F,", "-v", "ON_ERROR_STOP=1", "-f", "-"])
+        .stdin(Stdio::piped())
+        .stdout(Stdio::piped())
+        .stderr(Stdio::piped())
+        .spawn()
+        .and_then(|mut child| {
+            // Write on a separate thread: psql answers while it reads, so a
+            // parent that wrote everything first would block once the stdout
+            // pipe filled. A refused connection closes the pipe early; the
+            // exit status and stderr say why, so a write error is moot.
+            let mut stdin = child.stdin.take().expect("piped stdin");
+            let writer = std::thread::spawn(move || {
+                let _ = stdin.write_all(script.as_bytes());
+            });
+            let out = child.wait_with_output();
+            let _ = writer.join();
+            out
+        })
+        .map_err(|e| format!("could not run psql ({e})"))?;
+    if !out.status.success() {
+        return Err(format!(
+            "psql failed: {}",
+            String::from_utf8_lossy(&out.stderr).trim()
+        ));
+    }
+    Ok(String::from_utf8(out.stdout).expect("utf-8"))
+}
+
+fn main() {
+    let w = World::new(Fixture::generate(Spec::small()));
+    let b = w.prepare();
+    let qr = Qr::bind(&b, &w.witness()).expect("fits 256 slots");
+    let users = b.users;
+
+    // Tables in Prepared ordinals.
+    let mut s = String::from(
+        "CREATE TEMP TABLE m (u int, g int);\n\
+         CREATE TEMP TABLE sec (g int);\n\
+         CREATE TEMP TABLE nest (c int, p int);\n\
+         CREATE TEMP TABLE grant_ (g int, role int);\n",
+    );
+    s.push_str("INSERT INTO m VALUES ");
+    let mut first = true;
+    for u in 0..users {
+        for &g in b.user_groups.row(u) {
+            let _ = write!(s, "{}({u},{g})", if first { "" } else { "," });
+            first = false;
+        }
+        if u % 7 == 0 {
+            let _ = write!(s, "{}({u},NULL)", if first { "" } else { "," });
+            first = false;
+        }
+    }
+    s.push_str(";\nINSERT INTO sec VALUES ");
+    let mut first = true;
+    for g in 0..b.groups {
+        if bit(&b.security, g) {
+            let _ = write!(s, "{}({g})", if first { "" } else { "," });
+            first = false;
+        }
+    }
+    s.push_str(";\nINSERT INTO nest VALUES ");
+    for (k, &(c, p)) in b.nested.iter().enumerate() {
+        let _ = write!(s, "{}({c},{p})", if k == 0 { "" } else { "," });
+    }
+    s.push_str(";\nINSERT INTO grant_ VALUES ");
+    let mut first = true;
+    for g in 0..b.groups {
+        for role in ones(&[b.grants[g]]) {
+            let _ = write!(s, "{}({g},{role})", if first { "" } else { "," });
+            first = false;
+        }
+    }
+    s.push_str(";\n");
+    for q in [
+        "SELECT count(*) FROM m JOIN carries c ON c.g = m.g;",
+        "SELECT count(DISTINCT (m.u, c.role)) FROM m JOIN carries c ON c.g = m.g;",
+        "SELECT m.u, c.role, count(*) FROM m JOIN carries c ON c.g = m.g \
+         GROUP BY 1,2 ORDER BY 1,2;",
+    ] {
+        s.push_str(CTE);
+        s.push_str(q);
+        s.push('\n');
+    }
+
+    let stdout = match psql(s) {
+        Ok(o) => o,
+        Err(e) => {
+            println!("SKIP: {e}");
+            return;
+        }
+    };
+    let mut lines = stdout.lines().filter(|l| !l.is_empty());
+    let pg_bag: u64 = lines.next().expect("count(*)").parse().expect("count");
+    let pg_distinct: u64 = lines
+        .next()
+        .expect("count(distinct)")
+        .parse()
+        .expect("count");
+    // PostgreSQL's per-coordinate rows are read straight into a lane indexed
+    // `u * 64 + role` (an external reference's output, not a tuple answer).
+    let mut pg_lane = vec![0u16; users * 64];
+    for l in lines {
+        let f: Vec<usize> = l.split(',').map(|x| x.parse().expect("int")).collect();
+        pg_lane[f[0] * 64 + f[1]] = u16::try_from(f[2]).expect("count fits u16");
+    }
+
+    // Qr pages map user ordinal `u` to page `u / 256`, row `u % 256`
+    // (`Qr::page` reads `b.user_groups.row(u)` for `u` in the page), so page
+    // `p`, user `i` is ordinal `p * 256 + i`.
+    let c = Carriers::of(&qr);
+    let mut lane = vec![0u16; users * 64];
+    let (mut bag, mut distinct) = (0u64, 0u64);
+    for p in 0..qr.pages() {
+        let n = users.saturating_sub(p * 256).min(256);
+        let page = qr.page(&b, p).expect("page");
+        let m = Members::of(&qr, &page, n);
+        let out = compose(&m, &c, &w.witness(), Law::Count).expect("compose");
+        lane[p * 256 * 64..][..n * 64].copy_from_slice(&out.counts);
+        bag = add_bag(bag, out.bag).expect("no overflow");
+        distinct = add_bag(distinct, out.distinct).expect("no overflow");
+    }
+
+    if lane != pg_lane {
+        let at = lane.iter().zip(&pg_lane).position(|(a, b)| a != b);
+        panic!("per-coordinate lane differs from PostgreSQL, first at {at:?}");
+    }
+    assert_eq!(bag, pg_bag, "bag vs count(*)");
+    assert_eq!(distinct, pg_distinct, "distinct vs count(DISTINCT)");
+    assert!(
+        pg_bag > pg_distinct,
+        "the run never reached the Bag stratum: {pg_bag} vs {pg_distinct}"
+    );
+
+    let version = Command::new("psql")
+        .args(["-X", "-tA", "-c", "SHOW server_version"])
+        .output()
+        .map(|o| String::from_utf8_lossy(&o.stdout).trim().to_string())
+        .unwrap_or_else(|_| "?".into());
+    println!(
+        "PostgreSQL {version}: lane equal over {users} users x 64 roles; \
+         bag {bag} = count(*) {pg_bag}; distinct {distinct} = count(DISTINCT) {pg_distinct}"
+    );
+}
diff --git a/crates/dir-rmap-probe/src/compose.rs b/crates/dir-rmap-probe/src/compose.rs
new file mode 100644
index 000000000..8608430d4
--- /dev/null
+++ b/crates/dir-rmap-probe/src/compose.rs
@@ -0,0 +1,483 @@
+//! D-RFT-1, lab only: one typed composition over the DR256 register's shared
+//! slot coordinate.
+//!
+//! STATUS: PROBE. Nothing here is a production path.
+//!
+//! The composition is user -> group membership ([`Page::member`], slots)
+//! composed with role carriers ([`Qr::role_slots`], the slots whose reach
+//! carries role `r`) over the SAME 256-slot coordinate, bound by one [`Qr`]
+//! (one [`Witness`]). Per coordinate (user `i`, role `r`) the path mask is
+//!
+//! ```text
+//! member[i] & valid[i] & role_slots[r]
+//! ```
+//!
+//! Its population count is the Bag(N) multiplicity: the number of distinct
+//! DIRECT groups through which `r` is reached. Its existence is the Distinct
+//! answer: whether the path mask is non-empty. There is no value-directed
+//! gather, no semijoin, and no tuple: the answer lives in coordinate lanes
+//! indexed `i * 64 + r` and in a bit plane over the same coordinates, and
+//! every count, existence and equality is a mask-risc terminal.
+//!
+//! The join key is a slot. A bit set in `member` but clear in `valid` is a
+//! NULL join key: present payload, absent value. It matches nothing, as a
+//! NULL key matches nothing in SQL.
+//!
+//! Parallel paths combine by [`Law`]. `Count` and `Exists` are the same
+//! computation read at two strata (see [`Composed::answer`]). `Xor` is
+//! refused: parity loses an even number of paths, so the diamond (two direct
+//! groups reaching one role) would read zero, an answer neither stratum
+//! gives.
+
+use lance_graph_mask_risc::{
+    Foreign, Out, Planes, Program, Scratch, Value, execute_into, words_for,
+};
+use lance_graph_r2il::{FoldDialect, Val, procs};
+
+use crate::focus::{Row, SLOTS};
+use crate::prep::{Refusal, Witness};
+use crate::qr::{Page, Qr};
+
+/// How parallel paths combine.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum Law {
+    /// Number of paths (Bag(N)).
+    Count,
+    /// Whether any path exists (Distinct).
+    Exists,
+    /// Parity of the number of paths. Refused: an even number of parallel
+    /// paths reads zero.
+    Xor,
+}
+
+/// Which answer the two sides' cardinality witness guarantees.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum Stratum {
+    /// One side is functional (at most one member per user row, or at most
+    /// one carrier slot per role), so no coordinate has more than one path.
+    Distinct,
+    /// Both sides fan out: a coordinate can have several paths.
+    Bag,
+}
+
+/// Why a composition was refused.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum ComposeError {
+    /// The operands were bound to another version than `current`.
+    Stale,
+    /// The two operands were bound to different registers (witnesses).
+    SealMismatch,
+    /// [`Law::Xor`] is not a composition law here.
+    XorLaw,
+    /// A checked bag sum would have wrapped.
+    Overflow,
+    /// Any other refusal from the preparation layer.
+    Refusal(Refusal),
+}
+
+impl From<Refusal> for ComposeError {
+    /// `Refusal::Stale` is [`ComposeError::Stale`]; every other refusal is
+    /// carried as [`ComposeError::Refusal`].
+    fn from(r: Refusal) -> Self {
+        match r {
+            Refusal::Stale => ComposeError::Stale,
+            other => ComposeError::Refusal(other),
+        }
+    }
+}
+
+/// Left operand: one page's membership rows plus their presence plane.
+///
+/// A bit set in `member` but clear in `valid` is a NULL join key: present
+/// payload, absent value. It matches nothing.
+pub struct Members<'a> {
+    /// The register version the rows were bound against.
+    pub witness: &'a Witness,
+    /// Users in this page (at most 256).
+    pub users: usize,
+    /// Row `i`: user `i`'s direct security groups as slots.
+    pub member: &'a [Row; SLOTS],
+    /// Presence plane, same coordinates as `member`.
+    pub valid: [Row; SLOTS],
+}
+
+impl<'a> Members<'a> {
+    /// The page's rows with every key present (`valid == member`): 8 KiB of
+    /// presence plane is the only thing copied.
+    ///
+    /// The witness is the page's own, never the register's: a page resolved
+    /// against another register version then fails `compose`'s seal check
+    /// instead of being read through a different slot mapping. `qr` only
+    /// fixes the lifetime the rows are tied to.
+    pub fn of(_qr: &'a Qr, page: &'a Page, users: usize) -> Self {
+        assert!(users <= SLOTS, "a page holds at most {SLOTS} users");
+        Members {
+            witness: &page.witness,
+            users,
+            member: &page.member,
+            valid: page.member,
+        }
+    }
+}
+
+/// Right operand: role carriers over the same slots, held by the register
+/// in both orientations.
+pub struct Carriers<'a> {
+    /// The register version the carriers were bound against.
+    pub witness: &'a Witness,
+    /// Role -> the slots whose reach carries it.
+    pub role_slots: &'a [Row; 64],
+    /// Slot -> the roles its reach carries (one word per slot).
+    pub role: &'a [u64; SLOTS],
+}
+
+impl<'a> Carriers<'a> {
+    /// The register's role carriers.
+    pub fn of(qr: &'a Qr) -> Self {
+        Carriers {
+            witness: &qr.witness,
+            role_slots: &qr.role_slots,
+            role: &qr.role,
+        }
+    }
+}
+
+/// The composition's answer, in coordinates.
+pub struct Composed {
+    /// Which answer the cardinality witness guarantees.
+    pub stratum: Stratum,
+    /// Users in the page.
+    pub users: usize,
+    /// Bag multiplicity per coordinate `i * 64 + r`.
+    pub counts: Vec<u16>,
+    /// Distinct support: bit `i * 64 + r` set when the count is non-zero.
+    /// Word `i` is user `i`'s role set.
+    pub support: Vec<u64>,
+    /// Sum of `counts`, checked.
+    pub bag: u64,
+    /// Popcount of `support` (mask-risc `Count` over the support plane).
+    pub distinct: u64,
+    /// The cardinality witness: max path-mask fan on the left (members per
+    /// user row).
+    pub left_fan: u64,
+    /// The cardinality witness: max carrier slots per role on the right.
+    pub right_fan: u64,
+}
+
+impl Composed {
+    /// The answer the caller demands under `law`.
+    ///
+    /// The law only selects which ANSWER is read (`Count` -> [`Self::bag`],
+    /// `Exists` -> [`Self::distinct`]); it never selects a different
+    /// computation. `Xor` is refused.
+    pub fn answer(&self, law: Law) -> Result<u64, ComposeError> {
+        match law {
+            Law::Count => Ok(self.bag),
+            Law::Exists => Ok(self.distinct),
+            Law::Xor => Err(ComposeError::XorLaw),
+        }
+    }
+}
+
+/// Checked bag accumulation: `acc + n`, or [`ComposeError::Overflow`] where
+/// the sum would wrap.
+pub fn add_bag(acc: u64, n: u64) -> Result<u64, ComposeError> {
+    acc.checked_add(n).ok_or(ComposeError::Overflow)
+}
+
+/// The scalar a one-result r2il procedure left on the stack.
+fn take(stack: &mut Vec<Val>) -> u64 {
+    match stack.pop() {
+        Some(Val::Scalar(n)) if stack.is_empty() => {
+            u64::try_from(n).expect("a count or a 0/1 answer is non-negative")
+        }
+        other => panic!("an r2il procedure leaves one scalar, got {other:?}"),
+    }
+}
+
+/// An r2il procedure compiled once and replayed over planes of the same
+/// shape: a prepared statement.
+///
+/// A [`FoldDialect`] builds the mask-risc [`Program`] once from an r2il
+/// procedure over planes `0..k`; every later call runs that program over its
+/// own `k` zero-copy plane slices, with one scratch reused. Per call the
+/// cost is mask-risc execution alone, not program construction.
+struct Replay {
+    k: usize,
+    n_rows: usize,
+    program: Program,
+    scratch: Scratch<'static>,
+}
+
+impl Replay {
+    /// Compile `body` over `k` planes of `n_rows` rows. `keep` lends the
+    /// dialect a sink, for a body that ends in `KEEP`.
+    fn compile(
+        k: usize,
+        n_rows: usize,
+        keep: bool,
+        body: impl FnOnce(&mut FoldDialect<'_>, &mut Vec<Val>, &[u16]),
+    ) -> Self {
+        let zero = vec![0u64; words_for(n_rows)];
+        let mut sink = vec![0u64; words_for(n_rows)];
+        let masks = vec![zero.as_slice(); k];
+        let planes = Planes {
+            n_rows,
+            masks: &masks,
+            lanes: &[],
+        };
+        let ix: Vec<u16> = (0..k)
+            .map(|p| u16::try_from(p).expect("at most 256 planes"))
+            .collect();
+        let mut d = if keep {
+            FoldDialect::with_keep(&planes, &Foreign::NONE, &mut sink)
+        } else {
+            FoldDialect::new(&planes, &Foreign::NONE)
+        };
+        let mut stack = Vec::new();
+        body(&mut d, &mut stack, &ix);
+        let program = d.last_program().expect("the body folds").clone();
+        let scratch = Scratch::for_program(&program, n_rows).expect("scratch");
+        Replay {
+            k,
+            n_rows,
+            program,
+            scratch,
+        }
+    }
+
+    /// `procs::count_and` over `k` planes.
+    fn count_and(k: usize, n_rows: usize) -> Self {
+        Self::compile(k, n_rows, false, |d, stack, ix| {
+            procs::count_and(d, stack, ix).expect("compiles");
+            take(stack);
+        })
+    }
+
+    /// `procs::union` of `k` planes, then `procs::keep`.
+    fn keep_union(k: usize, n_rows: usize) -> Self {
+        Self::compile(k, n_rows, true, |d, stack, ix| {
+            procs::union(d, stack, ix).expect("compiles");
+            procs::keep(d, stack).expect("compiles");
+        })
+    }
+
+    /// `procs::intersect` of `k` planes, then `procs::keep`.
+    fn keep_intersect(k: usize, n_rows: usize) -> Self {
+        Self::compile(k, n_rows, true, |d, stack, ix| {
+            procs::intersect(d, stack, ix).expect("compiles");
+            procs::keep(d, stack).expect("compiles");
+        })
+    }
+
+    fn planes<'m>(&self, masks: &'m [&'m [u64]]) -> Planes<'m> {
+        assert_eq!(masks.len(), self.k, "compiled for {} planes", self.k);
+        Planes {
+            n_rows: self.n_rows,
+            masks,
+            lanes: &[],
+        }
+    }
+
+    /// Run a compiled count over `masks`.
+    fn count(&mut self, masks: &[&[u64]]) -> u64 {
+        let planes = self.planes(masks);
+        match execute_into(
+            &self.program,
+            &planes,
+            &Foreign::NONE,
+            &mut self.scratch,
+            Out::None,
+        ) {
+            Ok(Value::Count(n)) => n as u64,
+            other => panic!("a count program answers a Count, got {other:?}"),
+        }
+    }
+
+    /// Run a compiled keep over `masks`, writing the mask into `out`.
+    fn keep(&mut self, masks: &[&[u64]], out: &mut [u64]) {
+        let planes = self.planes(masks);
+        match execute_into(
+            &self.program,
+            &planes,
+            &Foreign::NONE,
+            &mut self.scratch,
+            Out::Mask(out),
+        ) {
+            Ok(Value::Mask(_)) => {}
+            other => panic!("a keep program answers a Mask, got {other:?}"),
+        }
+    }
+}
+
+/// Population count of `words` over `n_rows` rows (`procs::count_and` of
+/// one plane, i.e. `LOAD` then `PopCount`).
+fn plane_count(words: &[u64], n_rows: usize) -> u64 {
+    Replay::count_and(1, n_rows).count(&[words])
+}
+
+/// The set positions of `words`: the pointer table that inverts a plane to
+/// its local positions (the one thing the inversion materializes).
+fn positions(words: &[u64], out: &mut Vec<usize>) {
+    out.clear();
+    for (w, &word) in words.iter().enumerate() {
+        let mut bits = word;
+        while bits != 0 {
+            out.push(w * 64 + bits.trailing_zeros() as usize);
+            bits &= bits - 1;
+        }
+    }
+}
+
+/// The composition's prepared programs: every r2il procedure it runs,
+/// compiled once. Building one is the prepare step; [`Composer::compose`]
+/// is execution alone and reuses them (unions are compiled once per arity,
+/// on first use).
+pub struct Composer {
+    present_of: Replay,
+    path: Replay,
+    fan: Replay,
+    union_of: Vec<Option<Replay>>,
+}
+
+impl Default for Composer {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Composer {
+    /// Compile the composition's procedures.
+    pub fn new() -> Self {
+        Composer {
+            present_of: Replay::keep_intersect(2, SLOTS),
+            path: Replay::count_and(3, SLOTS),
+            fan: Replay::count_and(1, SLOTS),
+            union_of: (0..=SLOTS).map(|_| None).collect(),
+        }
+    }
+
+    /// Compose membership with role carriers.
+    ///
+    /// Order of refusal: `Xor` first (the law is wrong whatever the operands
+    /// are), then the operands' own seal ([`ComposeError::SealMismatch`]), then
+    /// the caller's `current` witness ([`ComposeError::Stale`]).
+    ///
+    /// `Law::Count` and `Law::Exists` return the same [`Composed`]; the caller
+    /// reads [`Composed::bag`] or [`Composed::distinct`] (or
+    /// [`Composed::answer`]).
+    pub fn compose(
+        &mut self,
+        left: &Members<'_>,
+        right: &Carriers<'_>,
+        current: &Witness,
+        law: Law,
+    ) -> Result<Composed, ComposeError> {
+        if law == Law::Xor {
+            return Err(ComposeError::XorLaw);
+        }
+        if left.witness != right.witness {
+            return Err(ComposeError::SealMismatch);
+        }
+        if left.witness != current {
+            return Err(ComposeError::Stale);
+        }
+        assert!(left.users <= SLOTS, "a page holds at most {SLOTS} users");
+        let users = left.users;
+
+        // Support first. A user's roles are the union of the role words of its
+        // present join keys (`member & valid`), so the only coordinates that can
+        // hold a path are the bits of that union; every other coordinate has
+        // count 0 by construction and is never evaluated. Each step is an r2il
+        // procedure compiled once and replayed over zero-copy row slices:
+        //   present_i  = keep(intersect(member_i, valid_i))       over 256 slots
+        //   support_i  = keep(union(role[s] for s in present_i))   over 64 roles
+        //   count(i,r) = count_and(member_i, valid_i, carriers_r)  for r in support_i
+        // The two pointer tables (present slots, supported roles) are the
+        // inversion from a plane to its local positions.
+        let mut counts = vec![0u16; users * 64];
+        let mut support = vec![0u64; users];
+        let mut bag = 0u64;
+        let mut left_fan = 0u64;
+        let mut present = [0u64; 4];
+        let (mut keys, mut roles): (Vec<usize>, Vec<usize>) = (Vec::new(), Vec::new());
+        let mut words: Vec<&[u64]> = Vec::with_capacity(SLOTS);
+        for i in 0..users {
+            let (m, v) = (&left.member[i][..], &left.valid[i][..]);
+            self.present_of.keep(&[m, v], &mut present);
+            positions(&present, &mut keys);
+            left_fan = left_fan.max(keys.len() as u64);
+            if keys.is_empty() {
+                continue;
+            }
+            words.clear();
+            words.extend(keys.iter().map(|&s| std::slice::from_ref(&right.role[s])));
+            self.union_of[keys.len()]
+                .get_or_insert_with(|| Replay::keep_union(keys.len(), 64))
+                .keep(&words, &mut support[i..=i]);
+            positions(&support[i..=i], &mut roles);
+            for &r in &roles {
+                let c = self.path.count(&[m, v, &right.role_slots[r][..]]);
+                // A path mask lives over 256 slots, so its count is at most 256.
+                counts[i * 64 + r] = u16::try_from(c).expect("a count over 256 slots fits u16");
+                bag = add_bag(bag, c)?;
+            }
+        }
+        let distinct = if users == 0 {
+            0
+        } else {
+            plane_count(&support, users * 64)
+        };
+
+        // Cardinality witness. Left: present join keys per user row (above).
+        // Right: carrier slots per role.
+        let mut right_fan = 0u64;
+        for carriers in right.role_slots.iter() {
+            right_fan = right_fan.max(self.fan.count(&[&carriers[..]]));
+        }
+
+        let stratum = if left_fan <= 1 || right_fan <= 1 {
+            Stratum::Distinct
+        } else {
+            Stratum::Bag
+        };
+        Ok(Composed {
+            stratum,
+            users,
+            counts,
+            support,
+            bag,
+            distinct,
+            left_fan,
+            right_fan,
+        })
+    }
+}
+
+/// [`Composer::compose`] with a freshly prepared [`Composer`].
+pub fn compose(
+    left: &Members<'_>,
+    right: &Carriers<'_>,
+    current: &Witness,
+    law: Law,
+) -> Result<Composed, ComposeError> {
+    Composer::new().compose(left, right, current, law)
+}
+
+/// Whether two planes differ over `n_rows` rows: `procs::differ`, i.e.
+/// `IntXor` then `ANY`. Both planes must be `n_rows.div_ceil(64)` words long.
+pub fn differ(a: &[u64], b: &[u64], n_rows: usize) -> bool {
+    let words = n_rows.div_ceil(64);
+    assert_eq!(a.len(), words, "plane a spans n_rows");
+    assert_eq!(b.len(), words, "plane b spans n_rows");
+    let masks = [a, b];
+    let planes = Planes {
+        n_rows,
+        masks: &masks,
+        lanes: &[],
+    };
+    let mut d = FoldDialect::new(&planes, &Foreign::NONE);
+    let mut stack = Vec::new();
+    procs::differ(&mut d, &mut stack, 0, 1).expect("two resident planes");
+    take(&mut stack) == 1
+}
diff --git a/crates/dir-rmap-probe/src/lib.rs b/crates/dir-rmap-probe/src/lib.rs
index 542ca6f70..4bc794bc6 100644
--- a/crates/dir-rmap-probe/src/lib.rs
+++ b/crates/dir-rmap-probe/src/lib.rs
@@ -17,6 +17,7 @@
 //! | Q  | [`qr`] | DR256 register: the security chain bound to 256 slots; nesting closed inside the tile; users x groups as a bit matrix |
 //! | Q  | [`pop`] | populations bound to one version: team focus, focus-restricted builds, relalg method folds, repair after edits |
 
+pub mod compose;
 pub mod fixture;
 pub mod focus;
 pub mod oracle;
diff --git a/crates/dir-rmap-probe/src/qr.rs b/crates/dir-rmap-probe/src/qr.rs
index da922cd9f..97d91bbc5 100644
--- a/crates/dir-rmap-probe/src/qr.rs
+++ b/crates/dir-rmap-probe/src/qr.rs
@@ -77,6 +77,9 @@ pub struct Page {
     pub roles: [u64; SLOTS],
     /// Ordinal -> slot lookups done while resolving this page.
     pub translations: usize,
+    /// The register version this page was resolved against. A page carries
+    /// it so it can never answer beside a register bound to another version.
+    pub witness: Witness,
 }
 
 impl Qr {
@@ -192,6 +195,7 @@ impl Qr {
             member,
             roles,
             translations,
+            witness: self.witness.clone(),
         })
     }
 
diff --git a/crates/dir-rmap-probe/tests/compose.rs b/crates/dir-rmap-probe/tests/compose.rs
new file mode 100644
index 000000000..7a3461b21
--- /dev/null
+++ b/crates/dir-rmap-probe/tests/compose.rs
@@ -0,0 +1,296 @@
+//! D-RFT-1: the typed composition (user -> group membership composed with
+//! role carriers over one slot coordinate) against an independent oracle.
+//!
+//! The oracle works from `Prepared` ordinals (a depth-first walk over the
+//! nesting CSR restricted to security groups), not from `Qr` (Warshall in
+//! the tile). Its output is a `Vec<u16>` lane in the same `i * 64 + r`
+//! coordinates, compared lane to lane; supports are compared with
+//! `compose::differ`. No tuple is built.
+
+use dir_rmap_probe::compose::{
+    Carriers, ComposeError, Law, Members, Stratum, add_bag, compose, differ,
+};
+use dir_rmap_probe::fixture::{Fixture, Kind, Spec};
+use dir_rmap_probe::prep::{Ordinals, Prepared, bit};
+use dir_rmap_probe::qr::{Page, Qr};
+use dir_rmap_probe::{Edit, World};
+use ogar_dir_core::{DirectoryScope, Guid128};
+
+/// One user, three security groups g0, g1, g2; user directly in g0 and g1
+/// (or only `direct`), g0 and g1 both nested in g2, g2 grants role 5.
+fn diamond(direct: Vec<(u32, u32)>) -> Fixture {
+    let mut spec = Spec::small();
+    spec.users = 1;
+    Fixture {
+        spec,
+        users: vec![Guid128([0x01; 16])],
+        groups: vec![
+            Guid128([0x10; 16]),
+            Guid128([0x11; 16]),
+            Guid128([0x12; 16]),
+        ],
+        kind: vec![Kind::Role; 3],
+        mail: vec![false; 3],
+        security: vec![Some(true); 3],
+        direct,
+        nested: vec![(0, 2), (1, 2)],
+        grants: vec![(2, 5)],
+        roles: vec![0, 1, 2],
+        locations: vec![],
+        teams: vec![],
+        attr: vec![0],
+        scope: DirectoryScope(Guid128([0x5C; 16])),
+    }
+}
+
+fn rig(f: Fixture) -> (World, Qr, Page) {
+    let w = World::new(f);
+    let b = w.prepare();
+    let qr = Qr::bind(&b, &w.witness()).expect("fits 256 slots");
+    let page = qr.page(&b, 0).expect("page 0");
+    (w, qr, page)
+}
+
+/// FAILS IF: the diamond (two direct groups reaching one role) is counted
+/// once or twice for Distinct, or the Bag count loses a path, or the
+/// stratum is not Bag when both sides fan out.
+#[test]
+fn diamond_counts_two_paths_once_distinct() {
+    let (w, qr, page) = rig(diamond(vec![(0, 0), (0, 1)]));
+    let m = Members::of(&qr, &page, 1);
+    let c = Carriers::of(&qr);
+    let out = compose(&m, &c, &w.witness(), Law::Count).unwrap();
+    assert_eq!(out.counts[5], 2);
+    assert_eq!(out.support[0] >> 5 & 1, 1);
+    assert_eq!(out.bag, 2);
+    assert_eq!(out.distinct, 1);
+    assert_eq!(out.stratum, Stratum::Bag);
+    assert_eq!(out.answer(Law::Count), Ok(2));
+    assert_eq!(out.answer(Law::Exists), Ok(1));
+}
+
+/// FAILS IF: parity (Xor) is accepted as a law, where the diamond would
+/// read 0.
+#[test]
+fn xor_is_refused() {
+    let (w, qr, page) = rig(diamond(vec![(0, 0), (0, 1)]));
+    let m = Members::of(&qr, &page, 1);
+    let c = Carriers::of(&qr);
+    assert_eq!(
+        compose(&m, &c, &w.witness(), Law::Xor).err(),
+        Some(ComposeError::XorLaw)
+    );
+    let out = compose(&m, &c, &w.witness(), Law::Count).unwrap();
+    assert_eq!(out.answer(Law::Xor), Err(ComposeError::XorLaw));
+}
+
+/// FAILS IF: a NULL join key (payload set, presence clear) still matches.
+#[test]
+fn null_key_matches_nothing() {
+    let (w, qr, page) = rig(diamond(vec![(0, 0), (0, 1)]));
+    let ord = Ordinals::of(&w.f, &w.view());
+    let s = qr.slot(ord.group[1]).expect("g1 is a security group");
+    let c = Carriers::of(&qr);
+    let full = Members::of(&qr, &page, 1);
+    assert_eq!(
+        compose(&full, &c, &w.witness(), Law::Count)
+            .unwrap()
+            .stratum,
+        Stratum::Bag,
+        "anti-vacuity: with both keys present the left side fans out"
+    );
+    let mut m = Members::of(&qr, &page, 1);
+    m.valid[0][s >> 6] &= !(1u64 << (s & 63));
+    let out = compose(&m, &c, &w.witness(), Law::Count).unwrap();
+    assert_eq!(out.counts[5], 1);
+    assert_eq!(out.bag, 1);
+    assert_eq!(out.distinct, 1);
+    // The left fan (member & valid) dropped to 1, so the rule yields Distinct.
+    assert_eq!(out.left_fan, 1);
+    assert_eq!(out.stratum, Stratum::Distinct);
+}
+
+/// FAILS IF: operands bound to different registers are composed.
+#[test]
+fn mismatched_seal_is_refused() {
+    let (wa, qa, pa) = rig(diamond(vec![(0, 0), (0, 1)]));
+    let (_wb, qb, _pb) = rig(diamond(vec![(0, 0), (0, 1)]));
+    let m = Members::of(&qa, &pa, 1);
+    let c = Carriers::of(&qb);
+    assert_eq!(
+        compose(&m, &c, &wa.witness(), Law::Count).err(),
+        Some(ComposeError::SealMismatch)
+    );
+}
+
+/// FAILS IF: a composition bound before an edit answers for the edited
+/// version.
+#[test]
+fn stale_binding_is_refused() {
+    let (mut w, qr, page) = rig(diamond(vec![(0, 0), (0, 1)]));
+    let m = Members::of(&qr, &page, 1);
+    let c = Carriers::of(&qr);
+    assert!(compose(&m, &c, &w.witness(), Law::Count).is_ok());
+    w.apply(&[Edit::AddDirect(0, 2)]);
+    assert_eq!(
+        compose(&m, &c, &w.witness(), Law::Count).err(),
+        Some(ComposeError::Stale)
+    );
+}
+
+/// FAILS IF: a page resolved before an edit is accepted when it is paired
+/// with a register bound after the edit. The page carries the witness it was
+/// resolved against, so `Members` cannot borrow the new register's witness
+/// for old rows.
+#[test]
+fn page_from_an_earlier_binding_is_refused() {
+    let (mut w, _qr_old, page_old) = rig(diamond(vec![(0, 0), (0, 1)]));
+    w.apply(&[Edit::AddDirect(0, 2)]);
+    let b = w.prepare();
+    let qr = Qr::bind(&b, &w.witness()).expect("fits 256 slots");
+    let c = Carriers::of(&qr);
+    let fresh = qr.page(&b, 0).expect("page 0");
+    assert!(
+        compose(&Members::of(&qr, &fresh, 1), &c, &w.witness(), Law::Count).is_ok(),
+        "anti-vacuity: a page resolved against this register composes"
+    );
+    assert_eq!(
+        compose(
+            &Members::of(&qr, &page_old, 1),
+            &c,
+            &w.witness(),
+            Law::Count
+        )
+        .err(),
+        Some(ComposeError::SealMismatch)
+    );
+}
+
+/// FAILS IF: a user with more present direct groups than r2il has scratch
+/// slots makes the support union panic, or the union loses a role. User in
+/// 12 security groups, group `g` grants role `g`.
+#[test]
+fn wide_support_union_composes() {
+    const G: u32 = 12;
+    let mut f = diamond((0..G).map(|g| (0, g)).collect());
+    f.groups = (0..G).map(|g| Guid128([0x20 + g as u8; 16])).collect();
+    f.kind = vec![Kind::Role; G as usize];
+    f.mail = vec![false; G as usize];
+    f.security = vec![Some(true); G as usize];
+    f.nested = vec![];
+    f.grants = (0..G).map(|g| (g, g as u8)).collect();
+    f.roles = (0..G).collect();
+    let (w, qr, page) = rig(f);
+    let out = compose(
+        &Members::of(&qr, &page, 1),
+        &Carriers::of(&qr),
+        &w.witness(),
+        Law::Count,
+    )
+    .expect("composes");
+    assert_eq!(out.left_fan, u64::from(G));
+    assert_eq!(out.support[0], (1u64 << G) - 1);
+    assert_eq!((out.bag, out.distinct), (u64::from(G), u64::from(G)));
+}
+
+/// FAILS IF: the bag sum wraps instead of refusing, or refuses one short of
+/// the boundary.
+#[test]
+fn overflow_is_refused() {
+    assert_eq!(add_bag(u64::MAX, 1).err(), Some(ComposeError::Overflow));
+    assert_eq!(add_bag(u64::MAX - 1, 1), Ok(u64::MAX));
+}
+
+/// FAILS IF: a functional left side (one direct group) is reported as the
+/// Bag stratum, or a coordinate exceeds one path.
+#[test]
+fn functional_side_is_distinct_stratum() {
+    let (w, qr, page) = rig(diamond(vec![(0, 0)]));
+    let m = Members::of(&qr, &page, 1);
+    let c = Carriers::of(&qr);
+    let out = compose(&m, &c, &w.witness(), Law::Count).unwrap();
+    assert_eq!(out.left_fan, 1);
+    assert_eq!(out.stratum, Stratum::Distinct);
+    assert!(out.counts.iter().all(|&n| n <= 1));
+    assert_eq!(out.counts[5], 1, "anti-vacuity: the role is reached");
+}
+
+/// Roles reachable upward from security group `d` through security groups
+/// only, by an explicit-stack walk over `b.up`.
+fn reach_roles(b: &Prepared, d: usize) -> u64 {
+    let mut seen = vec![false; b.groups];
+    let mut stack = vec![d];
+    seen[d] = true;
+    let mut roles = 0u64;
+    while let Some(h) = stack.pop() {
+        roles |= b.grants[h];
+        for &p in b.up.row(h) {
+            let p = p as usize;
+            if bit(&b.security, p) && !seen[p] {
+                seen[p] = true;
+                stack.push(p);
+            }
+        }
+    }
+    roles
+}
+
+/// The oracle lane for one page, in `i * 64 + r` coordinates.
+fn oracle_lane(b: &Prepared, first: usize, users: usize) -> Vec<u16> {
+    let mut lane = vec![0u16; users * 64];
+    for i in 0..users {
+        for &d in b.user_groups.row(first + i) {
+            let d = d as usize;
+            if !bit(&b.security, d) {
+                continue;
+            }
+            let roles = reach_roles(b, d);
+            for r in 0..64 {
+                lane[i * 64 + r] += u16::from(roles >> r & 1 == 1);
+            }
+        }
+    }
+    lane
+}
+
+/// FAILS IF: the composed counts or support differ from an independent
+/// computation over ordinals, or the fixture never reaches a multi-path
+/// coordinate (a vacuous run).
+#[test]
+fn random_fixture_matches_independent_oracle() {
+    let w = World::new(Fixture::generate(Spec::small()));
+    let b = w.prepare();
+    let qr = Qr::bind(&b, &w.witness()).expect("fits 256 slots");
+    let c = Carriers::of(&qr);
+    let (mut ones, mut multi, mut bag_total) = (0usize, 0usize, 0u64);
+    for p in 0..qr.pages() {
+        let users = qr.users.saturating_sub(p * 256).min(256);
+        let page = qr.page(&b, p).unwrap();
+        let m = Members::of(&qr, &page, users);
+        let out = compose(&m, &c, &w.witness(), Law::Count).unwrap();
+        let want = oracle_lane(&b, p * 256, users);
+        assert_eq!(out.counts, want, "page {p}: counts lane");
+        let mut want_support = vec![0u64; users];
+        for (i, word) in want_support.iter_mut().enumerate() {
+            for r in 0..64 {
+                *word |= u64::from(want[i * 64 + r] > 0) << r;
+            }
+        }
+        assert!(
+            !differ(&out.support, &want_support, users * 64),
+            "page {p}: support plane"
+        );
+        let want_bag = want
+            .iter()
+            .try_fold(0u64, |a, &n| add_bag(a, u64::from(n)))
+            .expect("no overflow");
+        assert_eq!(out.bag, want_bag, "page {p}: bag");
+        ones += want.iter().filter(|&&n| n == 1).count();
+        multi += want.iter().filter(|&&n| n >= 2).count();
+        bag_total += want_bag;
+    }
+    assert!(
+        ones > 0 && multi > 0,
+        "vacuous: coordinates with count 1 = {ones}, count >= 2 = {multi}, bag = {bag_total}"
+    );
+}
diff --git a/crates/lance-graph-r2il/src/lib.rs b/crates/lance-graph-r2il/src/lib.rs
index 813685434..8d3b8f665 100644
--- a/crates/lance-graph-r2il/src/lib.rs
+++ b/crates/lance-graph-r2il/src/lib.rs
@@ -34,21 +34,22 @@
 //!
 //! # Implemented, and refused by name
 //!
-//! Fold band: `VIA`, `SUM`, `MIN`, `MAX`, `GROUP_SUM`, `RANGE` and `KEEP`
-//! are wired. `KEY_RUNS`, `ANY`, `ALL`, `SCATTER_OR` and `BLEND` are
-//! refused as [`FoldError::Unimplemented`].
+//! Fold band: `VIA`, `SUM`, `MIN`, `MAX`, `GROUP_SUM`, `KEY_RUNS`, `RANGE`,
+//! `ANY`, `ALL` and `KEEP` are wired. `SCATTER_OR` and `BLEND` are refused as
+//! [`FoldError::Unimplemented`]. `ANY`/`ALL` push 1 or 0; `KEY_RUNS` pushes
+//! `COUNT(DISTINCT key)` over a key-ordered lane.
 //!
 //! `MIN`/`MAX` of an empty selection push the fold's identity, `i64::MAX` /
 //! `i64::MIN`: the same seed mask-risc's grouped `MIN`/`MAX` leave in an
 //! empty group. No `I32` value equals it, so a caller reads it as SQL NULL.
 //!
 //! R2IL band: `Load` (spaces 0 = U32, 1 = I32, 2 = U64, 3 = resident mask
-//! plane), `IntSub` (scalar/scalar), `IntAnd` (with survivor gating),
-//! `IntOr`, `IntNot`, `IntEqual`, `IntSLess` and `PopCount` are wired. `IntLess` and
+//! plane), `IntAdd`, `IntSub`, `IntMult`, `IntSDiv` (scalar/scalar, checked), `IntAnd` (with survivor gating),
+//! `IntOr`, `IntXor`, `IntNot`, `IntEqual`, `IntSLess` and `PopCount` are wired. `IntLess` and
 //! `IntLessEqual` are refused as [`FoldError::NoUnsignedLaneCompare`]
 //! (mask-risc has no unsigned lane compare); `IntSub` of two addresses as
 //! [`FoldError::LaneVersusLane`]; an unknown `Load` space as
-//! [`FoldError::UnknownLoadSpace`]; every other byte, `IntAdd` included, as
+//! [`FoldError::UnknownLoadSpace`]; every other byte as
 //! [`FoldError::Unimplemented`].
 //!
 //! # Boolean modes
@@ -60,7 +61,7 @@
 //!   gates a just-emitted predicate under its left operand (survivor gating)
 //!   instead of spending an `And`.
 //! - [`BoolMode::Fused`]: the calls build one [`BoolExpr`]; the first consumer
-//!   (`PopCount`, `SUM`, `GROUP_SUM`, `KEEP`) lowers it through mask-risc
+//!   (`PopCount`, `ANY`, `ALL`, `SUM`, `GROUP_SUM`, `KEEP`) lowers it through mask-risc
 //!   [`fuse()`] into ternlogs. Fewer passes, no survivor gating.
 //!
 //! Both modes keep the same rows. Predicates are emitted the same way in
@@ -69,6 +70,10 @@
 //! # Procedures
 //!
 //! [`procs`] holds named call sequences (`filter_lt`, `select_u32`, `count`,
+//! `exists`, `all`, `differ`, `intersect`, `union`, `keep`, `count_and`,
+//! the PostgreSQL primitives `is_null`, `is_not_null`, `and_between_i32`,
+//! `in_u32`, `count_valid`, `avg_parts`, `bool_or`, `bool_and`, `not_exists`,
+//! `except`, `count_distinct_sorted`, `count_union_all`,
 //! `sum`, `group_sum`, `margin`). They are Rust functions over [`Dialect::call`], not opcodes.
 //!
 //! # Slot limit
@@ -106,7 +111,7 @@ use lance_graph_mask_risc::{
     Value,
 };
 use ogar_loco::{Dialect, FnIndex};
-use ogar_r2il::{GROUP_SUM, KEEP, MAX, MIN, R2IL_BASE, RANGE, SUM, VIA};
+use ogar_r2il::{ALL, ANY, GROUP_SUM, KEEP, KEY_RUNS, MAX, MIN, R2IL_BASE, RANGE, SUM, VIA};
 
 // ── R2IL-band opcodes this dialect wires, by their VERIFIED ordinals ───────
 //
@@ -118,9 +123,20 @@ use ogar_r2il::{GROUP_SUM, KEEP, MAX, MIN, R2IL_BASE, RANGE, SUM, VIA};
 // `IntEqual`=27, `IntLess`=29, `IntSLess`=30, `IntLessEqual`=31,
 // `PopCount`=41.
 pub const LOAD: FnIndex = FnIndex(R2IL_BASE + 1);
+/// `IntAdd` = ordinal 9. Scalar + scalar only (two fold results); checked.
+pub const INT_ADD: FnIndex = FnIndex(R2IL_BASE + 9);
 pub const INT_SUB: FnIndex = FnIndex(R2IL_BASE + 10);
+/// `IntMult` = ordinal 11. Scalar × scalar only; checked.
+pub const INT_MULT: FnIndex = FnIndex(R2IL_BASE + 11);
+/// `IntSDiv` = ordinal 13. Scalar / scalar only, truncating toward zero (SQL
+/// integer division); division by zero is refused.
+pub const INT_S_DIV: FnIndex = FnIndex(R2IL_BASE + 13);
 pub const INT_AND: FnIndex = FnIndex(R2IL_BASE + 20);
 pub const INT_OR: FnIndex = FnIndex(R2IL_BASE + 21);
+/// `IntXor` = ordinal 22, arity 2 (confirmed against `R2ILFn::MNEMONICS`).
+/// Over two populations it lowers to `MaskOp::Xor`: the rows where exactly
+/// one side holds. `ANY` over it answers "do these masks differ".
+pub const INT_XOR: FnIndex = FnIndex(R2IL_BASE + 22);
 /// `IntNot` = ordinal 23, arity 1 (confirmed against `R2ILFn::MNEMONICS`).
 /// Over a population it lowers to `MaskOp::Not`.
 pub const INT_NOT: FnIndex = FnIndex(R2IL_BASE + 23);
@@ -222,6 +238,13 @@ pub enum FoldError {
     /// The fused mode's Boolean expression could not be lowered by mask-risc
     /// `fuse` (it needed a scratch slot past `u16::MAX`).
     Fuse(FuseError),
+    /// Scalar `IntSDiv` by zero. PostgreSQL raises `division_by_zero`; the
+    /// dialect refuses rather than answer.
+    DivisionByZero(FnIndex),
+    /// Scalar `IntAdd`/`IntMult` left `i64`. PostgreSQL raises
+    /// `numeric_value_out_of_range` for `bigint`; the dialect refuses rather
+    /// than wrap.
+    Overflow(FnIndex),
 }
 
 /// How a body's Boolean calls (`IntAnd`, `IntOr`, `IntNot` on masks) are
@@ -698,14 +721,15 @@ impl<'p> FoldDialect<'p> {
 
     /// [`Self::finalize_and_run`], asserting the terminal's known result
     /// shape (`Terminal::Count` → `Value::Count`, `Terminal::MaskedSumI32` →
-    /// `Value::SumI64`) — a caller-side invariant, not a new fallible path:
+    /// `Value::SumI64`, `Terminal::Any`/`All` → `Value::Bool` as 0 or 1) — a caller-side invariant, not a new fallible path:
     /// every call site below passes a terminal whose `Value` shape is fixed
     /// by construction.
     fn finalize_and_run_scalar(&mut self, terminal: Terminal) -> Result<i64, FoldError> {
         match self.finalize_and_run(terminal, OutKind::None)? {
             Value::Count(n) => Ok(i64::try_from(n).unwrap_or(i64::MAX)),
             Value::SumI64(n) => Ok(n),
-            other => panic!("terminal shape guarantees Count or SumI64, got {other:?}"),
+            Value::Bool(b) => Ok(i64::from(b)),
+            other => panic!("terminal shape guarantees Count, SumI64 or Bool, got {other:?}"),
         }
     }
 }
@@ -998,6 +1022,56 @@ impl Dialect for FoldDialect<'_> {
                 });
                 Ok(())
             }
+            // Like `IntOr`: one `Xor` pass in place, a `BoolExpr::Xor` when fused.
+            INT_XOR => {
+                let b = stack.pop().ok_or(FoldError::Underflow(f))?;
+                let a = stack.pop().ok_or(FoldError::Underflow(f))?;
+                if let (Val::Scalar(x), Val::Scalar(y)) = (a, b) {
+                    stack.push(Val::Scalar(x ^ y));
+                    return Ok(());
+                }
+                if !(is_mask(a) && is_mask(b)) {
+                    return Err(FoldError::WrongOperandKind(f));
+                }
+                if self.mode == BoolMode::Fused {
+                    return self.fuse_binary(a, b, stack, BoolExpr::Xor);
+                }
+                let operand = |v: Val, epoch_now: u32| match v {
+                    Val::Address(Addr::Plane(p)) => Ok(Operand::Plane(p)),
+                    Val::Slot { slot, epoch } if epoch == epoch_now => Ok(Operand::Scratch(slot)),
+                    _ => Err(FoldError::StaleSlot),
+                };
+                let (a, b) = (operand(a, self.epoch)?, operand(b, self.epoch)?);
+                let dst = self.fresh();
+                self.ops.push(MaskOp::Xor { a, b, dst });
+                stack.push(Val::Slot {
+                    slot: dst,
+                    epoch: self.epoch,
+                });
+                Ok(())
+            }
+            // Scalar arithmetic between fold results (`AVG` = `SUM` /
+            // `COUNT`, a bag product |L|·|R|). A lane operand has no scalar
+            // reading here: two lanes are `LaneVersusLane`, anything else
+            // `WrongOperandKind`.
+            INT_ADD | INT_MULT | INT_S_DIV => {
+                let b = stack.pop().ok_or(FoldError::Underflow(f))?;
+                let a = stack.pop().ok_or(FoldError::Underflow(f))?;
+                match (a, b) {
+                    (Val::Scalar(x), Val::Scalar(y)) => {
+                        let r = match f {
+                            INT_ADD => x.checked_add(y).ok_or(FoldError::Overflow(f))?,
+                            INT_MULT => x.checked_mul(y).ok_or(FoldError::Overflow(f))?,
+                            _ if y == 0 => return Err(FoldError::DivisionByZero(f)),
+                            _ => x.checked_div(y).ok_or(FoldError::Overflow(f))?,
+                        };
+                        stack.push(Val::Scalar(r));
+                        Ok(())
+                    }
+                    (Val::Address(_), Val::Address(_)) => Err(FoldError::LaneVersusLane(f)),
+                    _ => Err(FoldError::WrongOperandKind(f)),
+                }
+            }
             INT_SUB => {
                 let b = stack.pop().ok_or(FoldError::Underflow(f))?;
                 let a = stack.pop().ok_or(FoldError::Underflow(f))?;
@@ -1018,6 +1092,43 @@ impl Dialect for FoldDialect<'_> {
                 stack.push(Val::Scalar(n));
                 Ok(())
             }
+            // `COUNT(DISTINCT key)` under a mask, on a KEY-ORDERED `U32` lane:
+            // one per run of equal keys holding a selected row
+            // (`Terminal::CountKeyRunsU32`, O(1) state, no seen-set). An
+            // unordered lane is refused by mask-risc (`LaneNotOrdered`),
+            // never answered. Operands: mask, then the key lane.
+            KEY_RUNS => {
+                let key = stack.pop().ok_or(FoldError::Underflow(f))?;
+                let m = fold_mask(stack.pop().ok_or(FoldError::Underflow(f))?)
+                    .ok_or(FoldError::WrongOperandKind(f))?;
+                let Val::Address(Addr::Lane {
+                    idx: lane,
+                    kind: LaneKind::U32,
+                }) = key
+                else {
+                    return Err(FoldError::WrongOperandKind(f));
+                };
+                let mask = self.resolve_mask(m)?;
+                let n = self.finalize_and_run_scalar(Terminal::CountKeyRunsU32 { mask, lane })?;
+                stack.push(Val::Scalar(n));
+                Ok(())
+            }
+            // The Boolean reduction terminals: is any row selected, are all
+            // `n_rows` selected. mask-risc's `Any`/`All` stop at the first
+            // deciding word; the answer is pushed as 1 or 0.
+            ANY | ALL => {
+                let m = fold_mask(stack.pop().ok_or(FoldError::Underflow(f))?)
+                    .ok_or(FoldError::WrongOperandKind(f))?;
+                let mask = self.resolve_mask(m)?;
+                let terminal = if f == ANY {
+                    Terminal::Any { mask }
+                } else {
+                    Terminal::All { mask }
+                };
+                let b = self.finalize_and_run_scalar(terminal)?;
+                stack.push(Val::Scalar(b));
+                Ok(())
+            }
             SUM => {
                 let b = stack.pop().ok_or(FoldError::Underflow(f))?;
                 let a = stack.pop().ok_or(FoldError::Underflow(f))?;
diff --git a/crates/lance-graph-r2il/src/procs.rs b/crates/lance-graph-r2il/src/procs.rs
index 19d5e1585..1635a2e69 100644
--- a/crates/lance-graph-r2il/src/procs.rs
+++ b/crates/lance-graph-r2il/src/procs.rs
@@ -13,10 +13,11 @@
 //! [`Planes::lanes`](lance_graph_mask_risc::Planes::lanes).
 
 use ogar_loco::{Dialect, FnIndex};
-use ogar_r2il::{GROUP_SUM, KEEP, MAX, MIN, SUM};
+use ogar_r2il::{ALL, ANY, GROUP_SUM, KEEP, KEY_RUNS, MAX, MIN, SUM};
 
 use crate::{
-    FoldDialect, FoldError, Val, INT_AND, INT_EQUAL, INT_NOT, INT_SUB, INT_S_LESS, LOAD, POP_COUNT,
+    FoldDialect, FoldError, Val, INT_ADD, INT_AND, INT_EQUAL, INT_NOT, INT_OR, INT_SUB, INT_S_LESS,
+    INT_XOR, LOAD, POP_COUNT,
 };
 
 /// A selection that a procedure re-emits once per fold (see [`margin`]).
@@ -38,6 +39,14 @@ fn call(
     d.call(f, v, stack)
 }
 
+/// [`FoldDialect::call_owned`]: for a chain whose intermediate result this
+/// procedure created and consumes exactly once, so its slot can be reused.
+/// Only the chains inside one procedure use it; a value handed in on the
+/// caller's stack goes through [`call`].
+fn call_chain(d: &mut FoldDialect<'_>, f: FnIndex, stack: &mut Vec<Val>) -> Result<(), FoldError> {
+    d.call_owned(f, [0; 3], stack)
+}
+
 /// Push the address of lane `idx` in `space`.
 fn load(
     d: &mut FoldDialect<'_>,
@@ -109,6 +118,100 @@ pub fn count(d: &mut FoldDialect<'_>, stack: &mut Vec<Val>) -> Result<(), FoldEr
     call(d, POP_COUNT, [0; 3], stack)
 }
 
+/// Whether the mask on top of the stack holds any row. Pushes 1 or 0.
+///
+/// Calls: `ANY`.
+pub fn exists(d: &mut FoldDialect<'_>, stack: &mut Vec<Val>) -> Result<(), FoldError> {
+    call(d, ANY, [0; 3], stack)
+}
+
+/// Whether the mask on top of the stack holds all `n_rows` rows. Pushes 1
+/// or 0.
+///
+/// Calls: `ALL`.
+pub fn all(d: &mut FoldDialect<'_>, stack: &mut Vec<Val>) -> Result<(), FoldError> {
+    call(d, ALL, [0; 3], stack)
+}
+
+/// Whether the resident planes `a` and `b` differ in any row. Pushes 1 or 0.
+///
+/// This is mask equality as a fold: the planes are equal exactly when their
+/// `Xor` is empty. Nothing is decoded into rows.
+///
+/// Calls: `LOAD`, `LOAD`, `IntXor`, `ANY`.
+pub fn differ(
+    d: &mut FoldDialect<'_>,
+    stack: &mut Vec<Val>,
+    a: u16,
+    b: u16,
+) -> Result<(), FoldError> {
+    load(d, stack, SPACE_PLANE, a)?;
+    load(d, stack, SPACE_PLANE, b)?;
+    call(d, INT_XOR, [0; 3], stack)?;
+    exists(d, stack)
+}
+
+/// Push the conjunction of the resident planes `planes` as a mask. An
+/// empty list pushes nothing; the next fold then refuses with
+/// [`FoldError::Underflow`].
+///
+/// Calls: `LOAD`, then per further plane `LOAD`, `IntAnd`.
+pub fn intersect(
+    d: &mut FoldDialect<'_>,
+    stack: &mut Vec<Val>,
+    planes: &[u16],
+) -> Result<(), FoldError> {
+    for (k, &p) in planes.iter().enumerate() {
+        load(d, stack, SPACE_PLANE, p)?;
+        if k > 0 {
+            call_chain(d, INT_AND, stack)?;
+        }
+    }
+    Ok(())
+}
+
+/// Push the union of the resident planes `planes` as a mask, as
+/// [`intersect`] with `IntOr`.
+///
+/// Calls: `LOAD`, then per further plane `LOAD`, `IntOr`.
+pub fn union(
+    d: &mut FoldDialect<'_>,
+    stack: &mut Vec<Val>,
+    planes: &[u16],
+) -> Result<(), FoldError> {
+    for (k, &p) in planes.iter().enumerate() {
+        load(d, stack, SPACE_PLANE, p)?;
+        if k > 0 {
+            call_chain(d, INT_OR, stack)?;
+        }
+    }
+    Ok(())
+}
+
+/// Write the mask on top of the stack into the dialect's `KEEP` sink.
+/// Pushes nothing.
+///
+/// Calls: `KEEP`.
+pub fn keep(d: &mut FoldDialect<'_>, stack: &mut Vec<Val>) -> Result<(), FoldError> {
+    call(d, KEEP, [0; 3], stack)
+}
+
+/// How many rows the conjunction of the resident planes `planes` holds.
+/// Pushes the count.
+///
+/// This is the count of a composed path: with planes `[member, valid,
+/// carriers]` it is the number of join keys present on both sides.
+///
+/// Calls: [`intersect`], then `PopCount`.
+pub fn count_and(
+    d: &mut FoldDialect<'_>,
+    stack: &mut Vec<Val>,
+    planes: &[u16],
+) -> Result<(), FoldError> {
+    intersect(d, stack, planes)?;
+    count(d, stack)
+}
+
 /// `SUM(val_lane | mask)` for the mask on top of the stack. Pushes the sum.
 ///
 /// Calls: `LOAD`, `SUM`.
@@ -217,3 +320,206 @@ pub fn margin(
     sum(d, stack, cost)?;
     call(d, INT_SUB, [0; 3], stack)
 }
+
+// ── PostgreSQL primitives as call sequences ──────────────────────────────
+//
+// A column's NULLs are its presence plane: a resident mask with a bit set
+// where the value is present (Quack's model). Every procedure below either
+// leaves a mask on the stack (a WHERE predicate: KEEP it into a sink, and a
+// kept mask is what a Lance 13 masked read takes as its row-address
+// prefilter) or a scalar (an aggregate).
+
+/// `x IS NULL`: the complement of `x`'s presence plane `valid`. Pushes the
+/// mask.
+///
+/// Calls: `LOAD`, `IntNot`.
+pub fn is_null(d: &mut FoldDialect<'_>, stack: &mut Vec<Val>, valid: u16) -> Result<(), FoldError> {
+    load(d, stack, SPACE_PLANE, valid)?;
+    call(d, INT_NOT, [0; 3], stack)
+}
+
+/// `x IS NOT NULL`: `x`'s presence plane `valid`. Pushes the mask.
+///
+/// Calls: `LOAD`.
+pub fn is_not_null(
+    d: &mut FoldDialect<'_>,
+    stack: &mut Vec<Val>,
+    valid: u16,
+) -> Result<(), FoldError> {
+    load(d, stack, SPACE_PLANE, valid)
+}
+
+/// `x BETWEEN lo AND hi` on an `I32` lane, under the mask on top of the stack
+/// (pass the presence plane there for a nullable `x`: a NULL `x` is UNKNOWN,
+/// which `WHERE` drops). An empty range (`lo > hi`) keeps nothing, as in SQL.
+///
+/// Calls: [`and_cmp_i32`] twice.
+pub fn and_between_i32(
+    d: &mut FoldDialect<'_>,
+    stack: &mut Vec<Val>,
+    lane: u16,
+    lo: i32,
+    hi: i32,
+) -> Result<(), FoldError> {
+    and_cmp_i32(d, stack, lane, Order::Ge, lo)?;
+    and_cmp_i32(d, stack, lane, Order::Le, hi)
+}
+
+/// `x IN (v1, …, vn)` on a `U32` lane. Pushes the mask. An empty list keeps
+/// nothing (`x IN ()` is FALSE). A NULL in the list cannot be expressed here;
+/// that is K3 territory (Quack `sql_where`).
+///
+/// Calls: per value `LOAD`, `IntEqual`, `IntOr` (the first without `IntOr`);
+/// for an empty list `LOAD`, `IntEqual`, `IntNot`, `IntAnd` (x = 0 ∧ x ≠ 0).
+pub fn in_u32(
+    d: &mut FoldDialect<'_>,
+    stack: &mut Vec<Val>,
+    lane: u16,
+    values: &[u32],
+) -> Result<(), FoldError> {
+    if values.is_empty() {
+        for negate in [false, true] {
+            load(d, stack, SPACE_U32, lane)?;
+            stack.push(Val::Scalar(0));
+            call(d, INT_EQUAL, [0; 3], stack)?;
+            if negate {
+                call(d, INT_NOT, [0; 3], stack)?;
+            }
+        }
+        return call(d, INT_AND, [0; 3], stack);
+    }
+    for (k, &v) in values.iter().enumerate() {
+        load(d, stack, SPACE_U32, lane)?;
+        stack.push(Val::Scalar(i64::from(v)));
+        call(d, INT_EQUAL, [0; 3], stack)?;
+        if k > 0 {
+            call_chain(d, INT_OR, stack)?;
+        }
+    }
+    Ok(())
+}
+
+/// `COUNT(x)` for the mask on top of the stack: rows selected AND present.
+/// `COUNT(*)` is [`count`]. Pushes the count.
+///
+/// Calls: `LOAD`, `IntAnd`, `PopCount`.
+pub fn count_valid(
+    d: &mut FoldDialect<'_>,
+    stack: &mut Vec<Val>,
+    valid: u16,
+) -> Result<(), FoldError> {
+    load(d, stack, SPACE_PLANE, valid)?;
+    call(d, INT_AND, [0; 3], stack)?;
+    count(d, stack)
+}
+
+/// `AVG(x)` over an `I32` lane: pushes `SUM(x)` then `COUNT(x)`, both over
+/// `select ∧ valid`. The quotient is left to the caller because PostgreSQL's
+/// `avg(int)` is `numeric` (fractional) and R2IL has no exact rational; a
+/// count of 0 is SQL NULL. `select` is re-emitted per fold, as in [`margin`].
+///
+/// Calls: `select`, `LOAD`, `IntAnd`, `LOAD`, `SUM`; `select`, `LOAD`,
+/// `IntAnd`, `PopCount`.
+pub fn avg_parts(
+    d: &mut FoldDialect<'_>,
+    stack: &mut Vec<Val>,
+    select: &Select<'_>,
+    val_lane: u16,
+    valid: u16,
+) -> Result<(), FoldError> {
+    select(d, stack)?;
+    load(d, stack, SPACE_PLANE, valid)?;
+    call(d, INT_AND, [0; 3], stack)?;
+    sum(d, stack, val_lane)?;
+    select(d, stack)?;
+    count_valid(d, stack, valid)
+}
+
+/// `bool_or(flag)` over the mask on top of the stack, `flag` a resident plane
+/// of the rows where the boolean is TRUE: whether any selected row has it.
+/// Pushes 1 or 0. (An empty selection reads 0 here; PostgreSQL reads NULL,
+/// which the caller recovers from [`exists`] over the selection.)
+///
+/// Calls: `LOAD`, `IntAnd`, `ANY`.
+pub fn bool_or(d: &mut FoldDialect<'_>, stack: &mut Vec<Val>, flag: u16) -> Result<(), FoldError> {
+    load(d, stack, SPACE_PLANE, flag)?;
+    call(d, INT_AND, [0; 3], stack)?;
+    exists(d, stack)
+}
+
+/// `bool_and(flag)` (`every`) over the mask on top of the stack: whether no
+/// selected row lacks the flag. Pushes 1 or 0 (1 on an empty selection;
+/// PostgreSQL reads NULL there, as for [`bool_or`]).
+///
+/// Calls: `LOAD`, `IntNot`, `IntAnd`, `ANY`, then 1 − that (`IntSub`).
+pub fn bool_and(d: &mut FoldDialect<'_>, stack: &mut Vec<Val>, flag: u16) -> Result<(), FoldError> {
+    load(d, stack, SPACE_PLANE, flag)?;
+    call(d, INT_NOT, [0; 3], stack)?;
+    call(d, INT_AND, [0; 3], stack)?;
+    let missing = Val::Scalar(1);
+    exists(d, stack)?;
+    let any_missing = stack.pop().ok_or(FoldError::Underflow(ANY))?;
+    stack.push(missing);
+    stack.push(any_missing);
+    call(d, INT_SUB, [0; 3], stack)
+}
+
+/// `NOT EXISTS (…)` for the mask on top of the stack. Pushes 1 or 0.
+///
+/// Calls: `ANY`, then 1 − that (`IntSub`).
+pub fn not_exists(d: &mut FoldDialect<'_>, stack: &mut Vec<Val>) -> Result<(), FoldError> {
+    exists(d, stack)?;
+    let any = stack.pop().ok_or(FoldError::Underflow(ANY))?;
+    stack.push(Val::Scalar(1));
+    stack.push(any);
+    call(d, INT_SUB, [0; 3], stack)
+}
+
+/// `a EXCEPT b` over two resident planes of the same rows (set semantics on
+/// row identity): `a ∧ ¬b`. Pushes the mask. `INTERSECT` is [`intersect`],
+/// `UNION` is [`union`].
+///
+/// Calls: `LOAD`, `LOAD`, `IntNot`, `IntAnd`.
+pub fn except(
+    d: &mut FoldDialect<'_>,
+    stack: &mut Vec<Val>,
+    a: u16,
+    b: u16,
+) -> Result<(), FoldError> {
+    load(d, stack, SPACE_PLANE, a)?;
+    load(d, stack, SPACE_PLANE, b)?;
+    call(d, INT_NOT, [0; 3], stack)?;
+    call(d, INT_AND, [0; 3], stack)
+}
+
+/// `COUNT(DISTINCT key)` for the mask on top of the stack, on a KEY-ORDERED
+/// `U32` lane (the T0 key-order projection): one per run of equal keys that
+/// holds a selected row. An unordered lane is refused by mask-risc
+/// (`LaneNotOrdered`), never answered. Pushes the count.
+///
+/// Calls: `LOAD`, `KEY_RUNS`.
+pub fn count_distinct_sorted(
+    d: &mut FoldDialect<'_>,
+    stack: &mut Vec<Val>,
+    key_lane: u16,
+) -> Result<(), FoldError> {
+    load(d, stack, SPACE_U32, key_lane)?;
+    call(d, KEY_RUNS, [0; 3], stack)
+}
+
+/// `COUNT(*)` of two selections added (`UNION ALL` of two counts), checked.
+/// Pushes the sum.
+///
+/// Calls: `select_a`, `PopCount`, `select_b`, `PopCount`, `IntAdd`.
+pub fn count_union_all(
+    d: &mut FoldDialect<'_>,
+    stack: &mut Vec<Val>,
+    select_a: &Select<'_>,
+    select_b: &Select<'_>,
+) -> Result<(), FoldError> {
+    select_a(d, stack)?;
+    count(d, stack)?;
+    select_b(d, stack)?;
+    count(d, stack)?;
+    call(d, INT_ADD, [0; 3], stack)
+}
diff --git a/crates/lance-graph-r2il/tests/dialect.rs b/crates/lance-graph-r2il/tests/dialect.rs
index 03a689e1a..743a9587d 100644
--- a/crates/lance-graph-r2il/tests/dialect.rs
+++ b/crates/lance-graph-r2il/tests/dialect.rs
@@ -8,8 +8,8 @@
 
 use lance_graph_mask_risc::{words_for, Foreign, LaneRef, Planes};
 use lance_graph_r2il::{
-    Addr, FoldDialect, FoldError, Val, INT_AND, INT_EQUAL, INT_LESS, INT_LESS_EQUAL, INT_NOT,
-    INT_OR, INT_SUB, INT_S_LESS, LOAD, MAX_SLOTS, POP_COUNT,
+    Addr, FoldDialect, FoldError, Val, INT_ADD, INT_AND, INT_EQUAL, INT_LESS, INT_LESS_EQUAL,
+    INT_MULT, INT_NOT, INT_OR, INT_SUB, INT_S_DIV, INT_S_LESS, INT_XOR, LOAD, MAX_SLOTS, POP_COUNT,
 };
 use ogar_loco::{Dialect, FnIndex};
 use ogar_r2il::{
@@ -31,6 +31,10 @@ fn local_r2il_ordinals_name_the_ogar_r2il_opcodes() {
         (INT_SUB, "IntSub"),
         (INT_AND, "IntAnd"),
         (INT_OR, "IntOr"),
+        (INT_XOR, "IntXor"),
+        (INT_ADD, "IntAdd"),
+        (INT_MULT, "IntMult"),
+        (INT_S_DIV, "IntSDiv"),
         (INT_NOT, "IntNot"),
         (INT_EQUAL, "IntEqual"),
         (INT_LESS, "IntLess"),
@@ -80,10 +84,11 @@ fn every_named_refusal_is_still_produced() {
     let foreign = Foreign::NONE;
 
     // Fold-band bytes that are not wired, and one R2IL-band one. `IntOr` is
-    // wired since D-SQLP-2.
-    let int_add = FnIndex(R2IL_BASE + 9);
-    assert_eq!(mnemonic(int_add), "IntAdd");
-    for f in [KEY_RUNS, ANY, ALL, SCATTER_OR, BLEND, int_add] {
+    // wired since D-SQLP-2; `IntXor`, `ANY`, `ALL`, `KEY_RUNS`, `IntAdd`,
+    // `IntMult` and `IntSDiv` since D-RFT-1.
+    let int_rem = FnIndex(R2IL_BASE + 14);
+    assert_eq!(mnemonic(int_rem), "IntRem");
+    for f in [SCATTER_OR, BLEND, int_rem] {
         let mut d = FoldDialect::new(&planes, &foreign);
         let mut stack = vec![Val::Scalar(1), Val::Scalar(2)];
         assert_eq!(
@@ -136,6 +141,58 @@ fn every_named_refusal_is_still_produced() {
         Err(FoldError::StaleSlot)
     );
 
+    // ANY, ALL and IntXor are wired: a scalar where a mask belongs is a shape
+    // refusal, not Unimplemented.
+    for f in [ANY, ALL] {
+        let mut d = FoldDialect::new(&planes, &foreign);
+        let mut stack = vec![Val::Scalar(1)];
+        assert_eq!(
+            call(&mut d, f, [0; 3], &mut stack),
+            Err(FoldError::WrongOperandKind(f))
+        );
+    }
+    let mut d = FoldDialect::new(&planes, &foreign);
+    let mut stack = vec![i32_lane(0), Val::Scalar(1)];
+    assert_eq!(
+        call(&mut d, INT_XOR, [0; 3], &mut stack),
+        Err(FoldError::WrongOperandKind(INT_XOR))
+    );
+
+    // Scalar arithmetic: checked, never wrapped; a lane has no scalar reading.
+    for (f, a, b, want) in [
+        (INT_S_DIV, 7, 0, FoldError::DivisionByZero(INT_S_DIV)),
+        (INT_S_DIV, i64::MIN, -1, FoldError::Overflow(INT_S_DIV)),
+        (INT_ADD, i64::MAX, 1, FoldError::Overflow(INT_ADD)),
+        (INT_MULT, i64::MAX, 2, FoldError::Overflow(INT_MULT)),
+    ] {
+        let mut d = FoldDialect::new(&planes, &foreign);
+        let mut stack = vec![Val::Scalar(a), Val::Scalar(b)];
+        assert_eq!(call(&mut d, f, [0; 3], &mut stack), Err(want), "{f:?}");
+    }
+    for f in [INT_ADD, INT_MULT, INT_S_DIV] {
+        let mut d = FoldDialect::new(&planes, &foreign);
+        let mut stack = vec![i32_lane(0), i32_lane(0)];
+        assert_eq!(
+            call(&mut d, f, [0; 3], &mut stack),
+            Err(FoldError::LaneVersusLane(f))
+        );
+        let mut d = FoldDialect::new(&planes, &foreign);
+        let mut stack = vec![i32_lane(0), Val::Scalar(1)];
+        assert_eq!(
+            call(&mut d, f, [0; 3], &mut stack),
+            Err(FoldError::WrongOperandKind(f))
+        );
+    }
+    // KEY_RUNS takes a mask and a U32 key lane; an I32 lane is a shape refusal.
+    let mut d = FoldDialect::new(&planes, &foreign);
+    let mut stack = vec![];
+    call(&mut d, RANGE, [0, 10, 0], &mut stack).expect("RANGE");
+    stack.push(i32_lane(0));
+    assert_eq!(
+        call(&mut d, KEY_RUNS, [0; 3], &mut stack),
+        Err(FoldError::WrongOperandKind(KEY_RUNS))
+    );
+
     // VIA and GROUP_SUM are wired: a wrong operand kind is a shape refusal,
     // not Unimplemented.
     let mut d = FoldDialect::new(&planes, &foreign);
diff --git a/crates/lance-graph-r2il/tests/procs.rs b/crates/lance-graph-r2il/tests/procs.rs
index f88038002..66653a68f 100644
--- a/crates/lance-graph-r2il/tests/procs.rs
+++ b/crates/lance-graph-r2il/tests/procs.rs
@@ -6,10 +6,14 @@
 
 use lance_graph_mask_risc::{words_for, Foreign, LaneRef, MaskOp, Planes};
 use lance_graph_r2il::procs::{
-    and_cmp_i32, count, filter_lt, group_sum, margin, max, min, select_lt, select_u32, sum, Order,
+    all, and_between_i32, and_cmp_i32, avg_parts, bool_and, bool_or, count, count_and,
+    count_distinct_sorted, count_union_all, count_valid, differ, except, exists, filter_lt,
+    group_sum, in_u32, is_not_null, is_null, keep, margin, max, min, not_exists, select_lt,
+    select_u32, sum, union, Order,
 };
 use lance_graph_r2il::{
-    BoolMode, FoldDialect, FoldError, Val, INT_AND, INT_NOT, INT_OR, POP_COUNT,
+    BoolMode, FoldDialect, FoldError, Val, INT_ADD, INT_AND, INT_MULT, INT_NOT, INT_OR, INT_S_DIV,
+    INT_XOR, POP_COUNT,
 };
 use ogar_loco::Dialect;
 use ogar_r2il::KEEP;
@@ -499,3 +503,468 @@ fn ordered_compares_match_the_reference() {
         }
     }
 }
+
+// ── D-RFT-1 polyfill: IntXor, ANY, ALL and the procedures over them ────────
+
+/// A plane over `N` rows holding the rows `keep` selects; tail bits clear.
+fn plane(keep: impl Fn(usize) -> bool) -> Vec<u64> {
+    let mut m = vec![0u64; words_for(N)];
+    for r in (0..N).filter(|&r| keep(r)) {
+        m[r / 64] |= 1 << (r % 64);
+    }
+    m
+}
+
+/// The scalar a one-result call left on the stack.
+fn top(stack: &[Val]) -> i64 {
+    match stack {
+        [Val::Scalar(n)] => *n,
+        other => panic!("expected one scalar, got {other:?}"),
+    }
+}
+
+/// FAILS IF: `count_and` stops counting the conjunction of its planes, in
+/// either mode. Reference: popcount of the word-wise AND (a fold over words,
+/// no rows decoded). Each added plane must remove rows, or the test could
+/// not see a dropped `IntAnd`.
+#[test]
+fn count_and_counts_the_conjunction_in_both_modes() {
+    let a = plane(|r| r % 2 == 0);
+    let b = plane(|r| r % 3 == 0);
+    let c = plane(|r| r % 5 != 0);
+    let masks = [a.as_slice(), b.as_slice(), c.as_slice()];
+    let planes = Planes {
+        n_rows: N,
+        masks: &masks,
+        lanes: &[],
+    };
+    let reference = |ps: &[u16]| -> i64 {
+        (0..words_for(N))
+            .map(|w| {
+                ps.iter()
+                    .fold(u64::MAX, |acc, &p| acc & masks[usize::from(p)][w])
+                    .count_ones() as i64
+            })
+            .sum()
+    };
+    let cases: [&[u16]; 3] = [&[0], &[0, 1], &[0, 1, 2]];
+    let counts: Vec<i64> = cases.iter().map(|ps| reference(ps)).collect();
+    assert!(
+        counts.windows(2).all(|w| w[1] < w[0]),
+        "every plane must remove rows: {counts:?}"
+    );
+    for mode in [BoolMode::InPlace, BoolMode::Fused] {
+        for (ps, want) in cases.iter().zip(&counts) {
+            let mut d = FoldDialect::new(&planes, &Foreign::NONE).with_mode(mode);
+            let mut stack = vec![];
+            count_and(&mut d, &mut stack, ps).expect("count_and");
+            assert_eq!(top(&stack), *want, "{mode:?} {ps:?}");
+            assert_eq!(d.programs_run(), 1, "one fold");
+        }
+    }
+}
+
+/// FAILS IF: `differ` stops answering mask inequality, in either mode. The
+/// planes that differ do so in ONE row, the last, which sits in the tail
+/// word; an `IntXor` lowered as `And` or `Or`, or an `ANY` that skips the
+/// tail word, answers wrongly here.
+#[test]
+fn differ_is_mask_inequality_in_both_modes() {
+    let a = plane(|r| r % 7 == 0);
+    let same = plane(|r| r % 7 == 0);
+    let last = N - 1;
+    assert_ne!(last % 7, 0, "the flipped row must be clear in `a`");
+    let b = plane(|r| r % 7 == 0 || r == last);
+    let empty = plane(|_| false);
+    let masks = [
+        a.as_slice(),
+        same.as_slice(),
+        b.as_slice(),
+        empty.as_slice(),
+    ];
+    let planes = Planes {
+        n_rows: N,
+        masks: &masks,
+        lanes: &[],
+    };
+    for mode in [BoolMode::InPlace, BoolMode::Fused] {
+        for (x, y, want) in [(0, 1, 0), (0, 2, 1), (2, 0, 1), (3, 3, 0), (0, 3, 1)] {
+            let mut d = FoldDialect::new(&planes, &Foreign::NONE).with_mode(mode);
+            let mut stack = vec![];
+            differ(&mut d, &mut stack, x, y).expect("differ");
+            assert_eq!(top(&stack), want, "{mode:?} planes {x} vs {y}");
+        }
+    }
+}
+
+/// FAILS IF: `exists` or `all` stop answering over exactly `n_rows` rows.
+/// `full` holds all `N` rows and no tail bit; `all_but_one` misses only the
+/// last row, so an `ALL` that ignores the tail word answers 1 for it.
+#[test]
+fn exists_and_all_answer_over_n_rows() {
+    let empty = plane(|_| false);
+    let one = plane(|r| r == N - 1);
+    let full = plane(|_| true);
+    let all_but_one = plane(|r| r != N - 1);
+    let masks = [
+        empty.as_slice(),
+        one.as_slice(),
+        full.as_slice(),
+        all_but_one.as_slice(),
+    ];
+    let planes = Planes {
+        n_rows: N,
+        masks: &masks,
+        lanes: &[],
+    };
+    let run = |p: u16, f: fn(&mut FoldDialect<'_>, &mut Vec<Val>) -> Result<(), FoldError>| {
+        let mut d = FoldDialect::new(&planes, &Foreign::NONE);
+        let mut stack = vec![Val::Address(lance_graph_r2il::Addr::Plane(p))];
+        f(&mut d, &mut stack).expect("terminal");
+        top(&stack)
+    };
+    assert_eq!([0, 1, 2, 3].map(|p| run(p, exists)), [0, 1, 1, 1], "exists");
+    assert_eq!([0, 1, 2, 3].map(|p| run(p, all)), [0, 0, 1, 0], "all");
+}
+
+/// FAILS IF: scalar `IntXor` stops being bitwise exclusive or.
+#[test]
+fn int_xor_on_scalars_is_exclusive_or() {
+    let planes = Planes {
+        n_rows: N,
+        masks: &[],
+        lanes: &[],
+    };
+    let mut d = FoldDialect::new(&planes, &Foreign::NONE);
+    let mut stack = vec![Val::Scalar(0b1100), Val::Scalar(0b1010)];
+    d.call(INT_XOR, [0; 3], &mut stack).expect("xor");
+    assert_eq!(stack, vec![Val::Scalar(0b0110)]);
+}
+
+/// FAILS IF: `union` stops writing the union of its planes through `keep`,
+/// in either mode. Reference: the word-wise OR; equality by `differ`, not by
+/// decoding rows. Each added plane must add rows.
+#[test]
+fn union_keeps_the_union_in_both_modes() {
+    let a = plane(|r| r % 5 == 0);
+    let b = plane(|r| r % 7 == 0);
+    let c = plane(|r| r % 11 == 0);
+    let want: Vec<u64> = (0..words_for(N)).map(|w| a[w] | b[w] | c[w]).collect();
+    let partial: Vec<u64> = (0..words_for(N)).map(|w| a[w] | b[w]).collect();
+    for mode in [BoolMode::InPlace, BoolMode::Fused] {
+        let mut sink = vec![0u64; words_for(N)];
+        {
+            let masks = [a.as_slice(), b.as_slice(), c.as_slice()];
+            let planes = Planes {
+                n_rows: N,
+                masks: &masks,
+                lanes: &[],
+            };
+            let mut d = FoldDialect::with_keep(&planes, &Foreign::NONE, &mut sink).with_mode(mode);
+            let mut stack = vec![];
+            union(&mut d, &mut stack, &[0, 1, 2]).expect("union");
+            keep(&mut d, &mut stack).expect("keep");
+            assert!(stack.is_empty());
+        }
+        let masks = [sink.as_slice(), want.as_slice(), partial.as_slice()];
+        let planes = Planes {
+            n_rows: N,
+            masks: &masks,
+            lanes: &[],
+        };
+        for (other, expect) in [(1u16, 0i64), (2, 1)] {
+            let mut d = FoldDialect::new(&planes, &Foreign::NONE);
+            let mut stack = vec![];
+            differ(&mut d, &mut stack, 0, other).expect("differ");
+            assert_eq!(top(&stack), expect, "{mode:?} vs plane {other}");
+        }
+    }
+}
+
+/// FAILS IF: a union or intersection over more planes than the dialect has
+/// scratch slots (`MAX_SLOTS`) refuses instead of folding in
+/// `BoolMode::InPlace`. The chain consumes each intermediate once, so it
+/// reuses slots; a chain that allocated a fresh slot per `IntOr` needs
+/// `k - 1` and breaks at 9.
+///
+/// `BoolMode::Fused` is pinned as a refusal, not a panic: mask-risc's `fuse`
+/// gives every ternlog group its own slot, so a 24-leaf chain overflows the
+/// arena. That is the reference algebra's limit, recorded as open.
+#[test]
+fn wide_union_and_intersect_fit_the_scratch_arena() {
+    const K: usize = 3 * lance_graph_r2il::MAX_SLOTS;
+    let owned: Vec<Vec<u64>> = (0..K).map(|k| plane(|r| r % (k + 2) != 0)).collect();
+    let masks: Vec<&[u64]> = owned.iter().map(Vec::as_slice).collect();
+    let ix: Vec<u16> = (0..K as u16).collect();
+    let words = words_for(N);
+    let want_or: i64 = (0..words)
+        .map(|w| masks.iter().fold(0u64, |a, m| a | m[w]).count_ones() as i64)
+        .sum();
+    let want_and: i64 = (0..words)
+        .map(|w| masks.iter().fold(u64::MAX, |a, m| a & m[w]).count_ones() as i64)
+        .sum();
+    assert!(
+        want_and > 0 && want_or > want_and,
+        "anti-vacuity: {want_and} {want_or}"
+    );
+    let planes = Planes {
+        n_rows: N,
+        masks: &masks,
+        lanes: &[],
+    };
+    let mut d = FoldDialect::new(&planes, &Foreign::NONE);
+    let mut stack = vec![];
+    union(&mut d, &mut stack, &ix).expect("wide union");
+    count(&mut d, &mut stack).expect("count over the union");
+    assert_eq!(top(&stack), want_or, "union");
+    let mut d = FoldDialect::new(&planes, &Foreign::NONE);
+    let mut stack = vec![];
+    count_and(&mut d, &mut stack, &ix).expect("wide intersect");
+    assert_eq!(top(&stack), want_and, "intersect");
+
+    let mut d = FoldDialect::new(&planes, &Foreign::NONE).with_mode(BoolMode::Fused);
+    let mut stack = vec![];
+    union(&mut d, &mut stack, &ix).expect("fused union builds");
+    assert!(
+        matches!(count(&mut d, &mut stack), Err(FoldError::Exec(_))),
+        "fused wide chain: refused, not folded and not a panic"
+    );
+}
+
+// ── PostgreSQL primitives ────────────────────────────────────────────────
+//
+// Fixture: `x` (I32, `i % 50`) is NULL on every 9th row (presence plane
+// `VALID`); `key` (U32, `i % GROUPS`); a flag plane `FLAG` = rows where
+// `i % 3 == 0`; a key-ordered lane `SORTED` (`i / 10`). References are
+// computed from the fixture's own rule per row, never from the dialect.
+
+const VALID: u16 = 0;
+const FLAG: u16 = 1;
+const SORTED: u16 = 5;
+
+fn null_row(r: usize) -> bool {
+    r.is_multiple_of(9)
+}
+
+fn run_scalar(
+    planes: &Planes<'_>,
+    body: impl FnOnce(&mut FoldDialect<'_>, &mut Vec<Val>) -> Result<(), FoldError>,
+) -> i64 {
+    let mut d = FoldDialect::new(planes, &Foreign::NONE);
+    let mut stack = vec![];
+    body(&mut d, &mut stack).expect("body");
+    top(&stack)
+}
+
+/// FAILS IF: any PostgreSQL primitive stops matching its row-rule reference:
+/// `IS [NOT] NULL`, `BETWEEN` (with a NULL `x` dropped), `IN`, `IN ()`,
+/// `COUNT(x)` vs `COUNT(*)`, `AVG` parts, `bool_or`/`bool_and`, `NOT EXISTS`,
+/// `EXCEPT`, `COUNT(DISTINCT)` on a key-ordered lane, `UNION ALL` counts, and
+/// `IntAdd`/`IntMult`/`IntSDiv` on scalars.
+#[test]
+fn postgresql_primitives_match_row_rule_references() {
+    let f = fixture();
+    let sorted: Vec<u32> = (0..N).map(|i| (i / 10) as u32).collect();
+    let lanes = [
+        LaneRef::I32(&f.x),
+        LaneRef::I32(&f.y),
+        LaneRef::I32(&f.revenue),
+        LaneRef::I32(&f.cost),
+        LaneRef::U32(&f.key),
+        LaneRef::U32(&sorted),
+    ];
+    let valid = plane(|r| !null_row(r));
+    let flag = plane(|r| r % 3 == 0);
+    let all_rows = plane(|_| true);
+    let masks = [valid.as_slice(), flag.as_slice(), all_rows.as_slice()];
+    let planes = Planes {
+        n_rows: N,
+        masks: &masks,
+        lanes: &lanes,
+    };
+    let count_rows = |keep: &dyn Fn(usize) -> bool| (0..N).filter(|&r| keep(r)).count() as i64;
+
+    // IS NULL / IS NOT NULL.
+    let nulls = count_rows(&|r| null_row(r));
+    assert!(nulls > 0 && nulls < N as i64);
+    assert_eq!(
+        run_scalar(&planes, |d, s| {
+            is_null(d, s, VALID)?;
+            count(d, s)
+        }),
+        nulls
+    );
+    assert_eq!(
+        run_scalar(&planes, |d, s| {
+            is_not_null(d, s, VALID)?;
+            count(d, s)
+        }),
+        N as i64 - nulls
+    );
+
+    // BETWEEN over the presence plane: a NULL x is dropped.
+    let between = count_rows(&|r| !null_row(r) && (10..=20).contains(&f.x[r]));
+    let between_ignoring_null = count_rows(&|r| (10..=20).contains(&f.x[r]));
+    assert_ne!(
+        between, between_ignoring_null,
+        "the fixture must have NULL rows in range"
+    );
+    assert_eq!(
+        run_scalar(&planes, |d, s| {
+            is_not_null(d, s, VALID)?;
+            and_between_i32(d, s, X, 10, 20)?;
+            count(d, s)
+        }),
+        between
+    );
+    // An empty range keeps nothing.
+    assert_eq!(
+        run_scalar(&planes, |d, s| {
+            is_not_null(d, s, VALID)?;
+            and_between_i32(d, s, X, 20, 10)?;
+            count(d, s)
+        }),
+        0
+    );
+
+    // IN (…) and IN ().
+    assert_eq!(
+        run_scalar(&planes, |d, s| {
+            in_u32(d, s, KEY, &[1, 3])?;
+            count(d, s)
+        }),
+        count_rows(&|r| f.key[r] == 1 || f.key[r] == 3)
+    );
+    assert_eq!(
+        run_scalar(&planes, |d, s| {
+            in_u32(d, s, KEY, &[])?;
+            count(d, s)
+        }),
+        0
+    );
+
+    // COUNT(x) vs COUNT(*) under x < 17.
+    let star = count_rows(&|r| f.x[r] < 17);
+    let col = count_rows(&|r| f.x[r] < 17 && !null_row(r));
+    assert!(col < star);
+    assert_eq!(
+        run_scalar(&planes, |d, s| {
+            select_lt(d, s, X, 17)?;
+            count(d, s)
+        }),
+        star
+    );
+    assert_eq!(
+        run_scalar(&planes, |d, s| {
+            select_lt(d, s, X, 17)?;
+            count_valid(d, s, VALID)
+        }),
+        col
+    );
+
+    // AVG parts: SUM(x) and COUNT(x) over x < 17, NULL rows excluded.
+    let mut d = FoldDialect::new(&planes, &Foreign::NONE);
+    let mut stack = vec![];
+    let sel = |d: &mut FoldDialect<'_>, s: &mut Vec<Val>| select_lt(d, s, X, 17);
+    avg_parts(&mut d, &mut stack, &sel, X, VALID).expect("avg");
+    let want_sum: i64 = (0..N)
+        .filter(|&r| f.x[r] < 17 && !null_row(r))
+        .map(|r| i64::from(f.x[r]))
+        .sum();
+    assert_eq!(stack, vec![Val::Scalar(want_sum), Val::Scalar(col)]);
+
+    // bool_or / bool_and over a selection.
+    let bor = |t: i32| {
+        run_scalar(&planes, |d, s| {
+            select_lt(d, s, X, t)?;
+            bool_or(d, s, FLAG)
+        })
+    };
+    let band = |t: i32| {
+        run_scalar(&planes, |d, s| {
+            select_lt(d, s, X, t)?;
+            bool_and(d, s, FLAG)
+        })
+    };
+    // x < 1 selects rows 0, 50, 100, …: i % 50 == 0; 0 is flagged, 50 is not.
+    assert_eq!((bor(1), band(1)), (1, 0));
+    assert_eq!((bor(0), band(0)), (0, 1), "empty selection");
+
+    // NOT EXISTS.
+    assert_eq!(
+        run_scalar(&planes, |d, s| {
+            select_lt(d, s, X, 0)?;
+            not_exists(d, s)
+        }),
+        1
+    );
+    assert_eq!(
+        run_scalar(&planes, |d, s| {
+            select_lt(d, s, X, 1)?;
+            not_exists(d, s)
+        }),
+        0
+    );
+
+    // EXCEPT: valid ∧ ¬flag.
+    assert_eq!(
+        run_scalar(&planes, |d, s| {
+            except(d, s, VALID, FLAG)?;
+            count(d, s)
+        }),
+        count_rows(&|r| !null_row(r) && r % 3 != 0)
+    );
+
+    // COUNT(DISTINCT sorted) under x < 5: distinct i / 10 over rows with i % 50 < 5.
+    let mut seen = [false; N / 10 + 1];
+    for r in (0..N).filter(|&r| f.x[r] < 5) {
+        seen[r / 10] = true;
+    }
+    let distinct = seen.iter().filter(|&&b| b).count() as i64;
+    assert!(distinct < count_rows(&|r| f.x[r] < 5), "keys repeat");
+    assert_eq!(
+        run_scalar(&planes, |d, s| {
+            select_lt(d, s, X, 5)?;
+            count_distinct_sorted(d, s, SORTED)
+        }),
+        distinct
+    );
+
+    // UNION ALL of two counts.
+    let a = |d: &mut FoldDialect<'_>, s: &mut Vec<Val>| select_lt(d, s, X, 10);
+    let b = |d: &mut FoldDialect<'_>, s: &mut Vec<Val>| select_lt(d, s, Y, 5);
+    assert_eq!(
+        run_scalar(&planes, |d, s| count_union_all(d, s, &a, &b)),
+        count_rows(&|r| f.x[r] < 10) + count_rows(&|r| f.y[r] < 5)
+    );
+
+    // Scalar arithmetic.
+    for (op, x, y, want) in [
+        (INT_ADD, 7, 5, 12),
+        (INT_MULT, 7, -5, -35),
+        (INT_S_DIV, -7, 2, -3),
+    ] {
+        let mut d = FoldDialect::new(&planes, &Foreign::NONE);
+        let mut stack = vec![Val::Scalar(x), Val::Scalar(y)];
+        d.call(op, [0; 3], &mut stack).expect("arith");
+        assert_eq!(stack, vec![Val::Scalar(want)], "{op:?}");
+    }
+}
+
+/// FAILS IF: `COUNT(DISTINCT)` on an UNORDERED lane is answered instead of
+/// refused (`LaneNotOrdered`).
+#[test]
+fn count_distinct_refuses_an_unordered_lane() {
+    let f = fixture();
+    let lanes = lanes(&f);
+    let planes = Planes {
+        n_rows: N,
+        masks: &[],
+        lanes: &lanes,
+    };
+    let mut d = FoldDialect::new(&planes, &Foreign::NONE);
+    let mut stack = vec![];
+    select_lt(&mut d, &mut stack, X, 17).expect("select");
+    let err = count_distinct_sorted(&mut d, &mut stack, KEY).expect_err("key = i % 4 is unordered");
+    assert!(matches!(err, FoldError::Exec(_)), "{err:?}");
+}
diff --git a/crates/r2il-mask-abi-probe/tests/row_bridge.rs b/crates/r2il-mask-abi-probe/tests/row_bridge.rs
index 046605894..567499a7a 100644
--- a/crates/r2il-mask-abi-probe/tests/row_bridge.rs
+++ b/crates/r2il-mask-abi-probe/tests/row_bridge.rs
@@ -62,10 +62,12 @@
 //! physical terminal is chosen by the KEY operand's address kind),
 //! **`RANGE`** (`Pred::Range`, bounds as the call's two immediates) and
 //! **`KEEP`** (`Terminal::Keep` into a caller-lent mask sink) are wired.
-//! The other seven — `MIN`, `MAX`, `KEY_RUNS`, `ANY`, `ALL`, `SCATTER_OR`,
-//! `BLEND` — are refused as [`FoldError::Unimplemented`], named
-//! individually in the refusal, never silently coerced into one of the ones
-//! that exist.
+//! When this file was written the other seven — `MIN`, `MAX`, `KEY_RUNS`,
+//! `ANY`, `ALL`, `SCATTER_OR`, `BLEND` — were refused as
+//! [`FoldError::Unimplemented`]. `MIN`/`MAX` (D-SQLP-7) and `ANY`/`ALL`
+//! (D-RFT-1) are wired since; `lance-graph-r2il`'s crate doc is the current
+//! list. A refused byte is named individually, never silently coerced into
+//! one of the ones that exist.
 //!
 //! `RANGE`, `KEEP`, `IntNot` and `LOAD` space 3 (a RESIDENT mask plane as an
 //! address, read where it stands as `Operand::Plane`) were added in Round 4