boatramp_core/

orm.rs

//! A typed query AST and an injection-safe SQL compiler — the backing for the `orm`
//! handler binding (`boatramp:handlers/orm`).
//!
//! The compiler turns a typed [`Select`] / [`Insert`] / [`Update`] into a `?N`-placeholder
//! SQL string plus its bound [`SqlValue`] parameters, in order. It is **pure** (no I/O, no
//! wasm/wit deps) so it is fully unit-testable; the binding runs the result through the same
//! [`crate::sql::SqlTransaction`] the raw `sql-query` binding uses, which rewrites `?N` to the
//! engine's native dialect. That shares one execution + dialect substrate across bindings.
//!
//! # Expressiveness
//! [`Expr`] is a recursive scalar expression (column, bound value, aggregate, arithmetic,
//! a small allow-listed [`Func`] set, JSON key-path extraction, Postgres-only `pgvector`
//! distance, and a narrow correlated roll-up — a filtered aggregate over one named table) and
//! [`Predicate`] is a recursive boolean tree (`AND`/`OR`/`NOT` +
//! comparisons/`BETWEEN`/`IN`/`LIKE`/`IS NULL`). Selects add joins, `GROUP BY`/`HAVING`,
//! aliases, ordering and pagination; inserts/updates add `RETURNING`. General scalar
//! subqueries (beyond the correlated roll-up), CTEs and window functions are deliberately out
//! of scope — they go through the raw `sql-query` escape hatch.
//!
//! # Safety
//! - **Every value binds as a parameter** (`?N`); no value is ever formatted into the SQL.
//! - **Identifiers are validated** (`[A-Za-z_][A-Za-z0-9_]*`, optionally `table.column`) and
//!   emitted unquoted — an identifier that isn't a plain name is rejected, so a column/table
//!   name can't smuggle SQL. Function names come from the closed [`Func`] enum (never a
//!   free string), so they can't inject either.
//! - **UPDATE requires a filter** — an unbounded update is refused.
//!
//! # Isolation
//! The project/database boundary is the caller's (the binding opens a per-project database).
//! An optional per-query [`Scope`] (`column = value`) is the *in-site* row-tenancy seam: on a
//! read/update it is conjoined into the `WHERE`; on an insert it is forced into every row. It is
//! guest-declared here (the shim's `Scoped` model); a host-enforced-from-claims variant is a
//! later enhancement (see plans/PLAN-orm-wit.md §4).

use crate::sql::{Dialect, SqlValue};
use crate::tenancy::ResolvedScope;

// ---- expressions -----------------------------------------------------------

/// An aggregate function.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Agg {
    Count,
    Sum,
    Avg,
    Min,
    Max,
}

impl Agg {
    fn keyword(self) -> &'static str {
        match self {
            Self::Count => "count",
            Self::Sum => "sum",
            Self::Avg => "avg",
            Self::Min => "min",
            Self::Max => "max",
        }
    }
}

/// An arithmetic operator (rendered parenthesized, so precedence is explicit).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BinOp {
    Add,
    Sub,
    Mul,
    Div,
    Mod,
}

impl BinOp {
    fn symbol(self) -> &'static str {
        match self {
            Self::Add => "+",
            Self::Sub => "-",
            Self::Mul => "*",
            Self::Div => "/",
            Self::Mod => "%",
        }
    }
}

/// An allow-listed, dialect-portable scalar function. A closed enum (not a free string) so a
/// function name can never inject and only portable functions are reachable.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Func {
    Lower,
    Upper,
    Length,
    Trim,
    Abs,
    Round,
    Coalesce,
    /// `CURRENT_TIMESTAMP` (ANSI); takes no arguments.
    Now,
}

impl Func {
    /// The rendered SQL name, and the accepted argument arity as an inclusive `(min, max)`
    /// where `max == None` means variadic.
    fn spec(self) -> (&'static str, usize, Option<usize>) {
        match self {
            Self::Lower => ("lower", 1, Some(1)),
            Self::Upper => ("upper", 1, Some(1)),
            Self::Length => ("length", 1, Some(1)),
            Self::Trim => ("trim", 1, Some(1)),
            Self::Abs => ("abs", 1, Some(1)),
            Self::Round => ("round", 1, Some(2)),
            Self::Coalesce => ("coalesce", 2, None),
            Self::Now => ("current_timestamp", 0, Some(0)),
        }
    }
}

/// A `pgvector` distance metric. A closed enum, so the rendered operator is a compiler
/// constant (never a guest string) and can't inject. Postgres-only (see [`Expr::Distance`]).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Metric {
    /// Cosine distance (`<=>`).
    Cosine,
    /// Euclidean / L2 distance (`<->`).
    L2,
}

impl Metric {
    fn operator(self) -> &'static str {
        match self {
            Self::Cosine => "<=>",
            Self::L2 => "<->",
        }
    }
}

/// The argument of a correlated roll-up ([`Expr::RelatedAggregate`]): `*` (only valid for
/// `count`) or a single validated column. Deliberately not a full [`Expr`] — a correlated
/// aggregate takes a column or `*`, nothing free-form.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RelArg {
    /// `count(*)`.
    Star,
    /// `agg(<column>)`.
    Column(String),
}

/// A scalar expression: the leaf/branch type used in select lists, comparisons, `SET`,
/// `GROUP BY`, `ORDER BY` and join conditions.
#[derive(Debug, Clone, PartialEq)]
pub enum Expr {
    /// A column reference (`col` or `table.col`), validated + emitted unquoted.
    Column(String),
    /// A literal value — bound as a `?N` parameter, never formatted in.
    Value(SqlValue),
    /// `*`, valid only as the argument of `count(*)`.
    Star,
    /// An aggregate over an inner expression (use [`Expr::Star`] for `count(*)`).
    Aggregate(Agg, Box<Self>),
    /// A parenthesized binary arithmetic expression.
    Binary(BinOp, Box<Self>, Box<Self>),
    /// An allow-listed function call.
    Func(Func, Vec<Self>),
    /// Extract a text value from a JSON column by a key path (e.g. `["a", "b"]` ⇒ `$.a.b`).
    /// Rendered per-dialect (SQLite/MySQL `json_extract`, Postgres `#>>`); each key is
    /// validated as an identifier so the built path can't inject.
    JsonExtract(Box<Self>, Vec<String>),
    /// A `pgvector` distance between two vector expressions, rendered `(left <op> right)`.
    /// **Postgres-only** — SQLite/MySQL have no vector type, so it fails closed
    /// ([`OrmError::BadExpr`]); there is no correct portable fallback. Usable in a select
    /// list and in `ORDER BY` (nearest-neighbour search).
    Distance {
        left: Box<Self>,
        right: Box<Self>,
        metric: Metric,
    },
    /// A vector literal — a bracketed float list (`[0.1, 0.2, …]`) bound as a `?N` parameter
    /// and rendered `?N::vector`. The components are validated as finite numbers; the value
    /// binds (never formatted in), so it can't inject. **Postgres-only.**
    VectorLiteral(String),
    /// A filtered aggregate over a *named* related table, rendered as a scalar subquery
    /// `(SELECT agg(arg) FROM table WHERE <filter>)` — a correlated roll-up. The correlation
    /// to the outer row lives in `filter` (e.g. `child.fk = parent.pk`); unlike a
    /// `LEFT JOIN … GROUP BY` rewrite it never fans out, so several counts per row are just
    /// several select-list entries. Everything reachable is closed/validated: a closed [`Agg`],
    /// a [`RelArg`] column-or-`*`, an identifier-checked `table`, and a bound-parameter
    /// predicate — no arbitrary nested `FROM`, which keeps it mechanically scopable. This is
    /// the *only* subquery form; general scalar subqueries are deliberately not supported.
    RelatedAggregate {
        agg: Agg,
        arg: RelArg,
        table: String,
        filter: Box<Predicate>,
    },
    /// A `CASE WHEN <pred> THEN <expr> … [ELSE <expr>] END` (parenthesized). Each branch's
    /// condition reuses the predicate compiler (bound params). A boolean/comparison `ORDER BY`
    /// term is expressed portably as `ORDER BY CASE WHEN <cond> THEN 0 ELSE 1 END`.
    Case {
        branches: Vec<(Predicate, Self)>,
        otherwise: Option<Box<Self>>,
    },
    /// Extract a JSON value by a **dynamic/bound key**: `(base ->> key)` (key is an expression,
    /// e.g. a bound param — `labels ->> ?`). Postgres + SQLite; MySQL fails closed (its `->>`
    /// needs a `$.path`). Distinct from [`Expr::JsonExtract`], which takes a static key path.
    JsonExtractDyn(Box<Self>, Box<Self>),
    /// jsonb concat/merge `(left || right)` — **Postgres-only** (elsewhere `||` is string concat,
    /// so it fails closed). Used for `col = col || ?::jsonb` merge updates.
    JsonConcat(Box<Self>, Box<Self>),
    /// A **scalar subquery over a named table**: `(SELECT <column> FROM <table> WHERE <filter>)`.
    /// The narrow non-aggregate sibling of [`Expr::RelatedAggregate`] (single named table + a
    /// bound-parameter predicate — mechanically scopable, no arbitrary nested FROM). Used as the
    /// RHS of a comparison, e.g. `id = (SELECT head_version FROM pack WHERE …)`.
    RelatedScalar {
        column: String,
        table: String,
        filter: Box<Predicate>,
    },
    /// A host-resolved **"is this row the caller's own tenant?"** marker — a `0`/`1`-valued
    /// expression the guest builds *without naming the tenant column* (which is host-injected and
    /// hidden). During [`Select::force_scope`] it is lowered, using the same resolved scope the
    /// tenant predicate uses, to `CASE WHEN (<col> IS NOT NULL AND <col> = <own>) THEN 1 ELSE 0 END`
    /// — `1` for the tenant's own rows, `0` for the shared (`NULL`) baseline (or another tenant
    /// under a cross-tenant `all` read). Its purpose is the base-vs-override read: sort the tenant's
    /// override ahead of the shared base (`ORDER BY is_own DESC`) or select/filter on own-ness,
    /// without a raw `ORDER BY (tenant_id IS NOT NULL)`. **Fails closed:** if no own-tenant scope is
    /// applied (an unscoped/`disabled` function), it is never lowered and rendering it is an error.
    IsOwn,
}

impl Expr {
    /// Convenience: a column reference.
    pub fn col(name: impl Into<String>) -> Self {
        Self::Column(name.into())
    }
    /// Convenience: a bound literal.
    pub fn val(v: impl Into<SqlValue>) -> Self {
        Self::Value(v.into())
    }
}

// ---- predicates ------------------------------------------------------------

/// A comparison operator.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CmpOp {
    Eq,
    Ne,
    Lt,
    Le,
    Gt,
    Ge,
}

impl CmpOp {
    /// The SQL operator symbol (used by the compiler).
    fn symbol(self) -> &'static str {
        match self {
            Self::Eq => "=",
            Self::Ne => "<>",
            Self::Lt => "<",
            Self::Le => "<=",
            Self::Gt => ">",
            Self::Ge => ">=",
        }
    }
}

/// A recursive boolean predicate tree.
#[derive(Debug, Clone, PartialEq)]
pub enum Predicate {
    /// `AND` of all children (an empty list is the always-true identity `1 = 1`).
    And(Vec<Self>),
    /// `OR` of all children (an empty list is the always-false identity `1 = 0`).
    Or(Vec<Self>),
    /// Negation.
    Not(Box<Self>),
    /// `<left> <op> <right>`.
    Cmp { left: Expr, op: CmpOp, right: Expr },
    /// `<expr> [NOT] BETWEEN <low> AND <high>`.
    Between {
        expr: Expr,
        low: Expr,
        high: Expr,
        negated: bool,
    },
    /// `<expr> [NOT] IN (<values>)`. Empty `values` is the corresponding identity
    /// (`1 = 0` for `IN ()`, `1 = 1` for `NOT IN ()`).
    In {
        expr: Expr,
        values: Vec<Expr>,
        negated: bool,
    },
    /// `<expr> [NOT] LIKE <pattern>`; `insensitive` renders the portable
    /// `lower(<expr>) LIKE lower(<pattern>)` (no dialect-specific `ILIKE`).
    Like {
        expr: Expr,
        pattern: String,
        insensitive: bool,
        negated: bool,
    },
    /// `<expr> IS [NOT] NULL`.
    Null { expr: Expr, negated: bool },
    /// `<expr> [NOT] IN (SELECT <column> FROM <table> WHERE <filter>)` — a narrow single-named-
    /// table IN-subquery (the sibling of [`Expr::RelatedScalar`]; same safe-by-construction shape).
    InSubquery {
        expr: Expr,
        column: String,
        table: String,
        filter: Box<Self>,
        negated: bool,
    },
}

/// Build an `AND` of the given predicates.
pub fn all(preds: impl IntoIterator<Item = Predicate>) -> Predicate {
    Predicate::And(preds.into_iter().collect())
}
/// Build an `OR` of the given predicates.
pub fn any(preds: impl IntoIterator<Item = Predicate>) -> Predicate {
    Predicate::Or(preds.into_iter().collect())
}

// ---- select / insert / update ---------------------------------------------

/// The kind of join.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum JoinKind {
    Inner,
    Left,
}

/// A join: `<kind> JOIN <table>[ AS <alias>] ON <on>`.
#[derive(Debug, Clone, PartialEq)]
pub struct Join {
    pub kind: JoinKind,
    pub table: String,
    pub alias: Option<String>,
    pub on: Predicate,
}

/// A sort direction.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Direction {
    Asc,
    Desc,
}

/// An `ORDER BY` term over an expression.
#[derive(Debug, Clone, PartialEq)]
pub struct OrderBy {
    pub expr: Expr,
    pub dir: Direction,
}

/// A `SELECT`-list entry: an expression with an optional `AS <alias>`.
#[derive(Debug, Clone, PartialEq)]
pub struct SelectItem {
    pub expr: Expr,
    pub alias: Option<String>,
}

/// How a tenant [`Scope`] restricts rows for one operation. The host resolves this from the
/// per-function/site `db.read`/`db.write` grant (read modes on `SELECT`, write modes on
/// `INSERT`/`UPDATE`/`DELETE`); a guest never chooses it. `None`-grant (deny) is handled above
/// the compiler — a compiled query always carries a concrete mode.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum ScopeMode {
    /// `column = value` — the resolved tenant only.
    #[default]
    Own,
    /// `(column = value OR column IS NULL)` — the resolved tenant plus the shared/`NULL` baseline.
    OwnOrNull,
    /// `column IS NULL` — the shared/`NULL` baseline only (no tenant rows).
    NullOnly,
    /// No tenant predicate — cross-tenant. Only reachable with an explicit `all` grant under the
    /// operator posture ceiling (both enforced host-side, above this compiler).
    All,
}

/// Per-table tenant-key resolution for a [`Scope`] (PLAN-tenancy-principal D2/D3). Legacy / no
/// project schema ⇒ [`Uniform`](TableKeys::Uniform): every table scopes on [`Scope::column`].
/// A present project schema ⇒ [`PerTable`](TableKeys::PerTable): the authoritative `table →
/// `[`ResolvedScope`] map — `Column(col)` scopes that table on `col` (`TenantKeyed` identity tables
/// on their own PK), `Unscoped` a global table (no predicate), `TenantOrSession { tenant, session }`
/// the R3 anonymous-first disjunct on two disjoint columns; a table **absent** from the map is
/// refused ([`OrmError::TenancyUndeclared`], deny-by-default).
// Not `Eq`: `PerTableTarget` carries bound `SqlValue` literals (public-subset terms), and
// `SqlValue` is only `PartialEq` (a float variant) — same as `Scope`, which holds `TableKeys`.
#[derive(Debug, Clone, PartialEq, Default)]
pub enum TableKeys {
    #[default]
    Uniform,
    PerTable(std::collections::BTreeMap<String, ResolvedScope>),
    /// A **target read/write** (R4/D8): the same per-table tenant keys as `PerTable`, PLUS (when
    /// [`require_public`](TableKeys::PerTableTarget::require_public)) a per-table PUBLIC-subset
    /// confinement conjoined onto every accessed table. Built only by the host for a
    /// `TenancyClass::Target` fetch; never by a guest.
    PerTableTarget {
        keys: std::collections::BTreeMap<String, ResolvedScope>,
        public: std::collections::BTreeMap<String, Vec<PublicTermSql>>,
        /// **Target WRITE SET-allowlist (5b), deny-by-default.** The columns a target INSERT/UPDATE
        /// may set. **Empty ⇒ read-only** — any write force-scoped under this variant is refused
        /// ([`OrmError::TargetWriteNotGranted`]). Non-empty ⇒ an INSERT force-stamps `tenant = B` and
        /// (when `require_public`) the public-visibility columns and accepts ONLY these columns from
        /// the guest; an UPDATE confines its `WHERE` to `tenant = B AND <public>` and may set ONLY
        /// these columns; a DELETE is always refused. The tenant/visibility columns are never in this
        /// set, so a target write can neither change ownership nor flip a row's visibility.
        write: std::collections::BTreeSet<String>,
        /// Whether a per-table PUBLIC subset is **mandatory** (R4/D8 5c ruling A). `true` for the
        /// **anonymous** target sources (`domain`/`handle`): a table accessed with **no** declared
        /// public subset is refused ([`OrmError::PublicSubsetUndeclared`]) — for an unauthenticated
        /// actor the visibility predicate is the ONLY guard against reaching `B`'s private rows.
        /// `false` for a **`capability`-only** field: the host-verified, audience-bound capability
        /// (naming `tid = B` + the granted scope) IS the authorization, so a table with no declared
        /// subset confines to `tenant = B` alone (no visibility conjunct, no refusal) and the app's
        /// within-tenant per-client filter stays in-guest. A table that DOES declare a subset is still
        /// confined by it either way. Never `all`; still exactly one tenant `B`.
        require_public: bool,
    },
}

/// A lowered public-subset visibility term: a [`crate::tenancy::PublicTerm`] whose literal is
/// already a bound [`SqlValue`] (so it is always a parameter, never interpolated text). Conjoined
/// onto a target read to confine it to a table's public rows.
#[derive(Debug, Clone, PartialEq)]
pub enum PublicTermSql {
    /// `<column> <op> <bound value>`.
    Cmp {
        column: String,
        op: CmpOp,
        value: SqlValue,
    },
    /// `<column> IS [NOT] NULL`.
    Null { column: String, negated: bool },
}

/// Lower a host-held [`crate::tenancy::PublicPredicate`] (types-local literals) into the ORM's
/// bound-value [`PublicTermSql`] terms. Called by the host when building a target scope; a
/// `PublicLiteral` becomes a bound `SqlValue` (never interpolated).
pub fn lower_public_terms(pred: &crate::tenancy::PublicPredicate) -> Vec<PublicTermSql> {
    use crate::tenancy::{PublicCmp, PublicLiteral, PublicTerm};
    pred.terms
        .iter()
        .map(|t| match t {
            PublicTerm::Cmp { column, op, value } => {
                let op = match op {
                    PublicCmp::Eq => CmpOp::Eq,
                    PublicCmp::Ne => CmpOp::Ne,
                    PublicCmp::Lt => CmpOp::Lt,
                    PublicCmp::Le => CmpOp::Le,
                    PublicCmp::Gt => CmpOp::Gt,
                    PublicCmp::Ge => CmpOp::Ge,
                };
                let value = match value {
                    PublicLiteral::Bool(b) => SqlValue::Boolean(*b),
                    PublicLiteral::Int(n) => SqlValue::Integer(*n),
                    PublicLiteral::Text(s) => SqlValue::Text(s.clone()),
                };
                PublicTermSql::Cmp {
                    column: column.clone(),
                    op,
                    value,
                }
            }
            PublicTerm::Null { column, negated } => PublicTermSql::Null {
                column: column.clone(),
                negated: *negated,
            },
        })
        .collect()
}

/// A host-resolved in-site row-tenancy scope — the applied side of the resolved principal. `value`
/// is the resolved **own-tenant** fact (`None` ⇒ the actor has no tenant, e.g. a purely anonymous
/// `Session`-only request); `session` is the resolved anonymous-**session** fact (R3); `mode`
/// decides how the tenant axis restricts the operation; `keys` resolves the tenant **column(s) per
/// table** (the project schema — R2/R3/D2). Injected by the host on **every** query node (top-level,
/// `UNION` branch, `INSERT … SELECT` source), never guest-set. The fail-closed "no fact for a scope
/// that needs one" decision is made **per table** in the injector (a `Column` table with no tenant
/// value denies; a `TenantOrSession` table falls back to whichever axis fact is present).
#[derive(Debug, Clone, PartialEq)]
pub struct Scope {
    pub column: String,
    /// The resolved own-tenant value, or `None` for an anonymous (`Session`-only) actor.
    pub value: Option<SqlValue>,
    /// The resolved anonymous-session value (R3), or `None` when the request carries no session
    /// fact. Only consulted for a [`TableScope::TenantOrSession`](crate::tenancy::TableScope) table.
    pub session: Option<SqlValue>,
    pub mode: ScopeMode,
    /// Per-table key resolution; [`TableKeys::Uniform`] (the default) preserves the pre-schema
    /// single-column behavior (every table scopes on `column`).
    pub keys: TableKeys,
}

impl Scope {
    /// Resolve how `table` is scoped under the project schema (R2/R3/D2/D3): `Column(col)` ⇒ scope
    /// on `col`; `Unscoped` ⇒ no predicate; `TenantOrSession{tenant,session}` ⇒ the R3 disjunct;
    /// `Err(TenancyUndeclared)` ⇒ undeclared (deny-by-default). Legacy `Uniform` keys resolve every
    /// table to `Column(self.column)`, byte-identical to the pre-schema single-column behavior.
    fn resolve_table(&self, table: &str) -> Result<ResolvedScope, OrmError> {
        match &self.keys {
            TableKeys::Uniform => Ok(ResolvedScope::Column(self.column.clone())),
            TableKeys::PerTable(m) | TableKeys::PerTableTarget { keys: m, .. } => m
                .get(table)
                .cloned()
                .ok_or_else(|| OrmError::TenancyUndeclared(table.to_string())),
        }
    }

    /// The PUBLIC-subset confinement to conjoin for `table` under a **target read** (R4/D8):
    /// `Ok(None)` when the scope is not a target read (own/session — no public confinement, today's
    /// behavior). Under a target read where `require_public` (domain/handle, or any `target_or_null`),
    /// a table with **no** declared public subset is refused ([`OrmError::PublicSubsetUndeclared`],
    /// deny-by-default) and a declared subset is built as a qualified `AND` (each column qualified by
    /// `qualifier` for a join/subquery ref, so the confinement composes across every reachable table).
    /// Under a `capability`-only target (`!require_public`, ruling A) the subset is INERT ⇒ `Ok(None)`:
    /// confine to `tenant = B` alone (added separately by [`read_pred`](Self::read_pred)) — the
    /// host-verified, project-audience-bound, label-scoped capability plus the resolver's own in-guest
    /// per-`sub` filter IS the authorization. This holds whether or not the table DECLARES a subset,
    /// because a subset authored for the anonymous domain/handle funnel (e.g. `client_id IS NULL`)
    /// must not narrow a capability read — it would collide with the resolver's own filter and empty
    /// the result. An empty declared term list ⇒ no predicate (the schema loader rejects an empty
    /// declared subset). NB: `target_or_null` forces `require_public` (v0.4.8), so its shared
    /// NULL-base arm still visibility-gates even under a capability; the exemption is plain `target`.
    fn public_pred(
        &self,
        table: &str,
        qualifier: Option<&str>,
    ) -> Result<Option<Predicate>, OrmError> {
        let TableKeys::PerTableTarget {
            public,
            require_public,
            ..
        } = &self.keys
        else {
            return Ok(None);
        };
        // Ruling A (v0.4.4), made COMPLETE: on the capability axis the visibility subset is inert —
        // NO per-table subset is applied, not even a DECLARED one. `tenant = B` (applied separately)
        // + the capability + the resolver's own in-guest filter is the authorization.
        if !require_public {
            return Ok(None);
        }
        let terms = match public.get(table) {
            Some(t) => t,
            // domain/handle (or target_or_null): the visibility predicate is the ONLY guard for an
            // anonymous/base-arm actor, so a declared subset is mandatory (deny-by-default).
            None => return Err(OrmError::PublicSubsetUndeclared(table.to_string())),
        };
        let mut preds = Vec::with_capacity(terms.len());
        for term in terms {
            match term {
                PublicTermSql::Cmp { column, op, value } => {
                    ident(column)?;
                    preds.push(Predicate::Cmp {
                        left: Self::col_expr(column, qualifier),
                        op: *op,
                        right: Expr::Value(value.clone()),
                    });
                }
                PublicTermSql::Null { column, negated } => {
                    ident(column)?;
                    preds.push(Predicate::Null {
                        expr: Self::col_expr(column, qualifier),
                        negated: *negated,
                    });
                }
            }
        }
        Ok(match preds.len() {
            0 => None,
            1 => Some(preds.pop().unwrap()),
            _ => Some(Predicate::And(preds)),
        })
    }

    /// A (possibly-qualified) column expression `<qualifier>.column`.
    fn col_expr(column: &str, qualifier: Option<&str>) -> Expr {
        Expr::Column(match qualifier {
            Some(q) => format!("{q}.{column}"),
            None => column.to_string(),
        })
    }

    /// The **tenant-axis** predicate on `column` (optionally `<qual>.column`) for the resolved mode:
    /// `Ok(None)` for `All` (no predicate — cross-tenant); `NullOnly` needs no value; `Own`/`OwnOrNull`
    /// require a resolved own-tenant value and **fail closed** ([`OrmError::TenancyNoPrincipal`]) when
    /// there is none (a purely anonymous actor reading a plain tenant table). Never binds to another
    /// table's same-named column — it is qualified by `qual`.
    ///
    /// `base_inclusive` folds the shared `column IS NULL` base into the `Own` arm (`Or([col = v, col
    /// IS NULL])`) — the per-table [`ResolvedScope::TenantOrBase`] read, which reads a table's
    /// tenant-`NULL` base rows on *any* own/target read regardless of the field mode. It is a no-op for
    /// `OwnOrNull` (already OR-null), `NullOnly` (base only), and `All` (cross-tenant). It affects only
    /// the READ predicate; the write stamp is unchanged (a base-inclusive table stamps `tenant = v`,
    /// never `NULL`).
    fn tenant_pred(
        &self,
        column: &str,
        qualifier: Option<&str>,
        base_inclusive: bool,
    ) -> Result<Option<Predicate>, OrmError> {
        let is_null = Predicate::Null {
            expr: Self::col_expr(column, qualifier),
            negated: false,
        };
        let eq = |v: SqlValue| Predicate::Cmp {
            left: Self::col_expr(column, qualifier),
            op: CmpOp::Eq,
            right: Expr::Value(v),
        };
        Ok(match self.mode {
            ScopeMode::All => None,
            ScopeMode::NullOnly => Some(is_null),
            ScopeMode::Own => {
                let v = self.value.clone().ok_or(OrmError::TenancyNoPrincipal)?;
                Some(if base_inclusive {
                    Predicate::Or(vec![eq(v), is_null])
                } else {
                    eq(v)
                })
            }
            ScopeMode::OwnOrNull => {
                let v = self.value.clone().ok_or(OrmError::TenancyNoPrincipal)?;
                Some(Predicate::Or(vec![eq(v), is_null]))
            }
        })
    }

