rindle_wire/

normalize.rs

//! Normalized change events — the **second serializer** beside the engine's flat path
//! (`rindle::flatten`), for a local-first client that runs its own normalized base
//! tables + local IVM.
//!
//! See `NORMALIZED-CHANGES-DESIGN.md` for the full design. Where `rindle::flatten` keeps the
//! root→leaf **path** and drops a remove's subtree, [`NormalizeFold`] does the opposite:
//! it **drops the path, keeps only a table tag**, and **keeps the subtree on remove** (so
//! witnesses/children are decremented too, §4.3). The engine does not change — both
//! serializers consume the same `CaughtChange` stream the replica already delivers
//! (`writer.rs` hands out `Vec<(NodeId, Vec<CaughtChange>)>` per query).
//!
//! **Why this lives in `rindle-wire`, not the `rindle` core or `rindle-replica`:**
//! the sender set grew. This serializer was originally server-only — the browser receiver
//! is the TypeScript `NormalizedSync`, and even in normalized mode the wasm `Db` is fed
//! *flat* changes, never the fold — so it lived in the SQLite-linking replica. But a
//! *room* (`RINDLE-REALTIME-DESIGN.md` §2.4) is a normalized **publisher compiled to
//! wasm**, so the fold must link without SQLite and compile for `wasm32` — and since the
//! daemons and the room must speak byte-identical frames, the whole protocol (fold,
//! envelope, JSON codec) sits in this dedicated wire crate both depend on; the replica
//! re-exports it from here unchanged. It stays out of the `rindle` core for the same
//! reason as ever: the core is the engine, not a wire protocol. (Contrast
//! `rindle::flatten`, which the wasm client genuinely uses, so it earns its place in
//! core.)
//!
//! This module is Slice 1: the fold itself. It folds one transaction's `CaughtChange`s
//! into a flat **per-table set with intra-query refcounts** (the footprint) and emits this
//! tick's net membership deltas as table-tagged [`NormalizedOp`]s, deduplicated so a row
//! touched via several tree paths in one transaction collapses to **one** op (§4.1/§4.2).
//! The footprint **interns** its rows: it holds the `Arc`-shared `OwnedRow` the producing
//! pipeline already owns (no second copy), materializing a `WireRow` only for the rows it
//! actually emits. The protocol envelope / wire framing (`NormalizedPublisher`, the
//! `NormalizedHello`) is a later slice.

use std::cmp::Ordering;
use std::collections::{BTreeMap, BTreeSet, HashMap};

use rindle::value::{compare_values, values_identical, ColId, OwnedRow, OwnedValue};
use rindle::{normalize_pipeline_ast, query_local_slot_names};
use rindle::{
    Aggregate, Ast, BuildError, CaughtChange, CaughtNode, Condition, CorrelatedSubquery,
    CorrelatedSubqueryCondition, Correlation, ExistsOp, Lit, Op, System, ValuePosition, WireRow,
};

// ---------------------------------------------------------------------------
// The wire op (§3): table-tagged, path-free row deltas.
// ---------------------------------------------------------------------------

/// One normalized change. Self-identifies by **table name** — the client routes each row
/// to a base table directly, with no tree path and no slot map (§3). Rows are positional
/// (aligned to that table's column order). `OwnedValue` is serializable because the crate
/// enables `rindle`'s `serde` feature, so the op serializes for the wire / oracle directly.
#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
pub enum NormalizedOp {
    /// A row entered this query's footprint.
    Add { table: Box<str>, row: WireRow },
    /// A row left this query's footprint (the row carries ≥ its PK; a full row is fine).
    Remove { table: Box<str>, row: WireRow },
    /// A footprinted row's value changed (PK stable; `old` and `new` share a PK).
    Edit {
        table: Box<str>,
        old: WireRow,
        new: WireRow,
    },
}

// ---------------------------------------------------------------------------
// The table tree (§4.1): slot → child table, derived from the AST.
// ---------------------------------------------------------------------------

/// The query's **table tree**: each frame's base table plus, per relationship slot (in the
/// query-local `RelId` order), the child frame's tree. It is the path-free replacement for
/// [`PathSeg`](rindle::PathSeg): folding a `CaughtChange::Child` descends `children[rel]` to
/// learn the child's table, then throws the parent row away.
///
/// `children[i]` corresponds to slot `i` of [`query_local_slot_names`] — materialized
/// `related` first (in `ast.related` order), then EXISTS gating aliases in `where`-tree
/// pre-order. This is the same slot layout the dataflow and [`view_schema`] use, so a
/// `CaughtChange::Child`'s `rel` indexes straight into `children` by construction.
///
/// A slot is `None` when it is **pruned** (`exists_noSync`, `EXISTS-NOSYNC-DESIGN.md` §4): a
/// `system: Permissions` EXISTS gate whose witnesses must not enter the footprint (and so are
/// never synced). The slot **keeps its position** — dropping it would shift later indices out
/// of alignment with the dataflow `RelId`s — but the fold neither descends into it nor counts
/// its rows.
///
/// [`view_schema`]: rindle::view_schema
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct TableNode {
    /// This frame's base table name.
    pub table: Box<str>,
    /// One subtree per query-local relationship slot, in slot order; `None` ⇒ a pruned
    /// (`exists_noSync`) slot whose witnesses are excluded from the footprint.
    pub children: Vec<Option<TableNode>>,
}

/// Derive the [`TableNode`] tree for `ast`. A pure function of the AST — it needs only
/// table **names** (all present inline: the root `ast.table`, each subquery's `table`),
/// never the source schemas, so no `resolve` closure. Mirrors [`view_schema`]'s slot walk
/// exactly (normalize the frame, take [`query_local_slot_names`], resolve each slot name to
/// its subquery, recurse) but carries the **table** where the schema carries the alias.
///
/// [`view_schema`]: rindle::view_schema
pub fn table_tree(ast: &Ast) -> TableNode {
    // Frame-local normalization, exactly as the engine's `build_pipeline_internal` /
    // `view_schema` do per frame: it uniquifies this frame's EXISTS aliases so
    // `query_local_slot_names` reads the same names the dataflow `RelId`s resolve against.
    let ast = normalize_pipeline_ast(ast);
    let children = query_local_slot_names(&ast)
        .iter()
        .map(|name| {
            let name: &str = name;
            let (cs, origin) = slot_backing(&ast, name).unwrap_or_else(|| {
                panic!("rindle-replica internal invariant: slot {name:?} has no subquery")
            });
            if cs.system == Some(System::Permissions) {
                // `exists_noSync` (§4): a permission-provenance slot is pruned — `None` keeps
                // its position but excludes its witnesses from the footprint (never synced).
                None
            } else if cs.subquery.aggregate.is_some() && origin == SlotOrigin::WhereExists {
                // A `having_count` parent gate (`PARENT-AGGREGATE-FILTER-DESIGN.md` §3): the
                // aggregate is lowered to an EXISTS over a HAVING-filtered reduce, so THIS slot
                // holds the gate's witness child rows — not the reduce's `(group…, count)`
                // output. Re-tagging them `agg_table_name` (the `related` arm below) would ship
                // full-width child rows under the synthetic table's 2-column schema, and the
                // client's width check rejects the batch. Prune like `exists_noSync`: the count
                // the client needs is already synced by the DISPLAY `count_as` this gate was
                // cloned from (`Query::having_count` requires one to exist), and since
                // `agg_table_name` hashes neither `alias` nor `having`, both resolve to the SAME
                // `__agg_*` table — so the client evaluates the gate off those rows once
                // `rewriteAggregates` points it there (@rindle/normalized).
                None
            } else if cs.subquery.aggregate.is_some() {
                // A relationship **aggregate** (`AGGREGATE-SYNC-DESIGN.md` §3.2): the child
                // rows are never synced — the server ships the reduce's synthetic
                // `(group_key…, count)` row as a SYNTHETIC base table. Re-tag the slot to that
                // table (a leaf: the aggregate row carries no relationships) and do **not**
                // recurse into the child table. The fold then keys the synthetic row by the
                // group columns under this name with no special-casing — its PK rides the
                // publisher's advertised schema (`agg_table_schemas`), like any base table.
                Some(TableNode {
                    table: agg_table_name(cs),
                    children: vec![],
                })
            } else {
                Some(table_tree(&cs.subquery))
            }
        })
        .collect();
    TableNode {
        table: ast.table.clone(),
        children,
    }
}

// ---------------------------------------------------------------------------
// Synthetic aggregate tables (`AGGREGATE-SYNC-DESIGN.md` §3.1/§3.2)
// ---------------------------------------------------------------------------

/// A synthetic aggregate base table the server ships in place of an aggregate's child
/// rows: the reduce's `(group_key…, value)` output, given a base-table home on the client
/// (§3.2). Columns are `[child_field…, "count"]` (mirroring the engine's
/// `agg_relationship_reldef` / `Reduce::count_by` output), the leading `key_len` of which
/// are the group key (and the PK). Derived purely from the AST — there is no such table in
/// the DB — so [`agg_table_schemas`] feeds the publisher's `hello` + PK map for it.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct AggTable {
    /// The content-addressed table name (see [`agg_table_name`]).
    pub name: Box<str>,
    /// `[child_field…, "count"]` — the synthetic row's columns, positional.
    pub columns: Vec<Box<str>>,
    /// How many leading columns form the group key (and the primary key): `0..key_len`.
    pub key_len: usize,
}

/// Reject a relationship `sum`/`avg` aggregate anywhere in `ast`'s `related` tree — the
/// precomputed-sync path (synthetic aggregate base tables, `AGGREGATE-SYNC-DESIGN.md` §3)
/// is **count-only**: it ships the reduce's `(group…, count)` row as a base table and has
/// no wire encoding for a `sum`/`avg` value or its int-vs-real typing. A relationship
/// `sum`/`avg` still works for a **direct** engine read (the generic scalar projection);
/// it is only unsupported for sync materialization, so registration surfaces it as a
/// [`BuildError`] rather than silently minting a count-shaped synthetic table. Call this at
/// each sync-registration entry (the replica's `register_shared_query`, the room's
/// `materialize`) before any of [`table_tree`] / [`agg_table_schemas`] / [`NormalizeFold`]
/// walk the tree — that ordering is what makes the `Sum`/`Avg` arm of `hash_agg_subquery`
/// unreachable.
pub fn reject_unsupported_sync_aggregate(ast: &Ast) -> Result<(), BuildError> {
    for r in &ast.related {
        match &r.subquery.aggregate {
            Some(Aggregate::Sum(_)) | Some(Aggregate::Avg(_)) => {
                return Err(BuildError::Unsupported(
                    "relationship sum/avg aggregates are not supported for precomputed sync \
                     (count only) — read them directly instead",
                ));
            }
            // A relationship aggregate is a leaf (the builder forbids nested `related` under
            // one); only an ordinary materialized `related` recurses.
            Some(Aggregate::Count) => {}
            None => reject_unsupported_sync_aggregate(&r.subquery)?,
        }
    }
    Ok(())
}

