diff --git a/.claude/board/AGENT_LOG.md b/.claude/board/AGENT_LOG.md
index 5d4c077df..2acd93876 100644
--- a/.claude/board/AGENT_LOG.md
+++ b/.claude/board/AGENT_LOG.md
@@ -1,3 +1,8 @@
+## 2026-10-21 — 5+3 council: the tuple is a zero-copy violation (blast radius)
+
+- Five savants: prior-art-savant, iron-rule-savant, runtime-archaeologist (code truth), cascade-impact-savant (blast radius, Opus), creative-explorer-savant. Three reviewers on draft v2 only: overclaim-auditor (2 BLOCK, wording/grading, fixed), dilution-collapse-sentinel (0 BLOCK, 2 collapses restored), firewall-warden (0 BLOCK).
+- Result: `entries/2026-10-21-tuple-zero-copy-council.md`; follow-ups EXO-103..118 on the Linear board. No code changed. Written by the orchestrating main thread only.
+
 ## 2026-10-20 — D-TGB harvest and D-MPS-9 groundwork (Haiku)
 
 - Five runs: `substrate-mapper` over MORK `kernel/src/space.rs` (47 items, six verb bodies verbatim), over `mork-pathmap-probe/src` (267 items), over `lance-graph-quack/src` (201 rows); `harvest-scout` for the write-path symbols (GadamerRevision, pearl, MUL gates, `advance_on_gate`, `LanceCycleWriter`, `EpistemicState5`); `census-counter` (`nal_inventory.tsv` 115 rows / 115 ids, `KanbanColumn` 6, `RevisionKind` 9, `EvidentialEffect` 3, probe tests 42).