    /// The R3 **disjunct** read predicate for a `TenantOrSession` table: `Or` of the arms for
    /// whichever axis facts the request carries — `tenant = <own>` (if a tenant fact is present) and
    /// `session = <sid>` (if a session fact is present) — over the two **disjoint** columns. `All`
    /// mode ⇒ no predicate (cross-tenant). No fact at all ⇒ **deny** ([`TenancyNoPrincipal`]): a
    /// `TenantOrSession` read with neither an own nor a session identity fails closed rather than
    /// running unscoped. Each arm is a plain `col = value` (the session partition IS the
    /// tenant-`NULL` rows, so no extra NULL arm is added).
    fn disjunct_pred(
        &self,
        tenant_col: &str,
        session_col: &str,
        qualifier: Option<&str>,
    ) -> Result<Option<Predicate>, OrmError> {
        if matches!(self.mode, ScopeMode::All) {
            return Ok(None);
        }
        let eq = |column: &str, v: SqlValue| Predicate::Cmp {
            left: Self::col_expr(column, qualifier),
            op: CmpOp::Eq,
            right: Expr::Value(v),
        };
        let mut arms = Vec::new();
        if let Some(v) = self.value.clone() {
            arms.push(eq(tenant_col, v));
        }
        if let Some(s) = self.session.clone() {
            arms.push(eq(session_col, s));
        }
        match arms.len() {
            0 => Err(OrmError::TenancyNoPrincipal),
            1 => Ok(arms.pop()),
            _ => Ok(Some(Predicate::Or(arms))),
        }
    }

    /// The READ predicate to conjoin for `table` (qualified by `qualifier` in a join/subquery):
    /// dispatches on the per-table [`ResolvedScope`] — a plain tenant column, an `Unscoped` global
    /// (no predicate), or the R3 `TenantOrSession` disjunct. `Ok(None)` ⇒ no predicate (the table is
    /// global, or the mode is cross-tenant `All`). Undeclared / no-principal ⇒ fail closed.
    fn read_pred(
        &self,
        table: &str,
        qualifier: Option<&str>,
    ) -> Result<Option<Predicate>, OrmError> {
        let tenant = match self.resolve_table(table)? {
            ResolvedScope::Column(col) => {
                ident(&col)?;
                self.tenant_pred(&col, qualifier, false)?
            }
            ResolvedScope::Unscoped => None,
            ResolvedScope::TenantOrSession { tenant, session } => {
                ident(&tenant)?;
                ident(&session)?;
                self.disjunct_pred(&tenant, &session, qualifier)?
            }
            // Base-inclusive: fold the shared `tenant IS NULL` base into the read on any own/target
            // read (base_inclusive = true forces the OR-null form even under a plain `Own` mode).
            ResolvedScope::TenantOrBase { tenant } => {
                ident(&tenant)?;
                self.tenant_pred(&tenant, qualifier, true)?
            }
        };
        // R4/D8: under a TARGET read, additionally confine to the table's host-held PUBLIC subset
        // (deny-by-default if the table declares none). No-op under an own/session read. So a target
        // read of table `t` becomes `t.tenant = B AND <t's public predicate>`, composed per ref.
        let public = self.public_pred(table, qualifier)?;
        let mut out = Predicate::And(Vec::new());
        conjoin_front(&mut out, tenant);
        conjoin_front(&mut out, public);
        Ok(match out {
            Predicate::And(v) if v.is_empty() => None,
            p => Some(p),
        })
    }

    /// The `(column, value)` a scoped **WRITE** stamps/bounds for `table` — the actor's OWN axis:
    /// a plain tenant table (or `Uniform`) stamps `default_tenant_key = <own tenant>`; a
    /// `TenantOrSession` table stamps whichever single axis the actor holds (tenant if authenticated,
    /// else session) so an anonymous write lands in the session partition and an authenticated write
    /// in the tenant partition — never both, never cross. `Ok(None)` ⇒ `All` mode (no stamp — a
    /// posture-vetted cross-tenant write). Fail closed: an `Unscoped` (global) target
    /// ([`UnscopedWrite`]), an undeclared target ([`TenancyUndeclared`]), or a scoped write with no
    /// principal ([`TenancyNoPrincipal`]) are refused before any SQL.
    fn write_target(&self, table: &str) -> Result<Option<(String, SqlValue)>, OrmError> {
        // The tenant-axis stamp value for this mode (own/own+null → the resolved tenant; null → the
        // shared baseline; all → no stamp).
        let tenant_stamp = || -> Result<Option<SqlValue>, OrmError> {
            Ok(match self.mode {
                ScopeMode::All => None,
                ScopeMode::NullOnly => Some(SqlValue::Null),
                ScopeMode::Own | ScopeMode::OwnOrNull => {
                    Some(self.value.clone().ok_or(OrmError::TenancyNoPrincipal)?)
                }
            })
        };
        match self.resolve_table(table)? {
            ResolvedScope::Column(col) => Ok(tenant_stamp()?.map(|v| (col, v))),
            // A base-inclusive table WRITES exactly like a plain tenant `Column` table: stamp the
            // resolved tenant (never `NULL`). The base⊕own fold is read-only — a guest write can
            // neither create nor update a `NULL`-base row (base rows are operator-seeded via a
            // privileged `NullOnly`/`All` path, unreachable from a guest's `Own`-mode write).
            ResolvedScope::TenantOrBase { tenant } => Ok(tenant_stamp()?.map(|v| (tenant, v))),
            ResolvedScope::Unscoped => Err(OrmError::UnscopedWrite(table.to_string())),
            ResolvedScope::TenantOrSession { tenant, session } => {
                if matches!(self.mode, ScopeMode::All) {
                    return Ok(None);
                }
                // A TARGET write carries only the target tenant `B` (no session fact). Stamping
                // `tenant = B` onto a session-keyed row would silently claim an anon/session-owned row
                // for `B` and break the anon→promotion model, so refuse deny-by-default
                // (PLAN-delegable-capabilities, Stage A): a `TenantOrSession` table is written on the
                // caller's own/session-scoped path, never under a target scope.
                if self.is_target() {
                    return Err(OrmError::TargetWriteToSessionTable(table.to_string()));
                }
                // Prefer the tenant axis when authenticated; else the session axis for an anon write.
                if let Some(v) = self.value.clone() {
                    Ok(Some((tenant, v)))
                } else if let Some(s) = self.session.clone() {
                    Ok(Some((session, s)))
                } else {
                    Err(OrmError::TenancyNoPrincipal)
                }
            }
        }
    }

    /// Whether this scope is a **target** scope (`PerTableTarget` — reading/writing another tenant
    /// `B`'s public subset, R4/D8), vs. the caller's own.
    pub fn is_target(&self) -> bool {
        matches!(self.keys, TableKeys::PerTableTarget { .. })
    }

    /// The target-write SET-allowlist (5b), or `None` when this is not a target scope. An **empty**
    /// set means the target route is read-only (no `write` grant) — a write force-scoped under it is
    /// refused. Returned as `Some(&set)` for a target scope so a write path can tell "not a target"
    /// (own path) from "target, read-only" (refuse) from "target, may set these columns".
    fn target_write_allowlist(&self) -> Option<&std::collections::BTreeSet<String>> {
        match &self.keys {
            TableKeys::PerTableTarget { write, .. } => Some(write),
            _ => None,
        }
    }

    /// Whether a target read/write must apply the per-table PUBLIC visibility subset: `true` for an
    /// anonymous `domain`/`handle` source or any `target_or_null` (the shared `NULL`-base arm), where
    /// the subset is the only guard; `false` for a plain `capability`-only target (ruling A — the
    /// capability is the authorization, so the confinement is `tenant = B` alone). `false` when this
    /// is not a target scope. Read by the upsert confinement to admit an `ON CONFLICT DO UPDATE` only
    /// where the DO-UPDATE's `tenant = B` guard matches the (subset-less) plain-UPDATE confinement.
    fn target_require_public(&self) -> bool {
        matches!(
            &self.keys,
            TableKeys::PerTableTarget {
                require_public: true,
                ..
            }
        )
    }

    /// The resolved tenant column for `table` under a target scope (the per-table key, or `None` when
    /// this is not a target scope or the table has no plain-`Column` tenant key). Used to require that
    /// a target upsert's conflict target includes the tenant column.
    fn target_tenant_column(&self, table: &str) -> Option<String> {
        match &self.keys {
            TableKeys::PerTableTarget { keys, .. } => match keys.get(table) {
                Some(ResolvedScope::Column(c)) => Some(c.clone()),
                // A base-inclusive table has a plain tenant column too (its NULL rows are shared
                // base); a target upsert on it keys+guards on that column exactly like a Column table.
                Some(ResolvedScope::TenantOrBase { tenant }) => Some(tenant.clone()),
                _ => None,
            },
            _ => None,
        }
    }

    /// Refuse a **target** write to a `TenantOrSession` (anonymous-session-keyed) table. A target
    /// principal carries only the target tenant `B` (no session fact), so such a row could only be
    /// stamped `tenant = B` — silently claiming an anon/session-owned row for `B` and breaking the
    /// anon→promotion model. Called at the top of every target-write path (INSERT/UPDATE) so the
    /// refusal is **early and self-describing** rather than surfacing later as a public-subset error;
    /// a no-op for an own scope (a `TenantOrSession` write on the own/session path is legitimate).
    /// `write_target` keeps the equivalent guard as a fail-closed backstop.
    /// (PLAN-delegable-capabilities, Stage A.)
    fn assert_target_table_writable(&self, table: &str) -> Result<(), OrmError> {
        if !self.is_target() {
            return Ok(());
        }
        if let ResolvedScope::TenantOrSession { .. } = self.resolve_table(table)? {
            return Err(OrmError::TargetWriteToSessionTable(table.to_string()));
        }
        Ok(())
    }

    /// The `(column, value)` pairs a target **INSERT** must force so the inserted row lands in
    /// `table`'s PUBLIC subset (5b): each `column = <literal>` public term contributes `(column,
    /// literal)`, each `column IS NULL` term contributes `(column, NULL)`. A public term the host
    /// cannot pin to a single value (a range comparison, or `IS NOT NULL`) is not forceable — the
    /// host cannot guarantee publicness — so the INSERT is refused ([`PublicSubsetNotForceable`]).
    /// When `require_public` (domain/handle) a table with no declared subset is refused
    /// ([`PublicSubsetUndeclared`]); under a `capability`-only target (`!require_public`) an undeclared
    /// subset forces no visibility columns (the row is `tenant = B` + the guest's allowlisted columns —
    /// the capability is the authorization).
    fn public_force_cells(&self, table: &str) -> Result<Vec<(String, SqlValue)>, OrmError> {
        let TableKeys::PerTableTarget {
            public,
            require_public,
            ..
        } = &self.keys
        else {
            return Ok(Vec::new());
        };
        let terms = match public.get(table) {
            Some(t) => t,
            None if !require_public => return Ok(Vec::new()),
            None => return Err(OrmError::PublicSubsetUndeclared(table.to_string())),
        };
        let mut out = Vec::with_capacity(terms.len());
        for term in terms {
            match term {
                PublicTermSql::Cmp {
                    column,
                    op: CmpOp::Eq,
                    value,
                } => {
                    ident(column)?;
                    out.push((column.clone(), value.clone()));
                }
                PublicTermSql::Null {
                    column,
                    negated: false,
                } => {
                    ident(column)?;
                    out.push((column.clone(), SqlValue::Null));
                }
                // A range comparison or `IS NOT NULL` has no single value to stamp.
                PublicTermSql::Cmp { .. } | PublicTermSql::Null { .. } => {
                    return Err(OrmError::PublicSubsetNotForceable(table.to_string()))
                }
            }
        }
        Ok(out)
    }

    /// Assert a guest-supplied `column` is settable by a target write on `table` (5b). Two gates,
    /// both must pass: it is in the route's SET-allowlist, AND it is neither the tenant column nor a
    /// public-visibility column (the latter a defense-in-depth check so even an operator who wrongly
    /// listed the tenant/visibility column can't let a target write change ownership or flip
    /// visibility). A no-op (`Ok`) when this is not a target scope. Fail-closed
    /// ([`TargetWriteColumnDenied`]).
    fn assert_target_settable(&self, table: &str, column: &str) -> Result<(), OrmError> {
        let TableKeys::PerTableTarget {
            keys,
            public,
            write,
            ..
        } = &self.keys
        else {
            return Ok(());
        };
        let denied = || OrmError::TargetWriteColumnDenied(column.to_string());
        // Gate 0: a write-target column (an INSERT column / an UPDATE SET LHS) must be a BARE column
        // name — never `table.col`. A qualified name would (a) let `same_col` compare only the last
        // segment, so `published.x` could slip past the tenant/visibility check on base `x`, and (b)
        // render invalid SQL. Refuse it fail-closed at compile rather than emit a statement the DB
        // would reject.
        if column.contains('.') {
            return Err(denied());
        }
        // Gate 1: must be granted in the SET-allowlist.
        if !write.iter().any(|c| same_col(c, column)) {
            return Err(denied());
        }
        // Gate 2: never the tenant column (would change ownership).
        if let Some(rs) = keys.get(table) {
            let tenant_cols: &[&str] = match rs {
                ResolvedScope::Column(c) => &[c],
                ResolvedScope::TenantOrSession { tenant, session } => &[tenant, session],
                // The tenant column is off-limits to a guest SET; its NULL base rows are never
                // guest-writable, so a base-inclusive table protects the same one column.
                ResolvedScope::TenantOrBase { tenant } => &[tenant],
                ResolvedScope::Unscoped => &[],
            };
            if tenant_cols.iter().any(|t| same_col(t, column)) {
                return Err(denied());
            }
        }
        // Gate 2 (cont.): never a public-visibility column (would flip the row in/out of the subset).
        if let Some(terms) = public.get(table) {
            let is_public_col = terms.iter().any(|t| match t {
                PublicTermSql::Cmp { column: c, .. } | PublicTermSql::Null { column: c, .. } => {
                    same_col(c, column)
                }
            });
            if is_public_col {
                return Err(denied());
            }
        }
        Ok(())
    }
}

/// A `SELECT`.
#[derive(Debug, Clone, PartialEq)]
pub struct Select {
    pub table: String,
    pub table_alias: Option<String>,
    /// Empty ⇒ `SELECT *`.
    pub columns: Vec<SelectItem>,
    pub joins: Vec<Join>,
    pub filter: Option<Predicate>,
    pub scope: Option<Scope>,
    pub group_by: Vec<Expr>,
    pub having: Option<Predicate>,
    pub distinct: bool,
    /// `DISTINCT ON (<exprs>)` — **Postgres-only** (fails closed elsewhere). Non-empty takes
    /// precedence over `distinct`; empty ⇒ inactive.
    pub distinct_on: Vec<Expr>,
    pub order: Vec<OrderBy>,
    pub limit: Option<u32>,
    pub offset: Option<u32>,
    /// `UNION [ALL] <query>` — one level (the branch's own `union` is not rendered). Each side
    /// carries its own scope/filter, so both stay tenant-isolated.
    pub union: Option<Box<Union>>,
}

/// A `UNION [ALL]` branch of a [`Select`].
#[derive(Debug, Clone, PartialEq)]
pub struct Union {
    pub all: bool,
    pub query: Select,
}

/// A `column = <expr>` assignment (an INSERT cell or an UPDATE SET).
#[derive(Debug, Clone, PartialEq)]
pub struct Assignment {
    pub column: String,
    pub value: Expr,
}

/// One row's cells for an INSERT.
#[derive(Debug, Clone, PartialEq)]
pub struct RowValues {
    pub cells: Vec<Assignment>,
}

/// An `ON CONFLICT (<columns>) DO UPDATE SET <update>` (empty `update` ⇒ `DO NOTHING`).
#[derive(Debug, Clone, PartialEq)]
pub struct OnConflict {
    pub conflict_columns: Vec<String>,
    pub update: Vec<Assignment>,
}

/// An `INSERT` (single- or multi-row), optionally an upsert, optionally `RETURNING`.
#[derive(Debug, Clone, PartialEq)]
pub struct Insert {
    pub table: String,
    pub rows: Vec<RowValues>,
    pub conflict: Option<OnConflict>,
    /// Forces `column = value` into every inserted row (adds or overrides).
    pub scope: Option<Scope>,
    /// `RETURNING <items>` (empty ⇒ none). Not supported by every engine (e.g. MySQL).
    pub returning: Vec<SelectItem>,
    /// `INSERT INTO t (<columns>) <select>` — when set, rows come from a SELECT (`rows` ignored).
    /// Under a scoped write, [`Insert::force_scope`] read-scopes the source **and** host-forces the
    /// target tenant column (dropping any guest projection of it), so the written tenant can't be
    /// forged; without a scope (or `all`) the columns/projection are taken verbatim.
    pub from_select: Option<(Vec<String>, Box<Select>)>,
}

/// An `UPDATE`; `filter` is required (an unbounded update is refused).
#[derive(Debug, Clone, PartialEq)]
pub struct Update {
    pub table: String,
    pub set: Vec<Assignment>,
    pub filter: Predicate,
    pub scope: Option<Scope>,
    pub returning: Vec<SelectItem>,
}

/// A `DELETE`; `filter` is required (an unbounded delete is refused, mirroring [`Update`]).
#[derive(Debug, Clone, PartialEq)]
pub struct Delete {
    pub table: String,
    pub filter: Predicate,
    pub scope: Option<Scope>,
    pub returning: Vec<SelectItem>,
}

impl Insert {
    /// For a posture-vetted cross-tenant (`all`) INSERT with no host-injected stamp: the single,
    /// uniform LITERAL value of scope column `col` across every inserted row — the tenant the row(s)
    /// declare — or `None` when it isn't one well-defined literal (an `INSERT … SELECT` source, a
    /// row missing `col` or giving it a non-literal, or rows that disagree). Used ONLY to set the
    /// RLS tenant GUC to what the write targets (an `all` guest may write any one tenant, GUC-
    /// consistent); a `None` simply doesn't re-set the GUC, leaving it at whatever the per-transaction
    /// own/session set established (or unset if none) — under `all` that only over-restricts the write
    /// (the DB's `WITH CHECK` still confines it), never widens it. The DB is the final arbiter — a
    /// wrong value is rejected there.
    pub fn uniform_scope_value(&self, col: &str) -> Option<SqlValue> {
        if self.from_select.is_some() || self.rows.is_empty() {
            return None;
        }
        let mut found: Option<SqlValue> = None;
        for row in &self.rows {
            let cell = row.cells.iter().find(|a| same_col(&a.column, col))?;
            let v = match &cell.value {
                Expr::Value(v) => v.clone(),
                _ => return None, // a non-literal (expression/column) → not a single known tenant
            };
            match &found {
                None => found = Some(v),
                Some(prev) if *prev == v => {}
                Some(_) => return None, // rows declare different tenants → not a single value
            }
        }
        found
    }
}

impl Update {
    /// For a posture-vetted cross-tenant (`all`) UPDATE: the single tenant value the WHERE pins scope
    /// column `col` to — a top-level `col = <literal>` conjunct (optionally nested in `AND`s) — or
    /// `None` when the filter doesn't pin exactly one tenant (an `OR`/`IN`/range/no-op, or conjuncts
    /// pinning different values). Used ONLY to set the RLS GUC; a `None` doesn't re-set it, leaving the
    /// prior per-transaction value (or unset), which only over-restricts — under `all` the DB's `USING`
    /// then matches no row outside that tenant, so a genuinely multi-tenant UPDATE affects nothing. It
    /// must pin a single tenant to run; it never silently touches one tenant of a spanning filter.
    pub fn pinned_scope_value(&self, col: &str) -> Option<SqlValue> {
        fn find(pred: &Predicate, col: &str) -> Option<SqlValue> {
            match pred {
                Predicate::Cmp {
                    left,
                    op: CmpOp::Eq,
                    right,
                } => match (left, right) {
                    (Expr::Column(c), Expr::Value(v)) | (Expr::Value(v), Expr::Column(c))
                        if same_col(c, col) =>
                    {
                        Some(v.clone())
                    }
                    _ => None,
                },
                Predicate::And(children) => {
                    let mut found: Option<SqlValue> = None;
                    for ch in children {
                        if let Some(v) = find(ch, col) {
                            match &found {
                                None => found = Some(v),
                                Some(prev) if *prev == v => {}
                                Some(_) => return None, // contradictory pins → not a single tenant
                            }
                        }
                    }
                    found
                }
                // OR / NOT / IN / BETWEEN / LIKE / NULL / subquery don't pin exactly one tenant.
                _ => None,
            }
        }
        find(&self.filter, col)
    }
}

impl Select {
    /// Force a host-resolved `scope` onto this `SELECT` **and every nested read node** — its
    /// `UNION` branch — so a tenant scope reaches every row source (a union branch left unscoped
    /// would leak across tenants). Overwrites any pre-existing scope. This is the host's tenant
    /// injection point for reads; the guest never sets a scope of its own.
    pub fn force_scope(&mut self, scope: &Scope) -> Result<(), OrmError> {
        self.scope = Some(scope.clone());
        self.inject_subquery_scope(scope)?;
        if let Some(u) = self.union.as_mut() {
            u.query.force_scope(scope)?;
        }
        Ok(())
    }
}

impl Insert {
    /// Force the host-resolved tenant scope. `write` stamps the tenant column on a
    /// `VALUES`-based insert (per [`ScopeMode`]); for an `INSERT … SELECT`, the `read` scope is
    /// forced onto the source query (and its nested unions) so the selected rows stay
    /// tenant-isolated, **and** the target tenant column is host-forced too — any guest-supplied
    /// tenant column + its projection is dropped and re-appended bound to the resolved value, so a
    /// guest can't project another tenant's id into the write (a cross-tenant write forgery).
    /// `None` for an axis (cross-tenant `all`) clears that scope — the operation runs unscoped on
    /// that axis, by design (an `all` write's `stamp_value()` is `None`, so nothing is forced).
    pub fn force_scope(
        &mut self,
        write: Option<&Scope>,
        read: Option<&Scope>,
    ) -> Result<(), OrmError> {
        // R4/D8 target write (5b): confine BEFORE the own-write logic. A target INSERT accepts only
        // the SET-allowlisted columns from the guest and force-stamps the public-visibility columns,
        // so the inserted row lands in `B`'s public subset (the `tenant = B` stamp is applied by the
        // shared own-write path below, since `write_target` yields `B` for a target scope).
        if let Some(w) = write {
            if let Some(allow) = w.target_write_allowlist() {
                self.confine_target_insert(w, allow.is_empty())?;
            }
        }
        self.scope = write.cloned();
        // The write target's per-table stamp `(column, value)` — the actor's OWN axis (Stage 1/R3),
        // resolved once for the INSERT…SELECT tenant-projection re-append below. Resolving it enforces
        // deny-by-default at bind time (an undeclared target, an `Unscoped` target, or a scoped write
        // with no principal are refused). `None` ⇒ `all` mode (no stamp).
        let target: Option<(String, SqlValue)> = match write {
            Some(w) => w.write_target(&self.table)?,
            None => None,
        };
        // A subquery embedded in a row cell, an upsert `SET` expr, or a `RETURNING` item is a READ
        // of another table — scope it to that table so it can't read cross-tenant.
        if let Some(r) = read {
            for row in &mut self.rows {
                for cell in &mut row.cells {
                    inject_scope_expr(r, &mut cell.value)?;
                }
            }
            if let Some(c) = self.conflict.as_mut() {
                for a in &mut c.update {
                    inject_scope_expr(r, &mut a.value)?;
                }
            }
            for it in &mut self.returning {
                inject_scope_expr(r, &mut it.expr)?;
            }
        }
        if let Some((cols, src)) = self.from_select.as_mut() {
            match read {
                Some(r) => src.force_scope(r)?,
                None => src.scope = None,
            }
            // A scoped write owns the axis column written — never trust the guest's target
            // projection. Drop any guest-supplied owning-axis column (+ its aligned projection, in
            // the source and every union branch) and re-append it bound to the host value. The
            // column is the actor's per-table axis key (Stage 1/R3); `all` mode ⇒ `target` is `None`
            // ⇒ nothing is forced (a posture-vetted cross-tenant write).
            if let Some((column, v)) = &target {
                let column = column.clone();
                if let Some(i) = cols.iter().position(|c| same_col(c, &column)) {
                    cols.remove(i);
                    drop_projection_at(src, i);
                }
                let v = v.clone();
                cols.push(column);
                push_projection(
                    src,
                    SelectItem {
                        expr: Expr::Value(v),
                        alias: None,
                    },
                );
            }
        }
        Ok(())
    }