/// Collect a synthetic [`AggTable`] for **every** relationship `count` aggregate in `ast`,
/// recursively (a nested aggregate under a materialized `related` is included). The caller
/// (the replica consumer / the publisher's schema list) advertises these alongside the real
/// base-table schemas so the client can register + validate them.
///
/// The `where` tree is deliberately NOT walked. An aggregate does reach it — a `having_count`
/// parent gate is an EXISTS over a HAVING-filtered reduce — but that gate is a CLONE of a
/// materialized `related` `count_as` (`Query::having_count` requires one), and
/// [`agg_table_name`] hashes neither `alias` nor `having`, so it names the very table the
/// display aggregate already contributes here. Walking `where` would only re-derive a
/// duplicate. The gate's own slot ships nothing: [`table_tree`] prunes it, because a
/// `where`-EXISTS slot holds witness child rows rather than the reduce's output.
pub fn agg_table_schemas(ast: &Ast) -> Vec<AggTable> {
    let mut out = Vec::new();
    collect_agg_tables(ast, &mut out);
    out
}

fn collect_agg_tables(ast: &Ast, out: &mut Vec<AggTable>) {
    for r in &ast.related {
        if r.subquery.aggregate.is_some() {
            let mut columns: Vec<Box<str>> = r.correlation.child_field.clone();
            columns.push("count".into());
            out.push(AggTable {
                name: agg_table_name(r),
                columns,
                key_len: r.correlation.child_field.len(),
            });
        } else {
            collect_agg_tables(&r.subquery, out);
        }
    }
}

/// Rewrite an AST for a **client** engine reading synced aggregate tables — the Rust twin of
/// TypeScript `rewriteAggregates` (`packages/normalized/src/agg-table.ts`), byte-for-byte in
/// its choice of table name because both call [`agg_table_name`].
///
/// Each relationship `count` becomes a precomputed, source-backed singular relationship over
/// its synthetic table: read the server's count with a plain join + the same scalar projection
/// (`aggregate_precomputed`), never a `reduce`, which would recount already-aggregated rows.
/// Non-aggregate relationships recurse (a nested aggregate is rewritten too); the parent's
/// frame is otherwise untouched, so the view-schema slot order is preserved.
///
/// The `where` tree is rewritten the SAME way (`PARENT-AGGREGATE-FILTER-DESIGN.md` §3): a
/// `having_count` parent gate lowers to an EXISTS whose subquery CLONES the display `count_as`
/// and adds a post-aggregation `HAVING`. That is a relationship `count` like any other, so the
/// premise applies to it too — the client never recomputes a count, it lacks the child rows.
/// Left un-rewritten the gate reduces over child rows the server (rightly) does not sync, and
/// every parent fails it. Since [`agg_table_name`] hashes neither `alias` nor `having`, the
/// gate resolves to the SAME `__agg_*` table the display `count_as` already registers: the
/// rewrite costs no extra table and no extra rows. Its counterpart is the server pruning the
/// gate's witnesses from the footprint ([`table_tree`]) — that prune is sound only BECAUSE of
/// this rewrite.
///
/// # Why a Rust twin exists
/// The client rewrite used to live only in TypeScript, which meant no wire-bearing test tier
/// could **value**-grade an aggregate: grading re-runs the oracle over the reconstructed
/// footprint, and the footprint deliberately holds no child rows, so a recomputed `count` is
/// always 0. Every such tier therefore ran with aggregate generation switched off — the
/// blind spot both `having_count` bugs (the wire width collision and the planner flip) sat in.
/// With this, a harness can reconstruct the client's own AST and grade the answer.
pub fn rewrite_aggregates(ast: &Ast) -> Ast {
    rewrite_aggregates_with_local(ast, &|_| false)
}

/// [`rewrite_aggregates`], with the L1 local-table carve-out
/// (`201-LOCAL-ONLY-TABLES-DESIGN.md` §5.2): a count over a table `is_local` accepts is a
/// native IVM reduce with no server-authoritative `__agg_*` base, so it is left alone —
/// rewriting it would point the relationship at a synthetic table nothing ever feeds.
pub fn rewrite_aggregates_with_local(ast: &Ast, is_local: &dyn Fn(&str) -> bool) -> Ast {
    let mut out = ast.clone();
    out.r#where = ast
        .r#where
        .as_ref()
        .map(|c| rewrite_condition_aggregates(c, is_local));
    out.related = ast
        .related
        .iter()
        .map(|r| rewrite_relationship(r, is_local))
        .collect();
    out
}

/// The `where`-tree half: an EXISTS gate whose subquery carries an aggregate (the
/// `having_count` lowering) is rewritten by the very same [`rewrite_relationship`]; every
/// other condition recurses structurally. A `Simple` leaf holds no subquery.
fn rewrite_condition_aggregates(cond: &Condition, is_local: &dyn Fn(&str) -> bool) -> Condition {
    match cond {
        Condition::Simple(_) => cond.clone(),
        Condition::And { conditions } => Condition::And {
            conditions: conditions
                .iter()
                .map(|c| rewrite_condition_aggregates(c, is_local))
                .collect(),
        },
        Condition::Or { conditions } => Condition::Or {
            conditions: conditions
                .iter()
                .map(|c| rewrite_condition_aggregates(c, is_local))
                .collect(),
        },
        Condition::CorrelatedSubquery(csq) => {
            Condition::CorrelatedSubquery(CorrelatedSubqueryCondition {
                related: rewrite_relationship(&csq.related, is_local),
                ..csq.clone()
            })
        }
    }
}

fn rewrite_relationship(
    csq: &CorrelatedSubquery,
    is_local: &dyn Fn(&str) -> bool,
) -> CorrelatedSubquery {
    if csq.subquery.aggregate.is_some() && !is_local(&csq.subquery.table) {
        let mut subquery = Ast::new(&agg_table_name(csq));
        subquery.aggregate = csq.subquery.aggregate.clone();
        subquery.aggregate_precomputed = true;
        // The correlation is unchanged: the synthetic table's group columns are named after
        // the child correlation fields, so `child_field` still resolves against
        // `[child_field…, count]`.
        subquery.alias = csq.subquery.alias.clone();
        // The post-aggregation `HAVING` SURVIVES the rewrite — it is the gate's whole
        // predicate (`count > n`) and it addresses the reduce's OUTPUT column, which is
        // exactly what the synthetic row carries. A display `count_as` never has one, so
        // this is inert there.
        subquery.having = csq.subquery.having.clone();
        return CorrelatedSubquery {
            correlation: csq.correlation.clone(),
            subquery: Box::new(subquery),
            system: csq.system,
        };
    }
    CorrelatedSubquery {
        correlation: csq.correlation.clone(),
        subquery: Box::new(rewrite_aggregates_with_local(&csq.subquery, is_local)),
        system: csq.system,
    }
}

/// The synthetic base-table NAME for a relationship `count` aggregate (§3.1): a content
/// hash of the aggregate's **definition** — child table, kind, the group key (correlation
/// **child** fields), and the child `where` filter — so two queries with the *same*
/// definition share one table (cross-query refcount on the client) while a different filter
/// gets a *different* table (no `(table, pk)` collision). The **parent** correlation field
/// is excluded: the count for a given group key is the same whichever parent joins it.
///
/// # Byte protocol (the cross-language contract the TS client must reproduce, §3.1)
/// FNV-1a-64 (offset `0xcbf29ce484222325`, prime `0x100000001b3`); integers little-endian;
/// every string length-prefixed (`u32` byte-length + UTF-8). In order:
/// `u32(child_field.len)` then each `s(child_field[i])`; then the aggregate sub-AST identity
/// (`hash_agg_subquery`) — `s(table)`, `byte(kind)` (`count` = 1), the `where`
/// (`byte(0)` if absent, else `byte(1)` + `hash_condition`), and `u32(related.len)` with
/// each child's correlation + recursion. Rendered `"__agg_"` + 16 lowercase hex digits.
/// Only `count` reaches here — a relationship `sum`/`avg` is rejected before the sync path
/// ([`reject_unsupported_sync_aggregate`]).
pub fn agg_table_name(csq: &CorrelatedSubquery) -> Box<str> {
    let mut f = Fnv::new();
    f.u32(csq.correlation.child_field.len() as u32);
    for cf in &csq.correlation.child_field {
        f.s(cf);
    }
    hash_agg_subquery(&mut f, &csq.subquery);
    format!("__agg_{:016x}", f.0).into()
}

/// Hash the parts of an aggregate (sub)query that determine the count VALUE for a group:
/// the table, the aggregate kind, the `where` filter, and any nested materialized
/// `related` (a `where`-EXISTS witness can carry one). Deliberately excludes `alias`,
/// `select`, `order_by`, `start`, `limit`, `one` — none change a `count(*)` over the
/// (filtered) rows (the oracle rejects `start`/`limit`/nested-`related` in a count anyway).
///
/// Count-only: the synthetic-aggregate-table path has no encoding for a `sum`/`avg` value,
/// so [`reject_unsupported_sync_aggregate`] fails registration before a `sum`/`avg`
/// relationship aggregate can reach here (the `Sum`/`Avg` arms below are that invariant).
fn hash_agg_subquery(f: &mut Fnv, ast: &Ast) {
    f.s(&ast.table);
    f.byte(match ast.aggregate {
        Some(Aggregate::Count) => 1,
        None => 0,
        Some(Aggregate::Sum(_)) | Some(Aggregate::Avg(_)) => unreachable!(
            "relationship sum/avg is rejected before the sync path by \
             reject_unsupported_sync_aggregate"
        ),
    });
    match &ast.r#where {
        None => f.byte(0),
        Some(c) => {
            f.byte(1);
            hash_condition(f, c);
        }
    }
    f.u32(ast.related.len() as u32);
    for r in &ast.related {
        hash_correlation(f, &r.correlation);
        hash_agg_subquery(f, &r.subquery);
    }
}