diff --git a/.claude/board/entries/2026-10-21-coresearch-ms-vs-ns-folding.md b/.claude/board/entries/2026-10-21-coresearch-ms-vs-ns-folding.md
new file mode 100644
index 000000000..16e4a4b17
--- /dev/null
+++ b/.claude/board/entries/2026-10-21-coresearch-ms-vs-ns-folding.md
@@ -0,0 +1,124 @@
+# 2026-10-21 — Co-research: why the recent relalg / D-RU-6 probes measure milliseconds
+
+**Status:** EXPLORATION MAP. Nothing is adopted; the operator chooses. D-RU-6 stays Queued/Lab.
+
+## Question
+
+The operator: "Milliseconds is not normal in a substrate with nanoseconds being the normal." Why do relalg and the last 5–15 PRs measure 0.4–5,000 ms? Is folding missing? Do we still use CSR, adjacency scans, gathers or tuple decoding where a fold belongs? Side question: should the join key carry a little-endian ordered-lane contract?
+
+## Premise audit
+
+- **Partly wrong.** Each operation already runs at nanosecond scale: walk_mask sweeps at 0.59 ns per word, mask-risc words cost about 0.2 ns, relalg about 0.34 ns per lane-row. The milliseconds come from running 1e5–1e8 of them.
+- **Split.** Some work is intrinsic to the output a consumer asks for; some work no consumer asks for. Only the second is a missing fold.
+
+## Where the milliseconds go
+
+Each code path below was verified in the source, and the two main sites were re-read on the main thread. The numbers come from the board entries.
+
+**1. Tuple decoding in every D-RU-6 arm.** This is the dominant avoidable cost.
+- `tuples()` in `walk_mask.rs:232-249` and `walk_tile.rs:538-569` tests every column bit of every row and length. That costs O(n) per row, not O(set bits).
+- Each set bit becomes a `BTreeSet` insert, and NoWalk is built as a full complement.
+- edges(1024): 318 ms to decode against 0.16 ms count-only, about 2 M tuples at roughly 158 ns each. cyclic(300): about 1.2 s.
+- The count-only path already exists (`walk_tile::counts`, `:529-535`). Only the correctness oracle needs the tuples.
+
+**2. The crepe-plan-probe native terminal enumerates even for a count** (`native.rs:401-461`, reached from `pipeline.rs:189-214`).
+- For every non-Exists answer it forms the per-slot cartesian product over sides that emit, clones and sorts it, then inserts into a `BTreeMap`.
+- Sides that do not emit are already multiplied in rather than enumerated, so the fold exists only for those.
+- Preparation copies side rows into owned Vecs: 3.5 ms at 50 k rows.
+
+**3. Arm B (`walk_tile.rs:363-407`).** It spreads, broadcasts and runs a Ternlog over 1,024 words per contributing node, about 3,072 words each.
+- cyclic(300): 6.15 M calls in 3,828 ms, about 0.2 ns per word.
+- The cost per word is at the floor; the 256× amplification is not. `tile_gemm.rs:87-104` `BitMat::compose` does one 4-word OR per set bit (2.6–25 µs per 256² product), but it is used only as a test oracle.
+
+**4. Arm A and A′.**
+- Arm A gathers `delta[row(l2,y)]` at every adjacency bit, and sweeps all `lens·n·w` words every iteration: 230 M words at chain(700).
+- A′ removes the sweep but still scans every adjacency row: 14.6 M words.
+
+**5. Arm C.** It does the same row ORs as A′, plus allocation inside the hot loop: a fresh tile twice per changed tile per iteration, and `block_occ` and `next_occ` rebuilt every iteration.
+
+**6. relalg D-RA-1, U1 and U2: not a fold gap.**
+- Each query is one mask-risc program, with no tuples and no CSR. U2 runs 14 lanes over 1 M rows in 4.77 ms, about 0.34 ns per lane-row.
+- This is intrinsic to scanning a million rows. The only lever is skipping rows: D-CMR-2 runs and extents, or the `under` gate.
+
+**7. D-RMAP-DIR-1.** The milliseconds are in the baseline arm only, the dir-sim View walk. The fast arms run at 32–79 ns per user. The #1639 attest takes 7 ms and pays back after 2–3 queries.
+
+**Unused fold primitives.**
+- mask-risc offers `Pred{under}`, a fused `Ternlog` with an `Any` terminal, `CountKeyRunsU32` and `GroupPowerSumsI32`.
+- lane-fold `Query::check` refuses `wants_rows`.
+- `Refuse::PairList` is declared but raised nowhere in `lance-graph-lane-fold/src`, which is five files, all searched.
+
+**Answer.** Folding is missing at the **terminal**, not in the core loop. The probes and the native terminal materialize tuples for consumers that only need a count, an existence check or a complement. The remaining milliseconds are intrinsic row scans, plus lab-only amplification and allocation in arms B and C.
+
+## Exploration map
+
+Verdicts combine three reviewers on the crosswalk: a bridge architect, a firewall and fit critic, and a falsifier designer. Every probe is pre-registered.
+
+**Shared oracle rule for every D-RU-6 probe.**
+- The D-RU-6 correctness rule (`d-relalg-unified-v1.md:205`) stays: full `Walk` / `NoWalk` tuple sets against crepe and ascent.
+- Each fixture's oracle tuple sets are computed **once, untimed**, and encoded bijectively into reference planes. Timed arms compare planes word for word.
+- One untimed tuple-set check per fixture keeps the decoder honest.
+- Counts or fingerprints never replace the oracle.
+
+| idea | verdict | firewall | probe and kill condition |
+|---|---|---|---|
+| **X1+X2+X3** demand-typed plane terminal: Exists, Count or Rows. Exists and Count end at a popcount or plane comparison; NoWalk is counted as `extent AND NOT reach`; only Rows decodes | **ADOPT-NOW** (lab probes; no production change) | PASS (law 1, doctrine §6.2, law 10) | P-X2: any count or plane mismatch kills it. Count-only above 5 % of decode wall time on 3 or more of 7 checked fixtures kills it. P-X3 adds a dangling-node fixture, so the extent mask is shown load-bearing (can-fire) and edges(512) stays unchanged (silent) |
+| **X4** walk only set bits in `tuples()` when Rows is demanded | **ADOPT-NOW** | PASS. Decode reads only its own output plane, and law 10 lists "walk set bits". C's no-bit-scan audit covers execution, not decode | P-X4: under 2× decode speedup on 5 of 9 fixtures kills it, as does any tuple difference. Also kills if the time-vs-output intercept is not cut 5× or more |
+| **X6** native Count(Bag) = Σ over joint mask of Π side degrees; Count(Set) as Σ Π per-slot distinct counts | **PROBE** | Bag: PASS. Set: CONFLICT. It is legal only when the joint-slot key is in the head; otherwise it is law 7's count-distinct-across-factors. Resolve demand and head-key presence in the semantic plan first (law 1) | P-X6 on the D-MPS-8 star (n = 50 k) plus a multi-rule union with overlap. Any mismatch kills it, and so does under 10× on the star. The overlap twin must still route to dedup |
+| **X9** hoist allocation out of the hot loops of A′, B and C | **ADOPT-NOW** (probe hygiene) | PASS | P-X9: a counting global allocator must read 0 in the steady loop, and chain(700) must gain at least 15 % |
+| **X5** factorized Rows output: `(x, length, row bitmask)` borrowed from the plane, with optional `expand()` | **PROBE** | PASS, with law 7 kept | P-X5 over `--sweep`: kills if the wall-time slope against tuple count is not at least 10× flatter, or if the expansion fails the oracle |
+| **X8** `BitMat::compose` in arm B | **PROBE**, lab | CONFLICT: a per-bit write scatter inside one tile. Less severe than A's cross-table read, but it falls under the same open slot-binding ruling (`:234`) | P-X8: kills if under 3× on cyclic(300), or on any plane mismatch |
+| **X10** C's update through mask-risc `Ternlog` + `Any` | **PROBE** | PASS (no new opcode) | P-X10: kills if under 1.3× on C's loop |
+| **X11** push/pull per iteration | **PROBE**, lab | CONFLICT: a pull through reverse adjacency is a gather. A co-addressed pull exists as an inner product over the shared domain, popcount(rowₓ(A) ∧ row_z(δᵀ)) | P-X11 (P-SYS-4): kills if work per iteration does not decrease after the frontier peak, or if under 1.5× against A′ |
+| **X12** zero-zone and one-bit-word skip | **PARK** | PASS | Already capstone points 3 and 9 |
+| **X17** order contract on the join key | **PROBE** (correctness gate, contract) | PASS, conditionally: a separate join-key witness type with a numeric projection, never memcmp, never fused with a row-order digest. Correction: ADMISSION B requires an attested sorted key for **every** same-table key-range seek; frozen-codebook decoding is admitted separately, on an immutable mapping within the attested version, with no order requirement (`entries/2026-10-20-six-decisions.md` §1) | P-X17: binding a merge or seek join to a byte-ordered LE key (u16 1 vs 256) must be refused at bind. An equality or mask join on an identity key must bind without a witness |
+| **X7** min-length or tropical plane | **SKIP** for D-RU-6 | CONFLICT: it changes the answer (law 3) | Only as a separately declared existence or min-length query |
+| **X14** repeated squaring | **SKIP** | CONFLICT: it produces powers of two, not every length, and the source paper loses on sparse input | — |
+| **X13** row skipping in relalg | **SKIP** (already in flight) | PASS | D-CMR-2 |
+| **X15** Brie or B-tree relation storage | **SKIP** | TRAP: a stored second copy | — |
+| **X16** differential dataflow or DBSP | **SKIP** as a dependency | CONFLICT: indexed second copies. The concept already lives in the dirty-row rule | — |
+
+## On the little-endian join contract
+
+**Equality and mask joins need no order contract.** They need identity-canonical keys plus extent bounds.
+
+**Merge, seek and range joins need a total order.** ADMISSION B already demands an attested sorted key for every same-table key-range seek. Frozen-codebook decoding does not: it is admitted on an immutable mapping within the attested version, accounted for and never used as a relational lookup, so a codebook needs a mapping and version attestation, not an order witness (`entries/2026-10-20-six-decisions.md` §1; `plans/d-relalg-unified-v1.md:234`).
+
+**The contract must be numeric order over the LE image.** LE bytes are not memcmp-ordered: u16 256 sorts before 1. That is how `OrderedLaneWitness` defines it. Today the witness covers only sealed facet lanes. So the gap is a **join-key witness type**, checked at bind time and only for carriers that need order. It is not an attestation on every join key.
+
+## Not searched
+
+- No outside source was read in full.
+- SECONDHAND only: Factorized Databases, Beamer SC'12, Bagan–Durand–Grandjean / Segoufin, Umbra, Leapfrog Triejoin.
+- Failed fetches: the Kuzu blog (DNS failure) and the LAGraph BFS template (404).
+- Licences other than Soufflé's are from memory.
+- Inside: ogar-loco and r2il were not searched for an existing demand-typed terminal. lance-graph-quack and cypher-quack were not inspected.
+- The source of the finding that the two attestations must never be merged was not located in a bounded search.
+
+## Candidate SPEC v1 for the 5+3
+
+The ADOPT-NOW set, X1+X2+X3, X4 and X9, plus probe X6, form one coherent design: **a demand-typed terminal** with a plane oracle, in which tuples are built only for Rows. The operator chooses whether this goes to `/5plus3`.
+
+PR #1657 | STATUS: open | OUTCOME: milliseconds attributed per path; folding is missing at the terminal, not the core loop | OPEN: which ideas to take forward; join-key witness type; X8/X11 under the open slot-binding ruling
+
+## Correction (operator, 2026-10-21, after the map)
+
+**Ruling: no millisecond result is acceptable anywhere.** A single operation must finish in nanoseconds; a batch takes microseconds at most.
+
+1. **The tuple decode is the violation itself.** Making it faster (X4) does not satisfy the rule. X4 moves from ADOPT-NOW to **SKIP**, and the "decode cost" part of the 5+3 that followed was dropped.
+2. **lance-graph-mask-risc is the reference folding algebra.** It stays untouched until tests prove a new algebra also runs in nanoseconds.
+3. **The native count fold (X6) was implemented and reverted** (884abfd4..c00011ef, reverted in c49db8dc). On 50,000 centers it measured:
+
+   | case | time |
+   |---|---|
+   | Count(Bag) | 620 µs |
+   | Count(Bag), filtered | 1.2 ms |
+   | Count(Set) | 1.6 ms |
+   | Count(Set), filtered | 1.3 ms |
+
+   It also kept the enumerating terminal, at 243–310 ms. The causes were scalar loops instead of the ndarray polyfill, and a per-slot sort for Set. Any rework must be SIMD through ndarray and must avoid a per-slot sort, which depends on a sorting contract.
+4. **New open points (operator):**
+   - An ndarray plug for the contract crate, plug-and-play like `hotplug.rs`. The zero-dependency rule in `lance-graph-contract/Cargo.toml` is ruled obsolete.
+   - A sorting contract. mask-risc's `CountKeyRunsU32` checks key order at run time, and no attestation backs it.
+   - Whether the substrate is sorted to begin with.
+   - Whether a little-endian or sorting contract would slow folds down or prevent them.
+   - Why Quack does not consume the contract set algebra. Today only relalg and `lance-graph/src/fold` consume it; quack uses the contract only for `OrderedLaneWitness`.
diff --git a/.claude/board/entries/2026-10-21-tuple-zero-copy-council.md b/.claude/board/entries/2026-10-21-tuple-zero-copy-council.md
new file mode 100644
index 000000000..6b77d16a9
--- /dev/null
+++ b/.claude/board/entries/2026-10-21-tuple-zero-copy-council.md
@@ -0,0 +1,155 @@
+# 2026-10-21 — 5+3 council: the tuple is a zero-copy violation (blast radius)
+
+**Status:** FINDING. A policy ruling plus the classified sites it reaches. Nothing is implemented here. D-RU-6 stays Queued/Lab. This entry changes nothing in mask-risc and proposes no new op; any new op needs a law-6 insufficiency proof first. EXO-103 to EXO-118 are the follow-up issues on the Linear coordination board (EXO-50).
+
+## The model
+
+PowerShell `"{0}..{15}" -f $lane.split`: a format template names the positions of one value and reads them in place; it builds nothing. A ClassView carving (`6×(u8:u8)`, `4×(u8:u8:u8)`, `3×(u8:u8:u8:u8)`) is that template over the 16-byte facet. A fold acts on all positions at once (one compare → one mask). A **tuple** is the value split into owned, row-shaped value objects.
+
+## Decisions (policy, not derived facts)
+
+- **F1** (DECISION, lance-graph `CLAUDE.md` P0): no millisecond result anywhere. A single operation runs in nanoseconds, a batch in at most microseconds.
+- **F9** (DECISION, operator 2026-10-21, `CLAUDE.md` P0): a tuple, meaning an owned row-shaped copy of lane values, is a zero-copy violation. That holds as an intermediate and as a terminal, on any production or lab path, at any speed. F1 and F9 are separate grounds: copying breaks F9, and the per-row alloc/clone/sort separately breaks F1.
+- F3 and F7 are read through F9: a terminal's elected or demanded output is **coordinates** (a kept mask, `materialize*` row ids with O(n) stated), never value rows. Basis: `folding-doctrine.md` §1 and §5; `mask-risc/src/ir.rs` doc on `ScatterOrU32`: "the scattered mask is legal only when it IS the externally demanded result … never an intermediate".
+- mask-risc is the reference folding algebra and is frozen until new algebra proves itself in ns.
+
+## What is legal and what is not
+
+Named so no leg collapses into another:
+
+| leg | status |
+|---|---|
+| owned row-shaped value copy (`Vec<Vec<_>>`, `Vec<(Vec<_>,_)>`, tuple sets of values) | **violation** (F9) |
+| row-id population consumed by a further fold, filter, count or join | **violation** (F7) |
+| coordinates out (kept mask, `materialize*` row ids) | legal |
+| borrowed lens: an iterator or callback yielding coordinates or borrowed lane views, never owned values | legal |
+| mask/word scratch and transient tile routing (`folding-doctrine.md` §5) | legal; NOT a tuple |
+| fold of multiplicities (count(A×B)=count(A)·count(B), law 7) | legal |
+| enumeration in an independent test oracle (not the code it checks) | legal only there |
+| columnar lane copy at a cross-process or serialization membrane | **OPEN** (operator) |
+
+## Resolution
+
+- **R1.** In non-test code, no row-of-values or row-id population feeds a further fold, filter, count or join.
+- **R2.** No terminal returns copied row values; scalars (Count, Sum, Min, Max) and coordinates are fine. When a demand genuinely returns rows, the kept mask or row ids plus lane descriptors cross, and the consumer reads the lanes in place, or a borrowed lens yields coordinates or borrowed views.
+- **R2a.** No genuine row-demand gate exists yet. In cypher-quack `demand.rs`, classify fails closed to `Demand::Bindings` on unrecognised shapes. Bindings is also returned for non-row shapes, including `count(DISTINCT c)` with a cross-variable filter. These were observed at those sites; no closed search was run. So Bindings means "refused" today, not "rows requested". Refusal stays the default (`cypher-quack-seam.md`, CURRENT-CONTRACT). → EXO-106.
+- **R3.** Enumeration is licensed only as an independent test oracle (`zero-copy-lens-law.md`: old bodies kept as `#[cfg(test)]` oracles; coresearch shared-oracle rule). Gate it with `#[cfg(test)]` where possible. Where tests/ and examples/ link the normal library, as they currently do in crepe-plan-probe and mork-pathmap-probe, use a named-exemption guard in the style of mask-risc's `exactly_one_materialiser` (`mask-risc/src/exec.rs`). Searched: crepe-plan-probe `src/`; `tuples_over_mask` carries no cfg gate. mask-risc's public `reference_execute` / `reference_scratch` (`lib.rs` re-exports) is an operator decision only. → EXO-105, EXO-109.
+- **R4.** UPHELD for count, exists and total multiplicity: these need no tuple.
+  - `count(A×B)=count(A)·count(B)` is already implemented for non-emitting sides at `crepe-plan-probe/src/native.rs:446-449`.
+  - Distinct across factors uses `ScatterCountU32` (`ir.rs:364`), or `CountKeyRunsU32` on a key-ordered lane (`ir.rs:389`).
+  - OPEN for three cases (law 6: recorded, not designed; → EXO-108):
+    - GAP-A1, scalar distinct count over a projection;
+    - GAP-A2, set-semantics distinct output (a representative kept mask);
+    - GAP-B, bag count over emitting sides, the sum over set slots of the product of per-side fan-outs (only one-lane `MaskedSumI32` exists).
+- **R5.** crepe-plan-probe (lab):
+  - `cse.rs:393-424` and `pipeline.rs:189-219` answer Exists / Count(Set) / Count(Bag) from a tuple `BTreeMap`. This breaks R1 and R4 inside the lab. → EXO-103, EXO-116.
+  - `native.rs:420-479` (`tuples_over_mask`) is a violation in full, on four levels (see the correction below). No half of it is kept. → EXO-104.
+  - `NativeAnswer` (`native.rs:338`) has no reference outside its definition (searched: `crates/**/*.rs`).
+  - Nothing here is promoted out of the lab.
+
+## Replacements that exist (reuse, law 6)
+
+| need | mechanism |
+|---|---|
+| count / exists | mask-risc `Terminal::Count` / `Any`; relalg `Measure::Count` |
+| total multiplicity | law 7 product; mork-pathmap-probe `JoinExec::bag_count` (a scalar per-slot loop, not a mask-risc terminal) |
+| distinct across factors | `ScatterCountU32` (`ir.rs:364`); `CountKeyRunsU32` (`ir.rs:389`, key-ordered lane) |
+| coordinates out | `Terminal::Keep` (`ir.rs:321`); `materialize_rows` (mask-risc `exec.rs`, O(n) stated) |
+| coordinates-first ordering, precedent | lance-graph-report `render.rs` (Keep → row ids → lanes). Its final hydration into values is itself on the follow-up list. |
+| borrowed-lens shape, precedent | mork-pathmap-probe `for_each_answer` (2 sides only); lance-graph-report `exec.rs` `for_each_tuple` |
+
+None found in the crates searched: a public streaming set-bit iterator (crepe-plan-probe calls a local `for_each_set`), a multi-side multiplicity lens, and fan-out for multi-valued emitting sides. → EXO-107.
+
+## Blast radius
+
+The sweep covered non-test `src/` of every crate under `crates/` (548 pattern hits, 73 crates). Rows marked *read* were read in context; *hit-depth* rows were classified from the hit line and its surroundings. This is not a proof of absence.
+
+**Mandatory: crepe-plan-probe only** (3–5 src, 1 test, 5–8 example files). Its sites are listed under R5. The oracles are `reference.rs` and `lib.rs`. Examples time Count(Bag) through the tuple map (`examples/addressed.rs`, `addressed_read.rs`) and time oracle decode (walk_*, measure). Consumers searched: no `Cargo.toml` under `/home/user/*` depends on it, and no `.rs` file in `crates/` names `crepe_plan_probe` outside it.
+
+**No tuple site found by the sweep** (read) in mask-risc, relalg, quack, cypher-quack and lane-fold. Their `Vec<Vec<u64>>` uses are mask scratch. mask-risc's public reference executor is a licensed row-at-a-time oracle (R3). cypher-quack refuses Bindings.
+
+**Follow-up**, hit-depth unless marked:
+- **lance-graph-dir-sim `store.rs` `effective`** (read): Keep mask → `materialize_rows` → `BTreeSet<(Guid128,Guid128)>` → set difference. That is a row population feeding a set operation, in a non-test library path; whether a shipped binary reaches it was not checked. → EXO-117.
+  - **Priority inversion.** This is the only non-lab site, yet it is follow-up only. It is outside the lab refactor because versions have different row spaces plus an overlay, so mask AndNot does not apply directly. Its priority should be decided against the lab work, not after it.
+  - **OPEN (operator):** is a `(Guid,Guid)` pair set a coordinate set or a value tuple?
+- **mork-pathmap-probe:** `join.rs` `answers` → `BTreeMap<Row,u32>` (lab); `rmap.rs` edge/escape sets.
+- **lance-graph-report:** `result.rs` pivot `keys()` → `Vec<Vec<u32>>` (flat_map-clone shape).
+- **lance-graph-contract:** `graph_render.rs` `CypherExecutor` returns `Vec<Vec<CypherValue>>`. No implementor was found in this workspace or in the sibling repos checked: a dormant cross-repo surface.
+- **legacy:** lance-graph-cognitive `cypher_bridge.rs`; lance-graph core `graph/versioned.rs` edge-set difference and arigraph paths; holograph `query/executor.rs`; lance-graph-planner `adjacency/` (mostly build scratch).
+
+→ EXO-118. Eight more crepe-mentioning files were not read line by line.
+
+**Board/docs carrying ms-denominated figures from the tuple paths.** These get an append-only correction note, never a rewrite. The tuple-vs-fold timings stay as dated comparison evidence:
+- `entries/2026-10-21-u3-cross-rule-cse.md`;
+- `entries/2026-10-21-coresearch-ms-vs-ns-folding.md`;
+- `entries/2026-10-20-crepe-rule-optimizer-spike.md`;
+- `plans/crepe-rule-optimizer-spike-v1.md`;
+- STATUS_BOARD D-RU-3 and D-RU-6;
+- the three `entries/2026-10-21-d-ru-6-*.md`.
+
+→ EXO-110.
+
+## Rejected and narrowed objections (kept on record)
+
+- **"Someone pays the split at the membrane anyway."** Rejected in process: in the target design the consumer reads lanes in place. Narrowed across processes: at a serialization membrane (wire clients) lane bytes physically move. Columnar lane transport, never row tuples, is the only candidate shape there, and whether it is an F9 exception is OPEN for the operator.
+- **"A lens terminal is legal"** (v1). Narrowed: legal only when it yields coordinates or borrowed views. A lens that yields owned values is a violation.
+- **"A tuple is a violation because it is slow."** Superseded: it is a violation because it copies. Slowness is a separate ground (F1).
+
+## Gates
+
+- **G1:** every R has a verdict with file:line.
+- **G2:** H answered: ruled by F9 for owned value copies; legal legs named in the table.
+- **G3:** every site classified, with read depth stated; 8 files unread.
+- **G4:** this entry quotes no timing; board figures are named by file only.
+- **G5:** each violation names an existing replacement or a GAP (A1, A2, B, the set-bit iterator, the multi-side lens).
+- **G6:** no replacement yields owned values.
+
+## Council run
+
+| phase | who / what |
+|---|---|
+| 5 savants (Sonnet; S4 on Opus) | prior art, iron rules + fold laws, code truth, cascade / blast radius, different views |
+| 3 reviewers (Sonnet) | overclaim-auditor (2 BLOCK: absolute wording in §2 and §6, both re-worded and re-graded); dilution-collapse-sentinel (0 BLOCK; borrowed-lens and membrane legs restored); firewall-warden (0 BLOCK, all non-negotiables clean) |
+
+v1 → v2 changes: F9 added; R2 replaced; R2a, R3 independence and the gaps added; blast radius filled in.
+
+v2 → v3 changes:
+- **Wording:** absolutes scoped; F9 defined as owned row-shaped value copies.
+- **Legal legs:** borrowed lens and the membrane question restored; legal legs tabled. (v3 R5 said "emitting half"; corrected below.)
+- **Gaps:** GAP-A split into A1 and A2.
+- **Evidence:** citations corrected (`ir.rs:364`, `:389`); ms figures named by file only; scratch-file citations removed; dir-sim priority inversion flagged.
+- **Corrections:** board ms figures get append-only notes, never rewrites.
+
+## Correction (operator, 2026-10-21) — the tuple violates on four levels; no half is kept
+
+v3 said `tuples_over_mask` was a violation "in its emitting half" and that its count multiplication "survives as a fold". That was wrong. The function is a violation in full, on four levels:
+
+1. **Zero copy.** It copies lane values into owned rows, a second owner of substrate bytes (F9).
+2. **Folding.** It enumerates where a fold belongs. Count, exists and multiplicity are answered by decoding rows, not by folding mask words into a terminal.
+3. **Natural order.** The substrate already holds its rows in a natural order: slot order of the mask and the lanes, readable in place at zero copy. `out.sort()` plus merge discards that order and imposes a value order on decoded copies, so the substrate's own order is ignored. It also costs O(n log n). Order belongs to the substrate's coordinates, never to decoded rows.
+4. **Time.** The per-row alloc, clone and sort is what produced the millisecond results (F1).
+
+The multiplication of side counts at `native.rs:446-449` is not a part of the tuple builder to keep. The fold it computes (law 7, count(A×B)=count(A)·count(B)) exists on its own. It belongs at the terminal, over the kept mask in natural order, not inside a function that builds rows. "The fold-shaped half can be kept" was a collapse in the other direction: it filed a valid law under an invalid carrier.
+
+## Correction (operator, 2026-10-21): no boundaries
+
+The legal-legs table above is superseded where it allows anything to cross a
+boundary. There are no boundaries; the whole substrate is zero copy.
+
+- Struck: "coordinates out … legal", "borrowed lens … legal" as a
+  boundary leg, and the OPEN row "columnar lane copy at a cross-process or
+  serialization membrane". There is no membrane at which values are copied.
+- The only "exception" is the inversion, global to local, and it is still
+  zero copy: it materializes the pointer table `$i` and nothing else.
+- A "tuple" is a zero-copy fold template: `"{0}{1}" -f $laneA[$i],$laneB[$i]`.
+  A `(Guid, Guid)` pair held as a value (dir-sim, EXO-117) is the violation;
+  two lanes read at `$i` is the correct shape.
+- Oracles included: tests compare answers by folds through the template
+  (crepe-plan-probe `Answers::{total, matching, for_each}`), never by
+  expanding rows.
+- Open, found by the same rule: `mork_pathmap_probe::join::Side::prepare`
+  copies the non-key column into `vals`. Under this ruling it should hold
+  only `$i` (row positions per slot) and read `rows[$i]` in place. Its
+  `prepare` stage fails the no-ms gate at n = 10 000 and 50 000 (the
+  `no-semijoin` and `none` arms of `crepe-plan-probe` example `measure`).
+
diff --git a/.claude/board/entries/README.md b/.claude/board/entries/README.md
index b5018bf36..1d2b31da1 100644
--- a/.claude/board/entries/README.md
+++ b/.claude/board/entries/README.md
@@ -25,12 +25,13 @@ 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.
 