    /// Confine a **target INSERT / upsert** (5b): the guest may set ONLY the route's SET-allowlisted
    /// columns, and the host force-stamps the table's public-visibility columns so the inserted row
    /// lands in `B`'s public subset. A **capability-axis `ON CONFLICT … DO UPDATE` upsert** is
    /// supported (own↔target parity): the INSERT arm is confined exactly as a plain target INSERT,
    /// and the DO UPDATE arm is confined to the write allowlist here + guarded to `tenant = B` at
    /// compile (`render_conflict`, via the write stamp), so the conflict-row update can never touch
    /// another tenant's row. Refused fail-closed on: a read-only target (`empty_allowlist`), an
    /// `INSERT … SELECT`, an upsert on an anonymous (`require_public`) target (the DO-UPDATE guard
    /// carries no visibility subset — see below), an upsert whose conflict target omits the tenant
    /// column, a guest cell outside the allowlist (or the tenant/visibility columns), or a public
    /// subset that cannot be forced to a concrete row. (`tenant = B` is stamped by the shared
    /// own-write path.)
    fn confine_target_insert(
        &mut self,
        scope: &Scope,
        empty_allowlist: bool,
    ) -> Result<(), OrmError> {
        // A target write may not touch a TenantOrSession (anon-session) table — refuse early and
        // self-describingly, before the allowlist/public-subset checks (Stage A).
        scope.assert_target_table_writable(&self.table)?;
        if empty_allowlist {
            return Err(OrmError::TargetWriteNotGranted(self.table.clone()));
        }
        if self.from_select.is_some() {
            return Err(OrmError::TargetWriteUnsupported("INSERT … SELECT"));
        }
        // A confined `ON CONFLICT … DO UPDATE` upsert IS a valid target write on the CAPABILITY axis
        // (the request's `embedSubmitSurvey` latest-wins pattern; own↔target parity). Two hard
        // requirements keep it sound; a target upsert that fails either is refused fail-closed:
        if let Some(oc) = &self.conflict {
            // (a) Capability axis only. On an anonymous `domain`/`handle` source (or any
            //     `target_or_null`) a plain UPDATE confines to `tenant = B AND <public subset>`, but
            //     the DO-UPDATE guard is `tenant = B` alone (render_conflict has no subset), so such
            //     an upsert could update `B`'s NON-public row on a conflict. Refuse it (unchanged
            //     from before — only the capability axis, whose plain UPDATE is likewise `tenant = B`
            //     alone post-ruling-A, is newly admitted).
            if scope.target_require_public() {
                return Err(OrmError::TargetWriteUnsupported(
                    "ON CONFLICT upsert on an anonymous domain/handle/target_or_null target",
                ));
            }
            // A conflict-target column must be a BARE name (never `t.col`): a qualified name is
            // invalid in an `ON CONFLICT` inference clause AND could let a `.`-spelling slip past the
            // tenant-key check below on `same_col`'s last-segment compare. Refuse it self-describingly
            // at compile (mirrors gate 0 on the SET LHS in `assert_target_settable`).
            for c in &oc.conflict_columns {
                if c.contains('.') {
                    return Err(OrmError::TargetWriteColumnDenied(c.clone()));
                }
            }
            // (b) The conflict target must include the tenant column, so a conflict is always the
            //     target tenant's OWN row: `B`'s INSERT still lands when another tenant holds the
            //     same tenant-agnostic natural key, and there is no cross-tenant no-op / existence
            //     oracle. (The `tenant = B` guard + dropping any SET on the tenant column are applied
            //     at compile by render_conflict via the write stamp — belt-and-suspenders here.)
            match scope.target_tenant_column(&self.table) {
                Some(tcol) if oc.conflict_columns.iter().any(|c| same_col(c, &tcol)) => {}
                _ => return Err(OrmError::TargetUpsertKeyMissingTenant(self.table.clone())),
            }
            // The DO UPDATE SET is confined exactly like a target UPDATE: only allowlisted, non-tenant,
            // non-visibility columns may be assigned.
            for a in &oc.update {
                scope.assert_target_settable(&self.table, &a.column)?;
            }
        }
        // Every guest-supplied cell must be a granted, non-tenant, non-visibility column.
        for row in &self.rows {
            for cell in &row.cells {
                scope.assert_target_settable(&self.table, &cell.column)?;
            }
        }
        // Force the public-visibility columns onto every row (deny-by-default / not-forceable checks
        // live in `public_force_cells`). Appended as host literals — the guest cannot have set them
        // (they're excluded by `assert_target_settable`), so there is no dup to reconcile.
        let forced = scope.public_force_cells(&self.table)?;
        for row in &mut self.rows {
            for (column, value) in &forced {
                row.cells.push(Assignment {
                    column: column.clone(),
                    value: Expr::Value(value.clone()),
                });
            }
        }
        Ok(())
    }
}

/// Remove the projection at index `i` from a `SELECT` and every one-level `UNION` branch, keeping
/// the branches' column counts aligned (used by [`Insert::force_scope`]).
fn drop_projection_at(s: &mut Select, i: usize) {
    if i < s.columns.len() {
        s.columns.remove(i);
    }
    if let Some(u) = s.union.as_mut() {
        drop_projection_at(&mut u.query, i);
    }
}

/// Append `item` to a `SELECT`'s projection and every one-level `UNION` branch (so both sides of
/// a union source stamp the same host tenant value).
fn push_projection(s: &mut Select, item: SelectItem) {
    s.columns.push(item.clone());
    if let Some(u) = s.union.as_mut() {
        push_projection(&mut u.query, item);
    }
}

/// Conjoin `add` (if any) as the FIRST conjunct of `filter` (`filter := add AND filter`). A
/// no-op empty-`AND` existing filter is replaced outright, so the scope doesn't trail a spurious
/// `AND 1 = 1`.
fn conjoin_front(filter: &mut Predicate, add: Option<Predicate>) {
    let Some(a) = add else { return };
    if matches!(filter, Predicate::And(v) if v.is_empty()) {
        *filter = a;
    } else {
        let existing = std::mem::replace(filter, Predicate::And(Vec::new()));
        *filter = Predicate::And(vec![a, existing]);
    }
}

/// Lower an [`Expr::IsOwn`] marker to a concrete `0`/`1` rank using the resolved `scope` — the
/// same host-resolved tenant `value` the scope predicate uses. `CASE WHEN (<col> IS NOT NULL AND
/// <col> = <own>) THEN 1 ELSE 0 END`: `1` for the caller's own rows, `0` for the shared `NULL`
/// baseline (and for other tenants under a cross-tenant `all` read). The `IS NOT NULL` guard keeps
/// it a proper boolean (never `NULL`) so `ORDER BY … DESC` is portable (own sorts first) across
/// every dialect. The column is unqualified — the base-vs-override read this serves is single-table.
fn own_rank_expr(scope: &Scope) -> Expr {
    // No resolved own-tenant value (a purely anonymous actor) ⇒ nothing ranks as "own" ⇒ constant 0.
    let Some(value) = scope.value.clone() else {
        return Expr::Value(SqlValue::Integer(0));
    };
    let col = || Expr::Column(scope.column.clone());
    let own = Predicate::And(vec![
        Predicate::Null {
            expr: col(),
            negated: true,
        },
        Predicate::Cmp {
            left: col(),
            op: CmpOp::Eq,
            right: Expr::Value(value),
        },
    ]);
    Expr::Case {
        branches: vec![(own, Expr::Value(SqlValue::Integer(1)))],
        otherwise: Some(Box::new(Expr::Value(SqlValue::Integer(0)))),
    }
}

/// Conjoin the tenant scope for a **subquery's inner `table`** onto its `filter`, resolving that
/// table exactly as the top-level [`Select::scope_where_pred`] does (via [`Scope::read_pred`]): the
/// declared per-table column, the R3 `TenantOrSession` disjunct, **no** predicate for an `Unscoped`
/// reference table, and **refuse** an undeclared table or a scoped ref with no principal
/// (deny-by-default). This is what makes a subquery no weaker than a top-level FROM/JOIN ref — under
/// a `PerTable` schema a subquery can neither reach an undeclared table nor be scoped on the wrong
/// column. Legacy `Uniform` keys resolve to `scope.column` for every table (pre-schema behavior).
fn conjoin_subquery_scope(
    scope: &Scope,
    table: &str,
    filter: &mut Predicate,
) -> Result<(), OrmError> {
    conjoin_front(filter, scope.read_pred(table, Some(table))?);
    Ok(())
}

/// Walk an expression and inject the tenant scope into every **narrow subquery**'s inner filter,
/// qualified to that subquery's own table (`<subtable>.col`) and keyed on that table's declared
/// per-table column, so a subquery can neither read another tenant's rows nor reach an undeclared
/// table. Recurses into a subquery's filter first (nested subqueries scope their own tables). The
/// correctness twin of [`Select::scope_where_pred`] for the subquery surface — including its
/// deny-by-default, so `Err(TenancyUndeclared)` propagates out and the query fails closed. Also
/// lowers any [`Expr::IsOwn`] marker here (where the resolved `scope` is in hand) — so an
/// unlowered `IsOwn` reaching the renderer means no scope was applied, and it fails closed.
fn inject_scope_expr(scope: &Scope, e: &mut Expr) -> Result<(), OrmError> {
    match e {
        Expr::IsOwn => *e = own_rank_expr(scope),
        Expr::RelatedAggregate { table, filter, .. }
        | Expr::RelatedScalar { table, filter, .. } => {
            inject_scope_pred(scope, filter)?;
            conjoin_subquery_scope(scope, table, filter)?;
        }
        Expr::Aggregate(_, inner) | Expr::JsonExtract(inner, _) => inject_scope_expr(scope, inner)?,
        Expr::Binary(_, l, r) | Expr::JsonExtractDyn(l, r) | Expr::JsonConcat(l, r) => {
            inject_scope_expr(scope, l)?;
            inject_scope_expr(scope, r)?;
        }
        Expr::Distance { left, right, .. } => {
            inject_scope_expr(scope, left)?;
            inject_scope_expr(scope, right)?;
        }
        Expr::Func(_, args) => {
            for a in args.iter_mut() {
                inject_scope_expr(scope, a)?;
            }
        }
        Expr::Case {
            branches,
            otherwise,
        } => {
            for (when, then) in branches {
                inject_scope_pred(scope, when)?;
                inject_scope_expr(scope, then)?;
            }
            if let Some(e) = otherwise {
                inject_scope_expr(scope, e)?;
            }
        }
        Expr::Column(_) | Expr::Value(_) | Expr::Star | Expr::VectorLiteral(_) => {}
    }
    Ok(())
}

/// Walk a predicate and inject the tenant scope into every narrow subquery (see
/// [`inject_scope_expr`]). Propagates `Err(TenancyUndeclared)` from an undeclared subquery table.
fn inject_scope_pred(scope: &Scope, p: &mut Predicate) -> Result<(), OrmError> {
    match p {
        Predicate::InSubquery {
            expr,
            table,
            filter,
            ..
        } => {
            inject_scope_expr(scope, expr)?;
            inject_scope_pred(scope, filter)?;
            conjoin_subquery_scope(scope, table, filter)?;
        }
        Predicate::And(v) | Predicate::Or(v) => {
            for c in v.iter_mut() {
                inject_scope_pred(scope, c)?;
            }
        }
        Predicate::Not(inner) => inject_scope_pred(scope, inner)?,
        Predicate::Cmp { left, right, .. } => {
            inject_scope_expr(scope, left)?;
            inject_scope_expr(scope, right)?;
        }
        Predicate::Between {
            expr, low, high, ..
        } => {
            inject_scope_expr(scope, expr)?;
            inject_scope_expr(scope, low)?;
            inject_scope_expr(scope, high)?;
        }
        Predicate::In { expr, values, .. } => {
            inject_scope_expr(scope, expr)?;
            for v in values.iter_mut() {
                inject_scope_expr(scope, v)?;
            }
        }
        Predicate::Like { expr, .. } | Predicate::Null { expr, .. } => {
            inject_scope_expr(scope, expr)?;
        }
    }
    Ok(())
}

impl Select {
    /// Inject the tenant scope into every narrow subquery this SELECT embeds — across ALL of its
    /// expr/pred-bearing fields (projection, `DISTINCT ON`, filter, having, group-by, order, and
    /// join `ON`s) — so a subquery's own table is scoped, not just the outer FROM. Called by
    /// [`Select::force_scope`] after setting the scope. Must stay exhaustive over the Expr/Predicate
    /// fields: a missed field is a cross-tenant subquery leak.
    fn inject_subquery_scope(&mut self, scope: &Scope) -> Result<(), OrmError> {
        for it in &mut self.columns {
            inject_scope_expr(scope, &mut it.expr)?;
        }
        for e in &mut self.distinct_on {
            inject_scope_expr(scope, e)?;
        }
        if let Some(f) = self.filter.as_mut() {
            inject_scope_pred(scope, f)?;
        }
        if let Some(h) = self.having.as_mut() {
            inject_scope_pred(scope, h)?;
        }
        for e in &mut self.group_by {
            inject_scope_expr(scope, e)?;
        }
        for o in &mut self.order {
            inject_scope_expr(scope, &mut o.expr)?;
        }
        for j in &mut self.joins {
            inject_scope_pred(scope, &mut j.on)?;
        }
        Ok(())
    }
}

impl Update {
    /// Force a host-resolved write `scope` (conjoined into `WHERE`), also scoping any subquery in
    /// the `SET` exprs, filter, and `RETURNING` items. Overwrites any prior scope.
    ///
    /// **R4/D8 target write (5b):** under a target scope the guest may set ONLY the route's
    /// SET-allowlisted columns (never the tenant or a visibility column), and the `WHERE` is confined
    /// to `tenant = B AND <public>` — the tenant half via [`single_scope_pred`] at compile, the
    /// public half conjoined here — so an UPDATE can touch ONLY `B`'s already-public rows and cannot
    /// flip a row in or out of the public subset. A read-only target (empty allowlist) is refused.
    pub fn force_scope(&mut self, scope: &Scope) -> Result<(), OrmError> {
        if let Some(allow) = scope.target_write_allowlist() {
            // A target write may not touch a TenantOrSession (anon-session) table (Stage A).
            scope.assert_target_table_writable(&self.table)?;
            if allow.is_empty() {
                return Err(OrmError::TargetWriteNotGranted(self.table.clone()));
            }
            for a in &self.set {
                scope.assert_target_settable(&self.table, &a.column)?;
            }
            // Confine to the public subset (the `tenant = B` half is added by the compiler via
            // `single_scope_pred`). `public_pred` already enforces deny-by-default for an anonymous
            // (`require_public`) target — a subset-less table there returns `Err(PublicSubsetUndeclared)`.
            // A `None` here therefore means the capability-only exemption (5c ruling A): no visibility
            // conjunct, so the UPDATE is confined to `tenant = B` alone (+ the SET-allowlist above),
            // exactly like the capability read/INSERT paths.
            if let Some(pred) = scope.public_pred(&self.table, None)? {
                conjoin_front(&mut self.filter, Some(pred));
            }
        }
        self.scope = Some(scope.clone());
        for a in &mut self.set {
            inject_scope_expr(scope, &mut a.value)?;
        }
        inject_scope_pred(scope, &mut self.filter)?;
        for it in &mut self.returning {
            inject_scope_expr(scope, &mut it.expr)?;
        }
        Ok(())
    }
}

impl Delete {
    /// Force a host-resolved write `scope` (conjoined into `WHERE`), also scoping any subquery in
    /// the filter and `RETURNING` items. Overwrites any prior scope. **A DELETE under a target scope
    /// is always refused (5b):** target writes are INSERT/UPDATE only — a cross-tenant delete is
    /// never granted.
    pub fn force_scope(&mut self, scope: &Scope) -> Result<(), OrmError> {
        if scope.is_target() {
            return Err(OrmError::TargetDeleteRefused(self.table.clone()));
        }
        self.scope = Some(scope.clone());
        inject_scope_pred(scope, &mut self.filter)?;
        for it in &mut self.returning {
            inject_scope_expr(scope, &mut it.expr)?;
        }
        Ok(())
    }
}

/// Why compilation failed.
#[derive(Debug, Clone, PartialEq, thiserror::Error)]
pub enum OrmError {
    /// An identifier was not a plain `[A-Za-z_][A-Za-z0-9_]*` (optionally `table.column`) name.
    #[error("invalid identifier: {0:?}")]
    InvalidIdentifier(String),
    /// The query was structurally empty (no rows to insert, no columns to set, …).
    #[error("empty query: {0}")]
    Empty(&'static str),
    /// A function was called with the wrong number of arguments, or `*` was used outside
    /// `count(*)`.
    #[error("bad expression: {0}")]
    BadExpr(&'static str),
    /// A scoped query touched a table with **no** entry in the project's [`TenancySchema`]
    /// (deny-by-default, PLAN-tenancy-principal D3): "no key" and "forgot the key" are
    /// indistinguishable, so the safe collapse is to refuse rather than run it unscoped or wrongly
    /// scoped. `Unscoped` is the explicit, reviewed "this table is global"; an absent table is a
    /// misconfiguration the host surfaces (the binding names the component + marker site).
    #[error("tenancy: table {0:?} has no declared scope (deny-by-default)")]
    TenancyUndeclared(String),
    /// A guest WRITE (INSERT/UPDATE/DELETE) targeted a table declared `Unscoped` (global reference
    /// data). Reads of an `Unscoped` table are global by design, but writes are **deny-by-default**
    /// (a shared-data write is a cross-tenant blast — the [`TableScope::Unscoped`](crate::tenancy::TableScope::Unscoped)
    /// contract), so the host refuses them rather than running the write unbounded-by-tenant.
    #[error("tenancy: table {0:?} is Unscoped (global reference); guest writes are refused (deny-by-default)")]
    UnscopedWrite(String),
    /// A scoped read/write needed a resolved principal (an own-tenant value, or — for a
    /// `TenantOrSession` table — at least one of the tenant/session facts) but the request carried
    /// none. Fail closed: the query is refused rather than run unscoped. (The single-column raw-SQL
    /// path reports the equivalent `TenantDenied::NoSource` at the binding.)
    #[error("tenancy: no resolved principal for a scoped operation (deny-by-default)")]
    TenancyNoPrincipal,
    /// A **target read** (R4/D8) touched a table with **no** declared public subset in the project
    /// schema. A target scope may only read rows that satisfy each accessed table's host-held public
    /// predicate, so a table (root or any joined/subquery ref) that declares none is refused —
    /// deny-by-default, the strict analog of [`TenancyUndeclared`]. This is what keeps a target read
    /// from ever reaching another tenant's PRIVATE rows through an un-confined table.
    #[error(
        "tenancy: table {0:?} has no declared public subset for a target read (deny-by-default)"
    )]
    PublicSubsetUndeclared(String),
    /// A WRITE (INSERT/UPDATE) was force-scoped under a **target** scope whose SET-allowlist is empty
    /// — i.e. a target route with no `write` grant is read-only (5b, deny-by-default). Refused before
    /// any SQL.
    #[error("tenancy: target route {0:?} has no write grant (read-only; deny-by-default)")]
    TargetWriteNotGranted(String),
    /// A target write tried to set a column that is not in the route's SET-allowlist — the tenant
    /// column, a public-visibility column, or any other un-granted column. Refused fail-closed so a
    /// target write can never change ownership, flip visibility, or touch a non-granted column.
    #[error("tenancy: target write may not set column {0:?} (not in the write allowlist)")]
    TargetWriteColumnDenied(String),
    /// A `DELETE` was attempted under a target scope. Target writes are INSERT/UPDATE only; a target
    /// DELETE is always refused (a cross-tenant delete is never granted).
    #[error("tenancy: a target-tenant DELETE is refused (target writes are INSERT/UPDATE only)")]
    TargetDeleteRefused(String),
    /// A target INSERT could not force a table's public subset to a concrete row: a public term that
    /// is not `column = <literal>` or `column IS NULL` (e.g. a range or `IS NOT NULL`) has no single
    /// value to stamp, so the host cannot guarantee the inserted row lands in the public subset —
    /// refused (deny-by-default). Such a subset is read-/update-only, never target-insertable.
    #[error("tenancy: target INSERT cannot force table {0:?} into its public subset (a non-equality/non-null public term); refused")]
    PublicSubsetNotForceable(String),
    /// A target write used a shape the confinement does not support: an `INSERT … SELECT`, an
    /// `ON CONFLICT` upsert on an anonymous (domain/handle/`target_or_null`) target, or a `promote`.
    /// These could reach rows outside the public subset (a selected source; or a conflict-row DO
    /// UPDATE whose `tenant = B` guard carries no visibility subset, so an anonymous upsert could
    /// touch `B`'s non-public row). A **capability**-axis `ON CONFLICT … DO UPDATE` upsert IS
    /// supported (own↔target parity — see [`Insert::confine_target_insert`]); the rest are refused.
    #[error("tenancy: unsupported target write shape ({0}); target writes are a plain INSERT, a confined UPDATE, or a capability-axis ON CONFLICT DO UPDATE upsert")]
    TargetWriteUnsupported(&'static str),
    /// A capability-axis target upsert (`ON CONFLICT … DO UPDATE`) whose conflict target does NOT
    /// include the table's tenant column. Required so a conflict is always a same-tenant (`B`) row:
    /// otherwise `B`'s INSERT could conflict with another tenant `A`'s row on a tenant-agnostic
    /// natural key, and the `tenant = B`-guarded DO UPDATE would silently no-op — dropping `B`'s
    /// write AND leaking that `A` holds that key (a cross-tenant existence oracle). Refused
    /// fail-closed; add the tenant column to the conflict target (e.g. `(tenant_id, …)`).
    #[error("tenancy: a target upsert's ON CONFLICT target must include the tenant column for table {0:?} (so a conflict is always the target tenant's own row)")]
    TargetUpsertKeyMissingTenant(String),
    /// A target write (INSERT/UPDATE) touched a `TenantOrSession` (anonymous-session-keyed) table. A
    /// target principal carries only the target tenant `B` (no session fact), so the host cannot write
    /// such a row session-scoped — it could only stamp `tenant = B`, which would silently claim an
    /// anon/session-owned row for `B` and break the anon→promotion model. Refused deny-by-default:
    /// write a `TenantOrSession` table on the caller's own/session-scoped path, never under a target
    /// scope. (PLAN-delegable-capabilities, Stage A.)
    #[error("tenancy: target write may not touch TenantOrSession table {0:?} (no session fact under a target scope — write it on the session-scoped path)")]
    TargetWriteToSessionTable(String),
}

/// The compiled statement: `?N` SQL plus its bound parameters, in placeholder order.
pub type Compiled = (String, Vec<SqlValue>);

/// Validate a plain identifier or a `table.column` qualified one. Emitted unquoted, so this
/// is the *only* thing standing between a caller-supplied name and the SQL text.
fn ident(name: &str) -> Result<&str, OrmError> {
    let ok = |s: &str| {
        let mut cs = s.chars();
        matches!(cs.next(), Some(c) if c == '_' || c.is_ascii_alphabetic())
            && s.chars().all(|c| c == '_' || c.is_ascii_alphanumeric())
    };
    let valid = match name.split_once('.') {
        Some((t, c)) => !t.is_empty() && !c.is_empty() && ok(t) && ok(c),
        None => ok(name),
    };
    if valid {
        Ok(name)
    } else {
        Err(OrmError::InvalidIdentifier(name.to_string()))
    }
}

/// Whether two identifiers name the **same column** the way the engines resolve unquoted names:
/// ASCII-case-insensitively, ignoring a leading `table.` qualifier. Used by the tenant-scope
/// guards so a guest can't dodge them by re-spelling the tenant column (`TENANT_ID`, `t.tenant_id`)
/// — the DB would still resolve it to the tenant column, but a naive `==` would miss it.
fn same_col(a: &str, b: &str) -> bool {
    let base = |s: &str| s.rsplit('.').next().unwrap_or(s).to_ascii_lowercase();
    base(a) == base(b)
}

/// Accumulates the parameter list and mints `?N` placeholders in order.
#[derive(Default)]
struct Params(Vec<SqlValue>);

impl Params {
    fn bind(&mut self, v: SqlValue) -> String {
        self.0.push(v);
        format!("?{}", self.0.len())
    }
}

/// Render a scalar expression, binding any literals.
fn render_expr(e: &Expr, params: &mut Params, dialect: Dialect) -> Result<String, OrmError> {
    Ok(match e {
        Expr::Column(name) => ident(name)?.to_string(),
        Expr::Value(v) => params.bind(v.clone()),
        Expr::Star => {
            return Err(OrmError::BadExpr(
                "`*` is only valid as the count(*) argument",
            ))
        }
        Expr::Aggregate(agg, inner) => {
            let arg = match inner.as_ref() {
                Expr::Star if *agg == Agg::Count => "*".to_string(),
                Expr::Star => return Err(OrmError::BadExpr("`*` is only valid as count(*)")),
                other => render_expr(other, params, dialect)?,
            };
            format!("{}({arg})", agg.keyword())
        }
        Expr::Binary(op, l, r) => {
            format!(
                "({} {} {})",
                render_expr(l, params, dialect)?,
                op.symbol(),
                render_expr(r, params, dialect)?
            )
        }
        Expr::Func(f, args) => {
            let (name, min, max) = f.spec();
            if args.len() < min || max.is_some_and(|m| args.len() > m) {
                return Err(OrmError::BadExpr("function called with the wrong arity"));
            }
            if args.is_empty() {
                // Nullary (`current_timestamp`) renders without parentheses (ANSI form).
                name.to_string()
            } else {
                let rendered: Result<Vec<String>, _> = args
                    .iter()
                    .map(|a| render_expr(a, params, dialect))
                    .collect();
                format!("{name}({})", rendered?.join(", "))
            }
        }
        Expr::JsonExtract(inner, path) => {
            if path.is_empty() {
                return Err(OrmError::BadExpr("json extract needs at least one key"));
            }
            // Each key is validated as an identifier — the built path can't inject.
            for k in path {
                ident(k)?;
            }
            let base = render_expr(inner, params, dialect)?;
            match dialect {
                // Postgres: `(base) #>> '{a,b}'` — keys validated, safe to inline (there is
                // no portable way to bind a `text[]` path here).
                Dialect::Postgres => format!("({base}) #>> '{{{}}}'", path.join(",")),
                // SQLite/MySQL: `json_extract(base, ?N)` with the `$.a.b` path bound.
                Dialect::Sqlite | Dialect::Mysql => {
                    let p = params.bind(SqlValue::Text(format!("$.{}", path.join("."))));
                    format!("json_extract({base}, {p})")
                }
            }
        }
        Expr::Distance {
            left,
            right,
            metric,
        } => {
            if dialect != Dialect::Postgres {
                return Err(OrmError::BadExpr("vector distance is Postgres-only"));
            }
            format!(
                "({} {} {})",
                render_expr(left, params, dialect)?,
                metric.operator(),
                render_expr(right, params, dialect)?,
            )
        }
        Expr::VectorLiteral(v) => {
            if dialect != Dialect::Postgres {
                return Err(OrmError::BadExpr("vector literals are Postgres-only"));
            }
            let p = params.bind(SqlValue::Text(vector_literal(v)?));
            // The `::vector` cast rides through the `?N` placeholder normaliser unchanged.
            format!("{p}::vector")
        }
        Expr::RelatedAggregate {
            agg,
            arg,
            table,
            filter,
        } => {
            let arg_sql = match arg {
                RelArg::Star if *agg == Agg::Count => "*".to_string(),
                RelArg::Star => return Err(OrmError::BadExpr("`*` is only valid as count(*)")),
                RelArg::Column(c) => ident(c)?.to_string(),
            };
            let table_sql = ident(table)?;
            // The correlated filter reuses the ordinary predicate compiler (bound params); the
            // WHERE clause delimits it, so it renders unparenthesised (`nested = false`).
            let where_sql = render_pred(filter, params, false, dialect)?;
            format!(
                "(SELECT {}({arg_sql}) FROM {table_sql} WHERE {where_sql})",
                agg.keyword()
            )
        }
        Expr::RelatedScalar {
            column,
            table,
            filter,
        } => {
            let col_sql = ident(column)?;
            let table_sql = ident(table)?;
            let where_sql = render_pred(filter, params, false, dialect)?;
            format!("(SELECT {col_sql} FROM {table_sql} WHERE {where_sql})")
        }
        Expr::JsonExtractDyn(base, key) => {
            if matches!(dialect, Dialect::Mysql) {
                return Err(OrmError::BadExpr(
                    "dynamic-key json extract (->> <bound>) is not supported on MySQL",
                ));
            }
            format!(
                "({} ->> {})",
                render_expr(base, params, dialect)?,
                render_expr(key, params, dialect)?,
            )
        }
        Expr::JsonConcat(left, right) => {
            if dialect != Dialect::Postgres {
                return Err(OrmError::BadExpr("json concat (||) is Postgres-only"));
            }
            format!(
                "({} || {})",
                render_expr(left, params, dialect)?,
                render_expr(right, params, dialect)?,
            )
        }
        Expr::Case {
            branches,
            otherwise,
        } => {
            if branches.is_empty() {
                return Err(OrmError::BadExpr("CASE has no WHEN branches"));
            }
            let mut s = String::from("CASE");
            for (when, then) in branches {
                // Params bind in textual order: each WHEN before its THEN, branches in order,
                // ELSE last — matching how `render_pred`/`render_expr` push placeholders.
                let w = render_pred(when, params, false, dialect)?;
                let t = render_expr(then, params, dialect)?;
                s.push_str(&format!(" WHEN {w} THEN {t}"));
            }
            if let Some(e) = otherwise {
                let e = render_expr(e, params, dialect)?;
                s.push_str(&format!(" ELSE {e}"));
            }
            s.push_str(" END");
            format!("({s})")
        }
        // Reaching here means the marker was never lowered — i.e. no own-tenant scope was applied
        // to this query (an unscoped / `disabled` / cross-tenant-`all`-without-value function). Fail
        // closed rather than emit an unscoped ranking.
        Expr::IsOwn => {
            return Err(OrmError::BadExpr(
                "is_own()/own_first() requires an own-tenant (own or own+null) read scope",
            ))
        }
    })
}

/// Validate a `pgvector` literal — a bracketed, comma-separated list of finite numbers
/// (`[0.1, 0.2]`) — returning it whitespace-normalised. The result binds as a parameter, so
/// this is a data-quality gate (a clear early error over a Postgres runtime failure), not an
/// injection defence.
fn vector_literal(s: &str) -> Result<String, OrmError> {
    let inner = s
        .trim()
        .strip_prefix('[')
        .and_then(|x| x.strip_suffix(']'))
        .ok_or(OrmError::BadExpr(
            "vector literal must be a bracketed list like [0.1, 0.2]",
        ))?;
    if inner.trim().is_empty() {
        return Err(OrmError::BadExpr(
            "vector literal must have at least one component",
        ));
    }
    let mut parts = Vec::new();
    for part in inner.split(',') {
        let p = part.trim();
        let f: f64 = p
            .parse()
            .map_err(|_| OrmError::BadExpr("vector literal component is not a number"))?;
        if !f.is_finite() {
            return Err(OrmError::BadExpr("vector literal component must be finite"));
        }
        parts.push(p);
    }
    Ok(format!("[{}]", parts.join(",")))
}

/// Render a predicate; `nested` parenthesizes a compound (`AND`/`OR`) so precedence is explicit.
fn render_pred(
    p: &Predicate,
    params: &mut Params,
    nested: bool,
    dialect: Dialect,
) -> Result<String, OrmError> {
    let compound = |body: String| {
        if nested {
            format!("({body})")
        } else {
            body
        }
    };
    Ok(match p {
        Predicate::And(ps) => {
            if ps.is_empty() {
                "1 = 1".to_string()
            } else {
                let parts: Result<Vec<String>, _> = ps
                    .iter()
                    .map(|c| render_pred(c, params, true, dialect))
                    .collect();
                compound(parts?.join(" AND "))
            }
        }
        Predicate::Or(ps) => {
            if ps.is_empty() {
                "1 = 0".to_string()
            } else {
                let parts: Result<Vec<String>, _> = ps
                    .iter()
                    .map(|c| render_pred(c, params, true, dialect))
                    .collect();
                compound(parts?.join(" OR "))
            }
        }
        Predicate::Not(inner) => format!("NOT {}", render_pred(inner, params, true, dialect)?),
        Predicate::Cmp { left, op, right } => format!(
            "{} {} {}",
            render_expr(left, params, dialect)?,
            op.symbol(),
            render_expr(right, params, dialect)?
        ),
        Predicate::Between {
            expr,
            low,
            high,
            negated,
        } => format!(
            "{} {}BETWEEN {} AND {}",
            render_expr(expr, params, dialect)?,
            if *negated { "NOT " } else { "" },
            render_expr(low, params, dialect)?,
            render_expr(high, params, dialect)?
        ),
        Predicate::In {
            expr,
            values,
            negated,
        } => {
            if values.is_empty() {
                // `IN ()` is a syntax error; render the matching identity.
                if *negated { "1 = 1" } else { "1 = 0" }.to_string()
            } else {
                let lhs = render_expr(expr, params, dialect)?;
                let ph: Result<Vec<String>, _> = values
                    .iter()
                    .map(|v| render_expr(v, params, dialect))
                    .collect();
                format!(
                    "{lhs} {}IN ({})",
                    if *negated { "NOT " } else { "" },
                    ph?.join(", ")
                )
            }
        }
        Predicate::Like {
            expr,
            pattern,
            insensitive,
            negated,
        } => {
            let neg = if *negated { "NOT " } else { "" };
            let lhs = render_expr(expr, params, dialect)?;
            let pat = params.bind(SqlValue::Text(pattern.clone()));
            if *insensitive {
                // Portable case-insensitive LIKE (no dialect-specific ILIKE).
                format!("lower({lhs}) {neg}LIKE lower({pat})")
            } else {
                format!("{lhs} {neg}LIKE {pat}")
            }
        }
        Predicate::Null { expr, negated } => format!(
            "{} IS {}NULL",
            render_expr(expr, params, dialect)?,
            if *negated { "NOT " } else { "" }
        ),
        Predicate::InSubquery {
            expr,
            column,
            table,
            filter,
            negated,
        } => {
            let lhs = render_expr(expr, params, dialect)?;
            let col_sql = ident(column)?;
            let table_sql = ident(table)?;
            let where_sql = render_pred(filter, params, false, dialect)?;
            let not = if *negated { "NOT " } else { "" };
            format!("{lhs} {not}IN (SELECT {col_sql} FROM {table_sql} WHERE {where_sql})")
        }
    })
}

/// Render the `WHERE` body from a pre-built scope predicate + optional filter (scope conjoined
/// first). The scope predicate is built by the caller — single-table for UPDATE/DELETE
/// ([`single_scope_pred`]), multi-table-qualified for a SELECT with joins
/// ([`Select::scope_where_pred`]).
fn render_where(
    scope_pred: Option<Predicate>,
    filter: Option<&Predicate>,
    params: &mut Params,
    dialect: Dialect,
) -> Result<Option<String>, OrmError> {
    // An empty `AND` filter is a no-op (always true) — drop it so it never adds a spurious
    // `AND 1 = 1`. (An empty `OR` means "match nothing" and is kept.)
    let filter = filter.filter(|f| !matches!(f, Predicate::And(v) if v.is_empty()));
    // A lone clause renders directly (no wrapping `AND`, so a top-level `AND`/`OR` filter
    // isn't spuriously parenthesized); scope + filter conjoin as `scope AND (filter)`.
    let combined = match (scope_pred, filter) {
        (None, None) => return Ok(None),
        (Some(s), None) => s,
        (None, Some(f)) => f.clone(),
        (Some(s), Some(f)) => Predicate::And(vec![s, f.clone()]),
    };
    Ok(Some(render_pred(&combined, params, false, dialect)?))
}

/// The single-table scope predicate for an UPDATE/DELETE `WHERE`, bounding the write to the actor's
/// OWN partition via [`Scope::write_target`]: a plain tenant table bounds `tenant_col = <own>` (or
/// `tenant_col IS NULL` for the explicit null-baseline grant); a `TenantOrSession` table bounds the
/// single axis the actor holds (`tenant_col = T` authenticated, else `session_col = S`). `Ok(None)`
/// ⇒ `All` (no bound) or no forced scope. An `Unscoped` target is refused (`UnscopedWrite`), an
/// undeclared one too (`TenancyUndeclared`), and a scoped write with no principal
/// (`TenancyNoPrincipal`) — deny-by-default.
fn single_scope_pred(scope: Option<&Scope>, table: &str) -> Result<Option<Predicate>, OrmError> {
    let Some(s) = scope else { return Ok(None) };
    match s.write_target(table)? {
        None => Ok(None), // All — no bound
        Some((col, value)) => {
            ident(&col)?;
            let col_expr = Expr::Column(col);
            // A NULL stamp value is the explicit null-baseline grant ⇒ `IS NULL`; any real tenant /
            // session value ⇒ `= value`. (Own/session values are never NULL, so this is unambiguous.)
            let pred = if matches!(value, SqlValue::Null) {
                Predicate::Null {
                    expr: col_expr,
                    negated: false,
                }
            } else {
                Predicate::Cmp {
                    left: col_expr,
                    op: CmpOp::Eq,
                    right: Expr::Value(value),
                }
            };
            Ok(Some(pred))
        }
    }
}

/// Render a select list (empty ⇒ `*`).
fn render_select_items(
    items: &[SelectItem],
    params: &mut Params,
    dialect: Dialect,
) -> Result<String, OrmError> {
    if items.is_empty() {
        return Ok("*".to_string());
    }
    let parts: Result<Vec<String>, _> = items
        .iter()
        .map(|it| {
            let e = render_expr(&it.expr, params, dialect)?;
            Ok::<String, OrmError>(match &it.alias {
                Some(a) => format!("{e} AS {}", ident(a)?),
                None => e,
            })
        })
        .collect();
    Ok(parts?.join(", "))
}

/// Render a `RETURNING` clause, if any.
fn render_returning(
    items: &[SelectItem],
    params: &mut Params,
    dialect: Dialect,
) -> Result<String, OrmError> {
    if items.is_empty() {
        Ok(String::new())
    } else {
        Ok(format!(
            " RETURNING {}",
            render_select_items(items, params, dialect)?
        ))
    }
}

impl Select {
    /// A `SELECT * FROM <table>` to refine with the public fields.
    pub fn from(table: impl Into<String>) -> Self {
        Self {
            table: table.into(),
            table_alias: None,
            columns: Vec::new(),
            joins: Vec::new(),
            filter: None,
            scope: None,
            group_by: Vec::new(),
            having: None,
            distinct: false,
            distinct_on: Vec::new(),
            order: Vec::new(),
            limit: None,
            offset: None,
            union: None,
        }
    }