fn hash_correlation(f: &mut Fnv, c: &Correlation) {
    f.u32(c.parent_field.len() as u32);
    for p in &c.parent_field {
        f.s(p);
    }
    f.u32(c.child_field.len() as u32);
    for c2 in &c.child_field {
        f.s(c2);
    }
}

fn hash_condition(f: &mut Fnv, cond: &Condition) {
    match cond {
        Condition::Simple(sc) => {
            f.byte(0);
            f.s(op_str(sc.op));
            hash_value_position(f, &sc.left);
            hash_value_position(f, &sc.right);
        }
        Condition::And { conditions } => {
            f.byte(1);
            f.u32(conditions.len() as u32);
            for c in conditions {
                hash_condition(f, c);
            }
        }
        Condition::Or { conditions } => {
            f.byte(2);
            f.u32(conditions.len() as u32);
            for c in conditions {
                hash_condition(f, c);
            }
        }
        Condition::CorrelatedSubquery(cs) => {
            f.byte(3);
            f.byte(match cs.op {
                ExistsOp::Exists => 1,
                ExistsOp::NotExists => 0,
            });
            // A permission vs client witness gates different rows ⇒ a different count.
            f.byte(match cs.related.system {
                Some(System::Permissions) => 1,
                Some(System::Client) => 2,
                Some(System::Test) => 3,
                None => 0,
            });
            hash_correlation(f, &cs.related.correlation);
            hash_agg_subquery(f, &cs.related.subquery);
        }
    }
}

fn hash_value_position(f: &mut Fnv, vp: &ValuePosition) {
    match vp {
        ValuePosition::Column { name } => {
            f.byte(0);
            f.s(name);
        }
        ValuePosition::Literal { value } => {
            f.byte(1);
            hash_lit(f, value);
        }
    }
}

fn hash_lit(f: &mut Fnv, lit: &Lit) {
    match lit {
        Lit::Null => f.byte(0),
        Lit::Bool(b) => {
            f.byte(1);
            f.byte(*b as u8);
        }
        // A distinct tag + exact le-bytes: two distinct big i64s must never share a
        // hash (a false-shared query identity corrupts; a missed dedup against the
        // same value spelled `Number` is merely a second registration).
        Lit::Int(i) => {
            f.byte(5);
            for b in i.to_le_bytes() {
                f.byte(b);
            }
        }
        Lit::Number(n) => {
            f.byte(2);
            for b in n.to_le_bytes() {
                f.byte(b);
            }
        }
        Lit::Str(s) => {
            f.byte(3);
            f.s(s);
        }
        Lit::Array(xs) => {
            f.byte(4);
            f.u32(xs.len() as u32);
            for x in xs {
                hash_lit(f, x);
            }
        }
    }
}

/// The wire string for a comparison operator (matches `Op`'s serde rename — the bytes the
/// TS twin hashes). Kept here, not on `Op`, so the cross-language contract lives beside the
/// hash that depends on it.
fn op_str(op: Op) -> &'static str {
    match op {
        Op::Eq => "=",
        Op::Ne => "!=",
        Op::Lt => "<",
        Op::Le => "<=",
        Op::Gt => ">",
        Op::Ge => ">=",
        Op::Is => "IS",
        Op::IsNot => "IS NOT",
        Op::Like => "LIKE",
        Op::NotLike => "NOT LIKE",
        Op::ILike => "ILIKE",
        Op::NotILike => "NOT ILIKE",
        Op::In => "IN",
        Op::NotIn => "NOT IN",
    }
}

/// FNV-1a-64 over the length-prefixed byte protocol that names a synthetic aggregate table
/// ([`agg_table_name`]) — a local twin of `normalize_protocol::Fnv` / `wire_schema::Fnv`.
struct Fnv(u64);

impl Fnv {
    fn new() -> Fnv {
        Fnv(0xcbf2_9ce4_8422_2325) // FNV-1a 64 offset basis
    }
    fn byte(&mut self, b: u8) {
        self.0 ^= b as u64;
        self.0 = self.0.wrapping_mul(0x0000_0100_0000_01b3); // FNV-1a 64 prime
    }
    fn u32(&mut self, v: u32) {
        for b in v.to_le_bytes() {
            self.byte(b);
        }
    }
    /// Length-prefixed string — disambiguates concatenations (`"ab"+"c"` ≠ `"a"+"bc"`).
    fn s(&mut self, s: &str) {
        self.u32(s.len() as u32);
        for &b in s.as_bytes() {
            self.byte(b);
        }
    }
}

/// Resolve a query-local slot name to the [`CorrelatedSubquery`] that backs it, with the same
/// precedence [`query_local_slot_names`] uses: a materialized `related` alias first
/// (last-writer-wins on a duplicate, mirroring `view_schema`'s `.rev().find`), else an EXISTS
/// gating subquery in the `where` tree. `ast` must already be [`normalize_pipeline_ast`]-
/// normalized. The returned `CorrelatedSubquery` carries both the child `subquery` and its
/// `system` provenance (the prune signal), so `table_tree` reads them off one value.
///
/// This is a by-**name** lookup, so it does not duplicate the builder's slot *ordering*
/// (that rides on the public [`query_local_slot_names`], the single source of truth): the
/// EXISTS aliases are unique after uniquify, so any traversal returns the same match.
fn slot_backing<'a>(ast: &'a Ast, name: &str) -> Option<(&'a CorrelatedSubquery, SlotOrigin)> {
    if let Some(c) = ast
        .related
        .iter()
        .rev()
        .find(|c| c.subquery.alias.as_deref() == Some(name))
    {
        return Some((c, SlotOrigin::Related));
    }
    ast.r#where
        .as_ref()
        .and_then(|w| find_exists_csq(w, name))
        .map(|c| (c, SlotOrigin::WhereExists))
}

/// Which half of the frame a slot's backing subquery came from. The two are indistinguishable
/// by shape — `having_count` builds its gate by CLONING a materialized `related` aggregate —
/// so the provenance has to ride along with the lookup ([`slot_backing`]) rather than be
/// re-derived from the subquery.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum SlotOrigin {
    /// A materialized `related` relationship: its rows ARE the slot's dataflow contents.
    Related,
    /// An EXISTS gate in the `where` tree: the slot's dataflow contents are the gate's
    /// **witness child rows**, never a materialized projection of the subquery.
    WhereExists,
}

/// Search a `where` tree for the EXISTS [`CorrelatedSubquery`] aliased `name` (skipping a
/// `limit 0` gate, which claims no slot — though such an alias never reaches here, since
/// [`query_local_slot_names`] already excludes it).
fn find_exists_csq<'a>(cond: &'a Condition, name: &str) -> Option<&'a CorrelatedSubquery> {
    match cond {
        Condition::CorrelatedSubquery(c) => {
            let cs = &c.related;
            (cs.subquery.limit != Some(0) && cs.subquery.alias.as_deref() == Some(name))
                .then_some(cs)
        }
        Condition::And { conditions } | Condition::Or { conditions } => {
            conditions.iter().find_map(|c| find_exists_csq(c, name))
        }
        Condition::Simple(_) => None,
    }
}

// ---------------------------------------------------------------------------
// Projection (PROJECTION-SUPPORT-DESIGN.md §5.2): the columns a query needs synced
// per base table, for project-at-emit.
// ---------------------------------------------------------------------------

/// What columns this query needs synced for a base table. `All` ⇒ a `'*'` frame (no
/// `select`) referenced the table, so it must sync every column; `Names` ⇒ the explicit
/// set of column names the query structurally reads for the table (its `required_cols`
/// at the name level — `select` ∪ `where`-leaf ∪ `order_by` ∪ `start` ∪ correlation
/// fields). Unioned across every frame that references the same table (any `All` wins).
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum ReqCols {
    All,
    Names(BTreeSet<Box<str>>),
}

impl ReqCols {
    fn merge(&mut self, other: ReqCols) {
        match (self, other) {
            (ReqCols::All, _) => {}
            (slot @ ReqCols::Names(_), ReqCols::All) => *slot = ReqCols::All,
            (ReqCols::Names(a), ReqCols::Names(b)) => a.extend(b),
        }
    }
}

/// The columns each base table in `ast`'s footprint needs synced (§5.2). A pure function of
/// the AST (column **names**, resolved to indices by the caller against the table schema).
/// Mirrors [`table_tree`]'s slot walk + prune (`exists_noSync` slots are never synced, so
/// they contribute nothing). The result drives the projected wire schema and project-at-emit.
pub fn required_columns_by_table(ast: &Ast) -> BTreeMap<Box<str>, ReqCols> {
    let mut out = BTreeMap::new();
    collect_required(ast, &[], &mut out);
    out
}

fn collect_required(ast: &Ast, inherited: &[Box<str>], out: &mut BTreeMap<Box<str>, ReqCols>) {
    let ast = normalize_pipeline_ast(ast);
    let mut req = match &ast.select {
        // A `'*'` frame needs every column of its table.
        None => ReqCols::All,
        Some(sel) => ReqCols::Names(sel.iter().cloned().collect()),
    };
    if let ReqCols::Names(names) = &mut req {
        // Correlation child fields this frame is keyed on (it is a child of some parent).
        for c in inherited {
            names.insert(c.clone());
        }
        // `where`-leaf columns.
        if let Some(w) = &ast.r#where {
            where_leaf_names(w, names);
        }
        // `order_by` fields.
        for op in &ast.order_by {
            names.insert(op.field().into());
        }
        // `start`-bound columns.
        if let Some(b) = &ast.start {
            for cell in &b.row {
                names.insert(cell.0.clone());
            }
        }
    }
    // Merge this frame's requirement for its table.
    match out.get_mut(&ast.table) {
        Some(existing) => existing.merge(req),
        None => {
            out.insert(ast.table.clone(), req);
        }
    }
    // Walk the relationship slots (same layout + prune as `table_tree`): a parent needs each
    // child's `parent_field`; the child frame needs its `child_field` (inherited).
    for name in query_local_slot_names(&ast).iter() {
        let name: &str = name;
        let Some((cs, origin)) = slot_backing(&ast, name) else {
            continue;
        };
        if cs.system == Some(System::Permissions) {
            continue; // pruned (`exists_noSync`) — never synced
        }
        if cs.subquery.aggregate.is_some() && origin == SlotOrigin::WhereExists {
            continue; // pruned (`having_count` gate witnesses, see `table_tree`) — never synced
        }
        if let Some(ReqCols::Names(names)) = out.get_mut(&ast.table) {
            for pf in &cs.correlation.parent_field {
                names.insert(pf.clone());
            }
        }
        collect_required(&cs.subquery, &cs.correlation.child_field, out);
    }
}