-407 entries, 2026-08-06 .. 2026-10-21.
+409 entries, 2026-08-06 .. 2026-10-21.
 
 | date | entry id | finding | file |
 |---|---|---|---|
 | 2026-10-21 | `D-RU-3` |  | [2026-10-21-u3-cross-rule-cse.md](2026-10-21-u3-cross-rule-cse.md) |
 | 2026-10-21 | `D-RU-2` |  | [2026-10-21-u2-let-planes.md](2026-10-21-u2-let-planes.md) |
+| 2026-10-21 | `tuple-zero-copy-council` |  | [2026-10-21-tuple-zero-copy-council.md](2026-10-21-tuple-zero-copy-council.md) |
 | 2026-10-21 | `D-RU-6-LAB-1` |  | [2026-10-21-d-ru-6-tile-walk-probe.md](2026-10-21-d-ru-6-tile-walk-probe.md) |
 | 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) |
@@ -39,6 +40,7 @@ exactly one of them, never both.
 | 2026-10-21 | `D-CMR-5` |  | [2026-10-21-d-cmr-5-rescope.md](2026-10-21-d-cmr-5-rescope.md) |
 | 2026-10-21 | `D-CMR-2b` |  | [2026-10-21-d-cmr-2b-scratch-probe.md](2026-10-21-d-cmr-2b-scratch-probe.md) |
 | 2026-10-21 | `D-CMR-2` |  | [2026-10-21-d-cmr-2-lowering.md](2026-10-21-d-cmr-2-lowering.md) |
+| 2026-10-21 | `D-RU-6` |  | [2026-10-21-coresearch-ms-vs-ns-folding.md](2026-10-21-coresearch-ms-vs-ns-folding.md) |
 | 2026-10-20 | `D-RU-1` |  | [2026-10-20-u1-jc-batch.md](2026-10-20-u1-jc-batch.md) |
 | 2026-10-20 | `thought-git-replay-api-write-rubicon` |  | [2026-10-20-thought-git-replay-api-write-rubicon.md](2026-10-20-thought-git-replay-api-write-rubicon.md) |
 | 2026-10-20 | `ADR-2026-10-20-SIX` |  | [2026-10-20-six-decisions.md](2026-10-20-six-decisions.md) |
diff --git a/CLAUDE.md b/CLAUDE.md
index 3867c9c33..25b16121b 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -14,6 +14,73 @@
 
 `.grok/board/CLAUDE_LANE_FOLD_CAPSTONE.md`. Settled numbers, the ten open points, the refusal list. Do not re-derive. Do not cite 0.6 ns as a join.
 
+## P0 — no millisecond results, anywhere, ever
+
+**Nanoseconds for a single operation. At most microseconds for a batch. A
+result in milliseconds is a failed gate, never a measurement to report,
+discuss, commit or push.** This holds for lab crates, probes, examples,
+benchmarks and PR bodies alike.
+
+- DECISION (2026-10-21): an ms number means the work left the fold. The
+  substrate's reference is mask-risc / Quack, which fold over u64 mask words
+  into a terminal (`Count`, `Any`, `Keep`, …) in nanoseconds;
+  `folding-doctrine.md` §6.2 records 56 ns fused against 138 620 ns
+  materialized for the same query.
+- BASIS: every ms result seen so far came from leaving that path. Examples:
+  `crepe-plan-probe::native::tuples_over_mask` (a heap row per result, a copy
+  per combination, then sort and merge: 243–310 ms), and a count "fold" made of
+  scalar loops plus a per-slot sort (1.2–1.6 ms; pushed, then reverted in
+  `c49db8dc`). See `.claude/board/entries/2026-10-21-coresearch-ms-vs-ns-folding.md`.
+- **The mental model.** A value is read through a template, never split into
+  objects. PowerShell's `"{0}..{15}" -f $lane.split` builds nothing: it names
+  positions 0–15 of one value and reads them in place. A ClassView carving
+  (`6×(u8:u8)`, `4×(u8:u8:u8)`, `3×(u8:u8:u8:u8)`) is that template over the
+  16-byte facet. A fold acts on all positions at once (one compare → one mask).
+  A tuple, by contrast, is the value split into owned heap objects. It
+  discards the template, and it is where the milliseconds come from.
+- **A tuple is a violation on four levels** (2026-10-21), and no part of a
+  tuple builder is kept:
+  1. **Zero copy:** copying lane values into owned rows makes a second owner
+     of bytes the substrate already holds (`.claude/knowledge/zero-copy-lens-law.md`:
+     no escape hatch, no cost argument).
+  2. **Folding:** it enumerates where a fold belongs.
+  3. **Natural order:** the substrate already holds rows in its natural order
+     (mask/lane slot order, readable in place). Sorting decoded rows ignores
+     that order and imposes another one on copies. Order belongs to the
+     substrate's coordinates, never to decoded rows.
+  4. **Time:** the per-row alloc, clone and sort is where the milliseconds
+     come from.
+  It is a violation as an intermediate or a final answer, at any speed. A fold
+  that happens to sit inside a tuple builder (for example multiplying side
+  counts) is not "the good half": it belongs at the terminal, over the kept
+  mask.
+- **What follows for a session:**
+  - **Before reporting any timing,** check the unit. ms → stop; find where
+    the path left the mask-risc terminals.
+  - **Never decode a join or population into owned rows** (`Vec<Vec<_>>`,
+    `Vec<(Vec<_>, _)>`, tuple sets) to answer count, exists, sum, min, max or
+    group. Those are folds.
+  - **Never invent a "reference" or tuple decoder** that was not asked for.
+  - **mask-risc is the reference folding algebra.** Do not change it for a
+    new algebra until tests prove the new algebra also runs in nanoseconds.
+- SCOPE: every crate in this workspace, plus every number quoted in a board
+  entry, plan, PR or Linear comment.
+- **There are no boundaries** (2026-10-21). The whole substrate is zero copy:
+  no membrane, wire, process edge or "consumer side" where values get copied
+  out. The only "exception" is the **inversion, global to local**, and it is
+  still zero copy: the one thing it materializes is the pointer table `$i`
+  (the local positions). Everything else is read in place through `$i`.
+  - **A "tuple" is a zero-copy fold template,** never a value:
+    `"{0}{1}" -f $laneA[$i],$laneB[$i]` reads column 0 from lane A and column
+    1 from lane B at position `$i`. Whole-template form:
+    `"{$i[0]}..{$i[15]}" -f $table{$i[0]}..$table{$i[15]}`.
+  - **Violation:** a `(Guid, Guid)` pair stored or returned as a value (as in
+    dir-sim). **Correct:** two lanes, one pointer table, the template read at
+    `$i`.
+  - This holds for tests and oracles too: answers are compared by folds
+    through the template, never by expanding them into rows.
+- REVISIT WHEN: never on cost grounds.
+
 ## P0 — AdaWorldAPI forks ONLY, NEVER crates.io upstream
 
 **Always depend on the AdaWorldAPI fork of any crate that has one. NEVER use the
diff --git a/crates/crepe-plan-probe/examples/measure.rs b/crates/crepe-plan-probe/examples/measure.rs
index 88d5eb691..703bc602f 100644
--- a/crates/crepe-plan-probe/examples/measure.rs
+++ b/crates/crepe-plan-probe/examples/measure.rs
@@ -11,18 +11,22 @@
 //! Parity is checked against datafrog at every size and against the nested-loop
 //! reference up to 10,000 rows. Timings are medians of `REPS` runs.
 
-use std::collections::BTreeMap;
 use std::hint::black_box;
 use std::time::{Duration, Instant};
 
 use crepe_plan_probe::logic::{Labels, Rule, resolve};
 use crepe_plan_probe::mork_api::{admit, query_multi};
-use crepe_plan_probe::native::{StarPlan, Store, bindings, joint, prepare, prepare_semijoin, star};
+use crepe_plan_probe::native::{
+    DistinctGroup, RuleFilter, StarPlan, StarView, Store, add_distinct, canonical, joint, kept,
+    narrow_center, prepare, prepare_semijoin, star,
+};
 use crepe_plan_probe::optimize::{Demand, Options, ResultKind, optimize};
+use crepe_plan_probe::pipeline::Answers;
 use crepe_plan_probe::programs::family;
 use crepe_plan_probe::reference;
 use lance_graph_relalg::{Measure, Rel, normalize};
 use mork_pathmap_probe::Semantics;
+use mork_pathmap_probe::join::Side;
 
 const REPS: usize = 31;
 
@@ -183,23 +187,38 @@ fn main() {
                 programs += run.programs;
                 ops += prog.map_or(0, |p| p.ops.len());
             }
+            // The terminal: each rule's kept mask folded into the free
+            // column's distinct values (a value-space mask). No row is built.
             let answer = || {
-                let mut out: BTreeMap<Vec<u64>, u64> = BTreeMap::new();
+                let mut groups: Vec<DistinctGroup> = Vec::new();
                 for (s, p) in stars.iter().zip(&preps) {
-                    for (row, m) in bindings(p, s, &gens(s), false).unwrap() {
-                        *out.entry(row).or_insert(0) += m;
-                    }
+                    let (mut mask, _, _) = kept(p, s, &gens(s)).unwrap();
+                    let Some(filter) = RuleFilter::of(s, &opt.post_filter) else {
+                        continue;
+                    };
+                    narrow_center(&mut mask, &p.table.domain, &filter);
+                    let sides: Vec<&Side> = p.table.sides.iter().collect();
+                    let view = StarView {
+                        domain: &p.table.domain,
+                        sides: &sides,
+                        star: s,
+                    };
+                    add_distinct(&view, &mask, &filter, &mut groups).unwrap();
                 }
-                out.retain(|row, _| {
-                    row.iter()
-                        .zip(&opt.post_filter)
-                        .all(|(v, f)| f.is_none_or(|f| f == *v))
-                });
-                out.into_keys().collect::<Vec<_>>()
+                canonical(&mut groups);
+                groups
             };
             let terminal_t = time(answer);