    /// Compile to `?N` SQL + bound parameters for the given dialect. A `UNION` branch renders
    /// after the body, sharing the placeholder sequence (so binds stay in textual order).
    pub fn compile(&self, dialect: Dialect) -> Result<Compiled, OrmError> {
        let mut params = Params::default();
        let sql = self.render_into(&mut params, dialect)?;
        Ok((sql, params.0))
    }

    /// The scope predicate to conjoin into this SELECT's `WHERE`. With **no joins** it's the
    /// single-table (unqualified) predicate. With joins, the per-mode predicate is applied to the
    /// FROM table plus each **INNER** join, qualified by its alias-or-name, so a guest can't read a
    /// joined table's cross-tenant rows through the projection (a join to a table lacking the tenant
    /// column then fails closed at the DB, not leaks). A **LEFT-OUTER** join is confined in its own
    /// `ON` at [`Select::force_scope`] instead — a WHERE predicate on the nullable side would
    /// collapse the LEFT JOIN to an INNER JOIN — so it is skipped here. `all`/no-scope ⇒ `None`.
    fn scope_where_pred(&self) -> Result<Option<Predicate>, OrmError> {
        let Some(scope) = &self.scope else {
            return Ok(None);
        };
        if self.joins.is_empty() {
            // Single table: its per-table read predicate (the tenant column, the R3 disjunct, or
            // `None` for an `Unscoped` global; deny-by-default / no-principal fail closed).
            return scope.read_pred(&self.table, None);
        }
        // Joined: the FROM table and each INNER-joined table are scoped on their OWN resolved key,
        // qualified by alias-or-name, so a guest can't read a joined table's cross-tenant rows
        // through the projection. LEFT-joined tables are confined in their own `ON` (force_scope),
        // NOT here. A ref whose table is undeclared fails closed (deny-by-default); an `Unscoped`
        // ref adds no predicate (it is global by declaration).
        let refs: Vec<(&str, &str)> = std::iter::once((
            self.table.as_str(),
            self.table_alias.as_deref().unwrap_or(&self.table),
        ))
        .chain(
            self.joins
                .iter()
                .filter(|j| matches!(j.kind, JoinKind::Inner))
                .map(|j| (j.table.as_str(), j.alias.as_deref().unwrap_or(&j.table))),
        )
        .collect();
        let mut parts: Vec<Predicate> = Vec::with_capacity(refs.len());
        for (table, qual) in refs {
            ident(qual)?;
            if let Some(p) = scope.read_pred(table, Some(qual))? {
                parts.push(p);
            }
        }
        Ok((!parts.is_empty()).then_some(Predicate::And(parts)))
    }

    /// Render the full SELECT (body + any UNION branch) into the shared `params`. Reused by
    /// `INSERT … SELECT` so a source select shares the outer placeholder sequence. Module-private
    /// because `Params` is (Insert::compile, same module, is the other caller).
    fn render_into(&self, params: &mut Params, dialect: Dialect) -> Result<String, OrmError> {
        let mut sql = self.render_body(params, dialect)?;
        if let Some(u) = &self.union {
            let kw = if u.all { "UNION ALL" } else { "UNION" };
            let branch = u.query.render_body(params, dialect)?;
            sql.push_str(&format!(" {kw} {branch}"));
        }
        Ok(sql)
    }

    /// Render one SELECT body (no UNION) into the shared `params`.
    fn render_body(&self, params: &mut Params, dialect: Dialect) -> Result<String, OrmError> {
        let table = ident(&self.table)?;

        // The DISTINCT clause renders before the select list so any bound params order correctly.
        let distinct = if !self.distinct_on.is_empty() {
            if dialect != Dialect::Postgres {
                return Err(OrmError::BadExpr("DISTINCT ON is Postgres-only"));
            }
            let cols = self
                .distinct_on
                .iter()
                .map(|e| render_expr(e, &mut *params, dialect))
                .collect::<Result<Vec<_>, _>>()?;
            format!("DISTINCT ON ({}) ", cols.join(", "))
        } else if self.distinct {
            "DISTINCT ".to_string()
        } else {
            String::new()
        };
        let select_list = render_select_items(&self.columns, &mut *params, dialect)?;
        let mut sql = format!("SELECT {distinct}{select_list} FROM {table}");
        if let Some(a) = &self.table_alias {
            sql.push_str(&format!(" AS {}", ident(a)?));
        }

        for j in &self.joins {
            let jt = ident(&j.table)?;
            let kw = match j.kind {
                JoinKind::Inner => "JOIN",
                JoinKind::Left => "LEFT JOIN",
            };
            sql.push_str(&format!(" {kw} {jt}"));
            if let Some(a) = &j.alias {
                sql.push_str(&format!(" AS {}", ident(a)?));
            }
            // A **LEFT**-joined table is confined in its OWN `ON`, not the top-level `WHERE`
            // (`scope_where_pred` scopes the FROM table + INNER joins there and deliberately SKIPS
            // LEFT joins): a WHERE predicate on the nullable side would collapse the LEFT JOIN to an
            // INNER JOIN — dropping the driving row when there is no match, so a
            // `COALESCE(joined.col, driving.col)` fallback would never fire. Conjoined into the `ON`,
            // an unmatched / other-tenant row instead becomes `NULL` — never a cross-tenant bleed
            // (the tenant/public gate is AND-ed into the join condition, and the structured predicate
            // renders with correct precedence, so a top-level OR in the guest `ON` cannot widen past
            // the gate). Applied at RENDER (not `force_scope`), so confinement never depends on which
            // entry point set the scope. INNER joins are confined in the WHERE by `scope_where_pred`.
            let qual = j.alias.as_deref().unwrap_or(&j.table);
            let on = match (&self.scope, j.kind) {
                (Some(scope), JoinKind::Left) => match scope.read_pred(&j.table, Some(qual))? {
                    Some(conf) => Predicate::And(vec![j.on.clone(), conf]),
                    None => j.on.clone(),
                },
                _ => j.on.clone(),
            };
            sql.push_str(&format!(
                " ON {}",
                render_pred(&on, &mut *params, false, dialect)?
            ));
        }

        if let Some(w) = render_where(
            self.scope_where_pred()?,
            self.filter.as_ref(),
            &mut *params,
            dialect,
        )? {
            sql.push_str(&format!(" WHERE {w}"));
        }

        if !self.group_by.is_empty() {
            let terms: Result<Vec<String>, _> = self
                .group_by
                .iter()
                .map(|e| render_expr(e, &mut *params, dialect))
                .collect();
            sql.push_str(&format!(" GROUP BY {}", terms?.join(", ")));
        }

        if let Some(h) = &self.having {
            sql.push_str(&format!(
                " HAVING {}",
                render_pred(h, &mut *params, false, dialect)?
            ));
        }

        if !self.order.is_empty() {
            let terms: Result<Vec<String>, _> = self
                .order
                .iter()
                .map(|o| {
                    let e = render_expr(&o.expr, &mut *params, dialect)?;
                    let d = match o.dir {
                        Direction::Asc => "ASC",
                        Direction::Desc => "DESC",
                    };
                    Ok::<String, OrmError>(format!("{e} {d}"))
                })
                .collect();
            sql.push_str(&format!(" ORDER BY {}", terms?.join(", ")));
        }

        if let Some(n) = self.limit {
            sql.push_str(&format!(" LIMIT {n}"));
        }
        if let Some(n) = self.offset {
            sql.push_str(&format!(" OFFSET {n}"));
        }

        Ok(sql)
    }
}

impl Insert {
    /// Compile to `?N` SQL + bound parameters for the given dialect.
    pub fn compile(&self, dialect: Dialect) -> Result<Compiled, OrmError> {
        let table = ident(&self.table)?;
        let mut params = Params::default();

        // The write target's per-table stamp `(column, value)` — the actor's OWN axis (tenant when
        // authenticated, the anon session for a `TenantOrSession` table otherwise; PLAN R2/R3).
        // Resolving it enforces deny-by-default for BOTH the VALUES and INSERT…SELECT forms — an
        // undeclared target, an `Unscoped` (global) target, or a scoped write with no principal are
        // refused here. `None` ⇒ `all` mode (no stamp — a posture-vetted cross-tenant write) or no
        // forced scope.
        let stamp: Option<(String, SqlValue)> = match self.scope.as_ref() {
            Some(s) => s.write_target(&self.table)?,
            None => None,
        };

        // INSERT … SELECT: rows come from a source query sharing the placeholder sequence.
        if let Some((cols, select)) = &self.from_select {
            let col_sql = cols
                .iter()
                .map(|c| ident(c).map(str::to_string))
                .collect::<Result<Vec<_>, _>>()?;
            if col_sql.is_empty() {
                return Err(OrmError::Empty("insert-select has no columns"));
            }
            let select_sql = select.render_into(&mut params, dialect)?;
            let mut sql = format!("INSERT INTO {table} ({}) {select_sql}", col_sql.join(", "));
            sql.push_str(&render_conflict(
                self.conflict.as_ref(),
                table,
                stamp.as_ref().map(|(c, v)| (c.as_str(), v)),
                &mut params,
                dialect,
            )?);
            sql.push_str(&render_returning(&self.returning, &mut params, dialect)?);
            return Ok((sql, params.0));
        }

        if self.rows.is_empty() {
            return Err(OrmError::Empty("insert has no rows"));
        }

        // Column set: from the first row (+ the scope column if forced), in a stable order.
        // Every row is coerced to exactly these columns; the scope value overrides.
        let mut columns: Vec<String> = Vec::new();
        for a in &self.rows[0].cells {
            let c = ident(&a.column)?.to_string();
            if !columns.contains(&c) {
                columns.push(c);
            }
        }
        // The resolved stamp (own tenant, the null baseline, or the anon session value) forces its
        // per-table column into every row. `all` mode / no forced scope stamps nothing (`stamp` is
        // `None`). The column match is case/qualifier-insensitive (`same_col`) so a guest can't
        // smuggle its own value into the stamped column by re-spelling it (`TENANT_ID`, `t.tenant_id`).
        if let Some((column, _)) = &stamp {
            let c = ident(column)?.to_string();
            if !columns.iter().any(|existing| same_col(existing, &c)) {
                columns.push(c);
            }
        }
        if columns.is_empty() {
            return Err(OrmError::Empty("insert row has no columns"));
        }

        let mut value_groups: Vec<String> = Vec::new();
        for row in &self.rows {
            let mut ph: Vec<String> = Vec::with_capacity(columns.len());
            for col in &columns {
                // The scope forces its column to the resolved stamp; otherwise take the row's
                // cell expr, else NULL.
                if let Some((column, value)) = &stamp {
                    if same_col(column, col) {
                        ph.push(params.bind(value.clone()));
                        continue;
                    }
                }
                match row.cells.iter().find(|a| same_col(&a.column, col)) {
                    Some(a) => ph.push(render_expr(&a.value, &mut params, dialect)?),
                    None => ph.push(params.bind(SqlValue::Null)),
                }
            }
            value_groups.push(format!("({})", ph.join(", ")));
        }

        let mut sql = format!(
            "INSERT INTO {table} ({}) VALUES {}",
            columns.join(", "),
            value_groups.join(", ")
        );

        sql.push_str(&render_conflict(
            self.conflict.as_ref(),
            table,
            stamp.as_ref().map(|(c, v)| (c.as_str(), v)),
            &mut params,
            dialect,
        )?);
        sql.push_str(&render_returning(&self.returning, &mut params, dialect)?);
        Ok((sql, params.0))
    }
}

/// Render an `ON CONFLICT (...) DO NOTHING|UPDATE SET ...` clause (empty when `None`). The
/// DO UPDATE assignments bind params, so it takes the shared [`Params`].
///
/// Under a tenant scope with a stampable value (own/null — `all` bounds nothing), the DO UPDATE is
/// **bounded to the tenant's own rows** so a guest upsert can't overwrite another tenant's row via
/// a conflict on a non-tenant-partitioned key, and any assignment targeting the scope column is
/// **dropped** so the tenant of an existing row is never reassigned. MySQL's `ON DUPLICATE KEY
/// UPDATE` can't carry that bound, so a scoped upsert on MySQL is refused (fail-closed).
fn render_conflict(
    conflict: Option<&OnConflict>,
    table: &str,
    stamp: Option<(&str, &SqlValue)>,
    params: &mut Params,
    dialect: Dialect,
) -> Result<String, OrmError> {
    let Some(oc) = conflict else {
        return Ok(String::new());
    };
    let conflict_cols = oc
        .conflict_columns
        .iter()
        .map(|c| ident(c).map(str::to_string))
        .collect::<Result<Vec<_>, _>>()?;
    // The resolved write stamp `(column, value)` that must bound the upsert (own/session/null →
    // a predicate; `all` / no-scope → `None`, nothing to guard).
    let guard = stamp;
    let do_nothing = || format!(" ON CONFLICT ({}) DO NOTHING", conflict_cols.join(", "));
    if oc.update.is_empty() {
        return Ok(do_nothing());
    }
    if guard.is_some() && matches!(dialect, Dialect::Mysql) {
        return Err(OrmError::BadExpr(
            "a tenant-scoped upsert (ON CONFLICT DO UPDATE) is unsupported on MySQL \
             (ON DUPLICATE KEY UPDATE cannot be bounded to the tenant's rows)",
        ));
    }
    // Drop any assignment to the stamped (tenant/session) column: a guest upsert never reassigns an
    // existing row's owning axis. If that leaves nothing to update, degrade to DO NOTHING.
    let sets = oc
        .update
        .iter()
        .filter(|a| guard.is_none_or(|(col, _)| !same_col(&a.column, col)))
        .map(|a| {
            let c = ident(&a.column)?;
            Ok::<String, OrmError>(format!("{c} = {}", render_expr(&a.value, params, dialect)?))
        })
        .collect::<Result<Vec<_>, _>>()?;
    if sets.is_empty() {
        return Ok(do_nothing());
    }
    let mut clause = format!(
        " ON CONFLICT ({}) DO UPDATE SET {}",
        conflict_cols.join(", "),
        sets.join(", ")
    );
    if let Some((col, value)) = guard {
        ident(col)?;
        // Bound the DO UPDATE to the actor's own partition (Postgres/SQLite support a trailing
        // WHERE), keyed on the stamped column — `= value`, or `IS NULL` for the null baseline.
        // **Qualify with the target table** (`<table>.<col>`): inside `DO UPDATE` the target table
        // and the `excluded` pseudo-relation both expose the tenant column, so a bare `<col>` is
        // ambiguous on Postgres (`column reference "<col>" is ambiguous`) and the whole upsert fails.
        // `excluded` is never the guard's subject — the guard bounds the row being *updated* — so
        // target-qualifying is always correct. (The target table is `ident`-validated by the caller.)
        let col_expr = Expr::Column(format!("{table}.{col}"));
        let pred = if matches!(value, SqlValue::Null) {
            Predicate::Null {
                expr: col_expr,
                negated: false,
            }
        } else {
            Predicate::Cmp {
                left: col_expr,
                op: CmpOp::Eq,
                right: Expr::Value(value.clone()),
            }
        };
        clause.push_str(&format!(
            " WHERE {}",
            render_pred(&pred, params, false, dialect)?
        ));
    }
    Ok(clause)
}

impl Update {
    /// Compile to `?N` SQL + bound parameters for the given dialect. An empty `filter` is
    /// refused (no unbounded update).
    pub fn compile(&self, dialect: Dialect) -> Result<Compiled, OrmError> {
        if self.set.is_empty() {
            return Err(OrmError::Empty("update has no assignments"));
        }
        // Guard against an effectively-unbounded update: an empty `AND`/`OR` filter renders to
        // a tautology, so with no tenant scope it would touch every row. Refuse it. (A scope
        // keeps the update bounded, so an empty filter + scope is allowed.)
        let empty_filter =
            matches!(&self.filter, Predicate::And(v) | Predicate::Or(v) if v.is_empty());
        if empty_filter && self.scope.is_none() {
            return Err(OrmError::Empty(
                "update has an empty filter (unbounded update refused)",
            ));
        }
        let table = ident(&self.table)?;
        let mut params = Params::default();

        // A scoped write never reassigns the owning-axis column: drop any `SET <axis column> = …`
        // (case/qualifier-insensitively) so a guest can't donate its own rows into another tenant's
        // (or session's) partition (mirrors the ON CONFLICT DO UPDATE guard). The column is the
        // actor's per-table axis key (Stage 1/R3): `tenant_id` (or the identity PK) authenticated,
        // the `session_id` for an anon `TenantOrSession` write. `all` mode ⇒ no drop; an `Unscoped`
        // or undeclared or no-principal write is refused (via `write_target`). The WHERE still bounds
        // the update to own rows; this bounds what it may *change*.
        let scope_col: Option<String> = match self.scope.as_ref() {
            Some(s) => s.write_target(&self.table)?.map(|(col, _)| col),
            None => None,
        };
        // SET binds before WHERE so placeholder order matches the parameter order.
        let sets: Result<Vec<String>, _> = self
            .set
            .iter()
            .filter(|a| {
                scope_col
                    .as_deref()
                    .is_none_or(|col| !same_col(&a.column, col))
            })
            .map(|a| {
                let c = ident(&a.column)?;
                Ok::<String, OrmError>(format!(
                    "{c} = {}",
                    render_expr(&a.value, &mut params, dialect)?
                ))
            })
            .collect();
        let sets = sets?;
        if sets.is_empty() {
            return Err(OrmError::Empty(
                "update has no assignments left after dropping the tenant column",
            ));
        }
        let set_sql = sets.join(", ");

        let where_sql = render_where(
            single_scope_pred(self.scope.as_ref(), &self.table)?,
            Some(&self.filter),
            &mut params,
            dialect,
        )?
        .ok_or(OrmError::Empty(
            "update has an empty filter (unbounded update refused)",
        ))?;

        let mut sql = format!("UPDATE {table} SET {set_sql} WHERE {where_sql}");
        sql.push_str(&render_returning(&self.returning, &mut params, dialect)?);
        Ok((sql, params.0))
    }
}

impl Delete {
    /// Compile to `?N` SQL + bound parameters. An empty `filter` with no scope is refused
    /// (no unbounded delete), exactly as [`Update::compile`]. A scope keeps it bounded, so an
    /// empty filter + scope is allowed.
    pub fn compile(&self, dialect: Dialect) -> Result<Compiled, OrmError> {
        let empty_filter =
            matches!(&self.filter, Predicate::And(v) | Predicate::Or(v) if v.is_empty());
        if empty_filter && self.scope.is_none() {
            return Err(OrmError::Empty(
                "delete has an empty filter (unbounded delete refused)",
            ));
        }
        let table = ident(&self.table)?;
        let mut params = Params::default();
        let where_sql = render_where(
            single_scope_pred(self.scope.as_ref(), &self.table)?,
            Some(&self.filter),
            &mut params,
            dialect,
        )?
        .ok_or(OrmError::Empty(
            "delete has an empty filter (unbounded delete refused)",
        ))?;
        let mut sql = format!("DELETE FROM {table} WHERE {where_sql}");
        sql.push_str(&render_returning(&self.returning, &mut params, dialect)?);
        Ok((sql, params.0))
    }
}

/// Compile the deny-by-default **`promote`** verb (PLAN-tenancy-principal D7): claim a returning
/// visitor's anonymous-session rows for their now-authenticated tenant. It is a **distinct
/// host-mediated verb**, never an [`AccessMode`](crate::tenancy::AccessMode) or a guest-authored
/// UPDATE — the guest can name neither the session value nor the cross-axis NULL.
///
/// Requires `table` to be a [`TableScope::TenantOrSession`](crate::tenancy::TableScope) table AND
/// the [`Scope`] to carry BOTH a tenant fact `T` (`value`) and a session fact `S` (`session`) — else
/// refused ([`OrmError::TenancyNoPrincipal`] / [`OrmError::BadExpr`]). Lowers to:
///
/// ```sql
/// UPDATE <table> SET <tenant_col> = T WHERE <session_col> = S AND <tenant_col> IS NULL
/// ```
///
/// The `<tenant_col> IS NULL` match is the **anti-widening guard**: promotion can only claim rows
/// not yet owned by any tenant, never re-home another tenant's rows into `T`. It is non-escalating,
/// idempotent, and race-safe — a second promotion (or a concurrent one that lost) matches nothing.
pub fn compile_promote(scope: &Scope, table: &str, dialect: Dialect) -> Result<Compiled, OrmError> {
    // `promote` is an OWN-axis session→tenant claim; it is meaningless (and unsafe) under a target
    // scope. Refuse it fail-closed so a target route can never move another tenant's rows.
    if scope.is_target() {
        return Err(OrmError::TargetWriteUnsupported("promote"));
    }
    let (tenant_col, session_col) = match scope.resolve_table(table)? {
        ResolvedScope::TenantOrSession { tenant, session } => (tenant, session),
        _ => {
            return Err(OrmError::BadExpr(
                "promote requires a TenantOrSession table (an anonymous-first table)",
            ))
        }
    };
    ident(&tenant_col)?;
    ident(&session_col)?;
    // BOTH facts are mandatory — promotion is the authenticated claim of one's own anon session.
    let tenant = scope.value.clone().ok_or(OrmError::TenancyNoPrincipal)?;
    let session = scope.session.clone().ok_or(OrmError::TenancyNoPrincipal)?;
    let promote = Update {
        table: table.to_string(),
        // set the tenant column to the resolved tenant fact.
        set: vec![Assignment {
            column: tenant_col.clone(),
            value: Expr::val(tenant),
        }],
        // match this session's not-yet-owned rows only (the anti-widening guard).
        filter: Predicate::And(vec![
            Predicate::Cmp {
                left: Expr::Column(session_col),
                op: CmpOp::Eq,
                right: Expr::val(session),
            },
            Predicate::Null {
                expr: Expr::Column(tenant_col),
                negated: false,
            },
        ]),
        // The promotion IS the scope; no additional host force_scope (and the tenant-column SET is
        // the sanctioned reassignment-from-NULL, so the usual reassignment SET-drop must NOT fire).
        scope: None,
        returning: vec![],
    };
    promote.compile(dialect)
}

/// A parent-referencing derived-tenant write ([`compile_attach_reference`], PLAN-tenancy-principal
/// 5d): insert a row into `child` whose tenant is DERIVED from a `parent` row reachable under the
/// caller's current confined scope, gated by `parent.<ref_column> = ref_value`.
#[derive(Debug, Clone, PartialEq)]
pub struct AttachReference {
    /// The table the new row is inserted into.
    pub child: String,
    /// The referenced parent table (read under the caller's scope).
    pub parent: String,
    /// The parent selector column: `parent.<ref_column> = <ref_value>` picks the referenced row.
    pub ref_column: String,
    /// The parent selector value (guest-supplied — a row selector INSIDE the caller's confined scope,
    /// never a tenant value).
    pub ref_value: SqlValue,
    /// The child's non-tenant column assignments (guest values). Under a target scope each column
    /// must be in the route's SET-allowlist (never the tenant or a visibility column).
    pub set: Vec<Assignment>,
}

/// Compile `attach_reference` (5d): the host-mediated **derived-tenant** write. Lowers to
///
/// ```sql
/// INSERT INTO <child> (<set cols…>, <child tenant col> [, <child public cols…>])
/// SELECT <set vals…>, <parent tenant col> [, <public literals…>]
/// FROM <parent> WHERE <parent>.<ref> = ? AND <caller's scope on the parent>
/// ```
///
/// The child's tenant is **projected from the scope-confined parent**, never the guest — so it is
/// bounded by the caller's own reach (`own` → {A, NULL}; `target` → {B}, since the parent is confined
/// to `tenant = B AND <public>`). An unreachable parent selects zero rows ⇒ zero inserts (a
/// fail-closed no-op, never an oracle). Under a **target** scope the child's own public-visibility
/// columns are force-stamped and the guest `set` columns are gated by the target write-allowlist
/// ([`Scope::assert_target_settable`]), so the inserted row lands in the child's public subset. The
/// caller's WRITE grant is enforced above this (the binding's `scope_for(Write)` fails closed for a
/// read-only route — so a `handle`-resolved target, whose write axis is denied, can never reach here,
/// satisfying G1). `child`/`parent` must be plain `Column`-scoped tenant tables.
pub fn compile_attach_reference(
    scope: &Scope,
    spec: &AttachReference,
    dialect: Dialect,
) -> Result<Compiled, OrmError> {
    ident(&spec.ref_column)?;
    // Resolve the child + parent tenant columns — both must be plain tenant `Column` tables (a
    // derived-tenant write onto an identity/`Unscoped`/`TenantOrSession` table is out of scope).
    let child_tenant = match scope.resolve_table(&spec.child)? {
        ResolvedScope::Column(c) => c,
        _ => {
            return Err(OrmError::BadExpr(
                "attach_reference child must be a plain tenant table",
            ))
        }
    };
    let parent_tenant = match scope.resolve_table(&spec.parent)? {
        ResolvedScope::Column(c) => c,
        _ => {
            return Err(OrmError::BadExpr(
                "attach_reference parent must be a plain tenant table",
            ))
        }
    };
    ident(&child_tenant)?;
    ident(&parent_tenant)?;

    let is_target = scope.is_target();
    let mut columns: Vec<String> = Vec::with_capacity(spec.set.len() + 2);
    let mut projection: Vec<SelectItem> = Vec::with_capacity(spec.set.len() + 2);
    // The guest's non-tenant columns. The host DERIVES the child's tenant column from the parent, so
    // the guest may never name it (own OR target) — that would collide with (or try to forge) the
    // derived value. Under target, columns are additionally gated by the write-allowlist (never a
    // visibility column). Under own, they are only validated as identifiers (the caller writes its
    // own rows, exactly as a normal own INSERT).
    for a in &spec.set {
        if same_col(&a.column, &child_tenant) {
            return Err(OrmError::TargetWriteColumnDenied(a.column.clone()));
        }
        if is_target {
            scope.assert_target_settable(&spec.child, &a.column)?;
        } else {
            ident(&a.column)?;
        }
        columns.push(a.column.clone());
        projection.push(SelectItem {
            expr: a.value.clone(),
            alias: None,
        });
    }
    // The derived tenant: the child's tenant column is projected from the (scope-confined) parent's
    // tenant column — never a guest value.
    columns.push(child_tenant);
    projection.push(SelectItem {
        expr: Expr::Column(parent_tenant),
        alias: None,
    });
    // Under a target scope, force the child's public-visibility columns so the inserted row is itself
    // public (deny-by-default: a child table with no declared public subset is refused).
    if is_target {
        for (col, val) in scope.public_force_cells(&spec.child)? {
            columns.push(col);
            projection.push(SelectItem {
                expr: Expr::Value(val),
                alias: None,
            });
        }
    }
    // The source: SELECT <projection> FROM parent WHERE parent.<ref> = ?. Read-scoping it confines the
    // parent to the caller's reachable set (own: tenant = A [OR NULL]; target: tenant = B AND public),
    // so the projected parent tenant is bounded and an unreachable parent yields zero rows.
    let mut source = Select {
        columns: projection,
        filter: Some(Predicate::Cmp {
            left: Expr::Column(spec.ref_column.clone()),
            op: CmpOp::Eq,
            right: Expr::Value(spec.ref_value.clone()),
        }),
        ..Select::from(spec.parent.clone())
    };
    source.force_scope(scope)?;
    // The INSERT itself carries NO scope (`scope: None`) — the tenant is the projected parent's, not a
    // re-stamped scalar. (This is the sanctioned target INSERT…SELECT; a generic one is refused by
    // `Insert::force_scope` under a target scope.)
    let insert = Insert {
        table: spec.child.clone(),
        rows: vec![],
        conflict: None,
        scope: None,
        returning: vec![],
        from_select: Some((columns, Box::new(source))),
    };
    insert.compile(dialect)
}

#[cfg(test)]
mod rls_scope_value_tests {
    use super::*;