/// Collect the leaf column names a `where` tree references (either side of a `Simple`;
/// recursing `and`/`or`). An EXISTS subquery's correlation fields are gathered by the slot
/// walk, not here.
fn where_leaf_names(cond: &Condition, out: &mut BTreeSet<Box<str>>) {
    match cond {
        Condition::Simple(sc) => {
            for vp in [&sc.left, &sc.right] {
                if let rindle::ValuePosition::Column { name } = vp {
                    out.insert(name.clone());
                }
            }
        }
        Condition::And { conditions } | Condition::Or { conditions } => {
            for c in conditions {
                where_leaf_names(c, out);
            }
        }
        Condition::CorrelatedSubquery(_) => {}
    }
}

// ---------------------------------------------------------------------------
// The fold (§4): tree → table-tagged deltas with intra-query refcounts.
// ---------------------------------------------------------------------------

/// Table → primary-key column indices (into that table's row), supplied at construction.
pub type PkMap = HashMap<Box<str>, Vec<ColId>>;

/// One footprint entry: the latest row plus its **intra-query** refcount — the number of
/// tree positions this source row currently occupies within this one query (§4.2). The row
/// is the `Arc`-shared [`OwnedRow`] the pipeline already holds (interned, not re-copied).
#[derive(Clone, Debug)]
struct FootEntry {
    row: OwnedRow,
    rc: u32,
}

/// `table → (pk → entry)`. Persistent across ticks. Sorted maps so the emitted op order is
/// deterministic (useful for the differential oracle); the client applies a batch into one
/// transaction where per-(table,pk) collapse means order is not load-bearing.
type Footprint = BTreeMap<Box<str>, BTreeMap<PkKey, FootEntry>>;

/// This tick's touched rows, each carrying the row **as it was at tick start** (`None` ⇒
/// absent), captured lazily on first touch (an `Arc` clone — no copy). Diffing this against
/// the post-tick footprint yields the net per-row op (§4.1).
type Tick = BTreeMap<Box<str>, BTreeMap<PkKey, Option<OwnedRow>>>;

/// A primary key as a map key. [`OwnedValue`] deliberately has no derived `Ord`/`Eq` (it
/// would invite the wrong comparator); a PK is a homogeneous-per-column tuple, so we key
/// the maps by the engine's own [`compare_values`] total order — never panicking, since
/// within one table+column the cell types match.
#[derive(Clone, Debug)]
struct PkKey(Vec<OwnedValue>);

impl PartialEq for PkKey {
    fn eq(&self, o: &Self) -> bool {
        self.cmp(o) == Ordering::Equal
    }
}
impl Eq for PkKey {}
impl PartialOrd for PkKey {
    fn partial_cmp(&self, o: &Self) -> Option<Ordering> {
        Some(self.cmp(o))
    }
}
impl Ord for PkKey {
    fn cmp(&self, o: &Self) -> Ordering {
        for (a, b) in self.0.iter().zip(o.0.iter()) {
            match compare_values(a.as_ref(), b.as_ref()) {
                Ordering::Equal => {}
                ord => return ord,
            }
        }
        self.0.len().cmp(&o.0.len())
    }
}

/// The server-side serializer (§4): **one per registered normalized query**. Fold each
/// committed transaction's `CaughtChange`s with [`fold`](NormalizeFold::fold) to get that
/// tick's [`NormalizedOp`]s. The footprint persists across ticks; the snapshot is simply
/// the first fold over the hydrate batch (every row 0→1 ⇒ all `Add`s).
#[derive(Debug)]
pub struct NormalizeFold {
    tree: TableNode,
    pk: PkMap,
    footprint: Footprint,
}

impl NormalizeFold {
    /// Build a fold for the query `ast`, with `pk` giving each table's primary-key column
    /// indices (the server derives this from its source schemas; tests pass it inline).
    /// Panics if `pk` omits a table the query's tree can surface — a build-time config
    /// invariant, not a data-reachable error.
    pub fn new(ast: &Ast, pk: PkMap) -> NormalizeFold {
        let tree = table_tree(ast);
        validate_pk(&tree, &pk);
        // Pre-register every table the tree can surface with an empty per-table map. The fold
        // hot path then keys the footprint with `get_mut(&str)` (no `Box<str>` allocation per
        // touch) instead of `entry(table.into())`, which would allocate a fresh key on every
        // call even on a hit. Coverage matches `validate_pk`'s walk, so the lookups are
        // infallible. An empty per-table map is indistinguishable from an absent one to the
        // snapshot / emit readers (both yield no rows).
        let mut footprint: Footprint = BTreeMap::new();
        register_tables(&tree, &mut footprint);
        NormalizeFold {
            tree,
            pk,
            footprint,
        }
    }

    /// The query's table tree (slot → child table), derived from the AST.
    pub fn tree(&self) -> &TableNode {
        &self.tree
    }

    /// The current footprint as a seq-0 snapshot: one [`NormalizedOp::Add`] per live row,
    /// in deterministic (sorted by table, then PK) order. This baselines a **late
    /// subscriber** attaching to an already-hydrated *shared* query from the cached
    /// footprint, without re-running the engine pipeline (`RINDLE-SERVER-DESIGN.md` §11) —
    /// the membership is binary on the wire, so the intra-query refcount is not surfaced.
    pub fn footprint_snapshot(&self) -> Vec<NormalizedOp> {
        let mut ops = Vec::new();
        for (table, rows) in &self.footprint {
            for entry in rows.values() {
                ops.push(NormalizedOp::Add {
                    table: table.clone(),
                    row: entry.row.to_value_vec(),
                });
            }
        }
        ops
    }

    /// Fold one committed transaction (or the hydrate snapshot) into its net
    /// [`NormalizedOp`]s, updating the persistent footprint. An empty / no-membership-
    /// change tick returns `[]`.
    pub fn fold(&mut self, changes: &[CaughtChange]) -> Vec<NormalizedOp> {
        // Disjoint field borrows: the tree + pk map are read-only, the footprint mutates.
        let tree = &self.tree;
        let pk = &self.pk;
        let fp = &mut self.footprint;
        let mut tick: Tick = BTreeMap::new();
        for c in changes {
            fold_change(c, tree, pk, fp, Some(&mut tick));
        }
        emit(&tick, fp)
    }

    /// Update the footprint for one transaction **without** producing wire ops — the
    /// allocation-light path for a query nobody is currently receiving (a pinned, zero-
    /// subscriber materialization) and for the hydrate fold (whose ops the drain discards,
    /// baselining late subscribers off [`footprint_snapshot`](Self::footprint_snapshot)
    /// instead). It skips the per-tick diff map (the `Tick`), the tick-start row capture, and
    /// the `emit` pass entirely — exactly the allocations the wiki-demo RSS profiling traced
    /// the glibc retention to. The footprint it leaves is byte-identical to what
    /// [`fold`](Self::fold) would leave, so a later subscriber's snapshot is unaffected.
    pub fn fold_footprint_only(&mut self, changes: &[CaughtChange]) {
        let tree = &self.tree;
        let pk = &self.pk;
        let fp = &mut self.footprint;
        for c in changes {
            fold_change(c, tree, pk, fp, None);
        }
    }
}

/// Pre-register every table in `node`'s subtree with an empty per-table map (see
/// [`NormalizeFold::new`]). Mirrors [`validate_pk`]'s walk; pruned (`None`) slots carry no
/// table and are skipped.
fn register_tables(node: &TableNode, fp: &mut Footprint) {
    fp.entry(node.table.clone()).or_default();
    for c in node.children.iter().flatten() {
        register_tables(c, fp);
    }
}

/// Every table the tree can surface must have a registered PK (else its rows could not be
/// keyed). Validated once at construction so the fold's PK lookups are infallible.
fn validate_pk(node: &TableNode, pk: &PkMap) {
    assert!(
        pk.contains_key(&*node.table),
        "rindle-replica internal invariant: no primary key registered for table {:?}",
        node.table
    );
    // Pruned slots (`None`) carry no table — a permission table the client never sees need
    // not have a PK registered, so skip them.
    for c in node.children.iter().flatten() {
        validate_pk(c, pk);
    }
}

/// Direction of a subtree fold: a `CaughtChange::Add` increments every row it carries, a
/// `CaughtChange::Remove` decrements (the subtree **is** kept here — §4.3).
#[derive(Clone, Copy)]
enum Delta {
    Inc,
    Dec,
}

/// Fold one change: descend a `CaughtChange::Child` (dropping its parent row), or apply the
/// op at the reached table. `tick` is `Some` on the emitting path (capture tick-start rows
/// for the diff) and `None` on the footprint-only path (see
/// [`NormalizeFold::fold_footprint_only`]).
fn fold_change(
    c: &CaughtChange,
    tnode: &TableNode,
    pk: &PkMap,
    fp: &mut Footprint,
    tick: Option<&mut Tick>,
) {
    match c {
        CaughtChange::Child { rel, change, .. } => {
            // Descend the table tree; an unknown slot (shouldn't happen) or a pruned
            // (`None`, `exists_noSync`) slot is dropped — its witnesses never enter the
            // footprint.
            if let Some(Some(child)) = tnode.children.get(rel.ix()) {
                fold_change(change, child, pk, fp, tick);
            }
        }
        CaughtChange::Add(n) => fold_subtree(n, tnode, pk, fp, tick, Delta::Inc),
        CaughtChange::Remove(n) => fold_subtree(n, tnode, pk, fp, tick, Delta::Dec),
        CaughtChange::Edit { old, row } => {
            // CONTRACT (#13): an `Edit` is **pk-stable**. A pk change is row identity changing,
            // so it is split into Remove(old)+Add(new) at the engine ingress
            // (`ReplicaEngine::push_and_drain`) before the fold ever sees it. `set_row` keys by
            // pk, so a pk-changing Edit would key the NEW pk, miss the entry under the OLD pk,
            // and silently drop the change (leaking the footprint entry). Assert the invariant
            // in debug/tests rather than corrupt; the split upstream is what keeps it true.
            debug_assert!(
                pk_of(old, pk_cols(pk, &tnode.table)) == pk_of(row, pk_cols(pk, &tnode.table)),
                "NormalizeFold received a pk-changing Edit for table {:?} — contract violation \
                 (a pk change must arrive as Remove+Add; see ReplicaEngine::push_and_drain)",
                tnode.table,
            );
            let _ = old; // referenced only by the (debug-only) contract assertion above
            set_row(&tnode.table, row, pk, fp, tick);
        }
    }
}