-            let got = answer();
-            let parity = got == df && reference.as_ref().is_none_or(|r| *r == got);
+            // Parity against the oracles, outside the timed terminal, by
+            // folds: same count, and every oracle row answered once.
+            let got = Answers::Distinct(answer());
+            let agrees = |rows: &[Vec<u64>]| {
+                got.total(&[]) == rows.len() as u64
+                    && rows
+                        .iter()
+                        .all(|r| got.matching(&[], |c| r.get(c).copied()) == 1)
+            };
+            let parity = agrees(&df) && reference.as_deref().is_none_or(agrees);
             println!(
                 "{n},{name},{},{},{},{},{},{:.1},{},{:.1},{},{},{programs},{ops},{},{}",
                 plan_t.as_nanos(),
@@ -212,7 +231,7 @@ fn main() {
                 terminal_t.as_secs_f64() * 1e6,
                 preps.iter().map(|p| p.rows_prepared).sum::<usize>(),
                 opt.rules.len(),
-                got.len(),
+                got.total(&[]),
                 if parity { "ok" } else { "MISMATCH" },
             );
             assert!(parity, "answers differ from the oracles at n = {n}, {name}");
diff --git a/crates/crepe-plan-probe/examples/rule_cse.rs b/crates/crepe-plan-probe/examples/rule_cse.rs
index 38758a2d2..2fc0fbb90 100644
--- a/crates/crepe-plan-probe/examples/rule_cse.rs
+++ b/crates/crepe-plan-probe/examples/rule_cse.rs
@@ -41,8 +41,10 @@
 //! because the shared path cannot apply a per-rule semijoin.
 //!
 //! Two answers are timed. `Exists` is mask-only: it isolates the folds.
-//! `Bindings(Bag)` is the full answer with multiplicities; it adds reading
-//! the tuples, which is the same work in every arm.
+//! `Bindings(Bag)` hands out each rule's kept slots over the prepared tables;
+//! no row is built. The correctness checks, outside every timed region,
+//! compare answers by folds: a row is read through its head template over
+//! the lanes in place, never copied.
 
 use std::collections::BTreeSet;
 use std::time::Duration;
@@ -50,9 +52,9 @@ use std::time::Duration;
 use crepe::crepe;
 use crepe_plan_probe::cse::{Layout, SharedReport, evaluate_shared};
 use crepe_plan_probe::logic::{Atom, RelId, Rule, Term, VarId};
-use crepe_plan_probe::native::Store;
+use crepe_plan_probe::native::{PreparedStar, Store};
 use crepe_plan_probe::optimize::{Options, ResultKind};
-use crepe_plan_probe::pipeline::{Outcome, Report, demand, evaluate};
+use crepe_plan_probe::pipeline::{Answers, Outcome, Report, demand, evaluate};
 use mork_pathmap_probe::Semantics;
 
 /// Rewrites on, semijoin off: the arms differ only in sharing.
@@ -242,16 +244,54 @@ pub(crate) fn ascent_answer(w: &Workload) -> Tuples {
     prog.out.into_iter().collect()
 }
 
-pub(crate) fn as_set(o: &Outcome) -> Tuples {
+fn answers(o: &Outcome) -> &Answers {
     match o {
-        Outcome::Bindings(b) => b
-            .iter()
-            .map(|(r, _)| (r[0] as u32, r[1] as u32, r[2] as u32))
-            .collect(),
+        Outcome::Bindings(a) => a,
         other => panic!("expected bindings, got {other:?}"),
     }
 }
 
+/// Two answers agree: same total, and every row one of them reads (through
+/// its head template, over the lanes in place) is answered the same number
+/// of times by the other. Folds; no row is built.
+pub(crate) fn same(a: &Outcome, ta: &[PreparedStar], b: &Outcome, tb: &[PreparedStar]) -> bool {
+    let (a, b) = (answers(a), answers(b));
+    if a.total(ta) != b.total(tb) {
+        return false;
+    }
+    let mut ok = true;
+    a.for_each(ta, |get, _| {
+        let me = |c: usize| Some(get(c));
+        ok &= a.matching(ta, me) == b.matching(tb, me);
+    });
+    ok
+}
+
+/// The answer agrees with an oracle's set: every oracle tuple is answered,
+/// and every answered row is an oracle tuple.
+pub(crate) fn same_set(o: &Outcome, t: &[PreparedStar], set: &Tuples) -> bool {
+    let a = answers(o);
+    let mut ok = set.iter().all(|&(x, y, z)| {
+        let row = [x, y, z];
+        a.matching(t, |c| row.get(c).map(|&v| u64::from(v))) > 0
+    });
+    a.for_each(t, |get, _| {
+        ok &= set.contains(&(get(0) as u32, get(1) as u32, get(2) as u32));
+    });
+    ok
+}
+
+/// Against rows written out by a test: same total, each row answered its
+/// multiplicity.
+#[cfg(test)]
+pub(crate) fn same_bag(o: &Outcome, t: &[PreparedStar], want: &[(Vec<u64>, u64)]) -> bool {
+    let a = answers(o);
+    a.total(t) == want.iter().map(|(_, m)| m).sum::<u64>()
+        && want
+            .iter()
+            .all(|(row, m)| a.matching(t, |c| row.get(c).copied()) == *m)
+}
+
 fn arm_a(w: &Workload, kind: ResultKind) -> (Outcome, Report) {
     let d = demand(w.out_rel(), vec![None; 3]);
     evaluate(&w.rules, &d, kind, OPTS, &w.store()).expect("arm A")
@@ -294,16 +334,16 @@ fn main() {
         let bag = ResultKind::Bindings(Semantics::Bag);
 
         // Correctness, outside every timed region.
-        let (a, _) = arm_a(&w, bag);
-        let (d0, _) = arm_d(&w, bag, Layout::Components, false);
-        let (d, _) = arm_d(&w, bag, Layout::Components, true);
-        assert_eq!(a, d0, "D0 vs A");
-        assert_eq!(a, d, "D vs A");
+        let (a, ra) = arm_a(&w, bag);
+        let (d0, r0) = arm_d(&w, bag, Layout::Components, false);
+        let (d, rd) = arm_d(&w, bag, Layout::Components, true);
+        assert!(same(&a, &ra.tables, &d0, &r0.tables), "D0 vs A");
+        assert!(same(&a, &ra.tables, &d, &rd.tables), "D vs A");
         let checked = if oracle {
-            let set = as_set(&a);
-            assert_eq!(set, crepe_answer(&w), "A vs crepe");
-            assert_eq!(set, ascent_answer(&w), "A vs ascent");
-            format!("crepe = ascent = A = D0 = D ({} tuples)", set.len())
+            let set = crepe_answer(&w);
+            assert!(same_set(&a, &ra.tables, &set), "A vs crepe");
+            assert_eq!(set, ascent_answer(&w), "crepe vs ascent");
+            format!("crepe = ascent = A = D0 = D ({} oracle tuples)", set.len())
         } else {
             "A = D0 = D (oracles skipped)".to_string()
         };
@@ -358,7 +398,7 @@ fn main() {
                     t.1.push(r.execute);
                     split.0.push(r.share);
                     split.1.push(r.fold);
-                    split.2.push(r.tuples);
+                    split.2.push(r.terminal);
                     last = Some(r);
                 }
                 let r = last.expect("reps >= 1");
@@ -376,7 +416,7 @@ fn main() {
                     r.cost.plane_bytes
                 );
                 println!(
-                    "     execute = share {:.2}ms + folds {:.2}ms + tuples {:.2}ms",
+                    "     execute = share {:.2}ms + folds {:.2}ms + terminal {:.2}ms",
                     ms(median(split.0)),
                     ms(median(split.1)),
                     ms(median(split.2))
@@ -391,7 +431,7 @@ mod tests {
     use super::*;
     use crepe_plan_probe::cse::{CseRefusal, execute_shared, prepare_shared};
     use crepe_plan_probe::logic::RelId as Rid;
-    use crepe_plan_probe::pipeline::plan;
+    use crepe_plan_probe::pipeline::{PipelineRefusal, plan};
     use crepe_plan_probe::reference::eval;
     use mork_pathmap_probe::join::JoinRefusal;
 
@@ -421,28 +461,48 @@ mod tests {
         for shared in [0, RULES / 2, RULES * 9 / 10, RULES] {
             let w = workload(RULES, shared, CENTERS, ROWS);
             let bag = ResultKind::Bindings(Semantics::Bag);
-            let (a, _) = arm_a(&w, bag);
-            let (d0, _) = arm_d(&w, bag, Layout::Components, false);
-            let (d, _) = arm_d(&w, bag, Layout::Components, true);
-            let Outcome::Bindings(ab) = &a else {
-                panic!("bindings")
-            };
-            assert!(ab.len() > 50, "fixture too thin: {} tuples", ab.len());
+            let (a, ra) = arm_a(&w, bag);
+            let (d0, r0) = arm_d(&w, bag, Layout::Components, false);
+            let (d, rd) = arm_d(&w, bag, Layout::Components, true);
+            let want = reference_bag(&w);
+            assert!(want.len() > 50, "fixture too thin: {} rows", want.len());
             assert!(
-                ab.iter().any(|(_, m)| *m > 1),
+                want.iter().any(|(_, m)| *m > 1),
                 "fixture has no multiplicity above 1"
             );
-            assert_eq!(*ab, reference_bag(&w), "A vs reference, shared={shared}");
-            assert_eq!(d0, a, "D0 vs A, shared={shared}");
-            assert_eq!(d, a, "D vs A, shared={shared}");
-            let set = as_set(&a);
-            assert_eq!(set, crepe_answer(&w), "crepe, shared={shared}");
+            assert!(
+                same_bag(&a, &ra.tables, &want),
+                "A vs reference, shared={shared}"
+            );
+            assert!(
+                same(&d0, &r0.tables, &a, &ra.tables),
+                "D0 vs A, shared={shared}"
+            );
+            assert!(
+                same(&d, &rd.tables, &a, &ra.tables),
+                "D vs A, shared={shared}"
+            );
+            let set = crepe_answer(&w);
+            assert_eq!(set.len(), want.len(), "crepe, shared={shared}");
+            assert!(same_set(&a, &ra.tables, &set), "crepe, shared={shared}");
             assert_eq!(set, ascent_answer(&w), "ascent, shared={shared}");
-            for kind in [
-                ResultKind::Exists,
-                ResultKind::Count(Semantics::Set),
-                ResultKind::Count(Semantics::Bag),
-            ] {
+            // `Out(k, x, z)` has two free columns (x, z): a set count would
+            // need a seen-set over value pairs, so every arm refuses it.
+            let d = demand(w.out_rel(), vec![None; 3]);
+            let set_count = ResultKind::Count(Semantics::Set);
+            assert!(matches!(
+                evaluate(&w.rules, &d, set_count, OPTS, &w.store()),
+                Err(PipelineRefusal::Set { .. })
+            ));
+            for layout in [Layout::OneTable, Layout::Components] {
+                for sharing in [true, false] {
+                    assert!(matches!(
+                        evaluate_shared(&w.rules, &d, set_count, OPTS, &w.store(), layout, sharing),
+                        Err(CseRefusal::Pipeline(PipelineRefusal::Set { .. }))
+                    ));
+                }
+            }
+            for kind in [ResultKind::Exists, ResultKind::Count(Semantics::Bag)] {
                 let (want, _) = arm_a(&w, kind);
                 for layout in [Layout::OneTable, Layout::Components] {
                     assert_eq!(arm_d(&w, kind, layout, true).0, want, "{kind:?}");
@@ -467,11 +527,11 @@ mod tests {
         ] {
             let opts = Options { propagate, ..OPTS };
             let d = demand(w.out_rel(), bound.clone());
-            let (a, _) = evaluate(&w.rules, &d, bag, opts, &w.store()).expect("A");
-            let Outcome::Bindings(ab) = &a else {
-                panic!("bindings")
-            };
-            assert!(!ab.is_empty(), "demand {bound:?} selects nothing");
+            let (a, ra) = evaluate(&w.rules, &d, bag, opts, &w.store()).expect("A");
+            assert!(
+                answers(&a).total(&ra.tables) > 0,
+                "demand {bound:?} selects nothing"
+            );
             let store = w.store();
             let rows = |r: Rid| -> Vec<Vec<u64>> {
                 store
@@ -484,9 +544,9 @@ mod tests {
             let want: Vec<(Vec<u64>, u64)> = eval(&w.rules, w.out_rel(), &bound, &rows)
                 .into_iter()
                 .collect();
-            assert_eq!(*ab, want, "A vs reference, {bound:?}");
+            assert!(same_bag(&a, &ra.tables, &want), "A vs reference, {bound:?}");
             for sharing in [false, true] {
-                let (dd, _) = evaluate_shared(
+                let (dd, rdd) = evaluate_shared(
                     &w.rules,
                     &d,
                     bag,
@@ -496,7 +556,10 @@ mod tests {
                     sharing,
                 )
                 .expect("D");
-                assert_eq!(dd, a, "sharing={sharing}, {bound:?}, propagate={propagate}");
+                assert!(
+                    same(&dd, &rdd.tables, &a, &ra.tables),
+                    "sharing={sharing}, {bound:?}, propagate={propagate}"
+                );
             }
         }
     }
diff --git a/crates/crepe-plan-probe/src/cse.rs b/crates/crepe-plan-probe/src/cse.rs
index 70e535758..86d401993 100644
--- a/crates/crepe-plan-probe/src/cse.rs
+++ b/crates/crepe-plan-probe/src/cse.rs
@@ -18,8 +18,10 @@
 //!    [`share`] lifts every part two or more of the group's rules hold into
 //!    a let (`lance_graph_contract::set_program`);
 //! 4. [`run_program`] evaluates each let once into a plane and every rule
-//!    over the sides plus those planes. The tuples are read per rule with
-//!    [`tuples_over_mask`], exactly as the per-rule path reads them.
+//!    over the sides plus those planes. Each rule's kept mask then goes
+//!    through the same fold terminals as the per-rule path
+//!    ([`crate::native::bag_count`], [`crate::native::add_distinct`], …);
+//!    no row is built.
 //!
 //! Width is the price. A side is laid out densely over its table's domain,
 //! so a table pays per slot for every side it holds. [`Layout::OneTable`]
@@ -46,9 +48,12 @@ use mork_pathmap_probe::Semantics;
 use mork_pathmap_probe::join::{JoinRefusal, Side};
 
 use crate::logic::{Rule, VarId};
-use crate::native::{PreparedStar, SideKey, SidePlan, StarPlan, Store, prepare, tuples_over_mask};
+use crate::native::{
+    DistinctGroup, PreparedStar, RuleCoordinates, RuleFilter, SideKey, SidePlan, StarPlan,
+    StarView, Store, add_distinct, any_derivation, bag_count, canonical, narrow_center, prepare,
+};
 use crate::optimize::{Demand, Options, ResultKind};
-use crate::pipeline::{Outcome, PipelineRefusal, Planned, plan};
+use crate::pipeline::{Answers, Outcome, PipelineRefusal, Planned, plan};
 
 /// How the stars are laid out.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -217,7 +222,7 @@ pub enum CseRefusal {
 
 /// What one shared evaluation cost, each part on its own, summed over the
 /// groups.
-#[derive(Debug, Clone, Copy, Default)]
+#[derive(Debug, Clone, Default)]
 pub struct SharedReport {
     /// Tables laid out.
     pub groups: usize,
@@ -238,10 +243,14 @@ pub struct SharedReport {
     pub share: Duration,
     /// The folds: lets and rules.
     pub fold: Duration,
-    /// Reading the tuples over the kept masks.
-    pub tuples: Duration,
-    /// All of the run half: checks, share, folds, tuples.
+    /// The terminal folds over the kept masks.
+    pub terminal: Duration,
+    /// All of the run half: checks, share, folds, terminal.
     pub execute: Duration,
+    /// The shared tables, in group order: what [`Answers::Derivations`]
+    /// addresses. Filled by [`evaluate_shared`], which owns them;
+    /// [`execute_shared`] leaves it empty (the caller holds the tables).
+    pub tables: Vec<PreparedStar>,
 }
 
 /// The program a group runs: its rules over the table's sides, with the
@@ -283,6 +292,7 @@ pub fn evaluate_shared(
     let (outcome, mut report) = execute_shared(&planned, &shared, kind, &gens, sharing)?;
     report.plan = plan_cost;
     report.prepare = prepare_cost;
+    report.tables = shared.groups.into_iter().map(|g| g.prepared).collect();
     Ok((outcome, report))
 }
 
@@ -346,8 +356,10 @@ pub fn execute_shared(
         ..SharedReport::default()
     };
     let mut any = false;
-    let mut bag: BTreeMap<Vec<u64>, u64> = BTreeMap::new();
-    for (group, gens) in shared.groups.iter().zip(generations) {
+    let mut derivations = 0u64;
+    let mut groups: Vec<DistinctGroup> = Vec::new();
+    let mut coordinates: Vec<RuleCoordinates> = Vec::new();
+    for (gi, (group, gens)) in shared.groups.iter().zip(generations).enumerate() {
         let table = &group.prepared.table;
         if gens.len() != table.sides.len() {
             return Err(CseRefusal::Generations {
@@ -378,6 +390,7 @@ pub fn execute_shared(
         let tf = Instant::now();
         let run = run_program(&prog, version, &planes, &vec![measure; rules], &mut sinks)
             .map_err(CseRefusal::Program)?;
+        drop(sinks);
         report.fold += tf.elapsed();
 
         let tt = Instant::now();
@@ -385,17 +398,44 @@ pub fn execute_shared(
             any |= run.answers.iter().any(|a| *a != Answer::Count(0));
         } else {
             for (i, &j) in group.stars.iter().enumerate() {
+                let star = &planned.stars[j];
+                let Some(filter) = RuleFilter::of(star, &planned.optimized.post_filter) else {
+                    continue;
+                };
+                let mut mask = std::mem::take(&mut kept[i]);
+                narrow_center(&mut mask, &table.domain, &filter);
                 let sides: Vec<&Side> = group.rule_sides[i]
                     .iter()
                     .map(|&k| &table.sides[k])
                     .collect();
-                let star = &planned.stars[j];
-                for (row, m) in tuples_over_mask(&table.domain, &sides, star, &kept[i], true) {
-                    *bag.entry(row).or_insert(0) += m;
+                let view = StarView {
+                    domain: &table.domain,
+                    sides: &sides,
+                    star,
+                };
+                let rule = planned.optimized.rules[j].0;
+                match kind {
+                    ResultKind::Exists => any |= any_derivation(&view, &mask, &filter),
+                    ResultKind::Count(Semantics::Bag) => {
+                        derivations += bag_count(&view, &mask, &filter)
+                    }
+                    ResultKind::Count(Semantics::Set) | ResultKind::Bindings(Semantics::Set) => {
+                        add_distinct(&view, &mask, &filter, &mut groups).map_err(|why| {
+                            CseRefusal::Pipeline(PipelineRefusal::Set { rule, why })
+                        })?
+                    }
+                    ResultKind::Bindings(Semantics::Bag) => coordinates.push(RuleCoordinates {
+                        rule,
+                        table: gi,
+                        sides: group.rule_sides[i].clone(),
+                        star: star.clone(),
+                        filter,
+                        kept: mask,
+                    }),
                 }
             }
         }
-        report.tuples += tt.elapsed();
+        report.terminal += tt.elapsed();
         report.sides += table.sides.len();
         report.domain += n;
         report.rows_prepared += group.prepared.rows_prepared;
@@ -410,25 +450,19 @@ pub fn execute_shared(
         report.cost.max_scratch_slots = report.cost.max_scratch_slots.max(c.max_scratch_slots);
     }
     let tt = Instant::now();
-    let outcome = if mask_only {
-        Outcome::Exists(any)
-    } else {
-        bag.retain(|row, _| {
-            row.iter()
-                .zip(&planned.optimized.post_filter)
-                .all(|(v, f)| f.is_none_or(|f| f == *v))
-        });
-        match kind {
-            ResultKind::Exists => Outcome::Exists(!bag.is_empty()),
-            ResultKind::Count(Semantics::Set) => Outcome::Count(bag.len() as u64),
-            ResultKind::Count(Semantics::Bag) => Outcome::Count(bag.values().sum()),
-            ResultKind::Bindings(Semantics::Set) => {
-                Outcome::Bindings(bag.into_keys().map(|r| (r, 1)).collect())
-            }
-            ResultKind::Bindings(Semantics::Bag) => Outcome::Bindings(bag.into_iter().collect()),
+    canonical(&mut groups);
+    let outcome = match kind {
+        ResultKind::Exists => Outcome::Exists(any),
+        ResultKind::Count(Semantics::Bag) => Outcome::Count(derivations),
+        ResultKind::Count(Semantics::Set) => {
+            Outcome::Count(groups.iter().map(DistinctGroup::len).sum())
+        }
+        ResultKind::Bindings(Semantics::Set) => Outcome::Bindings(Answers::Distinct(groups)),
+        ResultKind::Bindings(Semantics::Bag) => {
+            Outcome::Bindings(Answers::Derivations(coordinates))
         }
     };
-    report.tuples += tt.elapsed();
+    report.terminal += tt.elapsed();
     report.execute = t.elapsed();
     Ok((outcome, report))
 }
diff --git a/crates/crepe-plan-probe/src/native.rs b/crates/crepe-plan-probe/src/native.rs
index 004a6d09a..e1568efd6 100644
--- a/crates/crepe-plan-probe/src/native.rs
+++ b/crates/crepe-plan-probe/src/native.rs
@@ -18,9 +18,19 @@
 //! ```
 //!
 //! is the set of `c` values every side has. That mask is all relalg computes.
-//! Existence and distinct-`c` counts read only it; tuple emission and bag
-//! counts are a separate terminal that reads each side at the slot the mask
-//! names, never at a position taken from another side's values.
+//! Every terminal folds over it in slot order and builds no row:
+//!
+//! - existence and the bag count (`Σ_slot Π_i |side_i at slot|`, law 7) read
+//!   the sides' lengths in place;
+//! - a set answer with one free head column scatters that column's values
+//!   into a mask over the value space; more free columns are refused;
+//! - bag bindings are handed out as coordinates (the kept slots) plus the
+//!   sides' rows borrowed in place; the caller applies the head template.
+//!
+//! A side is read at the slot the mask names, never at a position taken from
+//! another side's values. No terminal copies values into owned rows or sorts
+//! them: a tuple violates zero copy, folding and the substrate's natural
+//! (slot) order (lance-graph `CLAUDE.md`, P0).
 //!
 //! Anything else (a chain, a pattern of arity > 2, a side without `c`) is
 //! refused by name.
@@ -201,6 +211,7 @@ impl<'a> Store<'a> {
 
 /// A relationship table prepared for one plan's sides, with the witness that
 /// says which sides those are and which domain they share.
+#[derive(Debug, Clone)]
 pub struct PreparedStar {
     pub table: JoinTable,
     pub keys: Vec<SideKey>,
@@ -333,15 +344,6 @@ impl PreparedStar {
     }
 }
 
-/// What the native path answers.
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub enum NativeAnswer {
-    Exists(bool),
-    Count(u64),
-    /// Output tuples with their multiplicities (all 1 under set semantics).
-    Bindings(Vec<(Vec<u64>, u64)>),
-}
-
 /// The joint mask and its relalg run, after the witness and generation checks.
 pub fn joint(
     prep: &PreparedStar,
@@ -381,101 +383,452 @@ pub fn distinct_centers(
     }
 }
 
-/// Tuples over the joint mask. Reads each side at the slot the mask names.
-/// A side whose value the template does not use still multiplies the count of
-/// every tuple under bag semantics: each of its rows at that slot is one more
-/// derivation.
-pub fn bindings(
+/// The joint mask kept for the terminal, the relalg run, and the program
+/// that produced it. The mask is the answer's coordinates: slots of the
+/// shared domain, in natural order.
+pub fn kept(
     prep: &PreparedStar,
     plan: &StarPlan,
     current: &[u64],
-    bag: bool,
-) -> Result<Vec<(Vec<u64>, u64)>, NativeRefusal> {
-    bindings_run(prep, plan, current, bag).map(|(rows, _, _)| rows)
+) -> Result<(Vec<u64>, Run, Option<Program>), NativeRefusal> {
+    let mut mask = vec![0u64; words_for(prep.table.domain.len())];
+    let (run, program) = joint(prep, plan, current, Measure::Keep, Some(&mut mask))?;
+    Ok((mask, run, program))
 }
 
-/// Rows with multiplicities, the relalg run, and the program behind the mask.
-pub type BindingsRun = (Vec<(Vec<u64>, u64)>, Run, Option<Program>);
+/// One star read through its laid-out sides: the domain the kept mask
+/// addresses and, per plan side, the side holding it. Borrowed; nothing is
+/// copied out of the table.
+#[derive(Clone, Copy)]
+pub struct StarView<'a> {
+    pub domain: &'a JoinDomain,
+    pub sides: &'a [&'a Side],
+    pub star: &'a StarPlan,
+}
 
-/// [`bindings`], plus the relalg run and the program that produced the mask.
-pub fn bindings_run(
-    prep: &PreparedStar,
-    plan: &StarPlan,
-    current: &[u64],
-    bag: bool,
-) -> Result<BindingsRun, NativeRefusal> {
-    let n = prep.table.domain.len();
-    let mut mask = vec![0u64; words_for(n)];
-    let (run, program) = joint(prep, plan, current, Measure::Keep, Some(&mut mask))?;
-    let sides: Vec<&Side> = prep.table.sides.iter().collect();
-    let rows = tuples_over_mask(&prep.table.domain, &sides, plan, &mask, bag);
-    Ok((rows, run, program))
-}
-
-/// The output tuples of `plan` at the slots `mask` names, with
-/// multiplicities. `sides[i]` is the laid-out side for `plan.sides[i]`, over
-/// `domain`. Each side is read at the slot the mask names, never at a
-/// position taken from another side's values. Without `bag`, every
-/// multiplicity is 1.
-pub fn tuples_over_mask(
-    domain: &JoinDomain,
-    sides: &[&Side],
-    plan: &StarPlan,
-    mask: &[u64],
-    bag: bool,
-) -> Vec<(Vec<u64>, u64)> {
-    let n = domain.len();
-    let mut out: Vec<(Vec<u64>, u64)> = Vec::new();
-    for_each_set(mask, n, |slot| {
-        let center = u64::from(domain.id(slot));
-        let mut tuples: Vec<(Vec<Option<u64>>, u64)> = vec![(vec![None; plan.sides.len()], 1)];
-        for (i, s) in plan.sides.iter().enumerate() {
-            let vals = sides[i].at(slot);
-            if s.emit.is_some() {
-                tuples = tuples
-                    .into_iter()
-                    .flat_map(|(t, m)| {
-                        vals.iter().map(move |&v| {
-                            let mut t = t.clone();
-                            t[i] = Some(u64::from(v));
-                            (t, m)
-                        })
-                    })
-                    .collect();
-            } else {
-                let k = vals.len() as u64;
-                for t in &mut tuples {
-                    t.1 *= k;
+/// What the post filter asks of one rule, resolved against its head.
+/// `None` from [`RuleFilter::of`]: a head constant contradicts the filter,
+/// so the rule contributes nothing.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct RuleFilter {
+    /// The center must equal this value.
+    pub center: Option<u64>,
+    /// Side `i`'s value must equal `side[i]`.
+    pub side: Vec<Option<u64>>,
+}
+
+impl RuleFilter {
+    pub fn of(star: &StarPlan, post_filter: &[Option<u64>]) -> Option<RuleFilter> {
+        let mut f = RuleFilter {
+            center: None,
+            side: vec![None; star.sides.len()],
+        };
+        for (h, want) in star.head.iter().zip(post_filter) {
+            let Some(k) = *want else { continue };
+            match *h {
+                HeadTerm::Const(c) if c != k => return None,
+                HeadTerm::Const(_) => {}
+                HeadTerm::Center => match f.center {
+                    Some(prev) if prev != k => return None,
+                    _ => f.center = Some(k),
+                },
+                HeadTerm::Side(i) => match f.side[i] {
+                    Some(prev) if prev != k => return None,
+                    _ => f.side[i] = Some(k),
+                },
+            }
+        }
+        Some(f)
+    }
+}
+
+/// Narrows `kept` to the filtered center: one bit, or none when the value
+/// is outside the domain. A word-wise pass; no value is read.
+pub fn narrow_center(kept: &mut [u64], domain: &JoinDomain, filter: &RuleFilter) {
+    let Some(k) = filter.center else { return };
+    let slot = u32::try_from(k).ok().and_then(|v| domain.slot_of(v));
+    for (w, word) in kept.iter_mut().enumerate() {
+        *word &= match slot {
+            Some(s) if s >> 6 == w => 1 << (s & 63),
+            _ => 0,
+        };
+    }
+}
+
+/// Derivations side `i` contributes at `slot`: its row count there, or the
+/// rows equal to its filter value. Read in place.
+fn side_count(view: &StarView<'_>, filter: &RuleFilter, slot: usize, i: usize) -> u64 {
+    let vals = view.sides[i].at(slot);
+    match filter.side[i] {
+        None => vals.len() as u64,
+        Some(k) => vals.iter().filter(|&&v| u64::from(v) == k).count() as u64,
+    }
+}
+
+/// Derivations at `slot`: the product of every side's count (law 7). Zero
+/// when a filtered side has no matching row.
+fn slot_derivations(view: &StarView<'_>, filter: &RuleFilter, slot: usize) -> u64 {
+    (0..view.sides.len())
+        .map(|i| side_count(view, filter, slot, i))
+        .product()
+}
+
+/// Whether any kept slot yields a derivation. Stops at the first.
+pub fn any_derivation(view: &StarView<'_>, kept: &[u64], filter: &RuleFilter) -> bool {
+    if filter.side.iter().all(Option::is_none) {
+        return kept.iter().any(|&w| w != 0);
+    }
+    let n = view.domain.len();
+    for (wi, &w) in kept.iter().enumerate() {
+        let mut w = w;
+        while w != 0 {
+            let slot = wi * 64 + w.trailing_zeros() as usize;
+            if slot < n && slot_derivations(view, filter, slot) > 0 {
+                return true;
+            }
+            w &= w - 1;
+        }
+    }
+    false
+}
+
+/// The bag count over the kept slots, `Σ_slot Π_i count_i(slot)`, without
+/// building a row. Reads side lengths (and, for a filtered side, its
+/// values) in place, in slot order.
+pub fn bag_count(view: &StarView<'_>, kept: &[u64], filter: &RuleFilter) -> u64 {
+    let mut total = 0u64;
+    for_each_set(kept, view.domain.len(), |slot| {
+        total += slot_derivations(view, filter, slot);
+    });
+    total
+}
+
+/// Distinct answers whose head has at most one free column: every other
+/// position is a constant (a head constant, or a column the post filter
+/// fixes). The free column's distinct values are a mask over the value
+/// space (`ScatterOrU32`'s shape: the scattered mask is the demanded
+/// result). `values` is trimmed of trailing zero words; a template with no
+/// free column answers one row when `values == [1]`.
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
+pub struct DistinctGroup {
+    /// Per head position: `Some(c)` a constant, `None` the free column.
+    pub template: Vec<Option<u64>>,
+    pub values: Vec<u64>,
+}
+
+impl DistinctGroup {
+    /// Distinct rows in this group.
+    pub fn len(&self) -> u64 {
+        self.values.iter().map(|w| u64::from(w.count_ones())).sum()
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.values.iter().all(|&w| w == 0)
+    }
+}
+
+/// Why a set answer is not answered by a fold.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum SetRefusal {
+    /// More than one free head column: distinct rows over a projection need
+    /// a seen-set over value pairs, which no terminal provides (GAP-A2).
+    MultiColumn { free: usize },
+    /// Two rules' templates can share a row without being the same
+    /// template, so their union is not a union of value masks.
+    OverlappingTemplates,
+    /// A value too wide for a value-space mask (the free column must be a
+    /// u32).
+    ValueOutOfRange,
+}
+
+fn set_bit(values: &mut Vec<u64>, v: u64) {
+    let w = (v >> 6) as usize;
+    if values.len() <= w {
+        values.resize(w + 1, 0);
+    }
+    values[w] |= 1 << (v & 63);
+}
+
+fn disjoint(a: &[Option<u64>], b: &[Option<u64>]) -> bool {
+    a.iter()
+        .zip(b)
+        .any(|(x, y)| matches!((x, y), (Some(p), Some(q)) if p != q))
+}
+
+/// Adds one rule's distinct answers to `groups`. Fails closed: a head with
+/// two or more free columns, or a template that overlaps another group's
+/// without equalling it, is refused rather than deduplicated by sorting.
+pub fn add_distinct(
+    view: &StarView<'_>,
+    kept: &[u64],
+    filter: &RuleFilter,
+    groups: &mut Vec<DistinctGroup>,
+) -> Result<(), SetRefusal> {
+    let template: Vec<Option<u64>> = view
+        .star
+        .head
+        .iter()
+        .map(|h| match *h {
+            HeadTerm::Const(c) => Some(c),
+            HeadTerm::Center => filter.center,
+            HeadTerm::Side(i) => filter.side[i],
+        })
+        .collect();
+    let free: Vec<usize> = (0..template.len())
+        .filter(|&p| template[p].is_none())
+        .collect();
+    if free.len() > 1 {
+        return Err(SetRefusal::MultiColumn { free: free.len() });
+    }
+    let g = match groups.iter().position(|g| g.template == template) {
+        Some(g) => g,
+        None => {
+            if groups.iter().any(|g| !disjoint(&g.template, &template)) {
+                return Err(SetRefusal::OverlappingTemplates);
+            }
+            groups.push(DistinctGroup {
+                template,
+                values: Vec::new(),
+            });
+            groups.len() - 1
+        }
+    };
+    let n = view.domain.len();
+    let values = &mut groups[g].values;
+    match free.first().map(|&p| view.star.head[p]) {
+        None => {
+            if any_derivation(view, kept, filter) {
+                set_bit(values, 0);
+            }
+        }
+        Some(HeadTerm::Center) => for_each_set(kept, n, |slot| {
+            if slot_derivations(view, filter, slot) > 0 {
+                set_bit(values, u64::from(view.domain.id(slot)));
+            }
+        }),
+        Some(HeadTerm::Side(i)) => for_each_set(kept, n, |slot| {
+            let others = (0..view.sides.len())
+                .filter(|&j| j != i)
+                .all(|j| side_count(view, filter, slot, j) > 0);
+            if others {
+                for &v in view.sides[i].at(slot) {
+                    set_bit(values, u64::from(v));
                 }
             }
+        }),
+        Some(HeadTerm::Const(_)) => unreachable!("a constant is never free"),
+    }
+    Ok(())
+}
+
+/// Puts `groups` in canonical form: trailing zero words trimmed, empty
+/// groups dropped, sorted by template. Equal answers compare equal however
+/// the rules were planned.
+pub fn canonical(groups: &mut Vec<DistinctGroup>) {
+    for g in groups.iter_mut() {
+        while g.values.last() == Some(&0) {
+            g.values.pop();
         }
-        for (t, m) in tuples {
-            let row: Vec<u64> = plan
-                .head
-                .iter()
-                .map(|h| match *h {
-                    HeadTerm::Const(c) => c,
-                    HeadTerm::Center => center,
-                    HeadTerm::Side(i) => t[i].expect("emitting side"),
-                })
-                .collect();
-            out.push((row, m));
+    }
+    groups.retain(|g| !g.values.is_empty());
+    groups.sort();
+}
+
+/// One rule's derivations as coordinates: the rule, its star, its filter
+/// and the kept slots. The values stay in the table; read them with
+/// [`for_each_slot`].
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct RuleCoordinates {
+    pub rule: usize,
+    /// The table the kept slots address (the pipeline's star index, or the
+    /// shared path's group index).
+    pub table: usize,
+    /// For each plan side, its side index in that table.
+    pub sides: Vec<usize>,
+    pub star: StarPlan,
+    pub filter: RuleFilter,
+    pub kept: Vec<u64>,
+}
+
+/// One kept slot, borrowed: the center and every side's rows there. The
+/// derivations at the slot are the cartesian product of the emitting sides'
+/// filtered rows, each counted once per row of every non-emitting side; the
+/// caller applies the head template.
+pub struct SlotView<'a> {
+    pub slot: usize,
+    pub center: u32,
+    pub sides: &'a [&'a [u32]],
+    pub filter: &'a RuleFilter,
+    pub derivations: u64,
+}
+
+/// Calls `f` for every kept slot with at least one derivation, in slot
+/// order, with the sides' rows borrowed in place.
+pub fn for_each_slot(view: &StarView<'_>, rc: &RuleCoordinates, mut f: impl FnMut(&SlotView<'_>)) {
+    let mut sides: Vec<&[u32]> = vec![&[]; view.sides.len()];
+    for_each_set(&rc.kept, view.domain.len(), |slot| {
+        let derivations = slot_derivations(view, &rc.filter, slot);
+        if derivations == 0 {
+            return;
         }
+        for (i, s) in view.sides.iter().enumerate() {
+            sides[i] = s.at(slot);
+        }
+        f(&SlotView {
+            slot,
+            center: view.domain.id(slot),
+            sides: &sides,
+            filter: &rc.filter,
+            derivations,
+        });
     });
-    out.sort();
-    let mut merged: Vec<(Vec<u64>, u64)> = Vec::new();
-    for (row, m) in out {
-        match merged.last_mut() {
-            Some((r, c)) if *r == row => *c += m,
-            _ => merged.push((row, m)),
+}
+
+/// Derivations at the kept slots whose head reads as `pattern` (`None` at a
+/// column: any value). A fold, never an enumeration: per slot it is law 7
+/// with each side's count narrowed to the pattern's value. When the pattern
+/// pins the center, its global id is inverted to the local slot
+/// (`slot_of`, the one pointer table) and only that slot is read. With an
+/// all-`None` pattern this is [`bag_count`].
+pub fn matching(
+    view: &StarView<'_>,
+    rc: &RuleCoordinates,
+    pattern: impl Fn(usize) -> Option<u64>,
+) -> u64 {
+    let k = view.sides.len();
+    let mut center: Option<u64> = rc.filter.center;
+    let mut want: Vec<Option<u64>> = vec![None; k];
+    for (c, h) in view.star.head.iter().enumerate() {
+        let Some(v) = pattern(c) else { continue };
+        let slot = match *h {
+            HeadTerm::Const(x) if x != v => return 0,
+            HeadTerm::Const(_) => continue,
+            HeadTerm::Center => &mut center,
+            HeadTerm::Side(i) => &mut want[i],
+        };
+        match *slot {
+            Some(w) if w != v => return 0,
+            _ => *slot = Some(v),
+        }
+    }
+    for (i, w) in want.iter_mut().enumerate() {
+        match (*w, rc.filter.side[i]) {
+            (Some(a), Some(b)) if a != b => return 0,
+            (None, b) => *w = b,
+            _ => {}
         }
     }
-    if !bag {
-        for e in &mut merged {
-            e.1 = 1;
+    let at_slot = |slot: usize| -> u64 {
+        (0..k)
+            .map(|i| {
+                let vals = view.sides[i].at(slot);
+                match want[i] {
+                    None => vals.len() as u64,
+                    Some(t) => vals.iter().filter(|&&v| u64::from(v) == t).count() as u64,
+                }
+            })
+            .product()
+    };
+    if let Some(v) = center {
+        let Some(slot) = u32::try_from(v).ok().and_then(|y| view.domain.slot_of(y)) else {
+            return 0;
+        };
+        let kept = rc
+            .kept
+            .get(slot / 64)
+            .is_some_and(|w| w >> (slot % 64) & 1 == 1);
+        return if kept { at_slot(slot) } else { 0 };
+    }
+    let mut total = 0u64;
+    for_each_set(&rc.kept, view.domain.len(), |slot| total += at_slot(slot));
+    total
+}
+
+/// One derivation, read in place: the head template over the lanes at the
+/// side positions `at` (the `"{0}{1}" -f $laneA[$i],$laneB[$i]` shape).
+/// Nothing is copied; [`DerivationView::get`] reads one column.
+pub struct DerivationView<'a> {
+    pub center: u32,
+    head: &'a [HeadTerm],
+    sides: &'a [&'a [u32]],
+    at: &'a [usize],
+}
+
+impl DerivationView<'_> {
+    pub fn arity(&self) -> usize {
+        self.head.len()
+    }
+
+    pub fn get(&self, c: usize) -> u64 {
+        match self.head[c] {
+            HeadTerm::Const(k) => k,
+            HeadTerm::Center => u64::from(self.center),
+            HeadTerm::Side(i) => u64::from(self.sides[i][self.at[i]]),
         }
     }
-    merged
+}
+
+/// Every derivation at the kept slots, in slot order, as a template read over
+/// the side positions; `weight` counts the non-emitting sides' rows. Walks
+/// positions, never values: the only state is one position per side.
+pub fn for_each_derivation(
+    view: &StarView<'_>,
+    rc: &RuleCoordinates,
+    mut f: impl FnMut(&DerivationView<'_>, u64),
+) {
+    let k = view.sides.len();
+    let emitting: Vec<usize> = (0..k)
+        .filter(|&i| view.star.sides[i].emit.is_some())
+        .collect();
+    let mut at = vec![0usize; k];
+    let mut sides: Vec<&[u32]> = vec![&[]; k];
+    let keeps = |i: usize, v: u32| rc.filter.side[i].is_none_or(|t| t == u64::from(v));
+    for_each_set(&rc.kept, view.domain.len(), |slot| {
+        let center = view.domain.id(slot);
+        if rc.filter.center.is_some_and(|t| t != u64::from(center)) {
+            return;
+        }
+        for (i, s) in view.sides.iter().enumerate() {
+            sides[i] = s.at(slot);
+        }
+        let weight: u64 = (0..k)
+            .filter(|i| !emitting.contains(i))
+            .map(|i| sides[i].iter().filter(|&&v| keeps(i, v)).count() as u64)
+            .product();
+        if weight == 0 || emitting.iter().any(|&i| sides[i].is_empty()) {
+            return;
+        }
+        for &i in &emitting {
+            at[i] = 0;
+        }
+        loop {
+            if emitting.iter().all(|&i| keeps(i, sides[i][at[i]])) {
+                f(
+                    &DerivationView {
+                        center,
+                        head: &view.star.head,
+                        sides: &sides,
+                        at: &at,
+                    },
+                    weight,
+                );
+            }
+            // Odometer over the emitting sides' positions.
+            let mut d = 0;
+            while d < emitting.len() {
+                let i = emitting[d];
+                at[i] += 1;
+                if at[i] < sides[i].len() {
+                    break;
+                }
+                at[i] = 0;
+                d += 1;
+            }
+            if d == emitting.len() {
+                break;
+            }
+        }
+    });
 }
 
 /// True when the program contains an op that reads at a position taken from
diff --git a/crates/crepe-plan-probe/src/pipeline.rs b/crates/crepe-plan-probe/src/pipeline.rs
index eec8bef6e..e73ac4c82 100644
--- a/crates/crepe-plan-probe/src/pipeline.rs
+++ b/crates/crepe-plan-probe/src/pipeline.rs
@@ -1,7 +1,6 @@
 //! The stages wired together, with every intermediate product kept for the
 //! report: rules, rewrites, MORK patterns, star plans, mask-risc programs.
 
-use std::collections::BTreeMap;
 use std::time::{Duration, Instant};
 
 use lance_graph_mask_risc::Program;
@@ -12,10 +11,12 @@ use crate::lance_store::{LanceStore, LanceStoreError, SideIo, prepare_lance};
 use crate::logic::{RelId, Rule};
 use crate::mork_api::{MorkRefusal, QueryMulti, admit, query_multi};
 use crate::native::{
-    NativeRefusal, PreparedStar, StarPlan, Store, bindings_run, joint, prepare, prepare_semijoin,
-    star,
+    DerivationView, DistinctGroup, NativeRefusal, PreparedStar, RuleCoordinates, RuleFilter,
+    SetRefusal, StarPlan, StarView, Store, add_distinct, any_derivation, bag_count, canonical,
+    for_each_derivation, joint, kept, matching, narrow_center, prepare, prepare_semijoin, star,
 };
 use crate::optimize::{Demand, Optimized, Options, ResultKind, optimize};
+use mork_pathmap_probe::join::Side;
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum PipelineRefusal {
@@ -32,14 +33,125 @@ pub enum PipelineRefusal {
     /// The addressed arm refused a side (Arm B only): stale lane, unsorted
     /// table, a demand on the second column, or no attested lane.
     Addressed(String),
+    /// A set answer no fold gives (see [`SetRefusal`]).
+    Set {
+        rule: usize,
+        why: SetRefusal,
+    },
+}
+
+/// The answers themselves, never as owned rows.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum Answers {
+    /// Set semantics: per head template, the free column's distinct values
+    /// as a mask over the value space. Canonical (see
+    /// [`crate::native::canonical`]).
+    Distinct(Vec<DistinctGroup>),
+    /// Bag semantics: per rule, the kept slots. The rows stay in the
+    /// prepared tables ([`Report::tables`]); read them with
+    /// [`crate::native::for_each_slot`].
+    Derivations(Vec<RuleCoordinates>),
+}
+
+impl Answers {
+    /// Calls `f` with each rule's view over its table. The view holds
+    /// references to the lanes, never values.
+    fn rules(&self, tables: &[PreparedStar], mut f: impl FnMut(&StarView<'_>, &RuleCoordinates)) {
+        let Answers::Derivations(rules) = self else {
+            return;
+        };
+        for rc in rules {
+            let table = &tables[rc.table].table;
+            let sides: Vec<&Side> = rc.sides.iter().map(|&k| &table.sides[k]).collect();
+            let view = StarView {
+                domain: &table.domain,
+                sides: &sides,
+                star: &rc.star,
+            };
+            f(&view, rc);
+        }
+    }
+
+    /// Rows answered: derivations under bag semantics, distinct rows under
+    /// set semantics. A fold over the masks.
+    pub fn total(&self, tables: &[PreparedStar]) -> u64 {
+        match self {
+            Answers::Distinct(groups) => groups.iter().map(DistinctGroup::len).sum(),
+            Answers::Derivations(_) => {
+                let mut n = 0;
+                self.rules(tables, |v, rc| n += bag_count(v, &rc.kept, &rc.filter));
+                n
+            }
+        }
+    }
+
+    /// Rows reading as `pattern` (`None`: any value), counted like
+    /// [`Self::total`]. A fold; no row is built.
+    pub fn matching(&self, tables: &[PreparedStar], pattern: impl Fn(usize) -> Option<u64>) -> u64 {
+        match self {
+            Answers::Distinct(groups) => groups
+                .iter()
+                .map(|g| {
+                    let mut free = None;
+                    for (c, t) in g.template.iter().enumerate() {
+                        match (*t, pattern(c)) {
+                            (Some(k), Some(v)) if k != v => return 0,
+                            (None, p) => free = Some(p),
+                            _ => {}
+                        }
+                    }
+                    match free {
+                        // No free column: the group answers one row or none.
+                        None => g.len(),
+                        Some(None) => g.len(),
+                        Some(Some(v)) => {
+                            let bit = usize::try_from(v).unwrap_or(usize::MAX);
+                            u64::from(
+                                g.values
+                                    .get(bit / 64)
+                                    .is_some_and(|w| w >> (bit % 64) & 1 == 1),
+                            )
+                        }
+                    }
+                })
+                .sum(),
+            Answers::Derivations(_) => {
+                let mut n = 0;
+                self.rules(tables, |v, rc| n += matching(v, rc, &pattern));
+                n
+            }
+        }
+    }
+
+    /// Every answered row, read in place through its template, with its
+    /// weight; `get(c)` reads column `c`. Rows are not deduplicated across
+    /// slots or rules (that would need a seen-set).
+    pub fn for_each(&self, tables: &[PreparedStar], mut f: impl FnMut(&dyn Fn(usize) -> u64, u64)) {
+        match self {
+            Answers::Distinct(groups) => {
+                for g in groups {
+                    for (w, &word) in g.values.iter().enumerate() {
+                        let mut word = word;
+                        while word != 0 {
+                            let v = (w * 64 + word.trailing_zeros() as usize) as u64;
+                            f(&|c| g.template[c].unwrap_or(v), 1);
+                            word &= word - 1;
+                        }
+                    }
+                }
+            }
+            Answers::Derivations(_) => self.rules(tables, |v, rc| {
+                for_each_derivation(v, rc, |d: &DerivationView<'_>, w| f(&|c| d.get(c), w));
+            }),
+        }
+    }
 }
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum Outcome {
     Exists(bool),
     Count(u64),
-    /// Sorted tuples with multiplicities (1 under set semantics).
-    Bindings(Vec<(Vec<u64>, u64)>),
+    Bindings(Answers),
 }
 
 /// Costs, each measured on its own.
@@ -73,6 +185,9 @@ pub struct Report {
     /// mask-risc programs run (0 for a rule relalg answered symbolically).
     pub programs_run: usize,
     pub costs: Costs,
+    /// The prepared tables, in star order: what [`Answers::Derivations`]
+    /// addresses.
+    pub tables: Vec<PreparedStar>,
 }
 
 /// The cold half: rewrites and maps `rules` for `demand`; reads no row.
@@ -186,36 +301,60 @@ pub fn execute(
         }
         Outcome::Exists(any)
     } else {
-        let mut bag: BTreeMap<Vec<u64>, u64> = BTreeMap::new();
+        let mut any = false;
+        let mut derivations = 0u64;
+        let mut groups: Vec<DistinctGroup> = Vec::new();
+        let mut coordinates: Vec<RuleCoordinates> = Vec::new();
         for (j, ((i, _), (s, p))) in optimized
             .rules
             .iter()
             .zip(stars.iter().zip(&prepared))
             .enumerate()
         {
-            // One program per rule: the Keep mask that the bindings are read
-            // from; no separate count pass.
-            let (rows, run, prog) =
-                bindings_run(p, s, &generations[j], true).map_err(|w| native(*i, w))?;
+            // One program per rule: the Keep mask every terminal folds over.
+            let (mut mask, run, prog) = kept(p, s, &generations[j]).map_err(|w| native(*i, w))?;
             programs_run += run.programs;
             programs.extend(prog);
-            for (row, m) in rows {
-                *bag.entry(row).or_insert(0) += m;
+            let Some(filter) = RuleFilter::of(s, &optimized.post_filter) else {
+                continue;
+            };
+            narrow_center(&mut mask, &p.table.domain, &filter);
+            let sides: Vec<_> = p.table.sides.iter().collect();
+            let view = StarView {
+                domain: &p.table.domain,
+                sides: &sides,
+                star: s,
+            };
+            match kind {
+                ResultKind::Exists => any |= any_derivation(&view, &mask, &filter),
+                ResultKind::Count(Semantics::Bag) => {
+                    derivations += bag_count(&view, &mask, &filter)
+                }
+                ResultKind::Count(Semantics::Set) | ResultKind::Bindings(Semantics::Set) => {
+                    add_distinct(&view, &mask, &filter, &mut groups)
+                        .map_err(|why| PipelineRefusal::Set { rule: *i, why })?
+                }
+                ResultKind::Bindings(Semantics::Bag) => coordinates.push(RuleCoordinates {
+                    rule: *i,
+                    table: j,
+                    sides: (0..s.sides.len()).collect(),
+                    star: s.clone(),
+                    filter,
+                    kept: mask,
+                }),
             }
         }
-        bag.retain(|row, _| {
-            row.iter()
-                .zip(&optimized.post_filter)
-                .all(|(v, f)| f.is_none_or(|f| f == *v))
-        });
+        canonical(&mut groups);
         match kind {
-            ResultKind::Exists => Outcome::Exists(!bag.is_empty()),
-            ResultKind::Count(Semantics::Set) => Outcome::Count(bag.len() as u64),
-            ResultKind::Count(Semantics::Bag) => Outcome::Count(bag.values().sum()),
-            ResultKind::Bindings(Semantics::Set) => {
-                Outcome::Bindings(bag.into_keys().map(|r| (r, 1)).collect())
+            ResultKind::Exists => Outcome::Exists(any),
+            ResultKind::Count(Semantics::Bag) => Outcome::Count(derivations),
+            ResultKind::Count(Semantics::Set) => {
+                Outcome::Count(groups.iter().map(DistinctGroup::len).sum())
+            }
+            ResultKind::Bindings(Semantics::Set) => Outcome::Bindings(Answers::Distinct(groups)),
+            ResultKind::Bindings(Semantics::Bag) => {
+                Outcome::Bindings(Answers::Derivations(coordinates))
             }
-            ResultKind::Bindings(Semantics::Bag) => Outcome::Bindings(bag.into_iter().collect()),
         }
     };
     let execute_cost = t.elapsed();
@@ -233,6 +372,7 @@ pub fn execute(
             prepare: prepare_cost,
             execute: execute_cost,
         },
+        tables: prepared,
     };
     Ok((outcome, report))
 }
diff --git a/crates/crepe-plan-probe/tests/pipeline.rs b/crates/crepe-plan-probe/tests/pipeline.rs
index e18f80fc0..ea5d9f3de 100644
--- a/crates/crepe-plan-probe/tests/pipeline.rs
+++ b/crates/crepe-plan-probe/tests/pipeline.rs
@@ -7,11 +7,11 @@ use std::collections::BTreeMap;
 use crepe_plan_probe::logic::{Labels, RelId, Rule, resolve};
 use crepe_plan_probe::mork_api::{Expr, MorkRefusal, QueryMulti, admit, query_multi, render};
 use crepe_plan_probe::native::{
-    NativeRefusal, SideKey, Store, distinct_centers, fingerprint, has_foreign_hop, joint, prepare,
-    star,
+    NativeRefusal, PreparedStar, SetRefusal, SideKey, Store, distinct_centers, fingerprint,
+    has_foreign_hop, joint, prepare, star,
 };
 use crepe_plan_probe::optimize::{Demand, Options, ResultKind, Rewrite};
-use crepe_plan_probe::pipeline::{Outcome, PipelineRefusal, evaluate};
+use crepe_plan_probe::pipeline::{Answers, Outcome, PipelineRefusal, Report};
 use crepe_plan_probe::plan::Plan;
 use crepe_plan_probe::programs::{family, family_noisy, refused, star3, two_way};
 use crepe_plan_probe::{logic, reference};
@@ -20,6 +20,154 @@ use lance_graph_relalg::{Measure, Rel, normalize};
 use mork_pathmap_probe::join::{JoinRefusal, JoinTable};
 use mork_pathmap_probe::{Backend, JoinInput, RelationView, Semantics, bind, capability};
 
+// ---------------------------------------------------------------------------
+// Comparing answers by folds
+// ---------------------------------------------------------------------------
+//
+// No row is built, here or in the library. Bindings stay as the library
+// answered them (value masks, or kept slots over the prepared tables). Two
+// answers are equal when their totals agree and every row one of them reads
+// (through its head template, over the lanes in place) is answered the same
+// number of times by the other. Expected rows are written out by the test.
+
+#[derive(Debug, Clone, PartialEq)]
+enum Norm {
+    Exists(bool),
+    Count(u64),
+    Bindings(Bound),
+}
+
+/// A bindings answer with the tables its coordinates address.
+#[derive(Clone)]
+struct Bound {
+    answers: Answers,
+    tables: Vec<PreparedStar>,
+}
+
+impl Bound {
+    fn total(&self) -> u64 {
+        self.answers.total(&self.tables)
+    }
+
+    /// Rows reading as `pattern` (`None`: any value).
+    fn count(&self, pattern: &[Option<u64>]) -> u64 {
+        self.answers
+            .matching(&self.tables, |c| pattern.get(c).copied().flatten())
+    }
+}
+
+impl std::fmt::Debug for Bound {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "Bound {{ total: {} }}", self.total())
+    }
+}
+
+impl PartialEq for Bound {
+    fn eq(&self, other: &Bound) -> bool {
+        if self.total() != other.total() {
+            return false;
+        }
+        let mut same = true;
+        self.answers.for_each(&self.tables, |get, _| {
+            if same {
+                let me = |c: usize| Some(get(c));
+                same = self.answers.matching(&self.tables, me)
+                    == other.answers.matching(&other.tables, me);
+            }
+        });
+        same
+    }
+}
+
+/// Against rows the test writes out: same total, and each row answered
+/// exactly its multiplicity.
+impl PartialEq<Vec<(Vec<u64>, u64)>> for Bound {
+    fn eq(&self, want: &Vec<(Vec<u64>, u64)>) -> bool {
+        self.total() == want.iter().map(|(_, m)| m).sum::<u64>()
+            && want
+                .iter()
+                .all(|(row, m)| self.answers.matching(&self.tables, |c| row.get(c).copied()) == *m)
+    }
+}
+
+fn norm(o: &Outcome, tables: &[PreparedStar]) -> Norm {
+    match o {
+        Outcome::Exists(b) => Norm::Exists(*b),
+        Outcome::Count(n) => Norm::Count(*n),
+        Outcome::Bindings(a) => Norm::Bindings(Bound {
+            answers: a.clone(),
+            tables: tables.to_vec(),
+        }),
+    }
+}
+
+fn evaluate(
+    rules: &[Rule],
+    demand: &Demand,
+    kind: ResultKind,
+    opts: Options,
+    store: &Store<'_>,
+) -> Result<(Norm, Report), PipelineRefusal> {
+    crepe_plan_probe::pipeline::evaluate(rules, demand, kind, opts, store)
+        .map(|(o, rep)| (norm(&o, &rep.tables), rep))
+}
+
+async fn evaluate_lance(
+    rules: &[Rule],
+    demand: &Demand,
+    kind: ResultKind,
+    opts: Options,
+    store: &crepe_plan_probe::lance_store::LanceStore,
+) -> Result<
+    (
+        Norm,
+        Report,
+        Vec<Vec<crepe_plan_probe::lance_store::SideIo>>,
+    ),
+    PipelineRefusal,
+> {
+    crepe_plan_probe::pipeline::evaluate_lance(rules, demand, kind, opts, store)
+        .await
+        .map(|(o, rep, io)| (norm(&o, &rep.tables), rep, io))
+}
+
+type AddressedRun = (
+    Norm,
+    Report,
+    Vec<Vec<crepe_plan_probe::lance_store::SideIo>>,
+    Vec<Vec<crepe_plan_probe::addressed::Lookup>>,
+);
+
+async fn evaluate_addressed(
+    rules: &[Rule],
+    demand: &Demand,
+    kind: ResultKind,
+    opts: Options,
+    store: &crepe_plan_probe::lance_store::LanceStore,
+    lanes: &[crepe_plan_probe::addressed::AddressedRelation],
+) -> Result<AddressedRun, PipelineRefusal> {
+    crepe_plan_probe::pipeline::evaluate_addressed(rules, demand, kind, opts, store, lanes)
+        .await
+        .map(|(o, rep, io, l)| (norm(&o, &rep.tables), rep, io, l))
+}
+
+#[allow(clippy::too_many_arguments)]
+async fn evaluate_addressed_with(
+    rules: &[Rule],
+    demand: &Demand,
+    kind: ResultKind,
+    opts: Options,
+    store: &crepe_plan_probe::lance_store::LanceStore,
+    lanes: &[crepe_plan_probe::addressed::AddressedRelation],
+    policy: crepe_plan_probe::addressed::ReadPolicy,
+) -> Result<AddressedRun, PipelineRefusal> {
+    crepe_plan_probe::pipeline::evaluate_addressed_with(
+        rules, demand, kind, opts, store, lanes, policy,
+    )
+    .await
+    .map(|(o, rep, io, l)| (norm(&o, &rep.tables), rep, io, l))
+}
+
 const SET: Semantics = Semantics::Set;
 const BAG: Semantics = Semantics::Bag;
 
@@ -110,9 +258,9 @@ fn as_set(bag: &BTreeMap<Vec<u64>, u64>) -> Vec<(Vec<u64>, u64)> {
     bag.keys().map(|k| (k.clone(), 1)).collect()
 }
 
-fn bindings(o: Outcome) -> Vec<(Vec<u64>, u64)> {
+fn bindings(o: Norm) -> Bound {
     match o {
-        Outcome::Bindings(b) => b,
+        Norm::Bindings(b) => b,
         other => panic!("expected bindings, got {other:?}"),
     }
 }
@@ -409,9 +557,9 @@ fn f_set_and_bag_answers_are_not_confused() {
     );
     for opts in [Options::ALL, Options::NONE] {
         let count = |k| evaluate(&rules, &d, k, opts, &st).unwrap().0;
-        assert_eq!(count(ResultKind::Count(BAG)), Outcome::Count(derivations));
-        assert_eq!(count(ResultKind::Count(SET)), Outcome::Count(3));
-        assert_eq!(count(ResultKind::Exists), Outcome::Exists(true));
+        assert_eq!(count(ResultKind::Count(BAG)), Norm::Count(derivations));
+        assert_eq!(count(ResultKind::Count(SET)), Norm::Count(3));
+        assert_eq!(count(ResultKind::Exists), Norm::Exists(true));
     }
     // MORK stores sets: asked for a bag it refuses, never answers 1s.
     assert!(capability(Backend::Mork, BAG).is_err());
@@ -572,13 +720,24 @@ fn j_two_way_join_equals_mork_and_the_reference() {
         bound: vec![None, None, None],
     };
     let want_bag = reference::eval(&rules, d.goal, &d.bound, &raw(&labels, &st));
-    let (set, _) = evaluate(&rules, &d, ResultKind::Bindings(SET), Options::ALL, &st).unwrap();
+    // Three free head columns: distinct rows over them need a seen-set over
+    // value triples, which no fold gives. Refused, never deduplicated by a
+    // sort (GAP-A2).
+    for kind in [ResultKind::Bindings(SET), ResultKind::Count(SET)] {
+        assert!(
+            matches!(
+                evaluate(&rules, &d, kind, Options::ALL, &st),
+                Err(PipelineRefusal::Set {
+                    why: SetRefusal::MultiColumn { free: 3 },
+                    ..
+                })
+            ),
+            "{kind:?}"
+        );
+    }
     let (bag, _) = evaluate(&rules, &d, ResultKind::Bindings(BAG), Options::ALL, &st).unwrap();
-    assert_eq!(bindings(set.clone()), as_set(&want_bag));
-    assert_eq!(
-        bindings(bag),
-        want_bag.clone().into_iter().collect::<Vec<_>>()
-    );
+    let bag = bindings(bag);
+    assert_eq!(bag, want_bag.clone().into_iter().collect::<Vec<_>>());
 
     let input = JoinInput {
         r: RelationView {
@@ -591,14 +750,84 @@ fn j_two_way_join_equals_mork_and_the_reference() {
         },
     };
     let mut mork = bind(Backend::Mork, &input, SET).expect("MORK builds");
-    let mork_rows: Vec<(Vec<u64>, u64)> = mork
-        .answers(&input)
-        .expect("MORK answers")
-        .rows
-        .keys()
-        .map(|t| (t.iter().map(|&v| u64::from(v)).collect(), 1))
-        .collect();
-    assert_eq!(bindings(set), mork_rows, "MORK");
+    // MORK answers the set: exactly the bag's distinct rows, each answered.
+    let mork_rows = mork.answers(&input).expect("MORK answers").rows;
+    assert_eq!(mork_rows.len(), want_bag.len(), "MORK");
+    for t in mork_rows.keys() {
+        let n = bag
+            .answers
+            .matching(&bag.tables, |c| t.get(c).map(|&v| u64::from(v)));
+        assert_eq!(n, want_bag[&t.map(u64::from).to_vec()], "MORK row {t:?}");
+    }
+}
+
+/// FAILS IF: a demand on the join variable that is not pushed into the rule
+/// (propagation off, so it stays a post filter on the center) is not applied
+/// to the kept slots. Bag, count and exists would then answer every center.
+/// Both the per-star pipeline and the shared path are checked.
+#[test]
+fn j2_a_post_filter_on_the_center_narrows_the_kept_slots() {
+    use crepe_plan_probe::cse::{Layout, evaluate_shared};
+    let plan = two_way::crepe_plan();
+    let (labels, rules) = resolved(&plan);
+    let r: Vec<[u32; 2]> = (0..120u32).map(|i| [i, i % 17]).collect();
+    let s: Vec<[u32; 2]> = (0..90u32).map(|i| [i % 23, 500 + i]).collect();
+    let st = Store {
+        relations: vec![
+            (labels.id("R").unwrap(), &r, 1),
+            (labels.id("S").unwrap(), &s, 1),
+        ],
+    };
+    let goal = labels.id("T").unwrap();
+    let no_push = Options {
+        propagate: false,
+        ..Options::ALL
+    };
+    let all = reference::eval(&rules, goal, &[None, None, None], &raw(&labels, &st));
+    // y = 5 joins (R and S both hold it); y = 20 is only in S, so nothing.
+    for (y, nonempty) in [(5u64, true), (20, false)] {
+        let d = Demand {
+            goal,
+            bound: vec![None, Some(y), None],
+        };
+        let want: Vec<(Vec<u64>, u64)> =
+            reference::eval(&rules, goal, &d.bound, &raw(&labels, &st))
+                .into_iter()
+                .collect();
+        let derivations: u64 = want.iter().map(|(_, m)| m).sum();
+        assert_eq!(!want.is_empty(), nonempty, "fixture, y = {y}");
+        // Not vacuous: the filter drops most of the unfiltered answer.
+        assert!(want.len() * 3 < all.len(), "y = {y} keeps too much");
+        let (b, _) = evaluate(&rules, &d, ResultKind::Bindings(BAG), no_push, &st).unwrap();
+        assert_eq!(bindings(b), want, "pipeline bag, y = {y}");
+        let (c, _) = evaluate(&rules, &d, ResultKind::Count(BAG), no_push, &st).unwrap();
+        assert_eq!(c, Norm::Count(derivations), "pipeline count, y = {y}");
+        let (e, _) = evaluate(&rules, &d, ResultKind::Exists, no_push, &st).unwrap();
+        assert_eq!(e, Norm::Exists(nonempty), "pipeline exists, y = {y}");
+        for sharing in [false, true] {
+            let run = |kind| {
+                let (o, rep) =
+                    evaluate_shared(&rules, &d, kind, no_push, &st, Layout::Components, sharing)
+                        .unwrap();
+                norm(&o, &rep.tables)
+            };
+            assert_eq!(
+                bindings(run(ResultKind::Bindings(BAG))),
+                want,
+                "shared bag, y = {y}"
+            );
+            assert_eq!(
+                run(ResultKind::Count(BAG)),
+                Norm::Count(derivations),
+                "shared count"
+            );
+            assert_eq!(
+                run(ResultKind::Exists),
+                Norm::Exists(nonempty),
+                "shared exists"
+            );
+        }
+    }
 }
 
 // ---------------------------------------------------------------------------
@@ -838,7 +1067,6 @@ fn n_semijoin_intersects_two_restricted_sides() {
 mod stage3 {
     use super::*;
     use crepe_plan_probe::lance_store::{LanceStore, SideIo, membership_mask, prepare_lance};
-    use crepe_plan_probe::pipeline::evaluate_lance;
 
     async fn lance_family(labels: &Labels, f: &Family) -> (tempfile::TempDir, LanceStore) {
         let dir = tempfile::tempdir().unwrap();
@@ -1025,9 +1253,7 @@ mod arm_b {
     use crepe_plan_probe::addressed::{AddressedRefusal, AddressedRelation, attest, key};
     use crepe_plan_probe::addressed::{Lookup, Read, ReadPolicy};
     use crepe_plan_probe::lance_store::{LanceStore, SideIo};
-    use crepe_plan_probe::pipeline::{
-        PipelineRefusal, evaluate_addressed, evaluate_addressed_with, evaluate_lance,
-    };
+    use crepe_plan_probe::pipeline::PipelineRefusal;
     use lance_graph_contract::ordered_lane::WitnessError;
 
     struct Fixture {
@@ -1189,7 +1415,7 @@ mod arm_b {
         )
         .await
         .unwrap();
-        assert_eq!(out, Outcome::Count(0));
+        assert_eq!(out, Norm::Count(0));
         let scanned: usize = io.iter().flatten().map(|s| s.rows.rows_scanned).sum();
         assert_eq!(scanned, 0, "{io:?}");
     }
@@ -1215,7 +1441,7 @@ mod arm_b {
                 .await
                 .unwrap();
         assert_eq!(b, m);
-        assert_eq!(b, Outcome::Count(500));
+        assert_eq!(b, Norm::Count(500));
         let scanned: usize = io.iter().flatten().map(|s| s.rows.rows_scanned).sum();
         assert_eq!(scanned, 3 * 500);
         // The two unrestricted sides each sought the 500 distinct centers.
@@ -1246,8 +1472,8 @@ mod arm_b {
         // Not vacuous: z = 10 comes once through y = 1 and 2 x 2 times through
         // y = 2 (Parent(17, 2) twice, Allowed(2) twice); z = 12 only through
         // y = 2. Without the duplicate fact this is 1 + 2 + 2 = 5.
-        let Outcome::Bindings(rows) = b else { panic!() };
-        let mult = |z: u64| rows.iter().find(|(t, _)| t[1] == z).map_or(0, |(_, n)| *n);
+        let Norm::Bindings(rows) = b else { panic!() };
+        let mult = |z: u64| rows.count(&[None, Some(z)]);
         assert_eq!((mult(10), mult(12)), (5, 4), "{rows:?}");
     }
 
@@ -1337,7 +1563,7 @@ mod arm_b {
             bound: vec![Some(1), None, None],
         };
         let got = evaluate_addressed(&rules, &d, kind, Options::ALL, &lance, &lanes).await;
-        assert_eq!(got.unwrap().0, Outcome::Count(2));
+        assert_eq!(got.unwrap().0, Norm::Count(2));
     }
 
     /// B10: a demand constant above `u32::MAX` matches nothing on every path
@@ -1365,7 +1591,7 @@ mod arm_b {
         }
         // Not vacuous: the memory path answers nothing here.
         let (m, _) = evaluate(&rules, &d, ResultKind::Count(BAG), Options::ALL, &mem).unwrap();
-        assert_eq!(m, Outcome::Count(0));
+        assert_eq!(m, Norm::Count(0));
     }
 
     /// B11: with nothing bound there are no centers, every side is read in
@@ -1381,6 +1607,28 @@ mod arm_b {
             bound: vec![None, None],
         };
         for kind in kinds() {
+            // Unbound, the head has two free columns (x and z): a set answer
+            // is refused by both paths alike (GAP-A2).
+            if matches!(kind, ResultKind::Bindings(SET) | ResultKind::Count(SET)) {
+                let refused = |r: Result<_, PipelineRefusal>| {
+                    matches!(
+                        r,
+                        Err(PipelineRefusal::Set {
+                            why: SetRefusal::MultiColumn { free: 2 },
+                            ..
+                        })
+                    )
+                };
+                assert!(refused(
+                    evaluate(&rules, &d, kind, Options::ALL, &mem).map(|_| ())
+                ));
+                assert!(refused(
+                    evaluate_addressed(&rules, &d, kind, Options::ALL, &fx.lance, &fx.lanes)
+                        .await
+                        .map(|_| ())
+                ));
+                continue;
+            }
             let (m, _) = evaluate(&rules, &d, kind, Options::ALL, &mem).unwrap();
             let (b, _, _, _) =
                 evaluate_addressed(&rules, &d, kind, Options::ALL, &fx.lance, &fx.lanes)