    fn v(s: &str) -> SqlValue {
        SqlValue::Text(s.to_string())
    }
    fn cell(col: &str, val: SqlValue) -> Assignment {
        Assignment {
            column: col.into(),
            value: Expr::Value(val),
        }
    }
    fn insert(rows: Vec<Vec<Assignment>>) -> Insert {
        Insert {
            table: "t".into(),
            rows: rows.into_iter().map(|cells| RowValues { cells }).collect(),
            conflict: None,
            scope: None,
            returning: vec![],
            from_select: None,
        }
    }

    #[test]
    fn insert_uniform_scope_value_extracts_only_a_single_declared_tenant() {
        // Single row → the literal.
        assert_eq!(
            insert(vec![vec![cell("tenant_id", v("A")), cell("body", v("x"))]])
                .uniform_scope_value("tenant_id"),
            Some(v("A"))
        );
        // Multi-row agreeing → the shared literal.
        assert_eq!(
            insert(vec![
                vec![cell("tenant_id", v("A")), cell("body", v("x"))],
                vec![cell("tenant_id", v("A")), cell("body", v("y"))],
            ])
            .uniform_scope_value("tenant_id"),
            Some(v("A"))
        );
        // Rows disagree → None (fail-closed).
        assert_eq!(
            insert(vec![
                vec![cell("tenant_id", v("A"))],
                vec![cell("tenant_id", v("B"))],
            ])
            .uniform_scope_value("tenant_id"),
            None
        );
        // Missing column / non-literal → None.
        assert_eq!(
            insert(vec![vec![cell("body", v("x"))]]).uniform_scope_value("tenant_id"),
            None
        );
        assert_eq!(
            insert(vec![vec![cell("tenant_id", v("A")), cell("body", v("x"))]])
                .uniform_scope_value("id"),
            None
        );
        // Non-literal value → None.
        let non_lit = insert(vec![vec![Assignment {
            column: "tenant_id".into(),
            value: Expr::col("other"),
        }]]);
        assert_eq!(non_lit.uniform_scope_value("tenant_id"), None);
    }