/// inc/dec this node's row, then recurse into every materialized relationship — the whole
/// subtree, so an `Add` counts each descendant in and a `Remove` counts each out
/// (witnesses included, §4.3).
fn fold_subtree(
    n: &CaughtNode,
    tnode: &TableNode,
    pk: &PkMap,
    fp: &mut Footprint,
    mut tick: Option<&mut Tick>,
    delta: Delta,
) {
    let cols = pk_cols(pk, &tnode.table);
    match delta {
        Delta::Inc => inc(&tnode.table, &n.row, cols, fp, tick.as_deref_mut()),
        Delta::Dec => dec(&tnode.table, &n.row, cols, fp, tick.as_deref_mut()),
    }
    for (slot, children) in &n.relationships {
        // A pruned (`None`, `exists_noSync`) slot's witnesses are excluded from the footprint.
        if let Some(Some(child_tnode)) = tnode.children.get(slot.ix()) {
            for child in children {
                fold_subtree(child, child_tnode, pk, fp, tick.as_deref_mut(), delta);
            }
        }
    }
}

/// `inc(t, row)` (§4.1): membership refcount up; `0→1` is recorded for emission via the
/// captured tick-start state. The footprint **interns** the pipeline's `Arc` row — `clone`
/// is a refcount bump, not a deep copy. The per-table map is pre-registered
/// ([`NormalizeFold::new`]), so `get_mut` is infallible and allocates no `Box<str>` key.
fn inc(
    table: &str,
    row: &OwnedRow,
    pk_cols: &[ColId],
    fp: &mut Footprint,
    tick: Option<&mut Tick>,
) {
    let key = pk_of(row, pk_cols);
    let tbl = fp_table_mut(fp, table);
    if let Some(tick) = tick {
        capture_before(tick, table, &key, tbl);
    }
    let e = tbl.entry(key).or_insert_with(|| FootEntry {
        row: row.clone(),
        rc: 0,
    });
    e.rc += 1;
    e.row = row.clone();
}

/// `dec(t, row)` (§4.1): membership refcount down; `1→0` drops the entry. A `dec` of a row
/// not in the footprint is an inconsistent stream — defensively ignored.
fn dec(
    table: &str,
    row: &OwnedRow,
    pk_cols: &[ColId],
    fp: &mut Footprint,
    tick: Option<&mut Tick>,
) {
    let key = pk_of(row, pk_cols);
    let Some(tbl) = fp.get_mut(table) else {
        return;
    };
    if let Some(tick) = tick {
        capture_before(tick, table, &key, tbl);
    }
    if let Some(e) = tbl.get_mut(&key) {
        e.rc -= 1; // entries are never stored at rc 0, so this can't underflow
        if e.rc == 0 {
            tbl.remove(&key);
        }
    }
}

/// `set(table, new)` (§4.1): a footprinted row's value changed in place — PK is stable, so
/// `new`'s PK keys the existing entry; rc is untouched. An edit of an untracked row is an
/// inconsistent stream — defensively ignored (no membership synthesized).
fn set_row(table: &str, new: &OwnedRow, pk: &PkMap, fp: &mut Footprint, tick: Option<&mut Tick>) {
    let key = pk_of(new, pk_cols(pk, table));
    let Some(tbl) = fp.get_mut(table) else {
        return;
    };
    if let Some(tick) = tick {
        capture_before(tick, table, &key, tbl);
    }
    if let Some(e) = tbl.get_mut(&key) {
        e.row = new.clone();
    }
}

/// The per-table footprint map, keyed without allocating a `Box<str>` on the hot path: every
/// table the tree can surface is pre-registered at construction, so a `get_mut` hit is the
/// rule; the `entry(table.into())` fallback (one `Box<str>`) only fires for a table the walk
/// somehow missed (defensive — it never should).
fn fp_table_mut<'a>(fp: &'a mut Footprint, table: &str) -> &'a mut BTreeMap<PkKey, FootEntry> {
    if fp.contains_key(table) {
        fp.get_mut(table).expect("present")
    } else {
        fp.entry(table.into()).or_default()
    }
}

/// On the first touch of `(table, key)` this tick, record the row as it stood at tick start
/// (`None` ⇒ absent), as an `Arc` clone. Later touches don't overwrite it. Only called on the
/// emitting path (`tick` is `Some`).
fn capture_before(tick: &mut Tick, table: &str, key: &PkKey, tbl: &BTreeMap<PkKey, FootEntry>) {
    // Avoid a `Box<str>` table-key allocation when this tick already has the table (the
    // common case — a tick touches few tables, each repeatedly).
    let rows = if tick.contains_key(table) {
        tick.get_mut(table).expect("present")
    } else {
        tick.entry(table.into()).or_default()
    };
    rows.entry(key.clone())
        .or_insert_with(|| tbl.get(key).map(|e| e.row.clone()));
}

/// Diff each touched row's tick-start state against the post-tick footprint into the net op
/// for the row (§4.1): appeared ⇒ `Add`, vanished ⇒ `Remove`, present-but-changed ⇒
/// `Edit`, otherwise nothing. The interned `Arc` rows are materialized to a wire `WireRow`
/// only **here**, for the (small) set of rows actually emitted. Iterates the sorted maps,
/// so the op order is deterministic.
fn emit(tick: &Tick, fp: &Footprint) -> Vec<NormalizedOp> {
    let mut ops = Vec::new();
    for (table, rows) in tick {
        for (key, before) in rows {
            let after = fp.get(table).and_then(|t| t.get(key)).map(|e| &e.row);
            match (before.as_ref(), after) {
                (None, Some(new)) => ops.push(NormalizedOp::Add {
                    table: table.clone(),
                    row: new.to_value_vec(),
                }),
                (Some(old), None) => ops.push(NormalizedOp::Remove {
                    table: table.clone(),
                    row: old.to_value_vec(),
                }),
                (Some(old), Some(new)) => {
                    if !rows_equal(old, new) {
                        ops.push(NormalizedOp::Edit {
                            table: table.clone(),
                            old: old.to_value_vec(),
                            new: new.to_value_vec(),
                        });
                    }
                }
                (None, None) => {}
            }
        }
    }
    ops
}

/// Extract a row's primary key as a [`PkKey`].
fn pk_of(row: &OwnedRow, pk_cols: &[ColId]) -> PkKey {
    PkKey(pk_cols.iter().map(|&c| row.col(c).to_owned()).collect())
}

/// A table's PK columns. Infallible — [`validate_pk`] guarantees coverage at construction.
fn pk_cols<'a>(pk: &'a PkMap, table: &str) -> &'a [ColId] {
    pk.get(table)
        .map(Vec::as_slice)
        .unwrap_or_else(|| panic!("rindle-replica internal invariant: no PK for table {table:?}"))
}

/// Whole-row value equality (total, non-panicking, `null == null`): did the row change?
fn rows_equal(a: &OwnedRow, b: &OwnedRow) -> bool {
    a.len() == b.len()
        && a.cells()
            .zip(b.cells())
            .all(|(x, y)| values_identical(x, y))
}

#[cfg(test)]
mod tests {
    use super::*;
    use rindle::value::{owned_row, OwnedValue as V, RelId};
    use rindle::{
        CorrelatedSubquery, CorrelatedSubqueryCondition, Correlation, ExistsOp, SimpleCondition,
    };

    // ---- AST builders (only the fields `table_tree`/the fold read) ----

    fn correlation(parent: &str, child: &str) -> Correlation {
        Correlation {
            parent_field: vec![parent.into()],
            child_field: vec![child.into()],
        }
    }

    /// A materialized `related` child: table `table`, relationship alias `alias`.
    fn related(alias: &str, table: &str) -> CorrelatedSubquery {
        CorrelatedSubquery {
            correlation: correlation("id", "parent_id"),
            subquery: Box::new(Ast {
                table: table.into(),
                alias: Some(alias.into()),
                ..Default::default()
            }),
            system: None,
        }
    }

    /// A bare top-level `EXISTS(<table> as <alias>)` where-clause.
    fn exists_where(alias: &str, table: &str) -> Condition {
        Condition::CorrelatedSubquery(CorrelatedSubqueryCondition {
            related: related(alias, table),
            op: ExistsOp::Exists,
            flip: None,
            scalar: None,
            plan_id: None,
        })
    }

    // ---- CaughtChange builders ----

    fn cnode(row: Vec<V>, rels: Vec<(RelId, Vec<CaughtNode>)>) -> CaughtNode {
        CaughtNode {
            row: owned_row(row),
            relationships: rels.into_iter().collect(),
        }
    }
    fn leaf(row: Vec<V>) -> CaughtNode {
        cnode(row, vec![])
    }

    /// A simple PK map keyed on column 0 for the given tables.
    fn pk_col0(tables: &[&str]) -> PkMap {
        tables.iter().map(|t| ((*t).into(), vec![0usize])).collect()
    }

    // ---- helpers to read the emitted ops ----

    fn op_summary(op: &NormalizedOp) -> (String, &'static str, i64) {
        // (table, kind, pk-as-int) — every test row has an Int PK in column 0.
        let int = |r: &WireRow| match &r[0] {
            V::Int(i) => *i,
            o => panic!("expected Int pk, got {o:?}"),
        };
        match op {
            NormalizedOp::Add { table, row } => (table.to_string(), "add", int(row)),
            NormalizedOp::Remove { table, row } => (table.to_string(), "remove", int(row)),
            NormalizedOp::Edit { table, new, .. } => (table.to_string(), "edit", int(new)),
        }
    }
    fn summarize(ops: &[NormalizedOp]) -> Vec<(String, &'static str, i64)> {
        ops.iter().map(op_summary).collect()
    }

    // -------------------------------------------------------------------
    // table_tree derivation
    // -------------------------------------------------------------------

    #[test]
    fn table_tree_orders_related_before_exists() {
        // issue { comments } WHERE EXISTS(labels): slot 0 = related "comments" (table
        // "comment"), slot 1 = EXISTS gate "labels" (table "label"). The tree carries the
        // TABLE at each slot, not the alias.
        let ast = Ast {
            table: "issue".into(),
            related: vec![related("comments", "comment")],
            r#where: Some(exists_where("labels", "label")),
            ..Default::default()
        };
        let tree = table_tree(&ast);
        assert_eq!(
            tree,
            TableNode {
                table: "issue".into(),
                children: vec![
                    Some(TableNode {
                        table: "comment".into(),
                        children: vec![]
                    }),
                    Some(TableNode {
                        table: "label".into(),
                        children: vec![]
                    }),
                ],
            }
        );
    }

    #[test]
    fn table_tree_prunes_permission_exists_slot() {
        // `exists_noSync` (§4): issue WHERE EXISTS(labels) [client] AND
        // EXISTS_permissions(acl). The client `label` slot folds; the permission `acl` slot
        // is pruned to `None` (its witnesses never enter the footprint) but KEEPS its slot
        // position, so later slots stay aligned with the dataflow `RelId`s.
        let mut perm = related("acl_rel", "acl");
        perm.system = Some(rindle::System::Permissions);
        let ast = Ast {
            table: "issue".into(),
            r#where: Some(Condition::And {
                conditions: vec![
                    exists_where("labels", "label"),
                    Condition::CorrelatedSubquery(CorrelatedSubqueryCondition {
                        related: perm,
                        op: ExistsOp::Exists,
                        flip: None,
                        scalar: None,
                        plan_id: None,
                    }),
                ],
            }),
            ..Default::default()
        };
        let tree = table_tree(&ast);
        assert_eq!(
            tree,
            TableNode {
                table: "issue".into(),
                children: vec![
                    Some(TableNode {
                        table: "label".into(),
                        children: vec![]
                    }),
                    None, // the permission `acl` slot is pruned, position preserved
                ],
            }
        );
    }

    #[test]
    fn table_tree_prunes_permission_exists_under_or() {
        // A primary permission shape: `issue WHERE (EXISTS(labels) OR EXISTS_permissions(acl))`.
        // Under OR the slots are gathered in where-tree pre-order, so slot 0 = client `label`
        // (folded), slot 1 = permission `acl` (pruned to `None`). The pruning is keyed on the
        // by-name match `query_local_slot_names` also uses, so the `None` lands on the acl slot
        // and the client `label` slot is untouched — no mis-alignment, no leak, no corruption.
        let mut perm = related("acl_rel", "acl");
        perm.system = Some(rindle::System::Permissions);
        let ast = Ast {
            table: "issue".into(),
            r#where: Some(Condition::Or {
                conditions: vec![
                    exists_where("labels", "label"),
                    Condition::CorrelatedSubquery(CorrelatedSubqueryCondition {
                        related: perm,
                        op: ExistsOp::Exists,
                        flip: None,
                        scalar: None,
                        plan_id: None,
                    }),
                ],
            }),
            ..Default::default()
        };
        let tree = table_tree(&ast);
        assert_eq!(
            tree,
            TableNode {
                table: "issue".into(),
                children: vec![
                    Some(TableNode {
                        table: "label".into(),
                        children: vec![]
                    }),
                    None, // permission `acl` slot pruned, position preserved
                ],
            }
        );
    }

    #[test]
    fn table_tree_recurses_into_nested_related() {
        // issue { comments { authors } }: nested child tables resolved depth-first.
        let mut comments = related("comments", "comment");
        comments.subquery.related = vec![related("authors", "author")];
        let ast = Ast {
            table: "issue".into(),
            related: vec![comments],
            ..Default::default()
        };
        let tree = table_tree(&ast);
        assert_eq!(tree.table.as_ref(), "issue");
        assert_eq!(tree.children.len(), 1);
        let comment = tree.children[0].as_ref().unwrap();
        assert_eq!(comment.table.as_ref(), "comment");
        assert_eq!(comment.children.len(), 1);
        assert_eq!(
            comment.children[0].as_ref().unwrap().table.as_ref(),
            "author"
        );
    }

    // -------------------------------------------------------------------
    // The fold
    // -------------------------------------------------------------------

    /// playlist { tracks } — the canonical m2m shape: one track can sit under several
    /// playlists, so the same base `track` row is reached via multiple tree paths.
    fn playlist_tracks_ast() -> Ast {
        Ast {
            table: "playlist".into(),
            related: vec![related("tracks", "track")],
            ..Default::default()
        }
    }

    #[test]
    fn snapshot_emits_every_footprint_row_once() {
        let mut f = NormalizeFold::new(&playlist_tracks_ast(), pk_col0(&["playlist", "track"]));
        // Hydrate: playlist 1 with track 100.
        let ops = f.fold(&[CaughtChange::Add(cnode(
            vec![V::Int(1)],
            vec![(RelId(0), vec![leaf(vec![V::Int(100), V::str("a")])])],
        ))]);
        assert_eq!(
            summarize(&ops),
            vec![("playlist".into(), "add", 1), ("track".into(), "add", 100),]
        );
    }

    #[test]
    fn multi_path_add_collapses_to_one_add_with_rc2() {
        // §4.2: track 100 under BOTH playlist 1 and playlist 2 in one tick → ONE track add.
        let mut f = NormalizeFold::new(&playlist_tracks_ast(), pk_col0(&["playlist", "track"]));
        let track = || leaf(vec![V::Int(100), V::str("a")]);
        let ops = f.fold(&[
            CaughtChange::Add(cnode(vec![V::Int(1)], vec![(RelId(0), vec![track()])])),
            CaughtChange::Add(cnode(vec![V::Int(2)], vec![(RelId(0), vec![track()])])),
        ]);
        assert_eq!(
            summarize(&ops),
            vec![
                ("playlist".into(), "add", 1),
                ("playlist".into(), "add", 2),
                ("track".into(), "add", 100), // only once, despite two paths
            ]
        );
        // Footprint refcount collapsed the two paths.
        assert_eq!(f.footprint["track"].values().next().unwrap().rc, 2);
    }

    #[test]
    fn multi_path_remove_keeps_row_until_last_ref() {
        // Hydrate track 100 under playlists 1 and 2 (rc 2).
        let mut f = NormalizeFold::new(&playlist_tracks_ast(), pk_col0(&["playlist", "track"]));
        let track = || leaf(vec![V::Int(100), V::str("a")]);
        f.fold(&[
            CaughtChange::Add(cnode(vec![V::Int(1)], vec![(RelId(0), vec![track()])])),
            CaughtChange::Add(cnode(vec![V::Int(2)], vec![(RelId(0), vec![track()])])),
        ]);

        // Tick 2: playlist 1 drops the track (Child→Remove) → rc 2→1, NO track remove.
        let ops = f.fold(&[CaughtChange::Child {
            row: owned_row(vec![V::Int(1)]),
            rel: RelId(0),
            change: Box::new(CaughtChange::Remove(track())),
        }]);
        assert_eq!(summarize(&ops), vec![] as Vec<(String, &str, i64)>);
        assert_eq!(f.footprint["track"].values().next().unwrap().rc, 1);

        // Tick 3: playlist 2 drops it too → rc 1→0 → track remove emitted.
        let ops = f.fold(&[CaughtChange::Child {
            row: owned_row(vec![V::Int(2)]),
            rel: RelId(0),
            change: Box::new(CaughtChange::Remove(track())),
        }]);
        assert_eq!(summarize(&ops), vec![("track".into(), "remove", 100)]);
        assert!(!f.footprint.contains_key("track") || f.footprint["track"].is_empty());
    }

    #[test]
    fn remove_carries_the_subtree() {
        // §4.3: removing a parent decrements its children/witnesses too. Hydrate
        // playlist 1 { track 100 }, then Remove the whole playlist subtree.
        let mut f = NormalizeFold::new(&playlist_tracks_ast(), pk_col0(&["playlist", "track"]));
        let subtree = || {
            cnode(
                vec![V::Int(1)],
                vec![(RelId(0), vec![leaf(vec![V::Int(100)])])],
            )
        };
        f.fold(&[CaughtChange::Add(subtree())]);

        let ops = f.fold(&[CaughtChange::Remove(subtree())]);
        assert_eq!(
            summarize(&ops),
            vec![
                ("playlist".into(), "remove", 1),
                ("track".into(), "remove", 100), // the subtree row went too
            ]
        );
    }

    #[test]
    fn edit_dedups_across_multi_path() {
        // track 100 under playlists 1 and 2 (rc 2). An edit of its body arrives once per
        // ref (two Child→Edit). They collapse to ONE normalized edit; rc unchanged.
        let mut f = NormalizeFold::new(&playlist_tracks_ast(), pk_col0(&["playlist", "track"]));
        let v1 = || leaf(vec![V::Int(100), V::str("v1")]);
        f.fold(&[
            CaughtChange::Add(cnode(vec![V::Int(1)], vec![(RelId(0), vec![v1()])])),
            CaughtChange::Add(cnode(vec![V::Int(2)], vec![(RelId(0), vec![v1()])])),
        ]);

        let child_edit = |parent: i64| CaughtChange::Child {
            row: owned_row(vec![V::Int(parent)]),
            rel: RelId(0),
            change: Box::new(CaughtChange::Edit {
                old: owned_row(vec![V::Int(100), V::str("v1")]),
                row: owned_row(vec![V::Int(100), V::str("v2")]),
            }),
        };
        let ops = f.fold(&[child_edit(1), child_edit(2)]);
        assert_eq!(summarize(&ops), vec![("track".into(), "edit", 100)]);
        // The emitted edit carries old=v1, new=v2; rc still 2.
        match ops.into_iter().next().unwrap() {
            NormalizedOp::Edit { old, new, .. } => {
                assert!(values_identical(old[1].as_ref(), V::str("v1").as_ref()));
                assert!(values_identical(new[1].as_ref(), V::str("v2").as_ref()));
            }
            o => panic!("expected edit, got {o:?}"),
        }
        assert_eq!(f.footprint["track"].values().next().unwrap().rc, 2);
    }

    #[test]
    fn transient_add_then_remove_in_one_tick_is_a_noop() {
        // A row added and removed within the same transaction nets to nothing.
        let mut f = NormalizeFold::new(&playlist_tracks_ast(), pk_col0(&["playlist", "track"]));
        let ops = f.fold(&[
            CaughtChange::Add(cnode(vec![V::Int(1)], vec![])),
            CaughtChange::Remove(cnode(vec![V::Int(1)], vec![])),
        ]);
        assert_eq!(summarize(&ops), vec![] as Vec<(String, &str, i64)>);
        assert!(!f.footprint.contains_key("playlist") || f.footprint["playlist"].is_empty());
    }

    #[test]
    fn edit_unchanged_value_emits_nothing() {
        let mut f = NormalizeFold::new(&playlist_tracks_ast(), pk_col0(&["playlist", "track"]));
        f.fold(&[CaughtChange::Add(cnode(
            vec![V::Int(1), V::str("x")],
            vec![],
        ))]);
        // Edit that does not change any cell → no op.
        let ops = f.fold(&[CaughtChange::Edit {
            old: owned_row(vec![V::Int(1), V::str("x")]),
            row: owned_row(vec![V::Int(1), V::str("x")]),
        }]);
        assert_eq!(summarize(&ops), vec![] as Vec<(String, &str, i64)>);
    }

    // -------------------------------------------------------------------
    // Adversarial-review HIGH #13: pk-changing edits.
    //
    // The fold ASSUMES pk-stable edits by contract — a pk change arrives as Remove+Add because
    // the engine ingress (`ReplicaEngine::push_and_drain`) splits it before the fold sees one.
    // That end-to-end split (a real `UPDATE … SET <pk> = …` → Remove(old)+Add(new) in the
    // change stream) is covered by `tests/pk_edit_split.rs`. Here we only pin the pk-stable
    // path the fold actually handles.
    // -------------------------------------------------------------------

    #[test]
    fn pk_stable_edit_still_sets_in_place() {
        // The pk-stable path is unchanged: a value-only edit is one in-place Edit, rc untouched.
        let ast = Ast {
            table: "issue".into(),
            ..Default::default()
        };
        let mut f = NormalizeFold::new(&ast, pk_col0(&["issue"]));
        f.fold(&[CaughtChange::Add(leaf(vec![V::Int(1), V::str("a")]))]);
        let ops = f.fold(&[CaughtChange::Edit {
            old: owned_row(vec![V::Int(1), V::str("a")]),
            row: owned_row(vec![V::Int(1), V::str("b")]),
        }]);
        assert_eq!(summarize(&ops), vec![("issue".into(), "edit", 1)]);
        assert_eq!(f.footprint["issue"].values().next().unwrap().rc, 1);
    }

    #[test]
    fn normalized_op_serde_round_trips() {
        let op = NormalizedOp::Edit {
            table: "track".into(),
            old: vec![
                V::Int(1),
                V::str("v1"),
                V::Null,
                V::Bool(true),
                V::Float(1.5),
            ],
            new: vec![
                V::Int(1),
                V::str("v2"),
                V::Null,
                V::Bool(true),
                V::Float(1.5),
            ],
        };
        let json = serde_json::to_value(&op).expect("serialize");
        let back: NormalizedOp = serde_json::from_value(json.clone()).expect("deserialize");
        let json2 = serde_json::to_value(&back).expect("re-serialize");
        assert_eq!(json, json2);
    }

    // -------------------------------------------------------------------
    // Relationship aggregates → synthetic base tables (AGGREGATE-SYNC-DESIGN.md §3.2)
    // -------------------------------------------------------------------

    /// `issue { commentCount: count(comments) }` — comment.issue ↔ issue.id.
    fn issue_count_ast() -> Ast {
        let mut count = related("commentCount", "comment");
        count.subquery.aggregate = Some(Aggregate::Count);
        Ast {
            table: "issue".into(),
            related: vec![count],
            ..Default::default()
        }
    }

    fn issue_agg_ast(agg: Aggregate) -> Ast {
        let mut rel = related("commentAgg", "comment");
        rel.subquery.aggregate = Some(agg);
        Ast {
            table: "issue".into(),
            related: vec![rel],
            ..Default::default()
        }
    }

    #[test]
    fn reject_unsupported_sync_aggregate_gates_sum_and_avg() {
        // Count is the only sync-supported relationship aggregate; sum/avg are rejected at
        // registration (they work for a direct read but have no synthetic-table encoding).
        assert!(reject_unsupported_sync_aggregate(&issue_count_ast()).is_ok());
        assert!(reject_unsupported_sync_aggregate(&Ast {
            table: "issue".into(),
            ..Default::default()
        })
        .is_ok());
        assert!(
            reject_unsupported_sync_aggregate(&issue_agg_ast(Aggregate::Sum("score".into())))
                .is_err()
        );
        assert!(
            reject_unsupported_sync_aggregate(&issue_agg_ast(Aggregate::Avg("score".into())))
                .is_err()
        );

        // A sum nested under an ordinary materialized `related` is caught by the recursion.
        let mut parent = related("comments", "comment");
        parent.subquery.related = vec![{
            let mut inner = related("replyAvg", "reply");
            inner.subquery.aggregate = Some(Aggregate::Avg("len".into()));
            inner
        }];
        let nested = Ast {
            table: "issue".into(),
            related: vec![parent],
            ..Default::default()
        };
        assert!(reject_unsupported_sync_aggregate(&nested).is_err());
    }

    #[test]
    fn table_tree_retags_aggregate_slot_to_a_synthetic_table() {
        // The aggregate slot is re-tagged to the synthetic table — the child `comment` rows
        // are NOT surfaced (they are never synced), and the node is a leaf.
        let ast = issue_count_ast();
        let synthetic = agg_table_name(&ast.related[0]);
        let tree = table_tree(&ast);
        assert_eq!(tree.table.as_ref(), "issue");
        assert_eq!(tree.children.len(), 1);
        let child = tree.children[0].as_ref().unwrap();
        assert_eq!(child.table, synthetic);
        assert!(child.table.starts_with("__agg_"));
        assert_ne!(child.table.as_ref(), "comment");
        assert!(child.children.is_empty());
    }

    /// The `having_count` lowering (`PARENT-AGGREGATE-FILTER-DESIGN.md` §3): the display
    /// `count_as` stays a materialized `related`, and a CLONE of it — re-aliased, carrying the
    /// post-aggregation `HAVING` — rides the `where` tree as an EXISTS gate.
    fn issue_having_count_ast() -> Ast {
        let mut ast = issue_count_ast();
        let mut gate = ast.related[0].clone();
        gate.subquery.alias = Some("__having_commentCount".into());
        gate.subquery.having = Some(Condition::Simple(SimpleCondition {
            op: Op::Gt,
            left: ValuePosition::Column {
                name: "count".into(),
            },
            right: ValuePosition::Literal { value: Lit::Int(0) },
        }));
        ast.r#where = Some(Condition::CorrelatedSubquery(CorrelatedSubqueryCondition {
            related: gate,
            op: ExistsOp::Exists,
            flip: None,
            scalar: None,
            plan_id: None,
        }));
        ast
    }

    #[test]
    fn rewrite_aggregates_points_a_display_count_at_its_synthetic_table() {
        // The §3.3 rewrite: a relationship `count` becomes a precomputed singular join over the
        // synthetic table, NOT a reduce (which would recount already-aggregated rows). Alias
        // survives so the view slot keeps its name; the correlation is untouched because the
        // synthetic columns are named after the child correlation fields.
        let ast = issue_count_ast();
        let out = rewrite_aggregates(&ast);
        assert_eq!(out.related.len(), 1);
        let rel = &out.related[0];
        assert_eq!(rel.subquery.table, agg_table_name(&ast.related[0]));
        assert_eq!(rel.subquery.aggregate, Some(Aggregate::Count));
        assert!(rel.subquery.aggregate_precomputed);
        assert_eq!(rel.subquery.alias.as_deref(), Some("commentCount"));
        assert_eq!(rel.correlation, ast.related[0].correlation);
        // The child table's own filter/relationships are GONE — they are baked into the
        // synthetic table's identity (the name is a hash of them), not re-evaluated locally.
        assert!(rel.subquery.r#where.is_none());
        assert!(rel.subquery.related.is_empty());
    }

    #[test]
    fn rewrite_aggregates_rewrites_a_having_count_gate_in_the_where_tree() {
        // The half `rewriteAggregates` used to miss (and the whole reason `/thoughts/movies`
        // went blank): the gate is a relationship `count` like any other, so it must be
        // rewritten too. Left alone it would reduce over child rows the server deliberately
        // does not sync, and every parent would fail it.
        let ast = issue_having_count_ast();
        let out = rewrite_aggregates(&ast);
        let Some(Condition::CorrelatedSubquery(gate)) = out.r#where.as_ref() else {
            panic!("expected the gate EXISTS to survive the rewrite");
        };
        assert!(gate.related.subquery.table.starts_with("__agg_"));
        assert!(gate.related.subquery.aggregate_precomputed);
        // The HAVING is the gate's WHOLE predicate and addresses the reduce's output column,
        // which is exactly what the synthetic row carries — so it must survive.
        assert!(
            gate.related.subquery.having.is_some(),
            "dropping the HAVING turns the gate into a bare EXISTS"
        );
        // Same table as the display aggregate ⇒ the rewrite costs no extra sync.
        assert_eq!(gate.related.subquery.table, out.related[0].subquery.table);
    }

    #[test]
    fn rewrite_aggregates_is_a_cross_language_twin() {
        // The Rust half of the twin contract. `packages/normalized/test/agg-table.test.ts`
        // pins the SAME table name for the SAME aggregate definition through the TypeScript
        // `rewriteAggregates`; the two must agree byte-for-byte or a synced client reads a
        // table the server never fills. Same vector as
        // `agg_table_name_cross_language_vector`, reached through the rewrite instead of the
        // hash directly, so a rewrite that names the table some OTHER way is caught here.
        let ast = Ast {
            table: "issue".into(),
            related: vec![CorrelatedSubquery {
                correlation: correlation("id", "issue_id"),
                subquery: Box::new(Ast {
                    table: "comment".into(),
                    alias: Some("commentCount".into()),
                    aggregate: Some(Aggregate::Count),
                    ..Default::default()
                }),
                system: None,
            }],
            ..Default::default()
        };
        let out = rewrite_aggregates(&ast);
        assert_eq!(&*out.related[0].subquery.table, "__agg_725f6575e4f86856");
    }

    #[test]
    fn rewrite_aggregates_leaves_a_local_count_as_a_native_reduce() {
        // L1 (`201-LOCAL-ONLY-TABLES-DESIGN.md` §5.2): a count over a LOCAL child has no
        // server-authoritative `__agg_*` base, so rewriting it would point the relationship at
        // a synthetic table nothing ever feeds — empty forever.
        let ast = issue_count_ast();
        let out = rewrite_aggregates_with_local(&ast, &|t| t == "comment");
        assert_eq!(out.related[0].subquery.table.as_ref(), "comment");
        assert!(!out.related[0].subquery.aggregate_precomputed);
    }

    #[test]
    fn table_tree_prunes_a_having_count_gate_slot() {
        // The gate is an EXISTS over a HAVING-filtered reduce, so its slot holds WITNESS child
        // rows, not the reduce's `(group…, count)` output. Re-tagging them to the synthetic
        // table would ship full-width `comment` rows under its 2-column schema (the client's
        // width check rejects the batch). It is pruned to `None` — position preserved — while
        // the DISPLAY aggregate's slot still carries the synthetic table.
        let ast = issue_having_count_ast();
        let tree = table_tree(&ast);
        assert_eq!(tree.table.as_ref(), "issue");
        assert_eq!(tree.children.len(), 2, "display slot + gate slot");
        let display = tree.children[0].as_ref().expect("display slot survives");
        assert!(display.table.starts_with("__agg_"));
        assert_eq!(tree.children[1], None, "the gate slot is pruned");
    }

    #[test]
    fn a_having_count_gate_names_the_display_aggregate_table() {
        // Load-bearing for the prune: the client reads the gate off the DISPLAY aggregate's
        // rows, which only works because `agg_table_name` hashes neither `alias` nor `having`
        // — so the clone resolves to the very same synthetic table. If this ever diverges, the
        // pruned gate would have no rows to evaluate against and every parent would drop out.
        let ast = issue_having_count_ast();
        let Some(Condition::CorrelatedSubquery(gate)) = ast.r#where.as_ref() else {
            panic!("expected the gate EXISTS");
        };
        assert_eq!(
            agg_table_name(&gate.related),
            agg_table_name(&ast.related[0])
        );
    }

    #[test]
    fn a_having_count_gate_widens_no_projection() {
        // `collect_required` mirrors `table_tree`'s prune, so the gate contributes NOTHING to
        // the advertised projection: adding `.having(count > 0)` to a query must not widen the
        // footprint by one column. (`comment` is still requested — by the DISPLAY aggregate,
        // which recurses into its child; the gate just adds nothing on top.)
        assert_eq!(
            required_columns_by_table(&issue_having_count_ast()),
            required_columns_by_table(&issue_count_ast()),
        );
    }

    #[test]
    fn agg_table_schemas_derives_the_synthetic_schema() {
        let ast = issue_count_ast();
        let tables = agg_table_schemas(&ast);
        assert_eq!(tables.len(), 1);
        let t = &tables[0];
        assert_eq!(t.name, agg_table_name(&ast.related[0]));
        // `[child_field…, "count"]` (the `related` helper correlates on `parent_id`), group
        // key = the leading `child_field`.
        assert_eq!(
            t.columns.iter().map(|c| &**c).collect::<Vec<_>>(),
            vec!["parent_id", "count"]
        );
        assert_eq!(t.key_len, 1);
    }

    #[test]
    fn agg_table_name_cross_language_vector() {
        // The cross-language contract (AGGREGATE-SYNC-DESIGN.md §3.1): a canonical aggregate
        // definition must hash identically in Rust and the TS client twin (`@rindle/normalized`
        // `aggTableName`, pinned in `agg-table.test.ts`). A drift in either FNV byte protocol
        // breaks routing — client and server would name the synthetic table differently.
        // `count(comment)` grouped by `comment.issue_id`, no filter (the parent table is not
        // part of the hash, so the parent's table name is irrelevant).
        let csq = CorrelatedSubquery {
            correlation: correlation("id", "issue_id"),
            subquery: Box::new(Ast {
                table: "comment".into(),
                alias: Some("commentCount".into()),
                aggregate: Some(Aggregate::Count),
                ..Default::default()
            }),
            system: None,
        };
        assert_eq!(&*agg_table_name(&csq), "__agg_725f6575e4f86856");
    }

    #[test]
    fn agg_table_name_cross_language_vector_integer_literal_filter() {
        // The exact-int plane of the literal hash (design 226 Stage B): an integral
        // JSON literal deserializes as `Lit::Int` and hashes tag 5 + exact i64 LE
        // bytes, and the TS twin mirrors serde's integer-token rule (`agg-table.test.ts`
        // pins these SAME names). A drift here leaves the server advertising a
        // different synthetic table from the one the client rewrite reads — the local
        // aggregate would simply stay empty.
        let filtered = |value: Lit| CorrelatedSubquery {
            correlation: correlation("id", "issue_id"),
            subquery: Box::new(Ast {
                table: "comment".into(),
                aggregate: Some(Aggregate::Count),
                r#where: Some(Condition::Simple(SimpleCondition {
                    op: Op::Eq,
                    left: ValuePosition::Column {
                        name: "score".into(),
                    },
                    right: ValuePosition::Literal { value },
                })),
                ..Default::default()
            }),
            system: None,
        };
        // `score = 5` — the canonical small-integer literal every JS client produces.
        assert_eq!(
            &*agg_table_name(&filtered(Lit::Int(5))),
            "__agg_386c48c5334e7707"
        );
        // What a JS client's `2 ** 60` ACTUALLY arrives as: JS serializes the shortest
        // decimal that round-trips — the token "1152921504606847000", NOT the binary
        // value 1152921504606846976 — and serde parses that exact token as the i64.
        // The TS twin therefore hashes the wire token, and pins THIS name for `2 ** 60`.
        assert_eq!(
            &*agg_table_name(&filtered(Lit::Int(1152921504606847000))),
            "__agg_8eeccc486bcc3c3a"
        );
        // The binary-exact 2^60 is only producible by a bigint-capable producer (the
        // Rust fluent API); it is a DISTINCT identity from the JS token above — no TS
        // twin pin exists for it.
        assert_eq!(
            &*agg_table_name(&filtered(Lit::Int(1 << 60))),
            "__agg_3908e94b423e01c2"
        );
        // A non-integral literal stays the f64 plane (tag 2) on both sides.
        assert_eq!(
            &*agg_table_name(&filtered(Lit::Number(2.5))),
            "__agg_e504b86c532033e7"
        );
        // The property the int plane exists for: two adjacent i64s above 2^53 round to
        // the SAME f64 — under f64 hashing they'd share a synthetic table.
        assert_ne!(
            agg_table_name(&filtered(Lit::Int(1 << 60))),
            agg_table_name(&filtered(Lit::Int((1 << 60) + 1)))
        );
    }

    #[test]
    fn agg_table_name_is_stable_and_definition_sensitive() {
        let n1 = agg_table_name(&issue_count_ast().related[0]);
        // Stable across calls (same definition → same name → cross-query sharing).
        assert_eq!(n1, agg_table_name(&issue_count_ast().related[0]));
        assert!(n1.starts_with("__agg_"));

        // A different child filter → a different table (else issue 5's filtered and
        // unfiltered counts would collide on `(table, pk)` across two queries).
        let mut filtered = issue_count_ast();
        filtered.related[0].subquery.r#where = Some(Condition::Simple(SimpleCondition {
            op: Op::Ne,
            left: ValuePosition::Column {
                name: "body".into(),
            },
            right: ValuePosition::Literal {
                value: Lit::Str("x".into()),
            },
        }));
        assert_ne!(agg_table_name(&filtered.related[0]), n1);

        // A different group key (correlation child) → a different table.
        let mut regrouped = issue_count_ast();
        regrouped.related[0].correlation.child_field = vec!["author".into()];
        assert_ne!(agg_table_name(&regrouped.related[0]), n1);

        // A different child table → a different table.
        let mut other = issue_count_ast();
        other.related[0].subquery.table = "reaction".into();
        assert_ne!(agg_table_name(&other.related[0]), n1);

        // The PARENT correlation field is excluded — the count for a group key is the same
        // whichever parent joins it, so two parents share one synthetic table.
        let mut other_parent = issue_count_ast();
        other_parent.related[0].correlation.parent_field = vec!["other_id".into()];
        assert_eq!(agg_table_name(&other_parent.related[0]), n1);
    }

    #[test]
    fn fold_ships_the_aggregate_row_under_the_synthetic_table_not_the_child() {
        let ast = issue_count_ast();
        let synthetic = agg_table_name(&ast.related[0]);
        // The synthetic table's PK is the group column; `comment` is never registered.
        let pk: PkMap = [
            ("issue".into(), vec![0usize]),
            (synthetic.clone(), vec![0usize]),
        ]
        .into_iter()
        .collect();
        let mut f = NormalizeFold::new(&ast, pk);

        // Hydrate: issue 1 carries its reduce-output `(issue=1, count=2)` row as a Child.
        let ops = f.fold(&[CaughtChange::Add(cnode(
            vec![V::Int(1)],
            vec![(RelId(0), vec![leaf(vec![V::Int(1), V::Int(2)])])],
        ))]);
        // The synthetic table sorts before "issue" ('_' < 'i'); the agg row is tagged the
        // SYNTHETIC table, never "comment".
        assert_eq!(
            summarize(&ops),
            vec![
                (synthetic.to_string(), "add", 1),
                ("issue".into(), "add", 1),
            ]
        );

        // A later count shift for issue 1 (2 → 3): the reduce ships a pk-stable Edit of the
        // synthetic row (the count is the value, the group key is the PK).
        let ops = f.fold(&[CaughtChange::Child {
            row: owned_row(vec![V::Int(1)]),
            rel: RelId(0),
            change: Box::new(CaughtChange::Edit {
                old: owned_row(vec![V::Int(1), V::Int(2)]),
                row: owned_row(vec![V::Int(1), V::Int(3)]),
            }),
        }]);
        assert_eq!(summarize(&ops), vec![(synthetic.to_string(), "edit", 1)]);

        // The group dies (last comment removed): a Remove of the synthetic row.
        let ops = f.fold(&[CaughtChange::Child {
            row: owned_row(vec![V::Int(1)]),
            rel: RelId(0),
            change: Box::new(CaughtChange::Remove(leaf(vec![V::Int(1), V::Int(3)]))),
        }]);
        assert_eq!(summarize(&ops), vec![(synthetic.to_string(), "remove", 1)]);
    }
}