    #[test]
    fn update_pinned_scope_value_extracts_only_a_pinned_single_tenant() {
        let upd = |filter: Predicate| Update {
            table: "t".into(),
            set: vec![cell("body", v("x"))],
            filter,
            scope: None,
            returning: vec![],
        };
        // `col = 'A'` → A.
        assert_eq!(
            upd(Predicate::Cmp {
                left: Expr::col("tenant_id"),
                op: CmpOp::Eq,
                right: Expr::Value(v("A")),
            })
            .pinned_scope_value("tenant_id"),
            Some(v("A"))
        );
        // AND with an unrelated conjunct still pins.
        assert_eq!(
            upd(Predicate::And(vec![
                Predicate::Cmp {
                    left: Expr::col("tenant_id"),
                    op: CmpOp::Eq,
                    right: Expr::Value(v("A")),
                },
                Predicate::Cmp {
                    left: Expr::col("active"),
                    op: CmpOp::Eq,
                    right: Expr::Value(SqlValue::Boolean(true)),
                },
            ]))
            .pinned_scope_value("tenant_id"),
            Some(v("A"))
        );
        // OR does not pin → None (fail-closed).
        assert_eq!(
            upd(Predicate::Or(vec![
                Predicate::Cmp {
                    left: Expr::col("tenant_id"),
                    op: CmpOp::Eq,
                    right: Expr::Value(v("A")),
                },
                Predicate::Cmp {
                    left: Expr::col("tenant_id"),
                    op: CmpOp::Eq,
                    right: Expr::Value(v("B")),
                },
            ]))
            .pinned_scope_value("tenant_id"),
            None
        );
        // No equality on the column → None.
        assert_eq!(
            upd(Predicate::Cmp {
                left: Expr::col("active"),
                op: CmpOp::Eq,
                right: Expr::Value(SqlValue::Boolean(true)),
            })
            .pinned_scope_value("tenant_id"),
            None
        );
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn t(s: &str) -> SqlValue {
        SqlValue::Text(s.to_string())
    }
    fn cmp(col: &str, op: CmpOp, v: SqlValue) -> Predicate {
        Predicate::Cmp {
            left: Expr::Column(col.into()),
            op,
            right: Expr::Value(v),
        }
    }
    fn item(e: Expr) -> SelectItem {
        SelectItem {
            expr: e,
            alias: None,
        }
    }

    #[test]
    fn select_basic_where_order_limit() {
        let q = Select {
            columns: vec![item(Expr::col("id")), item(Expr::col("state"))],
            filter: Some(cmp("project_id", CmpOp::Eq, t("prj_1"))),
            order: vec![OrderBy {
                expr: Expr::col("created_at"),
                dir: Direction::Desc,
            }],
            limit: Some(10),
            ..Select::from("work_order")
        };
        let (sql, params) = q.compile(Dialect::Sqlite).unwrap();
        assert_eq!(
            sql,
            "SELECT id, state FROM work_order WHERE project_id = ?1 ORDER BY created_at DESC LIMIT 10"
        );
        assert_eq!(params, vec![t("prj_1")]);
    }

    #[test]
    fn scope_is_anded_and_bound_first() {
        let q = Select {
            filter: Some(cmp("kind", CmpOp::Eq, t("supplier"))),
            scope: Some(Scope {
                column: "tenant_id".into(),
                value: Some(t("ten_1")),
                session: None,
                mode: ScopeMode::Own,
                keys: TableKeys::Uniform,
            }),
            ..Select::from("party")
        };
        let (sql, params) = q.compile(Dialect::Sqlite).unwrap();
        assert_eq!(
            sql,
            "SELECT * FROM party WHERE tenant_id = ?1 AND kind = ?2"
        );
        assert_eq!(params, vec![t("ten_1"), t("supplier")]);
    }

    #[test]
    fn per_table_keys_scope_each_ref_on_its_own_column() {
        use std::collections::BTreeMap;
        // A settings-page read: storefront_config (Tenant -> tenant_id) LEFT JOIN the identity table
        // `tenant` (TenantKeyed -> its own PK `id`). The host injects the RIGHT column per ref (R2).
        let q = Select {
            table_alias: Some("sc".into()),
            joins: vec![Join {
                kind: JoinKind::Left,
                table: "tenant".into(),
                alias: Some("t".into()),
                on: Predicate::Cmp {
                    left: Expr::col("sc.tenant_id"),
                    op: CmpOp::Eq,
                    right: Expr::col("t.id"),
                },
            }],
            scope: Some(Scope {
                column: "tenant_id".into(),
                value: Some(t("acme")),
                session: None,
                mode: ScopeMode::Own,
                keys: TableKeys::PerTable(BTreeMap::from([
                    (
                        "storefront_config".to_string(),
                        ResolvedScope::Column("tenant_id".to_string()),
                    ),
                    (
                        "tenant".to_string(),
                        ResolvedScope::Column("id".to_string()),
                    ),
                ])),
            }),
            ..Select::from("storefront_config")
        };
        let (sql, params) = q.compile(Dialect::Sqlite).unwrap();
        assert!(
            sql.contains("sc.tenant_id = ?"),
            "base scoped on tenant_id: {sql}"
        );
        assert!(
            sql.contains("t.id = ?"),
            "identity table scoped on its own PK: {sql}"
        );
        assert_eq!(params, vec![t("acme"), t("acme")]);

        // An `Unscoped` join (reference data) adds NO tenant predicate; the base still scopes.
        let mut q2 = Select {
            table_alias: Some("sc".into()),
            joins: vec![Join {
                kind: JoinKind::Left,
                table: "countries".into(),
                alias: Some("c".into()),
                on: Predicate::Cmp {
                    left: Expr::col("sc.country"),
                    op: CmpOp::Eq,
                    right: Expr::col("c.code"),
                },
            }],
            ..Select::from("storefront_config")
        };
        q2.force_scope(&Scope {
            column: "tenant_id".into(),
            value: Some(t("acme")),
            session: None,
            mode: ScopeMode::Own,
            keys: TableKeys::PerTable(BTreeMap::from([
                (
                    "storefront_config".to_string(),
                    ResolvedScope::Column("tenant_id".to_string()),
                ),
                ("countries".to_string(), ResolvedScope::Unscoped),
            ])),
        })
        .unwrap();
        let (sql2, params2) = q2.compile(Dialect::Sqlite).unwrap();
        assert!(sql2.contains("sc.tenant_id = ?"), "sql2: {sql2}");
        // The `Unscoped` join binds NO tenant value — the sole bind is the base's own tenant — which
        // proves `countries` contributed no scope predicate (a substring check on the alias would
        // false-match `sc.tenant_id`).
        assert_eq!(
            params2,
            vec![t("acme")],
            "unscoped join adds no tenant predicate: {sql2}"
        );

        // An UNDECLARED table under a present schema is refused (deny-by-default, D3).
        let mut q3 = Select::from("secret_table");
        q3.force_scope(&Scope {
            column: "tenant_id".into(),
            value: Some(t("acme")),
            session: None,
            mode: ScopeMode::Own,
            keys: TableKeys::PerTable(BTreeMap::from([(
                "orders".to_string(),
                ResolvedScope::Column("tenant_id".to_string()),
            )])),
        })
        .unwrap();
        assert!(matches!(
            q3.compile(Dialect::Sqlite),
            Err(OrmError::TenancyUndeclared(tbl)) if tbl == "secret_table"
        ));
    }

    #[test]
    fn is_own_lowers_to_a_case_rank_and_orders_own_first() {
        // The base-vs-override read: `own+null` + `ORDER BY is_own DESC LIMIT 1` — the tenant's
        // override (own) sorts ahead of the shared base (NULL), without the guest naming tenant_id.
        let mut q = Select {
            columns: vec![item(Expr::col("body"))],
            filter: Some(cmp("key_name", CmpOp::Eq, t("k"))),
            order: vec![OrderBy {
                expr: Expr::IsOwn,
                dir: Direction::Desc,
            }],
            limit: Some(1),
            ..Select::from("knowledge_entry")
        };
        q.force_scope(&Scope {
            column: "tenant_id".into(),
            value: Some(t("acme")),
            session: None,
            mode: ScopeMode::OwnOrNull,
            keys: TableKeys::Uniform,
        })
        .unwrap();
        let (sql, params) = q.compile(Dialect::Sqlite).unwrap();
        assert_eq!(
            sql,
            "SELECT body FROM knowledge_entry WHERE (tenant_id = ?1 OR tenant_id IS NULL) \
             AND key_name = ?2 ORDER BY (CASE WHEN tenant_id IS NOT NULL AND tenant_id = ?3 \
             THEN ?4 ELSE ?5 END) DESC LIMIT 1"
        );
        // Scope value (own+null), filter, then the is_own rank (own value + 1/0), in textual order.
        assert_eq!(
            params,
            vec![
                t("acme"),
                t("k"),
                t("acme"),
                SqlValue::Integer(1),
                SqlValue::Integer(0)
            ]
        );
    }

    #[test]
    fn is_own_in_select_under_all_uses_the_resolved_own_value() {
        // Under a cross-tenant `all` read is_own means "MY own" (col = <own>), not "any non-base".
        let mut q = Select {
            columns: vec![item(Expr::IsOwn)],
            ..Select::from("t")
        };
        q.force_scope(&Scope {
            column: "tenant_id".into(),
            value: Some(t("acme")),
            session: None,
            mode: ScopeMode::All,
            keys: TableKeys::Uniform,
        })
        .unwrap();
        let (sql, params) = q.compile(Dialect::Sqlite).unwrap();
        assert_eq!(
            sql,
            "SELECT (CASE WHEN tenant_id IS NOT NULL AND tenant_id = ?1 THEN ?2 ELSE ?3 END) FROM t"
        );
        assert_eq!(
            params,
            vec![t("acme"), SqlValue::Integer(1), SqlValue::Integer(0)]
        );
    }

    #[test]
    fn is_own_without_a_scope_is_rejected() {
        // No force_scope ⇒ the marker is never lowered ⇒ fail closed at compile (never an unscoped
        // ranking that could leak whether other tenants exist).
        let q = Select {
            order: vec![OrderBy {
                expr: Expr::IsOwn,
                dir: Direction::Desc,
            }],
            ..Select::from("t")
        };
        let err = q.compile(Dialect::Sqlite).unwrap_err();
        assert!(
            matches!(err, OrmError::BadExpr(m) if m.contains("is_own")),
            "expected a fail-closed is_own error, got {err:?}"
        );
    }

    #[test]
    fn is_own_in_a_filter_does_not_subtract_the_scope_predicate() {
        // Using is_own() as a label in WHERE (`WHERE is_own() = 1`, "only my overrides") must keep
        // the independent host tenant predicate — the label can never remove a scope conjunct.
        let mut q = Select {
            filter: Some(Predicate::Cmp {
                left: Expr::IsOwn,
                op: CmpOp::Eq,
                right: Expr::Value(SqlValue::Integer(1)),
            }),
            ..Select::from("notes")
        };
        q.force_scope(&Scope {
            column: "tenant_id".into(),
            value: Some(t("acme")),
            session: None,
            mode: ScopeMode::OwnOrNull,
            keys: TableKeys::Uniform,
        })
        .unwrap();
        let (sql, _) = q.compile(Dialect::Sqlite).unwrap();
        // The host scope predicate is conjoined in FRONT, independent of the is_own label.
        assert!(
            sql.contains("(tenant_id = ?1 OR tenant_id IS NULL) AND"),
            "scope predicate must survive the is_own filter: {sql}"
        );
        assert!(
            sql.contains("CASE WHEN tenant_id IS NOT NULL AND tenant_id = ?2 THEN"),
            "is_own lowered to the own-rank CASE: {sql}"
        );
    }

    fn scoped_select(mode: ScopeMode) -> Select {
        Select {
            filter: Some(cmp("kind", CmpOp::Eq, t("supplier"))),
            scope: Some(Scope {
                column: "tenant_id".into(),
                value: Some(t("ten_1")),
                session: None,
                mode,
                keys: TableKeys::Uniform,
            }),
            ..Select::from("party")
        }
    }

    #[test]
    fn scope_mode_own_or_null_admits_the_shared_baseline() {
        let (sql, params) = scoped_select(ScopeMode::OwnOrNull)
            .compile(Dialect::Sqlite)
            .unwrap();
        assert_eq!(
            sql,
            "SELECT * FROM party WHERE (tenant_id = ?1 OR tenant_id IS NULL) AND kind = ?2"
        );
        assert_eq!(params, vec![t("ten_1"), t("supplier")]);
    }

    #[test]
    fn scope_mode_null_only_sees_only_the_baseline() {
        let (sql, params) = scoped_select(ScopeMode::NullOnly)
            .compile(Dialect::Sqlite)
            .unwrap();
        // The resolved tenant value is not bound at all — NULL-only never references it.
        assert_eq!(
            sql,
            "SELECT * FROM party WHERE tenant_id IS NULL AND kind = ?1"
        );
        assert_eq!(params, vec![t("supplier")]);
    }

    #[test]
    fn scope_mode_all_injects_no_tenant_predicate() {
        let (sql, params) = scoped_select(ScopeMode::All)
            .compile(Dialect::Sqlite)
            .unwrap();
        // `all` (cross-tenant) renders exactly as if unscoped — only the guest filter remains.
        assert_eq!(sql, "SELECT * FROM party WHERE kind = ?1");
        assert_eq!(params, vec![t("supplier")]);
    }

    #[test]
    fn force_scope_reaches_every_union_branch() {
        // A union whose branches start unscoped: force_scope must scope BOTH sides, or the
        // branch would leak across tenants.
        let branch = Select::from("archived_party");
        let mut q = Select {
            union: Some(Box::new(Union {
                all: false,
                query: branch,
            })),
            ..Select::from("party")
        };
        q.force_scope(&Scope {
            column: "tenant_id".into(),
            value: Some(t("ten_1")),
            session: None,
            mode: ScopeMode::Own,
            keys: TableKeys::Uniform,
        })
        .unwrap();
        let (sql, params) = q.compile(Dialect::Sqlite).unwrap();
        assert_eq!(
            sql,
            "SELECT * FROM party WHERE tenant_id = ?1 UNION SELECT * FROM archived_party WHERE tenant_id = ?2"
        );
        assert_eq!(params, vec![t("ten_1"), t("ten_1")]);
    }

    #[test]
    fn scoped_select_scopes_every_joined_table() {
        // A guest joins a victim table hoping to read its cross-tenant rows via the projection.
        // force_scope must scope the FROM table AND every joined table (qualified by alias/name).
        // A LEFT-joined victim is confined in its OWN `ON` (not the WHERE — that would collapse the
        // LEFT JOIN to an INNER JOIN): a cross-tenant victim row then fails the ON and yields NULL,
        // never leaking through the projection.
        let mut q = Select {
            table: "orders".into(),
            table_alias: Some("o".into()),
            columns: vec![item(Expr::col("v.secret"))],
            joins: vec![Join {
                kind: JoinKind::Left,
                table: "victim".into(),
                alias: Some("v".into()),
                on: Predicate::Cmp {
                    left: Expr::col("v.order_id"),
                    op: CmpOp::Eq,
                    right: Expr::col("o.id"),
                },
            }],
            ..Select::from("orders")
        };
        q.force_scope(&Scope {
            column: "tenant_id".into(),
            value: Some(t("ten_1")),
            session: None,
            mode: ScopeMode::Own,
            keys: TableKeys::Uniform,
        })
        .unwrap();
        let (sql, params) = q.compile(Dialect::Sqlite).unwrap();
        assert_eq!(
            sql,
            "SELECT v.secret FROM orders AS o LEFT JOIN victim AS v ON v.order_id = o.id \
             AND v.tenant_id = ?1 WHERE o.tenant_id = ?2"
        );
        assert_eq!(params, vec![t("ten_1"), t("ten_1")]);
    }

    #[test]
    fn scoped_returning_and_distinct_on_subqueries_are_scoped() {
        let sub = || Expr::RelatedScalar {
            column: "balance".into(),
            table: "victim".into(),
            filter: Box::new(Predicate::And(Vec::new())),
        };
        let scope = Scope {
            column: "tenant_id".into(),
            value: Some(t("ten_1")),
            session: None,
            mode: ScopeMode::Own,
            keys: TableKeys::Uniform,
        };
        // DELETE … RETURNING (subquery) — the RETURNING read must be scoped to victim.
        let mut del = Delete {
            table: "orders".into(),
            filter: cmp("id", CmpOp::Eq, t("o_1")),
            scope: None,
            returning: vec![item(sub())],
        };
        del.force_scope(&scope).unwrap();
        let (sql, _) = del.compile(Dialect::Sqlite).unwrap();
        assert!(
            sql.contains("RETURNING (SELECT balance FROM victim WHERE victim.tenant_id = ?"),
            "RETURNING subquery unscoped: {sql}"
        );
        // SELECT DISTINCT ON ((subquery)) — the DISTINCT ON read must be scoped too (PG).
        let mut sel = Select {
            columns: vec![item(Expr::col("id"))],
            distinct_on: vec![sub()],
            ..Select::from("orders")
        };
        sel.force_scope(&scope).unwrap();
        // The compiler emits portable `?N` placeholders (the backend rewrites to `$N` on PG).
        let (sql, _) = sel.compile(Dialect::Postgres).unwrap();
        assert!(
            sql.contains("DISTINCT ON ((SELECT balance FROM victim WHERE victim.tenant_id = ?"),
            "DISTINCT ON subquery unscoped: {sql}"
        );
    }

    #[test]
    fn scoped_select_scopes_a_subquerys_inner_table() {
        // A guest embeds a scalar subquery over another table; force_scope must scope the
        // subquery's OWN table so it can't read cross-tenant.
        let mut q = Select {
            columns: vec![item(Expr::RelatedScalar {
                column: "balance".into(),
                table: "victim".into(),
                filter: Box::new(Predicate::And(Vec::new())), // guest filter: none
            })],
            ..Select::from("orders")
        };
        q.force_scope(&Scope {
            column: "tenant_id".into(),
            value: Some(t("ten_1")),
            session: None,
            mode: ScopeMode::Own,
            keys: TableKeys::Uniform,
        })
        .unwrap();
        let (sql, params) = q.compile(Dialect::Sqlite).unwrap();
        // The subquery's WHERE is scoped to victim.tenant_id; the outer to orders (single table).
        assert_eq!(
            sql,
            "SELECT (SELECT balance FROM victim WHERE victim.tenant_id = ?1) \
             FROM orders WHERE tenant_id = ?2"
        );
        assert_eq!(params, vec![t("ten_1"), t("ten_1")]);
    }

    #[test]
    fn insert_select_cannot_forge_the_target_tenant() {
        // A guest projects a chosen tenant id into the target `tenant_id` column. force_scope must
        // drop that projection and re-bind the host-resolved own value — no cross-tenant forgery.
        let source = Select {
            columns: vec![
                item(Expr::val(t("VICTIM"))), // guest-chosen tenant id
                item(Expr::col("total")),
            ],
            ..Select::from("orders")
        };
        let mut ins = Insert {
            table: "orders".into(),
            rows: vec![],
            conflict: None,
            scope: None,
            returning: vec![],
            // `TENANT_ID` (case variant) must still be recognized as the tenant column + dropped.
            from_select: Some((vec!["TENANT_ID".into(), "total".into()], Box::new(source))),
        };
        let own = Scope {
            column: "tenant_id".into(),
            value: Some(t("OWN")),
            session: None,
            mode: ScopeMode::Own,
            keys: TableKeys::Uniform,
        };
        ins.force_scope(Some(&own), Some(&own)).unwrap();
        let (sql, params) = ins.compile(Dialect::Sqlite).unwrap();
        // The tenant column is re-appended last, bound to OWN; the source is read-scoped to OWN.
        assert_eq!(
            sql,
            "INSERT INTO orders (total, tenant_id) SELECT total, ?1 FROM orders WHERE tenant_id = ?2"
        );
        assert_eq!(params, vec![t("OWN"), t("OWN")]);
        assert!(
            !params.contains(&t("VICTIM")),
            "the forged tenant never binds"
        );
    }

    #[test]
    fn scoped_update_cannot_reassign_the_tenant() {
        // A guest tries to donate its own rows to another tenant: SET tenant_id = VICTIM. The
        // scope guard drops that assignment (case-insensitively) while the WHERE stays own-bound.
        let q = Update {
            table: "orders".into(),
            set: vec![
                Assignment {
                    column: "TENANT_ID".into(),
                    value: Expr::val(t("VICTIM")),
                },
                Assignment {
                    column: "status".into(),
                    value: Expr::val(t("paid")),
                },
            ],
            filter: cmp("id", CmpOp::Eq, t("o_1")),
            scope: Some(Scope {
                column: "tenant_id".into(),
                value: Some(t("OWN")),
                session: None,
                mode: ScopeMode::Own,
                keys: TableKeys::Uniform,
            }),
            returning: vec![],
        };
        let (sql, params) = q.compile(Dialect::Sqlite).unwrap();
        assert_eq!(
            sql,
            "UPDATE orders SET status = ?1 WHERE tenant_id = ?2 AND id = ?3"
        );
        assert_eq!(params, vec![t("paid"), t("OWN"), t("o_1")]);
        assert!(!params.contains(&t("VICTIM")));
    }

    #[test]
    fn scoped_upsert_drops_tenant_reassignment_and_bounds_the_do_update() {
        // A guest upsert tries to (a) reassign tenant_id to VICTIM on conflict and (b) overwrite
        // another tenant's row via a conflict on a non-tenant key. The scope guard must drop the
        // tenant reassignment and bound the DO UPDATE to own rows.
        let mut ins = Insert {
            table: "orders".into(),
            rows: vec![RowValues {
                cells: vec![Assignment {
                    column: "id".into(),
                    value: Expr::val(t("k")),
                }],
            }],
            conflict: Some(OnConflict {
                conflict_columns: vec!["id".into()],
                update: vec![
                    // Case/qualifier-respelled to dodge the drop — must still be caught.
                    Assignment {
                        column: "TENANT_ID".into(),
                        value: Expr::val(t("VICTIM")),
                    },
                    Assignment {
                        column: "total".into(),
                        value: Expr::val(SqlValue::Integer(999)),
                    },
                ],
            }),
            scope: None,
            returning: vec![],
            from_select: None,
        };
        let own = Scope {
            column: "tenant_id".into(),
            value: Some(t("OWN")),
            session: None,
            mode: ScopeMode::Own,
            keys: TableKeys::Uniform,
        };
        ins.force_scope(Some(&own), Some(&own)).unwrap();
        let (sql, params) = ins.compile(Dialect::Sqlite).unwrap();
        assert_eq!(
            sql,
            "INSERT INTO orders (id, tenant_id) VALUES (?1, ?2) \
             ON CONFLICT (id) DO UPDATE SET total = ?3 WHERE orders.tenant_id = ?4"
        );
        // The inserted row stamps OWN; the DO UPDATE is bounded to OWN; VICTIM never binds.
        assert_eq!(
            params,
            vec![t("k"), t("OWN"), SqlValue::Integer(999), t("OWN")]
        );
        assert!(!params.contains(&t("VICTIM")));
        // The same scoped upsert is refused on MySQL (no bounded DO UPDATE).
        assert!(matches!(
            ins.compile(Dialect::Mysql),
            Err(OrmError::BadExpr(_))
        ));
    }

    #[test]
    fn upsert_do_update_guard_is_target_table_qualified() {
        // The host-injected DO UPDATE partition guard names `<table>.<col>`, never a bare column:
        // inside `DO UPDATE` the target table AND the `excluded` pseudo-relation both expose the
        // tenant column, so a bare guard is ambiguous on Postgres and the whole upsert fails at
        // execution (construens' P48 cutover bug — reproduced on real PG 16). SQLite tolerates the
        // bare form, which is why this only surfaced on a Postgres backend.
        let build = |mode, value: Option<SqlValue>| {
            let mut ins = Insert {
                table: "module_config".into(),
                rows: vec![RowValues {
                    cells: vec![Assignment {
                        column: "module".into(),
                        value: Expr::val(t("m")),
                    }],
                }],
                // A NON-tenant conflict key: the guard is LOAD-BEARING here — it is the only thing
                // stopping a guest upsert from overwriting another tenant's row via the shared key,
                // so the fix must qualify it, not drop it.
                conflict: Some(OnConflict {
                    conflict_columns: vec!["module".into()],
                    update: vec![Assignment {
                        column: "enabled".into(),
                        value: Expr::col("excluded.enabled"),
                    }],
                }),
                scope: None,
                returning: vec![],
                from_select: None,
            };
            let s = Scope {
                column: "tenant_id".into(),
                value,
                session: None,
                mode,
                keys: TableKeys::Uniform,
            };
            ins.force_scope(Some(&s), Some(&s)).unwrap();
            ins
        };
        // own → `= value`, target-qualified — on BOTH the real (Postgres) backend and SQLite.
        for d in [Dialect::Postgres, Dialect::Sqlite] {
            let (sql, _) = build(ScopeMode::Own, Some(t("OWN"))).compile(d).unwrap();
            assert!(
                sql.ends_with(
                    "ON CONFLICT (module) DO UPDATE SET enabled = excluded.enabled \
                     WHERE module_config.tenant_id = ?3"
                ),
                "{d:?}: {sql}"
            );
        }
        // null baseline → `IS NULL`, also target-qualified (the identical ambiguity).
        let (sql, _) = build(ScopeMode::NullOnly, None)
            .compile(Dialect::Postgres)
            .unwrap();
        assert!(
            sql.ends_with(
                "ON CONFLICT (module) DO UPDATE SET enabled = excluded.enabled \
                 WHERE module_config.tenant_id IS NULL"
            ),
            "{sql}"
        );
    }

    #[test]
    fn insert_null_mode_stamps_null_all_mode_stamps_nothing() {
        let base = |mode| Insert {
            table: "audit_event".into(),
            rows: vec![RowValues {
                cells: vec![Assignment {
                    column: "detail".into(),
                    value: Expr::val(t("x")),
                }],
            }],
            conflict: None,
            scope: Some(Scope {
                column: "tenant_id".into(),
                value: Some(t("ten_1")),
                session: None,
                mode,
                keys: TableKeys::Uniform,
            }),
            returning: vec![],
            from_select: None,
        };
        // null-only write stamps NULL into the tenant column.
        let (sql, params) = base(ScopeMode::NullOnly).compile(Dialect::Sqlite).unwrap();
        assert_eq!(
            sql,
            "INSERT INTO audit_event (detail, tenant_id) VALUES (?1, ?2)"
        );
        assert_eq!(params, vec![t("x"), SqlValue::Null]);
        // all-mode write forces no tenant column — the guest's columns stand verbatim.
        let (sql, params) = base(ScopeMode::All).compile(Dialect::Sqlite).unwrap();
        assert_eq!(sql, "INSERT INTO audit_event (detail) VALUES (?1)");
        assert_eq!(params, vec![t("x")]);
    }

    #[test]
    fn nested_and_or_not_is_parenthesized() {
        // scope AND (state IN (..) AND (priority >= ? OR escalated = ?) AND NOT archived)
        let q = Select {
            filter: Some(all([
                Predicate::In {
                    expr: Expr::col("state"),
                    values: vec![Expr::val(t("po_linked")), Expr::val(t("awarded"))],
                    negated: false,
                },
                any([
                    cmp("priority", CmpOp::Ge, SqlValue::Integer(3)),
                    cmp("escalated", CmpOp::Eq, SqlValue::Boolean(true)),
                ]),
                Predicate::Not(Box::new(cmp(
                    "archived",
                    CmpOp::Eq,
                    SqlValue::Boolean(true),
                ))),
            ])),
            scope: Some(Scope {
                column: "tenant_id".into(),
                value: Some(t("ten_1")),
                session: None,
                mode: ScopeMode::Own,
                keys: TableKeys::Uniform,
            }),
            ..Select::from("order_to_network")
        };
        let (sql, params) = q.compile(Dialect::Sqlite).unwrap();
        assert_eq!(
            sql,
            "SELECT * FROM order_to_network WHERE tenant_id = ?1 AND (state IN (?2, ?3) AND (priority >= ?4 OR escalated = ?5) AND NOT archived = ?6)"
        );
        assert_eq!(
            params,
            vec![
                t("ten_1"),
                t("po_linked"),
                t("awarded"),
                SqlValue::Integer(3),
                SqlValue::Boolean(true),
                SqlValue::Boolean(true)
            ]
        );
    }

    #[test]
    fn group_by_having_with_aggregate_and_alias() {
        let q = Select {
            columns: vec![
                item(Expr::col("network_id")),
                SelectItem {
                    expr: Expr::Aggregate(Agg::Sum, Box::new(Expr::col("committed_minor"))),
                    alias: Some("total".into()),
                },
            ],
            group_by: vec![Expr::col("network_id")],
            having: Some(Predicate::Cmp {
                left: Expr::Aggregate(Agg::Sum, Box::new(Expr::col("committed_minor"))),
                op: CmpOp::Gt,
                right: Expr::val(SqlValue::Integer(1000)),
            }),
            order: vec![OrderBy {
                expr: Expr::col("total"),
                dir: Direction::Desc,
            }],
            ..Select::from("order_to_network")
        };
        let (sql, params) = q.compile(Dialect::Sqlite).unwrap();
        assert_eq!(
            sql,
            "SELECT network_id, sum(committed_minor) AS total FROM order_to_network GROUP BY network_id HAVING sum(committed_minor) > ?1 ORDER BY total DESC"
        );
        assert_eq!(params, vec![SqlValue::Integer(1000)]);
    }

    #[test]
    fn join_with_alias_and_column_ref_condition() {
        let q = Select {
            columns: vec![item(Expr::Aggregate(Agg::Count, Box::new(Expr::Star)))],
            joins: vec![Join {
                kind: JoinKind::Inner,
                table: "element".into(),
                alias: Some("e".into()),
                on: Predicate::Cmp {
                    left: Expr::col("order_to_network.element_id"),
                    op: CmpOp::Eq,
                    right: Expr::col("e.id"),
                },
            }],
            filter: Some(cmp("order_id", CmpOp::Eq, SqlValue::Integer(7))),
            ..Select::from("order_to_network")
        };
        let (sql, _) = q.compile(Dialect::Sqlite).unwrap();
        assert_eq!(
            sql,
            "SELECT count(*) FROM order_to_network JOIN element AS e ON order_to_network.element_id = e.id WHERE order_id = ?1"
        );
    }

    #[test]
    fn between_like_insensitive_and_notin() {
        let q = Select {
            filter: Some(all([
                Predicate::Between {
                    expr: Expr::col("amount"),
                    low: Expr::val(SqlValue::Integer(10)),
                    high: Expr::val(SqlValue::Integer(20)),
                    negated: false,
                },
                Predicate::Like {
                    expr: Expr::col("name"),
                    pattern: "ac%".into(),
                    insensitive: true,
                    negated: false,
                },
                Predicate::In {
                    expr: Expr::col("state"),
                    values: vec![Expr::val(t("void"))],
                    negated: true,
                },
            ])),
            ..Select::from("invoice")
        };
        let (sql, params) = q.compile(Dialect::Sqlite).unwrap();
        assert_eq!(
            sql,
            "SELECT * FROM invoice WHERE amount BETWEEN ?1 AND ?2 AND lower(name) LIKE lower(?3) AND state NOT IN (?4)"
        );
        assert_eq!(
            params,
            vec![
                SqlValue::Integer(10),
                SqlValue::Integer(20),
                t("ac%"),
                t("void")
            ]
        );
    }

    #[test]
    fn arithmetic_and_functions_in_select_and_set() {
        let q = Select {
            columns: vec![
                SelectItem {
                    expr: Expr::Func(Func::Lower, vec![Expr::col("email")]),
                    alias: Some("email_lc".into()),
                },
                item(Expr::Binary(
                    BinOp::Mul,
                    Box::new(Expr::col("qty")),
                    Box::new(Expr::val(SqlValue::Integer(2))),
                )),
                item(Expr::Func(
                    Func::Coalesce,
                    vec![Expr::col("nickname"), Expr::val(t("n/a"))],
                )),
            ],
            ..Select::from("account")
        };
        let (sql, params) = q.compile(Dialect::Sqlite).unwrap();
        assert_eq!(
            sql,
            "SELECT lower(email) AS email_lc, (qty * ?1), coalesce(nickname, ?2) FROM account"
        );
        assert_eq!(params, vec![SqlValue::Integer(2), t("n/a")]);
    }

    #[test]
    fn empty_in_and_not_in_are_identities() {
        let matches_none = Select {
            filter: Some(Predicate::In {
                expr: Expr::col("x"),
                values: vec![],
                negated: false,
            }),
            ..Select::from("t")
        };
        assert_eq!(
            matches_none.compile(Dialect::Sqlite).unwrap().0,
            "SELECT * FROM t WHERE 1 = 0"
        );
        let matches_all = Select {
            filter: Some(Predicate::In {
                expr: Expr::col("x"),
                values: vec![],
                negated: true,
            }),
            ..Select::from("t")
        };
        assert_eq!(
            matches_all.compile(Dialect::Sqlite).unwrap().0,
            "SELECT * FROM t WHERE 1 = 1"
        );
    }

    #[test]
    fn insert_with_scope_and_returning() {
        let q = Insert {
            table: "work_area".into(),
            rows: vec![RowValues {
                cells: vec![
                    Assignment {
                        column: "id".into(),
                        value: Expr::val(t("wa_1")),
                    },
                    Assignment {
                        column: "project_id".into(),
                        value: Expr::val(t("prj_1")),
                    },
                ],
            }],
            conflict: None,
            scope: Some(Scope {
                column: "tenant_id".into(),
                value: Some(t("ten_1")),
                session: None,
                mode: ScopeMode::Own,
                keys: TableKeys::Uniform,
            }),
            returning: vec![item(Expr::col("id"))],
            from_select: None,
        };
        let (sql, params) = q.compile(Dialect::Sqlite).unwrap();
        assert_eq!(
            sql,
            "INSERT INTO work_area (id, project_id, tenant_id) VALUES (?1, ?2, ?3) RETURNING id"
        );
        assert_eq!(params, vec![t("wa_1"), t("prj_1"), t("ten_1")]);
    }

    #[test]
    fn upsert_do_update_and_do_nothing() {
        let base = |update: Vec<Assignment>| Insert {
            table: "country_pack".into(),
            rows: vec![RowValues {
                cells: vec![
                    Assignment {
                        column: "country".into(),
                        value: Expr::val(t("US")),
                    },
                    Assignment {
                        column: "currency".into(),
                        value: Expr::val(t("USD")),
                    },
                ],
            }],
            conflict: Some(OnConflict {
                conflict_columns: vec!["tenant_id".into(), "country".into()],
                update,
            }),
            scope: None,
            returning: vec![],
            from_select: None,
        };
        let (sql_do, _) = base(vec![Assignment {
            column: "currency".into(),
            value: Expr::val(t("USD")),
        }])
        .compile(Dialect::Sqlite)
        .unwrap();
        assert_eq!(
            sql_do,
            "INSERT INTO country_pack (country, currency) VALUES (?1, ?2) ON CONFLICT (tenant_id, country) DO UPDATE SET currency = ?3"
        );
        let (sql_nothing, _) = base(vec![]).compile(Dialect::Sqlite).unwrap();
        assert_eq!(
            sql_nothing,
            "INSERT INTO country_pack (country, currency) VALUES (?1, ?2) ON CONFLICT (tenant_id, country) DO NOTHING"
        );
    }

    #[test]
    fn update_binds_set_before_where_and_supports_expr_set() {
        let q = Update {
            table: "counter".into(),
            set: vec![Assignment {
                column: "hits".into(),
                value: Expr::Binary(
                    BinOp::Add,
                    Box::new(Expr::col("hits")),
                    Box::new(Expr::val(SqlValue::Integer(1))),
                ),
            }],
            filter: cmp("id", CmpOp::Eq, t("c_1")),
            scope: Some(Scope {
                column: "tenant_id".into(),
                value: Some(t("ten_1")),
                session: None,
                mode: ScopeMode::Own,
                keys: TableKeys::Uniform,
            }),
            returning: vec![],
        };
        let (sql, params) = q.compile(Dialect::Sqlite).unwrap();
        assert_eq!(
            sql,
            "UPDATE counter SET hits = (hits + ?1) WHERE tenant_id = ?2 AND id = ?3"
        );
        assert_eq!(params, vec![SqlValue::Integer(1), t("ten_1"), t("c_1")]);
    }

    #[test]
    fn identifier_injection_is_rejected() {
        let q = Select {
            columns: vec![item(Expr::col("id; DROP TABLE users"))],
            ..Select::from("t")
        };
        assert!(matches!(
            q.compile(Dialect::Sqlite),
            Err(OrmError::InvalidIdentifier(_))
        ));
    }

    #[test]
    fn qualified_identifier_allowed() {
        let q = Select {
            columns: vec![item(Expr::col("t.id"))],
            ..Select::from("t")
        };
        assert_eq!(q.compile(Dialect::Sqlite).unwrap().0, "SELECT t.id FROM t");
    }

    #[test]
    fn function_arity_is_checked() {
        let q = Select {
            columns: vec![item(Expr::Func(Func::Lower, vec![]))],
            ..Select::from("t")
        };
        assert!(matches!(
            q.compile(Dialect::Sqlite),
            Err(OrmError::BadExpr(_))
        ));
    }

    #[test]
    fn update_with_empty_all_filter_is_refused() {
        let q = Update {
            table: "t".into(),
            set: vec![Assignment {
                column: "x".into(),
                value: Expr::val(SqlValue::Integer(1)),
            }],
            filter: Predicate::And(vec![]),
            scope: None,
            returning: vec![],
        };
        // An empty filter with no scope is an effectively-unbounded update → refused.
        assert!(matches!(
            q.compile(Dialect::Sqlite),
            Err(OrmError::Empty(_))
        ));
    }

    #[test]
    fn empty_filter_with_scope_is_allowed() {
        // A scope keeps it bounded, so an empty filter + scope compiles.
        let q = Update {
            table: "t".into(),
            set: vec![Assignment {
                column: "x".into(),
                value: Expr::val(SqlValue::Integer(1)),
            }],
            filter: Predicate::And(vec![]),
            scope: Some(Scope {
                column: "tenant_id".into(),
                value: Some(t("ten_1")),
                session: None,
                mode: ScopeMode::Own,
                keys: TableKeys::Uniform,
            }),
            returning: vec![],
        };
        assert_eq!(
            q.compile(Dialect::Sqlite).unwrap().0,
            "UPDATE t SET x = ?1 WHERE tenant_id = ?2"
        );
    }

    #[test]
    fn delete_by_predicate_compiles() {
        let q = Delete {
            table: "payment".into(),
            filter: cmp("id", CmpOp::Eq, t("pay_1")),
            scope: None,
            returning: vec![],
        };
        let (sql, params) = q.compile(Dialect::Sqlite).unwrap();
        assert_eq!(sql, "DELETE FROM payment WHERE id = ?1");
        assert_eq!(params, vec![t("pay_1")]);
    }

    #[test]
    fn delete_returning_renders() {
        // The one DELETE … RETURNING shape (consume-and-read a pending signup). `?N` is emitted
        // for every dialect — the backend rewrites to the engine's native placeholder.
        let q = Delete {
            table: "pending_signup".into(),
            filter: cmp("slug", CmpOp::Eq, t("acme")),
            scope: None,
            returning: vec![item(Expr::col("name")), item(Expr::col("password_hash"))],
        };
        assert_eq!(
            q.compile(Dialect::Postgres).unwrap().0,
            "DELETE FROM pending_signup WHERE slug = ?1 RETURNING name, password_hash"
        );
    }

    #[test]
    fn delete_with_empty_filter_is_refused() {
        // Empty filter, no scope → effectively-unbounded delete → refused (mirrors UPDATE).
        let q = Delete {
            table: "t".into(),
            filter: Predicate::And(vec![]),
            scope: None,
            returning: vec![],
        };
        assert!(matches!(
            q.compile(Dialect::Sqlite),
            Err(OrmError::Empty(_))
        ));
    }

    #[test]
    fn delete_empty_filter_with_scope_is_allowed() {
        // A scope keeps it bounded, so an empty filter + scope compiles (bulk clear within tenant).
        let q = Delete {
            table: "t".into(),
            filter: Predicate::And(vec![]),
            scope: Some(Scope {
                column: "tenant_id".into(),
                value: Some(t("ten_1")),
                session: None,
                mode: ScopeMode::Own,
                keys: TableKeys::Uniform,
            }),
            returning: vec![],
        };
        assert_eq!(
            q.compile(Dialect::Sqlite).unwrap().0,
            "DELETE FROM t WHERE tenant_id = ?1"
        );
    }

    #[test]
    fn delete_rejects_identifier_injection_in_table() {
        let q = Delete {
            table: "t; DROP TABLE users".into(),
            filter: cmp("id", CmpOp::Eq, t("x")),
            scope: None,
            returning: vec![],
        };
        assert!(matches!(
            q.compile(Dialect::Sqlite),
            Err(OrmError::InvalidIdentifier(_))
        ));
    }

    #[test]
    fn case_expression_renders_with_bound_params() {
        // CASE WHEN state = ? THEN 1 ELSE 0 END as a select item; params bind in textual order.
        let q = Select {
            columns: vec![item(Expr::Case {
                branches: vec![(
                    cmp("state", CmpOp::Eq, t("open")),
                    Expr::val(SqlValue::Integer(1)),
                )],
                otherwise: Some(Box::new(Expr::val(SqlValue::Integer(0)))),
            })],
            ..Select::from("t")
        };
        let (sql, params) = q.compile(Dialect::Sqlite).unwrap();
        assert_eq!(
            sql,
            "SELECT (CASE WHEN state = ?1 THEN ?2 ELSE ?3 END) FROM t"
        );
        assert_eq!(
            params,
            vec![t("open"), SqlValue::Integer(1), SqlValue::Integer(0)]
        );
    }

    #[test]
    fn distinct_on_renders_on_postgres_and_fails_closed_elsewhere() {
        let q = Select {
            distinct_on: vec![Expr::col("key")],
            columns: vec![item(Expr::col("key")), item(Expr::col("val"))],
            ..Select::from("consent_state")
        };
        assert_eq!(
            q.compile(Dialect::Postgres).unwrap().0,
            "SELECT DISTINCT ON (key) key, val FROM consent_state"
        );
        // No portable rewrite on SQLite/MySQL — fail closed.
        assert!(matches!(
            q.compile(Dialect::Sqlite),
            Err(OrmError::BadExpr(_))
        ));
    }

    #[test]
    fn empty_case_is_rejected() {
        let q = Select {
            columns: vec![item(Expr::Case {
                branches: vec![],
                otherwise: None,
            })],
            ..Select::from("t")
        };
        assert!(matches!(
            q.compile(Dialect::Sqlite),
            Err(OrmError::BadExpr(_))
        ));
    }

    #[test]
    fn json_extract_dyn_binds_the_key() {
        // labels ->> ?  (bound key). Postgres + SQLite render `->>`; MySQL fails closed.
        let q = Select {
            columns: vec![item(Expr::JsonExtractDyn(
                Box::new(Expr::col("labels")),
                Box::new(Expr::val(t("en"))),
            ))],
            ..Select::from("vocabulary_term")
        };
        for d in [Dialect::Postgres, Dialect::Sqlite] {
            assert_eq!(
                q.compile(d).unwrap().0,
                "SELECT (labels ->> ?1) FROM vocabulary_term"
            );
        }
        assert!(matches!(
            q.compile(Dialect::Mysql),
            Err(OrmError::BadExpr(_))
        ));
    }

    #[test]
    fn json_concat_merge_is_postgres_only() {
        // UPDATE request SET brief_state = brief_state || ?::jsonb WHERE id = ?
        let q = Update {
            table: "request".into(),
            set: vec![Assignment {
                column: "brief_state".into(),
                value: Expr::JsonConcat(
                    Box::new(Expr::col("brief_state")),
                    Box::new(Expr::val(SqlValue::Json("{\"a\":1}".into()))),
                ),
            }],
            filter: cmp("id", CmpOp::Eq, t("req_1")),
            scope: None,
            returning: vec![],
        };
        assert_eq!(
            q.compile(Dialect::Postgres).unwrap().0,
            "UPDATE request SET brief_state = (brief_state || ?1) WHERE id = ?2"
        );
        assert!(matches!(
            q.compile(Dialect::Sqlite),
            Err(OrmError::BadExpr(_))
        ));
    }

    #[test]
    fn union_renders_both_bodies_with_shared_params() {
        // slug-reservation check across two tables; the branches share the ?N sequence.
        let q = Select {
            columns: vec![item(Expr::col("slug"))],
            filter: Some(cmp("slug", CmpOp::Eq, t("acme"))),
            union: Some(Box::new(Union {
                all: false,
                query: Select {
                    columns: vec![item(Expr::col("slug"))],
                    filter: Some(cmp("slug", CmpOp::Eq, t("acme"))),
                    ..Select::from("reserved_slug")
                },
            })),
            ..Select::from("pending_signup")
        };
        let (sql, params) = q.compile(Dialect::Sqlite).unwrap();
        assert_eq!(
            sql,
            "SELECT slug FROM pending_signup WHERE slug = ?1 \
             UNION SELECT slug FROM reserved_slug WHERE slug = ?2"
        );
        assert_eq!(params, vec![t("acme"), t("acme")]);
    }

    #[test]
    fn insert_from_select_shares_params_and_carries_no_auto_scope() {
        // INSERT INTO ref (a, b) SELECT x, y FROM src WHERE id = ? (attach_reference shape).
        let q = Insert {
            table: "portfolio_ref".into(),
            rows: vec![],
            conflict: None,
            scope: None,
            returning: vec![],
            from_select: Some((
                vec!["a".into(), "b".into()],
                Box::new(Select {
                    columns: vec![item(Expr::col("x")), item(Expr::col("y"))],
                    filter: Some(cmp("id", CmpOp::Eq, t("pi_1"))),
                    ..Select::from("portfolio_item")
                }),
            )),
        };
        let (sql, params) = q.compile(Dialect::Sqlite).unwrap();
        assert_eq!(
            sql,
            "INSERT INTO portfolio_ref (a, b) SELECT x, y FROM portfolio_item WHERE id = ?1"
        );
        assert_eq!(params, vec![t("pi_1")]);
    }

    #[test]
    fn related_scalar_and_in_subquery_render() {
        // id = (SELECT head_version FROM pack WHERE id = ?1)
        let q = Select {
            columns: vec![item(Expr::col("id"))],
            filter: Some(Predicate::Cmp {
                left: Expr::col("id"),
                op: CmpOp::Eq,
                right: Expr::RelatedScalar {
                    column: "head_version".into(),
                    table: "pack".into(),
                    filter: Box::new(cmp("id", CmpOp::Eq, t("pk_1"))),
                },
            }),
            ..Select::from("pack_version")
        };
        assert_eq!(
            q.compile(Dialect::Sqlite).unwrap().0,
            "SELECT id FROM pack_version WHERE id = (SELECT head_version FROM pack WHERE id = ?1)"
        );

        // doc_id IN (SELECT id FROM document WHERE tenant_id = ?1)
        let q2 = Select {
            columns: vec![item(Expr::col("x"))],
            filter: Some(Predicate::InSubquery {
                expr: Expr::col("doc_id"),
                column: "id".into(),
                table: "document".into(),
                filter: Box::new(cmp("tenant_id", CmpOp::Eq, t("ten_1"))),
                negated: false,
            }),
            ..Select::from("access")
        };
        assert_eq!(
            q2.compile(Dialect::Sqlite).unwrap().0,
            "SELECT x FROM access WHERE doc_id IN (SELECT id FROM document WHERE tenant_id = ?1)"
        );
    }

    #[test]
    fn now_renders_without_parens() {
        let q = Select {
            columns: vec![item(Expr::Func(Func::Now, vec![]))],
            ..Select::from("t")
        };
        assert_eq!(
            q.compile(Dialect::Sqlite).unwrap().0,
            "SELECT current_timestamp FROM t"
        );
    }

    fn json_query() -> Select {
        Select {
            columns: vec![item(Expr::JsonExtract(
                Box::new(Expr::col("metadata")),
                vec!["status".into()],
            ))],
            filter: Some(Predicate::Cmp {
                left: Expr::JsonExtract(
                    Box::new(Expr::col("metadata")),
                    vec!["a".into(), "b".into()],
                ),
                op: CmpOp::Eq,
                right: Expr::val(t("x")),
            }),
            ..Select::from("doc")
        }
    }

    #[test]
    fn json_extract_sqlite_and_mysql_bind_the_path() {
        for d in [Dialect::Sqlite, Dialect::Mysql] {
            let (sql, params) = json_query().compile(d).unwrap();
            assert_eq!(
                sql,
                "SELECT json_extract(metadata, ?1) FROM doc WHERE json_extract(metadata, ?2) = ?3"
            );
            assert_eq!(params, vec![t("$.status"), t("$.a.b"), t("x")]);
        }
    }

    #[test]
    fn json_extract_postgres_inlines_the_validated_path() {
        let (sql, params) = json_query().compile(Dialect::Postgres).unwrap();
        assert_eq!(
            sql,
            "SELECT (metadata) #>> '{status}' FROM doc WHERE (metadata) #>> '{a,b}' = ?1"
        );
        assert_eq!(params, vec![t("x")]);
    }

    #[test]
    fn json_extract_key_injection_is_rejected() {
        let q = Select {
            columns: vec![item(Expr::JsonExtract(
                Box::new(Expr::col("m")),
                vec!["a'); DROP TABLE t--".into()],
            ))],
            ..Select::from("doc")
        };
        assert!(matches!(
            q.compile(Dialect::Postgres),
            Err(OrmError::InvalidIdentifier(_))
        ));
    }

    // ---- pgvector distance (Postgres-only) -----------------------------------

    fn knn_query() -> Select {
        // Nearest-neighbour: `ORDER BY embedding <=> [q] LIMIT k`.
        Select {
            columns: vec![item(Expr::col("id"))],
            order: vec![OrderBy {
                expr: Expr::Distance {
                    left: Box::new(Expr::col("embedding")),
                    right: Box::new(Expr::VectorLiteral("[0.1, 0.2, 0.3]".into())),
                    metric: Metric::Cosine,
                },
                dir: Direction::Asc,
            }],
            limit: Some(5),
            ..Select::from("doc")
        }
    }

    #[test]
    fn distance_orders_by_cosine_nearest_neighbour_on_postgres() {
        let (sql, params) = knn_query().compile(Dialect::Postgres).unwrap();
        assert_eq!(
            sql,
            "SELECT id FROM doc ORDER BY (embedding <=> ?1::vector) ASC LIMIT 5"
        );
        // The literal binds as a parameter (whitespace-normalised), never formatted in.
        assert_eq!(params, vec![t("[0.1,0.2,0.3]")]);
    }

    #[test]
    fn distance_l2_in_select_list_on_postgres() {
        let q = Select {
            columns: vec![
                item(Expr::col("id")),
                SelectItem {
                    expr: Expr::Distance {
                        left: Box::new(Expr::col("embedding")),
                        right: Box::new(Expr::VectorLiteral("[-1, 2e0, 3.5]".into())),
                        metric: Metric::L2,
                    },
                    alias: Some("dist".into()),
                },
            ],
            ..Select::from("doc")
        };
        let (sql, params) = q.compile(Dialect::Postgres).unwrap();
        assert_eq!(
            sql,
            "SELECT id, (embedding <-> ?1::vector) AS dist FROM doc"
        );
        assert_eq!(params, vec![t("[-1,2e0,3.5]")]);
    }

    #[test]
    fn distance_fails_closed_off_postgres() {
        for d in [Dialect::Sqlite, Dialect::Mysql] {
            assert!(
                matches!(knn_query().compile(d), Err(OrmError::BadExpr(_))),
                "vector distance must be rejected on {d:?}"
            );
        }
    }

    #[test]
    fn vector_literal_fails_closed_off_postgres() {
        for d in [Dialect::Sqlite, Dialect::Mysql] {
            let q = Select {
                columns: vec![item(Expr::VectorLiteral("[1, 2]".into()))],
                ..Select::from("doc")
            };
            assert!(
                matches!(q.compile(d), Err(OrmError::BadExpr(_))),
                "vector literal must be rejected on {d:?}"
            );
        }
    }

    #[test]
    fn malformed_vector_literal_is_rejected() {
        // No brackets, non-numeric, empty, unclosed, empty component, and non-finite
        // (`inf`/`NaN` parse as floats but must be refused).
        for bad in [
            "1,2",
            "[a, b]",
            "[]",
            "[1, 2",
            "[1,,2]",
            "[Infinity]",
            "[1, NaN]",
        ] {
            let q = Select {
                columns: vec![item(Expr::VectorLiteral(bad.to_string()))],
                ..Select::from("doc")
            };
            assert!(
                matches!(q.compile(Dialect::Postgres), Err(OrmError::BadExpr(_))),
                "expected {bad:?} to be rejected"
            );
        }
    }

    // ---- correlated roll-ups (related-aggregate) -----------------------------

    /// `agg(arg) FROM table WHERE child.fk = parent.pk [AND extra]` — the correlation is a
    /// column-to-column comparison in the subquery's filter.
    fn related(agg: Agg, arg: RelArg, table: &str, filter: Predicate) -> Expr {
        Expr::RelatedAggregate {
            agg,
            arg,
            table: table.into(),
            filter: Box::new(filter),
        }
    }
    fn correlate(fk: &str, pk: &str) -> Predicate {
        Predicate::Cmp {
            left: Expr::col(fk),
            op: CmpOp::Eq,
            right: Expr::col(pk),
        }
    }

    #[test]
    fn related_aggregate_single_correlated_count() {
        // construens subgraph-chain:701 — count of child rows per parent, no fan-out.
        let q = Select {
            columns: vec![
                item(Expr::col("id")),
                SelectItem {
                    expr: related(
                        Agg::Count,
                        RelArg::Star,
                        "element",
                        correlate("element.order_id", "work_order.id"),
                    ),
                    alias: Some("element_count".into()),
                },
            ],
            ..Select::from("work_order")
        };
        let (sql, params) = q.compile(Dialect::Postgres).unwrap();
        assert_eq!(
            sql,
            "SELECT id, (SELECT count(*) FROM element WHERE element.order_id = work_order.id) AS element_count FROM work_order"
        );
        assert!(params.is_empty());
    }

    #[test]
    fn related_aggregate_two_counts_bind_distinct_params_and_dont_fan_out() {
        // Two correlated counts in one SELECT — each its own subquery (no join, no fan-out),
        // and their bound filters take distinct `?N` in left-to-right order.
        let with_status = |child: &str, fk: &str, status: &str| {
            related(
                Agg::Count,
                RelArg::Star,
                child,
                Predicate::And(vec![
                    correlate(fk, "party.id"),
                    Predicate::Cmp {
                        left: Expr::col("status"),
                        op: CmpOp::Eq,
                        right: Expr::val(t(status)),
                    },
                ]),
            )
        };
        let q = Select {
            columns: vec![
                item(with_status("party_role", "party_role.party_id", "active")),
                item(with_status(
                    "party_qualification",
                    "party_qualification.party_id",
                    "valid",
                )),
            ],
            ..Select::from("party")
        };
        let (sql, params) = q.compile(Dialect::Postgres).unwrap();
        assert_eq!(
            sql,
            "SELECT \
             (SELECT count(*) FROM party_role WHERE party_role.party_id = party.id AND status = ?1), \
             (SELECT count(*) FROM party_qualification WHERE party_qualification.party_id = party.id AND status = ?2) \
             FROM party"
        );
        assert_eq!(params, vec![t("active"), t("valid")]);
    }

    #[test]
    fn related_aggregate_with_temporal_or_filter() {
        // construens subgraph-chain:731 — correlated count with a temporal `valid_to` filter.
        let q = Select {
            columns: vec![SelectItem {
                expr: related(
                    Agg::Count,
                    RelArg::Star,
                    "party_qualification",
                    Predicate::And(vec![
                        correlate("party_qualification.party_id", "party.id"),
                        Predicate::Or(vec![
                            Predicate::Null {
                                expr: Expr::col("valid_to"),
                                negated: false,
                            },
                            Predicate::Cmp {
                                left: Expr::col("valid_to"),
                                op: CmpOp::Gt,
                                right: Expr::val(t("2026-01-01")),
                            },
                        ]),
                    ]),
                ),
                alias: Some("active_quals".into()),
            }],
            ..Select::from("party")
        };
        let (sql, params) = q.compile(Dialect::Postgres).unwrap();
        assert_eq!(
            sql,
            "SELECT (SELECT count(*) FROM party_qualification WHERE party_qualification.party_id = party.id AND (valid_to IS NULL OR valid_to > ?1)) AS active_quals FROM party"
        );
        assert_eq!(params, vec![t("2026-01-01")]);
    }

    #[test]
    fn related_aggregate_max_over_a_column_is_portable() {
        // A correlated MAX over a column (construens subgraph-chain:1932 shape) — an ordinary
        // subquery, portable across engines (not Postgres-specific like vector distance).
        let q = Select {
            columns: vec![SelectItem {
                expr: related(
                    Agg::Max,
                    RelArg::Column("total_minor".into()),
                    "line_item",
                    correlate("line_item.order_id", "order_summary.id"),
                ),
                alias: Some("max_total".into()),
            }],
            ..Select::from("order_summary")
        };
        let (sql, _) = q.compile(Dialect::Sqlite).unwrap();
        assert_eq!(
            sql,
            "SELECT (SELECT max(total_minor) FROM line_item WHERE line_item.order_id = order_summary.id) AS max_total FROM order_summary"
        );
    }

    #[test]
    fn related_aggregate_star_is_count_only() {
        let q = Select {
            columns: vec![item(related(
                Agg::Sum,
                RelArg::Star,
                "t",
                correlate("t.fk", "p.id"),
            ))],
            ..Select::from("p")
        };
        assert!(matches!(
            q.compile(Dialect::Postgres),
            Err(OrmError::BadExpr(_))
        ));
    }

    #[test]
    fn related_aggregate_table_injection_is_rejected() {
        let q = Select {
            columns: vec![item(related(
                Agg::Count,
                RelArg::Star,
                "element; DROP TABLE users",
                correlate("element.order_id", "p.id"),
            ))],
            ..Select::from("p")
        };
        assert!(matches!(
            q.compile(Dialect::Postgres),
            Err(OrmError::InvalidIdentifier(_))
        ));
    }

    #[test]
    fn target_read_conjoins_the_public_subset_and_composes_across_joins() {
        use std::collections::BTreeMap;
        // A target read of tenant `B`: products (Tenant→tenant_id) LEFT JOIN reviews
        // (Tenant→tenant_id). Each ref is confined to `tenant_id = B` AND its OWN public predicate,
        // qualified per ref — so a target read can never reach B's private rows through any table.
        let keys = BTreeMap::from([
            (
                "products".to_string(),
                ResolvedScope::Column("tenant_id".to_string()),
            ),
            (
                "reviews".to_string(),
                ResolvedScope::Column("tenant_id".to_string()),
            ),
        ]);
        let public = BTreeMap::from([
            (
                "products".to_string(),
                vec![
                    PublicTermSql::Cmp {
                        column: "published".into(),
                        op: CmpOp::Eq,
                        value: SqlValue::Boolean(true),
                    },
                    PublicTermSql::Null {
                        column: "deleted_at".into(),
                        negated: false,
                    },
                ],
            ),
            (
                "reviews".to_string(),
                vec![PublicTermSql::Cmp {
                    column: "visible".into(),
                    op: CmpOp::Eq,
                    value: SqlValue::Boolean(true),
                }],
            ),
        ]);
        let mut q = Select {
            table_alias: Some("p".into()),
            joins: vec![Join {
                kind: JoinKind::Left,
                table: "reviews".into(),
                alias: Some("r".into()),
                on: Predicate::Cmp {
                    left: Expr::col("p.id"),
                    op: CmpOp::Eq,
                    right: Expr::col("r.product_id"),
                },
            }],
            ..Select::from("products")
        };
        q.force_scope(&Scope {
            column: "tenant_id".into(),
            value: Some(t("B")),
            session: None,
            mode: ScopeMode::Own,
            keys: TableKeys::PerTableTarget {
                keys: keys.clone(),
                public: public.clone(),
                write: std::collections::BTreeSet::new(),
                require_public: true,
            },
        })
        .unwrap();
        let (sql, _params) = q.compile(Dialect::Sqlite).unwrap();
        // Base ref `p`: tenant + its public terms, all qualified `p.`.
        assert!(sql.contains("p.tenant_id = ?"), "base tenant scope: {sql}");
        assert!(sql.contains("p.published = ?"), "base public term: {sql}");
        assert!(
            sql.contains("p.deleted_at IS NULL"),
            "base public null term: {sql}"
        );
        // Joined ref `r`: tenant + ITS OWN public term, qualified `r.` (composition across the join).
        assert!(
            sql.contains("r.tenant_id = ?"),
            "joined tenant scope: {sql}"
        );
        assert!(sql.contains("r.visible = ?"), "joined public term: {sql}");
    }

    #[test]
    fn target_read_of_a_table_with_no_public_subset_is_refused() {
        use std::collections::BTreeMap;
        // Deny-by-default: a target read touching a table that declares NO public subset is refused
        // (PublicSubsetUndeclared) — the strict analog of an undeclared tenant key, and what keeps a
        // target read from reaching a private table.
        let keys = BTreeMap::from([(
            "secret_table".to_string(),
            ResolvedScope::Column("tenant_id".to_string()),
        )]);
        let deny = Scope {
            column: "tenant_id".into(),
            value: Some(t("B")),
            session: None,
            mode: ScopeMode::Own,
            keys: TableKeys::PerTableTarget {
                keys,
                public: BTreeMap::new(), // no public subset for secret_table
                write: std::collections::BTreeSet::new(),
                require_public: true,
            },
        };
        let mut q = Select::from("secret_table");
        // The refusal surfaces at force_scope (join/subquery refs) or compile (base ref).
        let err = q
            .force_scope(&deny)
            .err()
            .or_else(|| q.compile(Dialect::Sqlite).err());
        assert!(
            matches!(&err, Some(OrmError::PublicSubsetUndeclared(t)) if t == "secret_table"),
            "expected PublicSubsetUndeclared, got {err:?}"
        );
    }

    #[test]
    fn own_read_is_unaffected_by_the_public_injection() {
        // An own read (PerTable, not PerTableTarget) conjoins NO public predicate — byte-identical
        // to pre-Stage-5. (Regression fence: the target path must not leak into the own path.)
        use std::collections::BTreeMap;
        let mut q = Select::from("products");
        q.force_scope(&Scope {
            column: "tenant_id".into(),
            value: Some(t("A")),
            session: None,
            mode: ScopeMode::Own,
            keys: TableKeys::PerTable(BTreeMap::from([(
                "products".to_string(),
                ResolvedScope::Column("tenant_id".to_string()),
            )])),
        })
        .unwrap();
        let (sql, _p) = q.compile(Dialect::Sqlite).unwrap();
        assert!(sql.contains("tenant_id = ?"));
        assert!(
            !sql.contains("published") && !sql.contains("IS NULL"),
            "own read must carry no public confinement: {sql}"
        );
    }

    #[test]
    fn tenant_or_base_folds_the_null_base_on_reads_but_not_writes() {
        // v0.4.16: `pack` is base-inclusive (its tenant-NULL rows are shared base); `product` is a
        // plain tenant table. On the OWN axis a read of `pack` folds `(tenant = A OR tenant IS NULL)`
        // — A's rows ⊕ the shared base — while `product` stays `tenant = A` alone. The fold is per
        // TABLE, so a query touching both confines each correctly (no field-level widening).
        use std::collections::BTreeMap;
        let scope = Scope {
            column: "tenant_id".into(),
            value: Some(t("A")),
            session: None,
            mode: ScopeMode::Own,
            keys: TableKeys::PerTable(BTreeMap::from([
                (
                    "pack".to_string(),
                    ResolvedScope::TenantOrBase {
                        tenant: "tenant_id".into(),
                    },
                ),
                (
                    "product".to_string(),
                    ResolvedScope::Column("tenant_id".into()),
                ),
            ])),
        };
        // Base-inclusive read: (tenant = A OR tenant IS NULL).
        let mut pack = Select::from("pack");
        pack.force_scope(&scope).unwrap();
        let (psql, _p) = pack.compile(Dialect::Sqlite).unwrap();
        assert!(
            psql.contains("tenant_id = ?")
                && psql.contains("tenant_id IS NULL")
                && psql.contains(" OR "),
            "base-inclusive read folds the NULL base: {psql}"
        );
        // Plain tenant read: tenant = A alone (no base fold — the fold is per-table, not per-field).
        let mut prod = Select::from("product");
        prod.force_scope(&scope).unwrap();
        let (dsql, _p) = prod.compile(Dialect::Sqlite).unwrap();
        assert!(
            dsql.contains("tenant_id = ?") && !dsql.contains("IS NULL"),
            "a plain tenant table alongside a base-inclusive one stays tenant-only: {dsql}"
        );
        // WRITE stamps the resolved tenant (never NULL) — a guest can't create a base row.
        let mut ins = Insert {
            table: "pack".into(),
            rows: vec![RowValues {
                cells: vec![Assignment {
                    column: "name".into(),
                    value: Expr::val(t("p")),
                }],
            }],
            conflict: None,
            scope: None,
            returning: vec![],
            from_select: None,
        };
        ins.force_scope(Some(&scope), Some(&scope)).unwrap();
        let (isql, iparams) = ins.compile(Dialect::Sqlite).unwrap();
        assert!(
            isql.contains("tenant_id"),
            "write stamps the tenant column: {isql}"
        );
        assert!(
            iparams.contains(&t("A")) && !iparams.contains(&SqlValue::Null),
            "write stamps tenant = A, never a NULL base row: {iparams:?}"
        );
    }

    #[test]
    fn target_read_capability_drops_a_declared_subset_confining_tenant_only() {
        // construens embed bug (ruling A made COMPLETE): under a capability-only target read
        // (require_public=false), a table that DECLARES a public subset (authored for the anonymous
        // funnel) must confine to `tenant = B` ALONE — the subset is NOT AND-ed on, so a resolver's
        // own within-tenant filter on that column isn't collided into the empty set. `tenant = B`
        // (applied separately) stays mandatory on the base ref AND every joined ref.
        use std::collections::BTreeMap;
        let keys = BTreeMap::from([
            (
                "products".to_string(),
                ResolvedScope::Column("tenant_id".to_string()),
            ),
            (
                "reviews".to_string(),
                ResolvedScope::Column("tenant_id".to_string()),
            ),
        ]);
        let public = BTreeMap::from([
            (
                "products".to_string(),
                vec![
                    PublicTermSql::Cmp {
                        column: "published".into(),
                        op: CmpOp::Eq,
                        value: SqlValue::Boolean(true),
                    },
                    PublicTermSql::Null {
                        column: "deleted_at".into(),
                        negated: false,
                    },
                ],
            ),
            (
                "reviews".to_string(),
                vec![PublicTermSql::Cmp {
                    column: "visible".into(),
                    op: CmpOp::Eq,
                    value: SqlValue::Boolean(true),
                }],
            ),
        ]);
        let cap = Scope {
            column: "tenant_id".into(),
            value: Some(t("B")),
            session: None,
            mode: ScopeMode::Own,
            keys: TableKeys::PerTableTarget {
                keys,
                public,
                write: std::collections::BTreeSet::new(),
                require_public: false, // capability axis (ruling A)
            },
        };
        let mut q = Select {
            table_alias: Some("p".into()),
            joins: vec![Join {
                kind: JoinKind::Inner,
                table: "reviews".into(),
                alias: Some("r".into()),
                on: Predicate::Cmp {
                    left: Expr::col("r.product_id"),
                    op: CmpOp::Eq,
                    right: Expr::col("p.id"),
                },
            }],
            ..Select::from("products")
        };
        q.force_scope(&cap).unwrap();
        let (sql, _p) = q.compile(Dialect::Sqlite).unwrap();
        assert!(
            sql.contains("p.tenant_id = ?"),
            "base tenant scope kept: {sql}"
        );
        assert!(
            sql.contains("r.tenant_id = ?"),
            "joined tenant scope kept: {sql}"
        );
        assert!(
            !sql.contains("published") && !sql.contains("deleted_at") && !sql.contains("visible"),
            "no declared subset term may appear on the capability axis: {sql}"
        );
    }

    // ---- 5b: target WRITES (INSERT + UPDATE with a SET-allowlist; DELETE refused) --------------

    /// A target-WRITE scope for `products`: tenant key `tenant_id`, public subset `published = true
    /// AND deleted_at IS NULL`, SET-allowlist = the given columns.
    fn target_write_scope(write: &[&str]) -> Scope {
        use std::collections::{BTreeMap, BTreeSet};
        Scope {
            column: "tenant_id".into(),
            value: Some(t("tenant_B")),
            session: None,
            mode: ScopeMode::Own,
            keys: TableKeys::PerTableTarget {
                keys: BTreeMap::from([(
                    "products".to_string(),
                    ResolvedScope::Column("tenant_id".to_string()),
                )]),
                public: BTreeMap::from([(
                    "products".to_string(),
                    vec![
                        PublicTermSql::Cmp {
                            column: "published".into(),
                            op: CmpOp::Eq,
                            value: SqlValue::Boolean(true),
                        },
                        PublicTermSql::Null {
                            column: "deleted_at".into(),
                            negated: false,
                        },
                    ],
                )]),
                write: write
                    .iter()
                    .map(ToString::to_string)
                    .collect::<BTreeSet<_>>(),
                require_public: true,
            },
        }
    }

    /// A **capability-axis** target-WRITE scope for `client_survey` (`require_public = false`, ruling
    /// A): tenant key `tenant_id`, NO declared public subset, SET-allowlist = the given columns. This
    /// is the axis on which an `ON CONFLICT … DO UPDATE` upsert is admitted.
    fn target_write_scope_cap(write: &[&str]) -> Scope {
        use std::collections::{BTreeMap, BTreeSet};
        Scope {
            column: "tenant_id".into(),
            value: Some(t("tenant_B")),
            session: None,
            mode: ScopeMode::Own,
            keys: TableKeys::PerTableTarget {
                keys: BTreeMap::from([(
                    "client_survey".to_string(),
                    ResolvedScope::Column("tenant_id".to_string()),
                )]),
                public: BTreeMap::new(), // capability axis: no visibility subset (ruling A)
                write: write
                    .iter()
                    .map(ToString::to_string)
                    .collect::<BTreeSet<_>>(),
                require_public: false,
            },
        }
    }

    fn target_insert(cells: Vec<Assignment>) -> Insert {
        Insert {
            table: "products".into(),
            rows: vec![RowValues { cells }],
            conflict: None,
            scope: None,
            returning: vec![],
            from_select: None,
        }
    }

    #[test]
    fn target_insert_forces_tenant_and_public_and_accepts_only_allowlisted_columns() {
        // The guest sets only the allowlisted `title`; the host force-stamps tenant=B, published=true,
        // deleted_at=NULL — so the inserted row lands squarely in B's public subset.
        let scope = target_write_scope(&["title"]);
        let mut ins = target_insert(vec![Assignment {
            column: "title".into(),
            value: Expr::val(t("Hello")),
        }]);
        ins.force_scope(Some(&scope), Some(&scope)).unwrap();
        let (sql, params) = ins.compile(Dialect::Sqlite).unwrap();
        assert!(sql.contains("tenant_id"), "{sql}");
        assert!(sql.contains("published"), "{sql}");
        assert!(sql.contains("deleted_at"), "{sql}");
        assert!(
            params.contains(&t("tenant_B")),
            "tenant forced to B: {params:?}"
        );
        assert!(
            params.contains(&SqlValue::Boolean(true)),
            "published forced true: {params:?}"
        );
        assert!(
            params.contains(&SqlValue::Null),
            "deleted_at forced NULL: {params:?}"
        );
        assert!(params.contains(&t("Hello")), "guest title kept: {params:?}");
    }

    #[test]
    fn target_insert_refuses_a_non_allowlisted_column() {
        // `price` is not in the SET-allowlist ⇒ refused (a target write may set only granted columns).
        let scope = target_write_scope(&["title"]);
        let mut ins = target_insert(vec![
            Assignment {
                column: "title".into(),
                value: Expr::val(t("x")),
            },
            Assignment {
                column: "price".into(),
                value: Expr::val(SqlValue::Integer(9)),
            },
        ]);
        let err = ins.force_scope(Some(&scope), Some(&scope)).unwrap_err();
        assert!(
            matches!(err, OrmError::TargetWriteColumnDenied(ref c) if c == "price"),
            "{err:?}"
        );
    }

    #[test]
    fn target_insert_refuses_setting_the_tenant_or_visibility_column() {
        // Even if the guest tries to set tenant_id or published directly, it's denied (they're never
        // in the allowlist; and `assert_target_settable` refuses them structurally regardless).
        for bad in ["tenant_id", "published", "deleted_at"] {
            let scope = target_write_scope(&["title", bad]); // even if wrongly granted...
            let mut ins = target_insert(vec![Assignment {
                column: bad.into(),
                value: Expr::val(t("x")),
            }]);
            let err = ins.force_scope(Some(&scope), Some(&scope)).unwrap_err();
            assert!(
                matches!(err, OrmError::TargetWriteColumnDenied(ref c) if c == bad),
                "{bad}: {err:?}"
            );
        }
    }

    #[test]
    fn target_insert_with_no_write_grant_is_refused() {
        let scope = target_write_scope(&[]); // read-only
        let mut ins = target_insert(vec![Assignment {
            column: "title".into(),
            value: Expr::val(t("x")),
        }]);
        let err = ins.force_scope(Some(&scope), Some(&scope)).unwrap_err();
        assert!(
            matches!(err, OrmError::TargetWriteNotGranted(ref t) if t == "products"),
            "{err:?}"
        );
    }

    #[test]
    fn target_write_to_a_tenant_or_session_table_is_refused() {
        use std::collections::{BTreeMap, BTreeSet};
        // `state_scope` is a TenantOrSession (anon-session) table. A target principal carries only
        // tenant B (no session fact), so a target write here could only stamp `tenant = B` — silently
        // claiming an anon/session-owned row for B and breaking anon→promotion. It must be refused
        // early + self-describingly (Stage A), NOT surfaced as a public-subset error and NOT silently
        // stamped. Both INSERT and UPDATE are covered.
        let scope = Scope {
            column: "tenant_id".into(),
            value: Some(t("tenant_B")),
            session: None,
            mode: ScopeMode::Own,
            keys: TableKeys::PerTableTarget {
                keys: BTreeMap::from([(
                    "state_scope".to_string(),
                    ResolvedScope::TenantOrSession {
                        tenant: "tenant_id".to_string(),
                        session: "session_id".to_string(),
                    },
                )]),
                public: BTreeMap::new(),
                write: BTreeSet::from(["note".to_string()]),
                require_public: true,
            },
        };
        // INSERT
        let mut ins = Insert {
            table: "state_scope".into(),
            rows: vec![RowValues {
                cells: vec![Assignment {
                    column: "note".into(),
                    value: Expr::val(t("x")),
                }],
            }],
            conflict: None,
            scope: None,
            returning: vec![],
            from_select: None,
        };
        let err = ins.force_scope(Some(&scope), Some(&scope)).unwrap_err();
        assert!(
            matches!(err, OrmError::TargetWriteToSessionTable(ref x) if x == "state_scope"),
            "INSERT should refuse a target write to a TenantOrSession table: {err:?}"
        );
        // UPDATE
        let mut upd = Update {
            table: "state_scope".into(),
            set: vec![Assignment {
                column: "note".into(),
                value: Expr::val(t("y")),
            }],
            filter: cmp("id", CmpOp::Eq, t("s1")),
            scope: None,
            returning: vec![],
        };
        let err = upd.force_scope(&scope).unwrap_err();
        assert!(
            matches!(err, OrmError::TargetWriteToSessionTable(ref x) if x == "state_scope"),
            "UPDATE should refuse a target write to a TenantOrSession table: {err:?}"
        );
    }

    #[test]
    fn target_insert_select_and_upsert_are_refused() {
        let scope = target_write_scope(&["title"]);
        let mut ins = target_insert(vec![Assignment {
            column: "title".into(),
            value: Expr::val(t("x")),
        }]);
        ins.from_select = Some((vec!["title".into()], Box::new(Select::from("products"))));
        assert!(matches!(
            ins.force_scope(Some(&scope), Some(&scope)).unwrap_err(),
            OrmError::TargetWriteUnsupported("INSERT … SELECT")
        ));
        // An upsert on an ANONYMOUS (require_public — domain/handle/target_or_null) target stays
        // refused: the DO-UPDATE `tenant = B` guard carries no visibility subset.
        let mut ins2 = target_insert(vec![Assignment {
            column: "title".into(),
            value: Expr::val(t("x")),
        }]);
        ins2.conflict = Some(OnConflict {
            conflict_columns: vec!["tenant_id".into(), "id".into()],
            update: vec![Assignment {
                column: "title".into(),
                value: Expr::col("excluded.title"),
            }],
        });
        assert!(matches!(
            ins2.force_scope(Some(&scope), Some(&scope)).unwrap_err(),
            OrmError::TargetWriteUnsupported(
                "ON CONFLICT upsert on an anonymous domain/handle/target_or_null target"
            )
        ));
    }

    #[test]
    fn target_upsert_capability_confines_do_update_and_guards_tenant_b() {
        // v0.4.15 (own↔target parity): a capability-axis `ON CONFLICT (tenant_id, …) DO UPDATE`
        // upsert compiles — the INSERT stamps tenant=B, the DO UPDATE SET is allowlisted, and the
        // conflict-update is guarded to `tenant = B` so it can never touch another tenant's row.
        // The natural-key cells (project_id, client_id) are inserted by the guest, so they're in the
        // write-allowlist alongside the mutated `score` (a target INSERT cell must be allowlisted).
        let scope = target_write_scope_cap(&["score", "comment", "project_id", "client_id"]);
        let mut ins = Insert {
            table: "client_survey".into(),
            rows: vec![RowValues {
                cells: vec![
                    Assignment {
                        column: "project_id".into(),
                        value: Expr::val(t("p1")),
                    },
                    Assignment {
                        column: "client_id".into(),
                        value: Expr::val(t("cli_a")),
                    },
                    Assignment {
                        column: "score".into(),
                        value: Expr::val(SqlValue::Integer(5)),
                    },
                ],
            }],
            conflict: Some(OnConflict {
                conflict_columns: vec!["tenant_id".into(), "project_id".into(), "client_id".into()],
                update: vec![Assignment {
                    column: "score".into(),
                    value: Expr::col("excluded.score"),
                }],
            }),
            scope: None,
            returning: vec![],
            from_select: None,
        };
        ins.force_scope(Some(&scope), Some(&scope)).unwrap();
        let (sql, params) = ins.compile(Dialect::Sqlite).unwrap();
        assert!(
            sql.contains("ON CONFLICT (tenant_id, project_id, client_id) DO UPDATE SET"),
            "upsert compiled: {sql}"
        );
        assert!(
            sql.contains("score = excluded.score"),
            "SET allowlisted col: {sql}"
        );
        assert!(
            sql.contains("client_survey.tenant_id ="),
            "DO UPDATE guarded to tenant = B (qualified): {sql}"
        );
        assert!(
            params.iter().filter(|p| **p == t("tenant_B")).count() >= 2,
            "tenant B stamped on INSERT AND bound in the DO-UPDATE guard: {params:?}"
        );
    }

    #[test]
    fn target_upsert_capability_requires_tenant_in_conflict_key() {
        // Deny-by-default: a capability upsert whose conflict target OMITS the tenant column is
        // refused — else B's INSERT could conflict with tenant A's row on a tenant-agnostic key.
        let scope = target_write_scope_cap(&["score", "project_id", "client_id"]);
        let mut ins = Insert {
            table: "client_survey".into(),
            rows: vec![RowValues {
                cells: vec![Assignment {
                    column: "score".into(),
                    value: Expr::val(SqlValue::Integer(5)),
                }],
            }],
            conflict: Some(OnConflict {
                conflict_columns: vec!["project_id".into(), "client_id".into()], // NO tenant_id
                update: vec![Assignment {
                    column: "score".into(),
                    value: Expr::col("excluded.score"),
                }],
            }),
            scope: None,
            returning: vec![],
            from_select: None,
        };
        assert!(matches!(
            ins.force_scope(Some(&scope), Some(&scope)).unwrap_err(),
            OrmError::TargetUpsertKeyMissingTenant(t) if t == "client_survey"
        ));
        // A `.`-qualified conflict-key column is refused self-describingly at compile (invalid ON
        // CONFLICT syntax + can't be allowed to slip past the tenant-key check on a last-segment match).
        let mut qual = Insert {
            table: "client_survey".into(),
            rows: vec![RowValues {
                cells: vec![Assignment {
                    column: "score".into(),
                    value: Expr::val(SqlValue::Integer(5)),
                }],
            }],
            conflict: Some(OnConflict {
                conflict_columns: vec!["cs.tenant_id".into(), "project_id".into()],
                update: vec![Assignment {
                    column: "score".into(),
                    value: Expr::col("excluded.score"),
                }],
            }),
            scope: None,
            returning: vec![],
            from_select: None,
        };
        assert!(matches!(
            qual.force_scope(Some(&scope), Some(&scope)).unwrap_err(),
            OrmError::TargetWriteColumnDenied(c) if c == "cs.tenant_id"
        ));
    }

    #[test]
    fn target_upsert_capability_refuses_a_non_allowlisted_or_tenant_do_update_set() {
        // The DO UPDATE SET is confined exactly like a target UPDATE: a non-allowlisted column, or the
        // tenant column, in the SET is refused (can't change ownership or touch a non-granted column).
        let scope = target_write_scope_cap(&["score"]);
        let mk = |set_col: &str| Insert {
            table: "client_survey".into(),
            rows: vec![RowValues {
                cells: vec![Assignment {
                    column: "score".into(),
                    value: Expr::val(SqlValue::Integer(5)),
                }],
            }],
            conflict: Some(OnConflict {
                conflict_columns: vec!["tenant_id".into(), "project_id".into()],
                update: vec![Assignment {
                    column: set_col.into(),
                    value: Expr::val(SqlValue::Integer(9)),
                }],
            }),
            scope: None,
            returning: vec![],
            from_select: None,
        };
        let mut bad_col = mk("secret"); // not in the allowlist
        assert!(matches!(
            bad_col.force_scope(Some(&scope), Some(&scope)).unwrap_err(),
            OrmError::TargetWriteColumnDenied(_)
        ));
        let mut tenant_set = mk("tenant_id"); // the tenant column itself
        assert!(matches!(
            tenant_set
                .force_scope(Some(&scope), Some(&scope))
                .unwrap_err(),
            OrmError::TargetWriteColumnDenied(_)
        ));
    }

    #[test]
    fn target_update_confines_to_the_public_subset_and_enforces_the_allowlist() {
        let scope = target_write_scope(&["title"]);
        let mut upd = Update {
            table: "products".into(),
            set: vec![Assignment {
                column: "title".into(),
                value: Expr::val(t("new")),
            }],
            filter: cmp("id", CmpOp::Eq, t("p1")),
            scope: None,
            returning: vec![],
        };
        upd.force_scope(&scope).unwrap();
        let (sql, params) = upd.compile(Dialect::Sqlite).unwrap();
        // WHERE = tenant = B AND (public terms) AND (guest filter).
        assert!(sql.contains("tenant_id = ?"), "tenant confinement: {sql}");
        assert!(sql.contains("published = ?"), "public confinement: {sql}");
        assert!(
            sql.contains("deleted_at IS NULL"),
            "public null confinement: {sql}"
        );
        assert!(sql.contains("SET title = ?"), "{sql}");
        assert!(params.contains(&t("tenant_B")), "{params:?}");
    }

    #[test]
    fn target_update_capability_no_subset_confines_tenant_only_not_refused() {
        // 5c ruling A: a capability-only (require_public=false) UPDATE on a table with NO declared
        // public subset confines to `tenant = B` (+ the SET-allowlist) — NOT refused (v0.4.3 bug the
        // review caught: it errored PublicSubsetUndeclared, breaking capability write-embeds).
        use std::collections::{BTreeMap, BTreeSet};
        let scope = Scope {
            column: "tenant_id".into(),
            value: Some(t("tenant_B")),
            session: None,
            mode: ScopeMode::Own,
            keys: TableKeys::PerTableTarget {
                keys: BTreeMap::from([(
                    "invoices".to_string(),
                    ResolvedScope::Column("tenant_id".into()),
                )]),
                public: BTreeMap::new(), // no subset for invoices
                write: BTreeSet::from(["amount".to_string()]),
                require_public: false, // capability
            },
        };
        let mut upd = Update {
            table: "invoices".into(),
            set: vec![Assignment {
                column: "amount".into(),
                value: Expr::val(SqlValue::Integer(5)),
            }],
            filter: cmp("id", CmpOp::Eq, t("inv1")),
            scope: None,
            returning: vec![],
        };
        upd.force_scope(&scope).unwrap();
        let (sql, _params) = upd.compile(Dialect::Sqlite).unwrap();
        assert!(
            sql.contains("tenant_id = ?"),
            "tenant=B confinement present: {sql}"
        );
        assert!(!sql.contains("published"), "no visibility conjunct: {sql}");
        assert!(sql.contains("SET amount = ?"), "{sql}");
        // A non-allowlisted column is still refused under the exemption.
        let mut bad = Update {
            table: "invoices".into(),
            set: vec![Assignment {
                column: "tenant_id".into(),
                value: Expr::val(t("evil")),
            }],
            filter: cmp("id", CmpOp::Eq, t("inv1")),
            scope: None,
            returning: vec![],
        };
        assert!(
            bad.force_scope(&scope).is_err(),
            "tenant column still un-settable"
        );
    }

    #[test]
    fn target_update_refuses_a_non_allowlisted_or_visibility_set() {
        for bad in ["price", "published", "tenant_id"] {
            let scope = target_write_scope(&["title"]);
            let mut upd = Update {
                table: "products".into(),
                set: vec![Assignment {
                    column: bad.into(),
                    value: Expr::val(t("x")),
                }],
                filter: cmp("id", CmpOp::Eq, t("p1")),
                scope: None,
                returning: vec![],
            };
            assert!(
                matches!(upd.force_scope(&scope).unwrap_err(), OrmError::TargetWriteColumnDenied(ref c) if c == bad),
                "{bad}"
            );
        }
    }

    #[test]
    fn target_write_refuses_a_qualified_column() {
        // A dotted write-target column (`published.x`) must be refused at compile — never rendered
        // (it would sneak past the last-segment `same_col` visibility check and emit invalid SQL).
        let scope = target_write_scope(&["title"]);
        let mut ins = target_insert(vec![Assignment {
            column: "published.x".into(),
            value: Expr::val(t("x")),
        }]);
        assert!(matches!(
            ins.force_scope(Some(&scope), Some(&scope)).unwrap_err(),
            OrmError::TargetWriteColumnDenied(ref c) if c == "published.x"
        ));
        let mut upd = Update {
            table: "products".into(),
            set: vec![Assignment {
                column: "title.y".into(),
                value: Expr::val(t("x")),
            }],
            filter: cmp("id", CmpOp::Eq, t("p1")),
            scope: None,
            returning: vec![],
        };
        assert!(matches!(
            upd.force_scope(&scope).unwrap_err(),
            OrmError::TargetWriteColumnDenied(ref c) if c == "title.y"
        ));
    }

    #[test]
    fn target_delete_is_always_refused() {
        let scope = target_write_scope(&["title"]);
        let mut del = Delete {
            table: "products".into(),
            filter: cmp("id", CmpOp::Eq, t("p1")),
            scope: None,
            returning: vec![],
        };
        assert!(matches!(
            del.force_scope(&scope).unwrap_err(),
            OrmError::TargetDeleteRefused(t) if t == "products"
        ));
    }

    #[test]
    fn target_promote_is_refused() {
        let scope = target_write_scope(&["title"]);
        assert!(matches!(
            compile_promote(&scope, "products", Dialect::Sqlite).unwrap_err(),
            OrmError::TargetWriteUnsupported("promote")
        ));
    }

    // ---- 5d: attach_reference (derived-tenant write) ------------------------------------------

    fn attach_spec() -> AttachReference {
        AttachReference {
            child: "favorites".into(),
            parent: "products".into(),
            ref_column: "id".into(),
            ref_value: t("prod_1"),
            set: vec![Assignment {
                column: "note".into(),
                value: Expr::val(t("nice")),
            }],
        }
    }

    #[test]
    fn attach_reference_own_derives_tenant_from_the_scoped_parent() {
        use std::collections::BTreeMap;
        // An OWN scope over `favorites` (child) + `products` (parent), both keyed on tenant_id.
        let scope = Scope {
            column: "tenant_id".into(),
            value: Some(t("A")),
            session: None,
            mode: ScopeMode::Own,
            keys: TableKeys::PerTable(BTreeMap::from([
                (
                    "favorites".to_string(),
                    ResolvedScope::Column("tenant_id".into()),
                ),
                (
                    "products".to_string(),
                    ResolvedScope::Column("tenant_id".into()),
                ),
            ])),
        };
        let (sql, params) =
            compile_attach_reference(&scope, &attach_spec(), Dialect::Sqlite).unwrap();
        // The child tenant is projected from the parent; the source is confined to the caller's own
        // tenant (so the derived tenant is bounded; an unreachable product selects nothing).
        assert_eq!(
            sql,
            "INSERT INTO favorites (note, tenant_id) SELECT ?1, tenant_id FROM products \
             WHERE tenant_id = ?2 AND id = ?3"
        );
        assert_eq!(params, vec![t("nice"), t("A"), t("prod_1")]);
    }

    #[test]
    fn attach_reference_target_confines_parent_to_b_public_and_forces_child_public() {
        // A TARGET write scope: parent `products` confined to B + public; child `favorites` gets its
        // tenant from the parent (=B) + its own public columns force-stamped; `note` is allowlisted.
        use std::collections::{BTreeMap, BTreeSet};
        let public_terms = vec![PublicTermSql::Cmp {
            column: "visible".into(),
            op: CmpOp::Eq,
            value: SqlValue::Boolean(true),
        }];
        let scope = Scope {
            column: "tenant_id".into(),
            value: Some(t("tenant_B")),
            session: None,
            mode: ScopeMode::Own,
            keys: TableKeys::PerTableTarget {
                keys: BTreeMap::from([
                    (
                        "favorites".to_string(),
                        ResolvedScope::Column("tenant_id".into()),
                    ),
                    (
                        "products".to_string(),
                        ResolvedScope::Column("tenant_id".into()),
                    ),
                ]),
                public: BTreeMap::from([
                    ("favorites".to_string(), public_terms.clone()),
                    (
                        "products".to_string(),
                        vec![PublicTermSql::Cmp {
                            column: "published".into(),
                            op: CmpOp::Eq,
                            value: SqlValue::Boolean(true),
                        }],
                    ),
                ]),
                write: BTreeSet::from(["note".to_string()]),
                require_public: true,
            },
        };
        let (sql, params) =
            compile_attach_reference(&scope, &attach_spec(), Dialect::Sqlite).unwrap();
        // Child gets note + tenant(from parent) + forced visible=true; the source (products) is
        // confined to tenant=B AND published=true (the base-table predicate is unqualified).
        assert!(
            sql.contains("INSERT INTO favorites (note, tenant_id, visible)"),
            "{sql}"
        );
        assert!(
            sql.contains("SELECT ?1, tenant_id, ?2 FROM products"),
            "{sql}"
        );
        assert!(
            sql.contains("published = ?") && sql.contains("tenant_id = ?"),
            "parent confined: {sql}"
        );
        assert!(sql.contains("AND id = ?"), "ref selector present: {sql}");
        assert!(
            params.contains(&t("tenant_B")),
            "parent confined to B: {params:?}"
        );
        assert!(
            params.contains(&SqlValue::Boolean(true)),
            "child visible forced + parent published: {params:?}"
        );
    }

    #[test]
    fn attach_reference_target_refuses_a_non_allowlisted_or_visibility_set() {
        use std::collections::{BTreeMap, BTreeSet};
        let scope = Scope {
            column: "tenant_id".into(),
            value: Some(t("tenant_B")),
            session: None,
            mode: ScopeMode::Own,
            keys: TableKeys::PerTableTarget {
                keys: BTreeMap::from([
                    (
                        "favorites".to_string(),
                        ResolvedScope::Column("tenant_id".into()),
                    ),
                    (
                        "products".to_string(),
                        ResolvedScope::Column("tenant_id".into()),
                    ),
                ]),
                public: BTreeMap::from([
                    (
                        "favorites".to_string(),
                        vec![PublicTermSql::Cmp {
                            column: "visible".into(),
                            op: CmpOp::Eq,
                            value: SqlValue::Boolean(true),
                        }],
                    ),
                    (
                        "products".to_string(),
                        vec![PublicTermSql::Cmp {
                            column: "published".into(),
                            op: CmpOp::Eq,
                            value: SqlValue::Boolean(true),
                        }],
                    ),
                ]),
                write: BTreeSet::from(["note".to_string()]),
                require_public: true,
            },
        };
        // `price` isn't allowlisted; `visible` is a visibility column — both refused.
        for bad in ["price", "visible", "tenant_id"] {
            let spec = AttachReference {
                set: vec![Assignment {
                    column: bad.into(),
                    value: Expr::val(t("x")),
                }],
                ..attach_spec()
            };
            assert!(
                matches!(
                    compile_attach_reference(&scope, &spec, Dialect::Sqlite).unwrap_err(),
                    OrmError::TargetWriteColumnDenied(ref c) if c == bad
                ),
                "{bad}"
            );
        }
    }

    #[test]
    fn attach_reference_own_refuses_the_guest_naming_the_tenant_column() {
        use std::collections::BTreeMap;
        // Even under OWN (no write-allowlist gating), the guest may not name the child's tenant
        // column in `set` — the host derives it from the parent; a guest value would collide/forge.
        let scope = Scope {
            column: "tenant_id".into(),
            value: Some(t("A")),
            session: None,
            mode: ScopeMode::Own,
            keys: TableKeys::PerTable(BTreeMap::from([
                (
                    "favorites".to_string(),
                    ResolvedScope::Column("tenant_id".into()),
                ),
                (
                    "products".to_string(),
                    ResolvedScope::Column("tenant_id".into()),
                ),
            ])),
        };
        let spec = AttachReference {
            set: vec![Assignment {
                column: "tenant_id".into(),
                value: Expr::val(t("VICTIM")),
            }],
            ..attach_spec()
        };
        assert!(matches!(
            compile_attach_reference(&scope, &spec, Dialect::Sqlite).unwrap_err(),
            OrmError::TargetWriteColumnDenied(ref c) if c == "tenant_id"
        ));
    }

    #[test]
    fn attach_reference_refuses_a_non_column_table() {
        use std::collections::BTreeMap;
        // `countries` is Unscoped ⇒ not a plain tenant table ⇒ refused as a child/parent.
        let scope = Scope {
            column: "tenant_id".into(),
            value: Some(t("A")),
            session: None,
            mode: ScopeMode::Own,
            keys: TableKeys::PerTable(BTreeMap::from([
                (
                    "favorites".to_string(),
                    ResolvedScope::Column("tenant_id".into()),
                ),
                ("countries".to_string(), ResolvedScope::Unscoped),
            ])),
        };
        let spec = AttachReference {
            parent: "countries".into(),
            ..attach_spec()
        };
        assert!(compile_attach_reference(&scope, &spec, Dialect::Sqlite).is_err());
    }
}