//! Per-node-kind walk dispatch (ADR-0024 §architecture).
//!
//! `walk_node` is the recursive workhorse that the public
//! `walk()` entry calls into for a `CommandNode`'s `shape`. It
//! tries to match `node` starting at `position`, mutating
//! `path` (matched terminals collected in declaration order) and
//! `per_byte` (highlight class assignments) as it goes.
//!
//! The return value distinguishes four cases:
//!
//! - `Matched { end }` — full match, walker consumed up to `end`.
//! - `NoMatch { … }` — node didn't engage at this position. For
//!   `Optional` and `Choice` callers this is benign (try the
//!   next branch / skip the optional); for `Seq` it's only
//!   benign on the first child.
//! - `Incomplete { … }` — node committed (consumed at least one
//!   terminal) but ran out of input. Surfaces as
//!   `WalkOutcome::Incomplete` at the top level.
//! - `Failed { … }` — node committed and a content validator
//!   rejected the value, or a hard structural failure occurred
//!   mid-shape. Surfaces as `WalkOutcome::Mismatch` or
//!   `WalkOutcome::ValidationFailed` at the top level.

use std::collections::HashMap;
use std::sync::{LazyLock, Mutex};

use crate::completion::TableColumn;
use crate::dsl::grammar::{HighlightClass, Node, ValidationError};
use crate::dsl::walker::context::WalkContext;
use crate::dsl::walker::lex_helpers::{
    consume_bare_path, consume_flag, consume_ident, consume_number_literal, consume_string_literal,
    skip_whitespace,
};
use crate::dsl::walker::outcome::{ByteClass, Expectation, MatchedItem, MatchedKind, MatchedPath};

/// Maximum nesting of `Node::Subgrammar` frames (ADR-0026 §1).
///
/// The stratified WHERE-expression grammar descends one
/// `Subgrammar` hop per precedence tier, plus a tier-stack per
/// parenthesised group, so this bounds real expression nesting
/// many parentheses deep — far past any hand-written filter.
/// Its purpose is purely a stack-overflow guard: input nested
/// past the cap (`((((…))))`) fails with a friendly
/// `expression_too_deep` error instead of recursing until the
/// process stack is exhausted.
pub const MAX_SUBGRAMMAR_DEPTH: usize = 64;

/// Memo cache for `Node::DynamicSubgrammar` resolution.
///
/// A factory builds a `Node` from the active `WalkContext`; the
/// resolved Node's combinator children are `&'static`, so it
/// must be `Box::leak`ed. Leaking per walk grows unbounded
/// under per-keystroke completion. Memoizing on the schema
/// state the factory reads means each *distinct* value-list
/// shape leaks exactly once — the total leak is bounded by the
/// number of distinct (table-columns × form) combinations, not
/// by keystroke count (handoff-13 §memoization).
#[derive(PartialEq, Eq, Hash)]
struct DynamicKey {
    /// The factory's function-pointer address. Distinguishes
    /// `column_value_list` from `current_column_value`.
    factory: usize,
    /// Every `WalkContext` field a dynamic factory may read.
    /// A superset of any single factory's true dependencies —
    /// sound (never returns a wrong shape), at worst slightly
    /// over-keyed (an extra leak when an unread field differs).
    current_table_columns: Option<Vec<TableColumn>>,
    current_column: Option<TableColumn>,
    user_listed_columns: Option<Vec<String>>,
}

static DYNAMIC_CACHE: LazyLock<Mutex<HashMap<DynamicKey, &'static Node>>> =
    LazyLock::new(|| Mutex::new(HashMap::new()));

/// Resolve a `DynamicSubgrammar` factory to a `&'static Node`,
/// reusing a previously-leaked Node when the factory's inputs
/// match a cached entry.
fn resolve_dynamic(factory: fn(&WalkContext) -> Node, ctx: &WalkContext) -> &'static Node {
    let key = DynamicKey {
        factory: factory as usize,
        current_table_columns: ctx.current_table_columns.clone(),
        current_column: ctx.current_column.clone(),
        user_listed_columns: ctx.user_listed_columns.clone(),
    };
    let mut cache = DYNAMIC_CACHE
        .lock()
        .expect("dynamic-subgrammar cache mutex poisoned");
    if let Some(&node) = cache.get(&key) {
        return node;
    }
    let resolved: &'static Node = Box::leak(Box::new(factory(ctx)));
    cache.insert(key, resolved);
    resolved
}

#[derive(Debug, Clone)]
pub enum NodeWalkResult {
    Matched {
        end: usize,
        /// Expectations contributed by Optional children that
        /// skipped (matched zero terminals). Walker callers
        /// merge these into the next failure's expected set so
        /// completion sees the full "what could have appeared
        /// here" union, not just the strictly-required next
        /// terminal.
        skipped: Vec<Expectation>,
    },
    /// Did not engage at this position. Caller decides whether
    /// this is benign (Optional, Choice fallthrough) or a hard
    /// failure (Seq mid-shape).
    NoMatch {
        position: usize,
        expected: Vec<Expectation>,
    },
    /// Committed and ran out of input.
    Incomplete {
        position: usize,
        expected: Vec<Expectation>,
    },
    /// Committed and hit a hard mismatch or validator failure.
    Failed { position: usize, kind: FailureKind },
}

const fn matched(end: usize) -> NodeWalkResult {
    NodeWalkResult::Matched {
        end,
        skipped: Vec::new(),
    }
}

#[derive(Debug, Clone)]
pub enum FailureKind {
    Mismatch { expected: Vec<Expectation> },
    Validation(ValidationError),
}

pub fn walk_node(
    source: &str,
    position: usize,
    node: &Node,
    ctx: &mut WalkContext,
    path: &mut MatchedPath,
    per_byte: &mut Vec<ByteClass>,
) -> NodeWalkResult {
    let pos = skip_whitespace(source, position);
    let result = walk_node_inner(source, pos, node, ctx, path, per_byte);
    // ADR-0024 §HintMode-per-node: `pending_hint_mode` records
    // the Hinted slot the cursor is currently inside. Any
    // successful match means the cursor advanced past whatever
    // slot was pending — clear it. This also undoes the leak
    // where a failed `Hinted` branch of a `Choice` sets the
    // mode and the `Choice` then matches via a different
    // branch: that branch's match clears the stale mode.
    if matches!(result, NodeWalkResult::Matched { .. }) {
        ctx.pending_hint_mode = None;
    }
    result
}

fn walk_node_inner(
    source: &str,
    pos: usize,
    node: &Node,
    ctx: &mut WalkContext,
    path: &mut MatchedPath,
    per_byte: &mut Vec<ByteClass>,
) -> NodeWalkResult {
    match node {
        Node::Word(word) => walk_word(source, pos, word, path, per_byte),
        Node::Punct(ch) => walk_punct(source, pos, *ch, path, per_byte),
        Node::Ident {
            source: src,
            role,
            validator,
            highlight_override,
            writes_table,
            writes_column,
            writes_user_listed_column,
            writes_table_alias,
            writes_cte_name,
            writes_projection_alias,
        } => walk_ident(
            source,
            pos,
            *src,
            role,
            *validator,
            *highlight_override,
            *writes_table,
            *writes_column,
            *writes_user_listed_column,
            *writes_table_alias,
            *writes_cte_name,
            *writes_projection_alias,
            ctx,
            path,
            per_byte,
        ),
        Node::NumberLit { validator } => walk_number_lit(source, pos, *validator, path, per_byte),
        Node::Literal(literal) => walk_literal(source, pos, literal, path, per_byte),
        Node::StringLit => walk_string_lit(source, pos, path, per_byte),
        Node::BlobLit => {
            // BlobLit terminals are declared but no current grammar
            // node uses them. Reaching this branch means a future
            // grammar declared a BlobLit without walker support
            // landing — surface as a hard failure so tests catch
            // it loudly rather than silently mis-parsing.
            NodeWalkResult::Failed {
                position: pos,
                kind: FailureKind::Mismatch { expected: vec![] },
            }
        }
        Node::Subgrammar(inner) => walk_subgrammar(source, pos, inner, ctx, path, per_byte),
        Node::ScopedSubgrammar(inner) => {
            walk_scoped_subgrammar(source, pos, inner, ctx, path, per_byte)
        }
        Node::DynamicSubgrammar(factory) => {
            // ADR-0024 §sub-grammars: resolve the inner Node at
            // walk time from the active `WalkContext`, then walk
            // it. The resolved Node's combinator children are
            // `&'static`, so a runtime-built Node has to be
            // leaked. `resolve_dynamic` memoizes on the schema
            // state the factory reads, so each *distinct*
            // value-list shape leaks once — the total leak is
            // bounded by schema size, not by keystroke count
            // (handoff-13 §memoization).
            let resolved = resolve_dynamic(*factory, ctx);
            walk_node(source, pos, resolved, ctx, path, per_byte)
        }
        Node::Lookahead(factory) => {
            // ADR-0024 §Phase D Form-C type-awareness: the
            // factory peeks the source at `pos` (e.g. to tell a
            // Form A column list from a Form C value list) and
            // returns the shape to walk. Not memoized — the
            // result depends on the source — but the factory
            // returns a small node (a Repeated, or a thin
            // DynamicSubgrammar wrapper that delegates to the
            // memoized `column_value_list`), so the per-walk
            // leak is a few bytes, not a whole typed tree.
            let resolved: &'static Node = Box::leak(Box::new(factory(ctx, source, pos)));
            walk_node(source, pos, resolved, ctx, path, per_byte)
        }
        Node::SetColumn(col) => {
            // ADR-0036 Phase 3b: zero-width — establish the active
            // column for the value position that follows, exactly as an
            // `Ident { writes_column: true }` would (current_column for
            // the typed slot's dispatch; pending_value_column for the
            // hint's "for `col`:" framing), but without consuming a
            // column identifier (VALUES positions are positional). The
            // following `SET_VALUE` slot reads `current_column`.
            let col: &crate::completion::TableColumn = col;
            ctx.current_column = Some(col.clone());
            ctx.pending_value_column = Some(col.name.clone());
            NodeWalkResult::Matched {
                end: pos,
                skipped: Vec::new(),
            }
        }
        Node::TypedValueSlot {
            ty,
            column_name,
            inner,
        } => {
            // ADR-0024 §Phase D §typed-value-slots. Tag the
            // pending column type so the hint resolver can emit
            // per-type prose at empty prefix. If a column name
            // is embedded (insert column_value_list path), tag
            // that too so the hint can mention the column by
            // name. Clear on successful inner match — positions
            // BETWEEN typed slots (post-comma, between values)
            // don't carry stale hint state.
            ctx.pending_value_type = Some(*ty);
            if let Some(name) = column_name {
                ctx.pending_value_column = Some((*name).to_string());
            }
            let result = walk_node(source, pos, inner, ctx, path, per_byte);
            if matches!(result, NodeWalkResult::Matched { .. }) {
                ctx.pending_value_type = None;
                ctx.pending_value_column = None;
            }
            result
        }
        Node::Hinted { mode, inner } => {
            // ADR-0024 §HintMode-per-node. Record the grammar's
            // declared hint mode so the hint resolver can read
            // it directly. The `walk_node` wrapper clears it on
            // any successful match (the cursor moved past the
            // slot), so a Hinted slot whose inner fails at EOF
            // leaves the mode set for the resolver to read.
            ctx.pending_hint_mode = Some(*mode);
            walk_node(source, pos, inner, ctx, path, per_byte)
        }
        Node::Concept { topic, inner } => {
            // Issue #37 / ADR clause-concept-hints D1. Walk the
            // inner clause, then record its covered byte span in
            // `ctx.concept_spans` iff the inner *committed* — i.e.
            // matched, ran out mid-clause (Incomplete), or hit a
            // hard failure after engaging (Failed). On `NoMatch`
            // the node never engaged (e.g. a `Choice` tried this
            // branch and it didn't apply), so record nothing — no
            // stale span. `start` is the post-whitespace position
            // the walk wrapper already resolved.
            let result = walk_node(source, pos, inner, ctx, path, per_byte);
            let end = match &result {
                NodeWalkResult::Matched { end, .. } => Some(*end),
                NodeWalkResult::Incomplete { position, .. }
                | NodeWalkResult::Failed { position, .. } => Some(*position),
                NodeWalkResult::NoMatch { .. } => None,
            };
            if let Some(end) = end {
                ctx.concept_spans
                    .push(crate::dsl::walker::context::ConceptSpan {
                        topic,
                        start: pos,
                        end,
                    });
            }
            result
        }
        Node::Flag(name) => walk_flag(source, pos, name, path, per_byte),
        Node::Repeated {
            inner,
            separator,
            min,
        } => walk_repeated(source, pos, inner, *separator, *min, ctx, path, per_byte),
        Node::BarePath => walk_bare_path(source, pos, path, per_byte),
        Node::Choice(children) => walk_choice(source, pos, children, ctx, path, per_byte),
        Node::Seq(children) => walk_seq(source, pos, children, ctx, path, per_byte),
        Node::Optional(child) => walk_optional(source, pos, child, ctx, path, per_byte),
    }
}

fn walk_word(
    source: &str,
    position: usize,
    word: &crate::dsl::grammar::Word,
    path: &mut MatchedPath,
    per_byte: &mut Vec<ByteClass>,
) -> NodeWalkResult {
    // First scan an identifier-shape token at `position`; if
    // none, we definitely don't have this keyword. If one, check
    // it against the word's primary + aliases.
    let Some((start, end)) = consume_ident(source, position) else {
        return NodeWalkResult::NoMatch {
            position,
            expected: vec![Expectation::Word(word.primary)],
        };
    };
    let candidate = &source[start..end];
    if word.matches(candidate) {
        path.push(MatchedItem {
            kind: MatchedKind::Word(word.primary),
            text: candidate.to_string(),
            span: (start, end),
        });
        per_byte.push(ByteClass {
            start,
            end,
            // A keyword may opt into a non-default colour via
            // `Word::type_keyword` (e.g. `double precision`, ADR-0022
            // Amendment 4). Plain keywords leave it `None`.
            class: word.highlight_override.unwrap_or(HighlightClass::Keyword),
        });
        NodeWalkResult::Matched {
            end,
            skipped: Vec::new(),
        }
    } else {
        NodeWalkResult::NoMatch {
            position,
            expected: vec![Expectation::Word(word.primary)],
        }
    }
}

fn walk_punct(
    source: &str,
    position: usize,
    ch: char,
    path: &mut MatchedPath,
    per_byte: &mut Vec<ByteClass>,
) -> NodeWalkResult {
    let bytes = source.as_bytes();
    if position < bytes.len() && bytes[position] == ch as u8 {
        // ADR-0033 Amendment 4: a `-` does not match when it begins an
        // adjacent `--`. The playground supports no `--` line comment, and
        // `--` is the DSL flag marker (e.g. `--all-rows`); treating `--`
        // as two minus operators silently mis-parsed `set x = 42
        // --all-rows` as arithmetic over phantom columns `all`/`rows`.
        // Refusing here makes the SQL expression stop, so the SQL shape
        // fails and dispatch falls back to the DSL flag. Spaced `- -3` is
        // unaffected (the dashes are not adjacent). `Node::Punct('-')` is
        // used only by the SQL expression grammar, so this is scoped to it.
        if ch == '-' && bytes.get(position + 1) == Some(&b'-') {
            return NodeWalkResult::NoMatch {
                position,
                expected: vec![Expectation::Punct(ch)],
            };
        }
        path.push(MatchedItem {
            kind: MatchedKind::Punct(ch),
            text: ch.to_string(),
            span: (position, position + 1),
        });
        per_byte.push(ByteClass {
            start: position,
            end: position + 1,
            class: HighlightClass::Punct,
        });
        matched(position + 1)
    } else {
        NodeWalkResult::NoMatch {
            position,
            expected: vec![Expectation::Punct(ch)],
        }
    }
}

#[allow(clippy::too_many_arguments)]
fn walk_ident(
    source: &str,
    position: usize,
    src: crate::dsl::grammar::IdentSource,
    role: &'static str,
    validator: Option<crate::dsl::grammar::IdentValidator>,
    highlight_override: Option<crate::dsl::grammar::HighlightClass>,
    writes_table: bool,
    writes_column: bool,
    writes_user_listed_column: bool,
    writes_table_alias: bool,
    writes_cte_name: bool,
    writes_projection_alias: bool,
    ctx: &mut WalkContext,
    path: &mut MatchedPath,
    per_byte: &mut Vec<ByteClass>,
) -> NodeWalkResult {
    let Some((start, end)) = consume_ident(source, position) else {
        return NodeWalkResult::NoMatch {
            position,
            expected: vec![Expectation::Ident { role, source: src }],
        };
    };
    let text = source[start..end].to_string();
    if let Some(v) = validator
        && let Err(err) = v(&text)
    {
        return NodeWalkResult::Failed {
            position: start,
            kind: FailureKind::Validation(err),
        };
    }
    // ADR-0024 §Phase D / ADR-0032 §10.1: schema-aware writes.
    // When the ident is a `Tables` source with `writes_table`,
    // resolve the matched name against the schema cache and:
    //   1. populate `current_table` / `current_table_columns`
    //      (preserved for DSL paths that read those fields
    //      directly);
    //   2. push a `TableBinding` onto the top `ScopeFrame`'s
    //      `from_scope` (ADR-0032 §10.1 — for SQL multi-table
    //      contexts).
    if writes_table && matches!(src, crate::dsl::grammar::IdentSource::Tables) {
        let resolved_columns: Vec<crate::completion::TableColumn> = ctx
            .schema
            .and_then(|s| s.columns_for_table(&text).map(<[_]>::to_vec))
            .unwrap_or_default();
        ctx.current_table = Some(text.clone());
        ctx.current_table_columns = if resolved_columns.is_empty() {
            None
        } else {
            Some(resolved_columns.clone())
        };
        if let Some(frame) = ctx.from_scope_stack.last_mut() {
            frame
                .from_scope
                .push(crate::dsl::walker::context::TableBinding {
                    table: text.clone(),
                    alias: None,
                    columns: resolved_columns,
                });
        }
    }
    // ADR-0032 §10.1: the optional `[ AS ] alias` slot on a
    // `from_clause` / `join_clause` table source. The flag is
    // expected on `IdentSource::NewName` slots; the just-pushed
    // binding (the most recent entry in the top frame's
    // `from_scope`) gets its alias set.
    if writes_table_alias
        && let Some(frame) = ctx.from_scope_stack.last_mut()
        && let Some(binding) = frame.from_scope.last_mut()
    {
        binding.alias = Some(text.clone());
    }
    // ADR-0032 §10.3 stage 1 + stage 2: push a placeholder
    // CteBinding into the top (outer) frame before the body's
    // ScopedSubgrammar pushes its own frame. The body can
    // self-reference the CTE name as a table source (WITH
    // RECURSIVE), and downstream CTE-name validators see the
    // binding. Then arm `pending_cte_harvest` so the next
    // ScopedSubgrammar (which is structurally guaranteed to be
    // the CTE body — no intervening scoped subgrammar in CTE
    // syntax) runs the harvest at body-frame exit.
    if writes_cte_name && let Some(frame) = ctx.from_scope_stack.last_mut() {
        frame
            .cte_bindings
            .push(crate::dsl::walker::context::CteBinding {
                name: text.clone(),
                columns: Vec::new(),
            });
        let placeholder_index = frame.cte_bindings.len() - 1;
        ctx.pending_cte_harvest = Some(crate::dsl::walker::context::PendingCteHarvest {
            placeholder_index,
            col_list: Vec::new(),
            cte_name: text.clone(),
            cte_name_span: (start, end),
        });
    }
    // ADR-0032 §10.3: the optional `(c1, c2, …)` rename list
    // between the cte name and `AS`. Each `cte_column` ident
    // appends to the pending harvest's col_list; the harvest
    // applies them as positional renames on the derived
    // columns.
    if role == "cte_column"
        && let Some(pending) = ctx.pending_cte_harvest.as_mut()
    {
        pending.col_list.push(text.clone());
    }
    // ADR-0032 §10.4: projection-list alias accumulator for
    // ORDER BY completion candidates.
    if writes_projection_alias && let Some(frame) = ctx.from_scope_stack.last_mut() {
        frame.projection_aliases.push(text.clone());
    }
    if writes_column && matches!(src, crate::dsl::grammar::IdentSource::Columns) {
        ctx.current_column = ctx.current_table_columns.as_ref().and_then(|cols| {
            cols.iter()
                .find(|c| c.name.eq_ignore_ascii_case(&text))
                .cloned()
        });
        // Surface the column name to the hint resolver too —
        // this is the `update <T> set <col>=` / `where <col>=`
        // path. The matching column's canonical name (from the
        // schema) wins over the user's spelling so the hint
        // mirrors what's in the schema.
        ctx.pending_value_column = ctx
            .current_column
            .as_ref()
            .map(|c| c.name.clone())
            .or_else(|| Some(text.clone()));
    }
    if writes_user_listed_column && matches!(src, crate::dsl::grammar::IdentSource::Columns) {
        // Form A: `insert into <T> (col1, col2, …)`. Append the
        // matched column name to user_listed_columns so the
        // inner `values (…)` slot list mirrors the user's
        // explicit selection. Schema-canonical name wins over
        // user's spelling so downstream lookups (typed slot
        // dispatch, hint rendering) are consistent.
        let canonical = ctx
            .current_table_columns
            .as_ref()
            .and_then(|cols| {
                cols.iter()
                    .find(|c| c.name.eq_ignore_ascii_case(&text))
                    .map(|c| c.name.clone())
            })
            .unwrap_or_else(|| text.clone());
        ctx.user_listed_columns
            .get_or_insert_with(Vec::new)
            .push(canonical);
    }
    path.push(MatchedItem {
        kind: MatchedKind::Ident { role, source: src },
        text,
        span: (start, end),
    });
    per_byte.push(ByteClass {
        start,
        end,
        // A type slot (and any future slot that wants a non-default
        // colour) overrides the otherwise-uniform Identifier class
        // (issue #8 / ADR-0022 Amendment 4).
        class: highlight_override.unwrap_or(HighlightClass::Identifier),
    });
    NodeWalkResult::Matched {
        end,
        skipped: Vec::new(),
    }
}

fn walk_string_lit(
    source: &str,
    position: usize,
    path: &mut MatchedPath,
    per_byte: &mut Vec<ByteClass>,
) -> NodeWalkResult {
    let Some(((start, end), content)) = consume_string_literal(source, position) else {
        return NodeWalkResult::NoMatch {
            position,
            expected: vec![Expectation::StringLit],
        };
    };
    path.push(MatchedItem {
        kind: MatchedKind::StringLit,
        text: content,
        span: (start, end),
    });
    per_byte.push(ByteClass {
        start,
        end,
        class: HighlightClass::String,
    });
    NodeWalkResult::Matched {
        end,
        skipped: Vec::new(),
    }
}

fn walk_literal(
    source: &str,
    position: usize,
    literal: &'static str,
    path: &mut MatchedPath,
    per_byte: &mut Vec<ByteClass>,
) -> NodeWalkResult {
    let bytes = source.as_bytes();
    let lit_bytes = literal.as_bytes();
    if position + lit_bytes.len() > bytes.len() {
        return NodeWalkResult::NoMatch {
            position,
            expected: vec![Expectation::Literal(literal)],
        };
    }
    if &bytes[position..position + lit_bytes.len()] != lit_bytes {
        return NodeWalkResult::NoMatch {
            position,
            expected: vec![Expectation::Literal(literal)],
        };
    }
    // Lookahead: if the literal is a single digit / alphabetic
    // run, the next byte must not extend it (so `1` doesn't
    // half-match `12`).
    let end = position + lit_bytes.len();
    let last = lit_bytes[lit_bytes.len() - 1];
    let last_is_word = last.is_ascii_alphanumeric() || last == b'_';
    if last_is_word && end < bytes.len() {
        let next = bytes[end];
        if next.is_ascii_alphanumeric() || next == b'_' {
            return NodeWalkResult::NoMatch {
                position,
                expected: vec![Expectation::Literal(literal)],
            };
        }
    }
    // Highlight class follows the literal's shape: digits get
    // Number; letters get Keyword; mixed defaults to Keyword.
    let class = if lit_bytes.iter().all(|b| b.is_ascii_digit()) {
        HighlightClass::Number
    } else {
        HighlightClass::Keyword
    };
    path.push(MatchedItem {
        kind: MatchedKind::Word(literal),
        text: literal.to_string(),
        span: (position, end),
    });
    per_byte.push(ByteClass {
        start: position,
        end,
        class,
    });
    NodeWalkResult::Matched {
        end,
        skipped: Vec::new(),
    }
}

fn walk_number_lit(
    source: &str,
    position: usize,
    validator: Option<crate::dsl::grammar::NumberValidator>,
    path: &mut MatchedPath,
    per_byte: &mut Vec<ByteClass>,
) -> NodeWalkResult {
    let Some((start, end)) = consume_number_literal(source, position) else {
        return NodeWalkResult::NoMatch {
            position,
            expected: vec![Expectation::NumberLit],
        };
    };
    let text = source[start..end].to_string();
    if let Some(v) = validator
        && let Err(err) = v(&text)
    {
        return NodeWalkResult::Failed {
            position: start,
            kind: FailureKind::Validation(err),
        };
    }
    path.push(MatchedItem {
        kind: MatchedKind::NumberLit,
        text,
        span: (start, end),
    });
    per_byte.push(ByteClass {
        start,
        end,
        class: HighlightClass::Number,
    });
    NodeWalkResult::Matched {
        end,
        skipped: Vec::new(),
    }
}

fn walk_flag(
    source: &str,
    position: usize,
    name: &'static str,
    path: &mut MatchedPath,
    per_byte: &mut Vec<ByteClass>,
) -> NodeWalkResult {
    let Some((start, end)) = consume_flag(source, position) else {
        return NodeWalkResult::NoMatch {
            position,
            expected: vec![Expectation::Flag(name)],
        };
    };
    // `consume_flag` guarantees `start..end` covers `--<body>`.
    let body = &source[start + 2..end];
    if body != name {
        return NodeWalkResult::NoMatch {
            position,
            expected: vec![Expectation::Flag(name)],
        };
    }
    path.push(MatchedItem {
        kind: MatchedKind::Flag(name),
        text: source[start..end].to_string(),
        span: (start, end),
    });
    per_byte.push(ByteClass {
        start,
        end,
        class: HighlightClass::Flag,
    });
    NodeWalkResult::Matched {
        end,
        skipped: Vec::new(),
    }
}

#[allow(clippy::too_many_arguments)]
fn walk_repeated(
    source: &str,
    position: usize,
    inner: &Node,
    separator: Option<&Node>,
    min: usize,
    ctx: &mut WalkContext,
    path: &mut MatchedPath,
    per_byte: &mut Vec<ByteClass>,
) -> NodeWalkResult {
    let mut cur = position;
    let mut count = 0_usize;
    let mut last_expected: Option<Vec<Expectation>> = None;
    // Trailing-optional expectations carried by the most recently
    // matched item — e.g. `asc`/`desc` after an ORDER BY sort
    // item, or a projection's `as` alias. Surfaced when the list
    // ends cleanly at an item boundary so completion still offers
    // the optional suffix the user could type next (handoff 31 —
    // the `desc` follow-up to F5). The separator itself is
    // deliberately NOT surfaced.
    let mut last_item_skipped: Vec<Expectation> = Vec::new();
    // Set when the loop stops because the separator did not match
    // at an item boundary (a clean end of list), as opposed to an
    // inner mismatch past an already-consumed separator. Only at a
    // clean boundary are the last item's trailing optionals valid
    // continuations at the cursor.
    let mut ended_at_item_boundary = false;
    loop {
        let saved_path_len = path.items.len();
        let saved_byte_len = per_byte.len();
        // Track whether the separator successfully consumed
        // before the inner attempt. Used below to distinguish
        // "user typed `,` then stopped at EOF — mid-typing the
        // next item" from "list naturally ended at the inner
        // boundary".
        let mut sep_consumed_to: Option<usize> = None;
        let result = if count == 0 {
            walk_node(source, cur, inner, ctx, path, per_byte)
        } else if let Some(sep) = separator {
            let sep_saved_path = path.items.len();
            let sep_saved_byte = per_byte.len();
            match walk_node(source, cur, sep, ctx, path, per_byte) {
                NodeWalkResult::Matched { end, .. } => {
                    sep_consumed_to = Some(end);
                    walk_node(source, end, inner, ctx, path, per_byte)
                }
                NodeWalkResult::NoMatch { .. } => {
                    path.items.truncate(sep_saved_path);
                    per_byte.truncate(sep_saved_byte);
                    ended_at_item_boundary = true;
                    break;
                }
                other => return other,
            }
        } else {
            walk_node(source, cur, inner, ctx, path, per_byte)
        };
        match result {
            NodeWalkResult::Matched { end, skipped } => {
                cur = end;
                count += 1;
                last_item_skipped = skipped;
            }
            NodeWalkResult::NoMatch {
                expected,
                position: inner_pos,
            } => {
                // Mid-typing-the-next-item recovery: if the
                // separator just consumed and the inner failed
                // at EOF, the user is partway through typing the
                // next item — propagate as Incomplete so the
                // outer walker classifies the input as
                // mid-typing rather than rolling the separator
                // back and producing a structural Mismatch at
                // the separator position.
                //
                // Without this branch, `insert into T (a, ` at
                // EOF would roll back the `,`, then the outer
                // `(`-list expected `)` at `cur`, see the
                // separator instead, and report a definite
                // error at the separator. Real users hit this
                // every time they type a comma and pause.
                if let Some(post_sep) = sep_consumed_to {
                    let post_ws = skip_whitespace(source, post_sep);
                    if post_ws >= source.len() {
                        return NodeWalkResult::Incomplete {
                            position: inner_pos,
                            expected,
                        };
                    }
                }
                path.items.truncate(saved_path_len);
                per_byte.truncate(saved_byte_len);
                last_expected = Some(expected);
                break;
            }
            other => return other,
        }
    }
    if count < min {
        return NodeWalkResult::NoMatch {
            position: cur,
            expected: last_expected.unwrap_or_default(),
        };
    }
    // The "could continue" expectations become this Repeated's
    // `skipped` set so the caller's expected-set surfaces them at
    // completion time. When the list ended cleanly at an item
    // boundary, that is the last item's trailing optionals (e.g.
    // `asc`/`desc`); otherwise it is whatever the final inner
    // attempt expected.
    let skipped = if ended_at_item_boundary {
        last_item_skipped
    } else {
        last_expected.unwrap_or_default()
    };
    NodeWalkResult::Matched { end: cur, skipped }
}

fn walk_bare_path(
    source: &str,
    position: usize,
    path: &mut MatchedPath,
    per_byte: &mut Vec<ByteClass>,
) -> NodeWalkResult {
    let Some((start, end)) = consume_bare_path(source, position) else {
        return NodeWalkResult::NoMatch {
            position,
            expected: vec![Expectation::BarePath],
        };
    };
    let text = source[start..end].to_string();
    path.push(MatchedItem {
        kind: MatchedKind::BarePath,
        text,
        span: (start, end),
    });
    per_byte.push(ByteClass {
        start,
        end,
        class: HighlightClass::String,
    });
    NodeWalkResult::Matched {
        end,
        skipped: Vec::new(),
    }
}

fn walk_choice(
    source: &str,
    position: usize,
    children: &[Node],
    ctx: &mut WalkContext,
    path: &mut MatchedPath,
    per_byte: &mut Vec<ByteClass>,
) -> NodeWalkResult {
    let mut all_expected: Vec<Expectation> = Vec::new();
    for child in children {
        let saved_path_len = path.items.len();
        let saved_byte_len = per_byte.len();
        match walk_node(source, position, child, ctx, path, per_byte) {
            m @ NodeWalkResult::Matched { .. } => return m,
            NodeWalkResult::NoMatch { expected, .. } => {
                path.items.truncate(saved_path_len);
                per_byte.truncate(saved_byte_len);
                merge_expected(&mut all_expected, expected);
            }
            other => return other,
        }
    }
    NodeWalkResult::NoMatch {
        position,
        expected: all_expected,
    }
}

fn walk_seq(
    source: &str,
    position: usize,
    children: &[Node],
    ctx: &mut WalkContext,
    path: &mut MatchedPath,
    per_byte: &mut Vec<ByteClass>,
) -> NodeWalkResult {
    let mut cur = position;
    let mut idx = 0;
    // Carries expectations from skipped-Optional children so
    // that a NoMatch on a later child reports the union of "you
    // could have typed any of these" — making the completion
    // engine see optional connectives that haven't been typed.
    let mut pending_skipped: Vec<Expectation> = Vec::new();
    for child in children {
        let path_before = path.items.len();
        match walk_node(source, cur, child, ctx, path, per_byte) {
            NodeWalkResult::Matched { end, skipped } => {
                if end == cur {
                    // Child matched zero terminals (Optional skipped,
                    // empty Repeated, empty Seq). Accumulate its
                    // would-be expectations into pending.
                    for e in skipped {
                        if !pending_skipped.contains(&e) {
                            pending_skipped.push(e);
                        }
                    }
                } else {
                    // Child consumed terminals — the "missing optional"
                    // window closed; reset the pending list.
                    pending_skipped.clear();
                    pending_skipped.extend(skipped);
                }
                cur = end;
                idx += 1;
            }
            NodeWalkResult::NoMatch {
                position,
                mut expected,
            } => {
                // Merge pending skipped-optional expectations with this
                // child's expected set.
                for e in std::mem::take(&mut pending_skipped) {
                    if !expected.contains(&e) {
                        expected.push(e);
                    }
                }
                if idx == 0 {
                    return NodeWalkResult::NoMatch { position, expected };
                }
                let post_ws = skip_whitespace(source, position);
                if post_ws >= source.len() {
                    return NodeWalkResult::Incomplete {
                        position: post_ws,
                        expected,
                    };
                }
                return NodeWalkResult::Failed {
                    position: post_ws,
                    kind: FailureKind::Mismatch { expected },
                };
            }
            NodeWalkResult::Incomplete {
                position,
                mut expected,
            } => {
                // Only merge the skipped-Optional expectations when
                // the Incomplete-producing child consumed nothing
                // (path didn't grow): the cursor still sits at the
                // optional boundary, so those optionals are genuine
                // alternatives. If the child committed terminals
                // (e.g. `order by` consumed, now awaiting a sort
                // item) the cursor has moved *past* the skipped
                // optionals — clauses positioned before this child
                // are no longer valid continuations, so dropping
                // `pending_skipped` keeps them out of the expected
                // set (handoff 30 §3.3, F5).
                let child_consumed = path.items.len() > path_before;
                if !child_consumed {
                    for e in std::mem::take(&mut pending_skipped) {
                        if !expected.contains(&e) {
                            expected.push(e);
                        }
                    }
                }
                return NodeWalkResult::Incomplete { position, expected };
            }
            NodeWalkResult::Failed { position, kind } => {
                return NodeWalkResult::Failed { position, kind };
            }
        }
    }
    NodeWalkResult::Matched {
        end: cur,
        skipped: pending_skipped,
    }
}

/// Issue #26: when an `Optional` is skipped (its inner didn't engage),
/// stash any `IntroProse` hint the inner left in `pending_hint_mode`
/// into the surviving slot before it is cleared by this empty match.
/// `position` is where the optional was skipped — the resolver compares
/// it to the cursor so the hint only shows while the cursor sits at that
/// optional, not after a later clause consumes input past it. Only
/// `IntroProse` is carried (it is the "introduce an optional position"
/// mode); `ProseOnly` / `ForceProse` mark active slots and reach the
/// resolver through the normal `pending_hint_mode` path.
const fn capture_skipped_intro_hint(ctx: &mut WalkContext, position: usize) {
    if let Some(crate::dsl::grammar::HintMode::IntroProse(key)) = ctx.pending_hint_mode {
        ctx.surviving_intro_hint = Some((key, position));
    }
}

fn walk_optional(
    source: &str,
    position: usize,
    child: &Node,
    ctx: &mut WalkContext,
    path: &mut MatchedPath,
    per_byte: &mut Vec<ByteClass>,
) -> NodeWalkResult {
    let saved_path_len = path.items.len();
    let saved_byte_len = per_byte.len();
    let result = walk_node(source, position, child, ctx, path, per_byte);
    let inner_committed = path.items.len() > saved_path_len;
    match result {
        m @ NodeWalkResult::Matched { .. } => m,
        NodeWalkResult::NoMatch { expected, .. } => {
            // Inner didn't engage at all — skip the Optional
            // but carry the inner's expectations so the caller's
            // expected-set sees them.
            capture_skipped_intro_hint(ctx, position);
            path.items.truncate(saved_path_len);
            per_byte.truncate(saved_byte_len);
            NodeWalkResult::Matched {
                end: position,
                skipped: expected,
            }
        }
        NodeWalkResult::Incomplete {
            position: p,
            expected,
        } if !inner_committed => {
            // Inner reported Incomplete without consuming
            // anything — same as NoMatch from the user's
            // perspective. Roll back and skip.
            capture_skipped_intro_hint(ctx, position);
            path.items.truncate(saved_path_len);
            per_byte.truncate(saved_byte_len);
            let _ = p;
            NodeWalkResult::Matched {
                end: position,
                skipped: expected,
            }
        }
        NodeWalkResult::Failed {
            kind: FailureKind::Mismatch { expected },
            ..
        } if !inner_committed => {
            // Inner reported Mismatch without consuming
            // anything — roll back and skip.
            path.items.truncate(saved_path_len);
            per_byte.truncate(saved_byte_len);
            NodeWalkResult::Matched {
                end: position,
                skipped: expected,
            }
        }
        // Inner committed (consumed at least one terminal) but
        // then ran out / hit a mismatch. Propagate the failure
        // up — the user is mid-typing the optional's content and
        // we'd lose their intent by rolling back. (Pre-fix
        // behavior matched chumsky's `or_not` rollback, but
        // that conflates "Form A in progress" with "Form C with
        // trailing junk" — see e.g. `insert into T (a, b, c)
        // values (1, 2, 3` losing the `values (…)` partial.)
        // Validation failures already propagate as a separate
        // branch below.
        propagated @ (NodeWalkResult::Incomplete { .. } | NodeWalkResult::Failed { .. }) => {
            propagated
        }
    }
}

/// Walk a `&'static Node` reference once (ADR-0026 §2).
///
/// The reference indirection is what lets a named `static`
/// grammar fragment recurse: `Seq` / `Choice` embed children by
/// value and so cannot close a cycle, but a `Subgrammar` node
/// holding a `&'static Node` can point back into an enclosing
/// fragment. The stratified WHERE-expression grammar's
/// `( or_expr )` branch and `not_expr` self-reference both
/// recurse this way.
///
/// `WalkContext::subgrammar_depth` counts active frames. Past
/// `MAX_SUBGRAMMAR_DEPTH` the walk fails with a friendly
/// `expression_too_deep` validation error rather than
/// overflowing the process stack. The depth is saved on entry
/// and restored on exit unconditionally, so a speculatively-
/// walked branch that a `Choice` later rolls back leaves the
/// counter clean.
fn walk_subgrammar(
    source: &str,
    pos: usize,
    inner: &'static Node,
    ctx: &mut WalkContext,
    path: &mut MatchedPath,
    per_byte: &mut Vec<ByteClass>,
) -> NodeWalkResult {
    let saved_depth = ctx.subgrammar_depth;
    ctx.subgrammar_depth += 1;
    if ctx.subgrammar_depth > MAX_SUBGRAMMAR_DEPTH {
        ctx.subgrammar_depth = saved_depth;
        return NodeWalkResult::Failed {
            position: pos,
            kind: FailureKind::Validation(ValidationError {
                message_key: "parse.custom.expression_too_deep",
                args: Vec::new(),
            }),
        };
    }
    let result = walk_node(source, pos, inner, ctx, path, per_byte);
    ctx.subgrammar_depth = saved_depth;
    result
}

/// Walk a `ScopedSubgrammar` reference once (ADR-0032 §10.2).
///
/// Pushes a fresh `ScopeFrame` onto `from_scope_stack` on
/// entry and pops it back on exit. The push/pop is
/// unconditional — a speculatively-walked branch that a
/// `Choice` later rolls back leaves the stack clean. Shares
/// the `subgrammar_depth` counter with the plain `Subgrammar`
/// variant so the depth cap fires uniformly.
fn walk_scoped_subgrammar(
    source: &str,
    pos: usize,
    inner: &'static Node,
    ctx: &mut WalkContext,
    path: &mut MatchedPath,
    per_byte: &mut Vec<ByteClass>,
) -> NodeWalkResult {
    let saved_depth = ctx.subgrammar_depth;
    ctx.subgrammar_depth += 1;
    if ctx.subgrammar_depth > MAX_SUBGRAMMAR_DEPTH {
        ctx.subgrammar_depth = saved_depth;
        return NodeWalkResult::Failed {
            position: pos,
            kind: FailureKind::Validation(ValidationError {
                message_key: "parse.custom.expression_too_deep",
                args: Vec::new(),
            }),
        };
    }
    // ADR-0032 §10.3 stage 2 — pick up a pending CTE harvest
    // request armed by the immediately-preceding cte_name ident.
    // Clear unconditionally: a non-matching body must not leave
    // stale state for a later unrelated ScopedSubgrammar.
    let pending_cte = ctx.pending_cte_harvest.take();
    ctx.from_scope_stack
        .push(crate::dsl::walker::context::ScopeFrame::default());
    let result = walk_node(source, pos, inner, ctx, path, per_byte);

    // Harvest happens only on a fully-matched body. Speculative
    // walks that NoMatch / Incomplete / Fail leave the placeholder
    // empty (the outer-frame state is also discarded in the
    // speculative path, so this is correct).
    if let (Some(req), NodeWalkResult::Matched { end, .. }) = (pending_cte, &result) {
        run_cte_harvest(ctx, path, source, pos, *end, &req);
    }

    ctx.from_scope_stack.pop();
    ctx.subgrammar_depth = saved_depth;
    result
}

/// Run the §10.3 stage-2 harvest after a CTE body's
/// `ScopedSubgrammar` matched, while the body's frame is still
/// on top of `from_scope_stack`.
///
/// Reads the body's projection items out of the matched path's
/// byte range, classifies each via the six derivation rules,
/// applies any `(col-list)` positional rename, and writes the
/// derived columns into the placeholder `CteBinding` in the
/// outer (now `len - 2`) frame.
fn run_cte_harvest(
    ctx: &mut WalkContext,
    path: &MatchedPath,
    _source: &str,
    body_start: usize,
    body_end: usize,
    req: &crate::dsl::walker::context::PendingCteHarvest,
) {
    use crate::dsl::walker::context::{CteColumn, ScopeFrame};
    use crate::dsl::walker::outcome::{MatchedItem, MatchedKind};

    // The body's frame is at the top of the stack while the
    // harvest runs. Need this for from_scope lookups in the
    // derivation rules.
    let body_frame: &ScopeFrame = match ctx.from_scope_stack.last() {
        Some(f) => f,
        None => return,
    };

    // Compute body_depth = paren-balance over path items strictly
    // before body_start. The `(` immediately preceding the body
    // is at the outer depth and increments to the body's depth;
    // body_start is INSIDE that paren.
    let mut prefix_depth: i32 = 0;
    for item in &path.items {
        if item.span.0 >= body_start {
            break;
        }
        match item.kind {
            MatchedKind::Punct('(') => prefix_depth += 1,
            MatchedKind::Punct(')') => prefix_depth -= 1,
            _ => {}
        }
    }
    let body_depth = prefix_depth;

    // The path items strictly inside the body byte range.
    let body_items: Vec<&MatchedItem> = path
        .items
        .iter()
        .filter(|i| i.span.0 >= body_start && i.span.1 <= body_end)
        .collect();

    // Track depth within the body. First leg's projection list
    // begins at the first body-depth SELECT and ends at the
    // first body-depth FROM/WHERE/etc OR set-op keyword OR end.
    let mut depth = body_depth;
    let mut select_idx: Option<usize> = None;
    let mut end_idx: usize = body_items.len();
    for (i, item) in body_items.iter().enumerate() {
        let cur = depth;
        match item.kind {
            MatchedKind::Punct('(') => depth += 1,
            MatchedKind::Punct(')') => depth -= 1,
            _ => {}
        }
        if cur != body_depth {
            continue;
        }
        match item.kind {
            MatchedKind::Word("select") if select_idx.is_none() => {
                select_idx = Some(i + 1); // start of projection list
            }
            MatchedKind::Word(
                "from" | "where" | "group" | "having" | "order" | "limit" | "offset" | "union"
                | "intersect" | "except",
            ) if select_idx.is_some() => {
                end_idx = i;
                break;
            }
            _ => {}
        }
    }
    let Some(start_idx) = select_idx else {
        return;
    };
    if start_idx >= end_idx {
        return;
    }

    // Split the projection-list slice into individual items by
    // commas at body_depth.
    let mut item_slices: Vec<&[&MatchedItem]> = Vec::new();
    let mut depth_scan = body_depth;
    let mut slice_start = start_idx;
    for i in start_idx..end_idx {
        let cur = depth_scan;
        match body_items[i].kind {
            MatchedKind::Punct('(') => depth_scan += 1,
            MatchedKind::Punct(')') => depth_scan -= 1,
            MatchedKind::Punct(',') if cur == body_depth => {
                item_slices.push(&body_items[slice_start..i]);
                slice_start = i + 1;
            }
            _ => {}
        }
    }
    if slice_start < end_idx {
        item_slices.push(&body_items[slice_start..end_idx]);
    }

    // Classify each projection item per ADR-0032 §10.3.
    let mut derived: Vec<CteColumn> = Vec::new();
    for slice in item_slices {
        classify_projection_item(slice, body_frame, &ctx.from_scope_stack, &mut derived);
    }

    // Apply (c1, c2, …) positional rename if provided. Types
    // are preserved; names overridden by the col_list. Arity
    // mismatch is emitted as `diagnostic.cte_arity_mismatch`
    // on the cte_name span before any padding/truncation so
    // the diagnostic carries the *true* derived count.
    if !req.col_list.is_empty() {
        let declared = req.col_list.len();
        let actual = derived.len();
        if declared != actual {
            use crate::dsl::walker::outcome::{Diagnostic, Severity};
            ctx.pending_diagnostics.push(Diagnostic {
                severity: Severity::Error,
                span: req.cte_name_span,
                message: crate::friendly::translate(
                    "diagnostic.cte_arity_mismatch",
                    &[
                        ("cte", &req.cte_name as &dyn std::fmt::Display),
                        ("declared", &declared as &dyn std::fmt::Display),
                        ("actual", &actual as &dyn std::fmt::Display),
                    ],
                ),
            });
        }
        for (i, name) in req.col_list.iter().enumerate() {
            if let Some(col) = derived.get_mut(i) {
                col.name = Some(name.clone());
            } else {
                // col_list has MORE entries than derived items —
                // synthesize a typeless slot with the declared
                // name so qualified-prefix completion still
                // surfaces it.
                derived.push(CteColumn {
                    name: Some(name.clone()),
                    type_: None,
                });
            }
        }
        // Truncate any extras when derived > declared, so the
        // CTE's externally visible arity matches the col-list
        // declaration. (The diagnostic above already captured
        // the original derived count.)
        if derived.len() > declared {
            derived.truncate(declared);
        }
    }

    // Write into the outer frame's placeholder.
    let stack_len = ctx.from_scope_stack.len();
    if stack_len >= 2
        && let Some(outer) = ctx.from_scope_stack.get_mut(stack_len - 2)
        && let Some(placeholder) = outer.cte_bindings.get_mut(req.placeholder_index)
    {
        placeholder.columns = derived;
    }
}

/// Classify one projection item by examining its leading
/// terminals and append its derived CteColumn(s) to `out`. The
/// six rules of ADR-0032 §10.3.
fn classify_projection_item(
    slice: &[&crate::dsl::walker::outcome::MatchedItem],
    body_frame: &crate::dsl::walker::context::ScopeFrame,
    scope_stack: &[crate::dsl::walker::context::ScopeFrame],
    out: &mut Vec<crate::dsl::walker::context::CteColumn>,
) {
    use crate::dsl::grammar::IdentSource;
    use crate::dsl::walker::context::CteColumn;
    use crate::dsl::walker::outcome::MatchedKind;

    // Strip an optional trailing `[AS] alias` from the slice so
    // shape detection can examine just the expression part.
    let (expr_slice, alias) = strip_trailing_alias(slice);

    // Rule 1: `*` — every column from body_frame.from_scope.
    // When a binding represents a CTE reference (its columns are
    // empty because it wasn't a base-table lookup), resolve
    // through to the in-scope CteBinding so nested CTEs project
    // correctly.
    if expr_slice.len() == 1 && matches!(expr_slice[0].kind, MatchedKind::Punct('*')) {
        for binding in &body_frame.from_scope {
            for col in expand_binding(binding, scope_stack) {
                out.push(col);
            }
        }
        return;
    }

    // Rule 2: `t.*` — every column from binding `t`.
    if expr_slice.len() == 3
        && matches!(
            expr_slice[0].kind,
            MatchedKind::Ident {
                role: "qualified_star_qualifier",
                ..
            }
        )
        && matches!(expr_slice[1].kind, MatchedKind::Punct('.'))
        && matches!(expr_slice[2].kind, MatchedKind::Punct('*'))
    {
        let qual = &expr_slice[0].text;
        if let Some(binding) = body_frame.from_scope.iter().find(|b| {
            b.alias
                .as_deref()
                .is_some_and(|a| a.eq_ignore_ascii_case(qual))
                || b.table.eq_ignore_ascii_case(qual)
        }) {
            for col in expand_binding(binding, scope_stack) {
                out.push(col);
            }
        }
        return;
    }

    // Rule 3: bare `col` — a single sql_expr_ident terminal.
    if expr_slice.len() == 1
        && matches!(
            expr_slice[0].kind,
            MatchedKind::Ident {
                source: IdentSource::Columns,
                role: "sql_expr_ident",
            }
        )
    {
        let col_text = &expr_slice[0].text;
        let resolved_type = resolve_bare_column_type_in_frame(body_frame, scope_stack, col_text);
        let name = alias.unwrap_or_else(|| col_text.clone());
        out.push(CteColumn {
            name: Some(name),
            type_: resolved_type,
        });
        return;
    }

    // Rule 4: qualified `t.col` — three-token shape with the
    // sql_expr_qualified_ref role on the tail ident.
    if expr_slice.len() == 3
        && matches!(
            expr_slice[0].kind,
            MatchedKind::Ident {
                source: IdentSource::Columns,
                role: "sql_expr_ident",
            }
        )
        && matches!(expr_slice[1].kind, MatchedKind::Punct('.'))
        && matches!(
            expr_slice[2].kind,
            MatchedKind::Ident {
                source: IdentSource::Columns,
                role: "sql_expr_qualified_ref",
            }
        )
    {
        let qual = &expr_slice[0].text;
        let col_text = &expr_slice[2].text;
        let resolved_type = resolve_qualified_column_type(body_frame, scope_stack, qual, col_text);
        let name = alias.unwrap_or_else(|| col_text.clone());
        out.push(CteColumn {
            name: Some(name),
            type_: resolved_type,
        });
        return;
    }

    // Rule 5 / 6: computed expression — name = alias if present,
    // else None. Type = None either way (ADR-0032 Amendment 1).
    out.push(CteColumn {
        name: alias,
        type_: None,
    });
}

/// Peel a trailing `[AS] <ident>` off the projection-item slice
/// if present. Returns (expr_slice_without_alias, Some(alias))
/// or (slice, None) if no alias is detected.
fn strip_trailing_alias<'a>(
    slice: &'a [&'a crate::dsl::walker::outcome::MatchedItem],
) -> (
    &'a [&'a crate::dsl::walker::outcome::MatchedItem],
    Option<String>,
) {
    use crate::dsl::grammar::IdentSource;
    use crate::dsl::walker::outcome::MatchedKind;

    if slice.is_empty() {
        return (slice, None);
    }
    let last = slice[slice.len() - 1];
    if matches!(
        last.kind,
        MatchedKind::Ident {
            source: IdentSource::NewName,
            role: "projection_alias",
        }
    ) {
        // Optional preceding `AS` keyword.
        if slice.len() >= 2 && matches!(slice[slice.len() - 2].kind, MatchedKind::Word("as")) {
            return (&slice[..slice.len() - 2], Some(last.text.clone()));
        }
        return (&slice[..slice.len() - 1], Some(last.text.clone()));
    }
    (slice, None)
}

fn resolve_bare_column_type_in_frame(
    frame: &crate::dsl::walker::context::ScopeFrame,
    scope_stack: &[crate::dsl::walker::context::ScopeFrame],
    column: &str,
) -> Option<crate::dsl::types::Type> {
    let mut found = None;
    for binding in &frame.from_scope {
        for col in expand_binding(binding, scope_stack) {
            if col
                .name
                .as_deref()
                .is_some_and(|n| n.eq_ignore_ascii_case(column))
            {
                if found.is_some() {
                    return None; // ambiguous — no type
                }
                found = col.type_;
            }
        }
    }
    found
}

fn resolve_qualified_column_type(
    frame: &crate::dsl::walker::context::ScopeFrame,
    scope_stack: &[crate::dsl::walker::context::ScopeFrame],
    qualifier: &str,
    column: &str,
) -> Option<crate::dsl::types::Type> {
    let binding = frame.from_scope.iter().find(|b| {
        b.alias
            .as_deref()
            .is_some_and(|a| a.eq_ignore_ascii_case(qualifier))
            || b.table.eq_ignore_ascii_case(qualifier)
    })?;
    expand_binding(binding, scope_stack)
        .into_iter()
        .find(|c| {
            c.name
                .as_deref()
                .is_some_and(|n| n.eq_ignore_ascii_case(column))
        })
        .and_then(|c| c.type_)
}

/// Resolve a `TableBinding` to its column list as `CteColumn`s.
///
/// Base-table bindings carry typed `TableColumn`s populated from
/// the schema cache — convert them directly. CTE-source bindings
/// (the binding's `columns` is empty because the FROM name
/// didn't match a base table) look up the matching `CteBinding`
/// in any in-scope frame and return its `columns` verbatim.
///
/// This is the bridge that lets a nested CTE's outer harvest see
/// the inner CTE's derived columns: the body's `FROM inner`
/// produces an empty-columns binding, but `expand_binding`
/// resolves it through the inner CteBinding (which has its
/// derived columns by the time the outer harvest runs, because
/// the inner body's harvest fires on inner-body exit, before the
/// outer body exits).
///
/// A self-reference inside a `WITH RECURSIVE` body sees the
/// placeholder (empty columns) and the resolution returns empty
/// — that's correct, since the harvest only fires on the
/// non-recursive (first) leg per §10.3.
fn expand_binding(
    binding: &crate::dsl::walker::context::TableBinding,
    scope_stack: &[crate::dsl::walker::context::ScopeFrame],
) -> Vec<crate::dsl::walker::context::CteColumn> {
    use crate::dsl::walker::context::CteColumn;

    if !binding.columns.is_empty() {
        return binding
            .columns
            .iter()
            .map(|c| CteColumn {
                name: Some(c.name.clone()),
                type_: Some(c.user_type),
            })
            .collect();
    }
    for frame in scope_stack.iter().rev() {
        if let Some(cte) = frame
            .cte_bindings
            .iter()
            .find(|c| c.name.eq_ignore_ascii_case(&binding.table))
        {
            return cte.columns.clone();
        }
    }
    Vec::new()
}

fn merge_expected(dst: &mut Vec<Expectation>, src: Vec<Expectation>) {
    for e in src {
        if !dst.contains(&e) {
            dst.push(e);
        }
    }
}

#[cfg(test)]
mod tests {
    use super::{
        DYNAMIC_CACHE, FailureKind, MAX_SUBGRAMMAR_DEPTH, NodeWalkResult, resolve_dynamic,
        walk_node,
    };
    use crate::dsl::grammar::{Node, Word};
    use crate::dsl::walker::context::WalkContext;
    use crate::dsl::walker::outcome::MatchedPath;

    // Recursive test grammar for the `Subgrammar` node
    // (ADR-0026 §2): `x` | `( <self> )`. `NESTED_GROUP` reaches
    // back to `NESTED` through `Subgrammar(&NESTED)` — the cycle
    // a by-value `Seq` slice could not express.
    static NESTED_GROUP: &[Node] = &[
        Node::Punct('('),
        Node::Subgrammar(&NESTED),
        Node::Punct(')'),
    ];
    static NESTED_CHOICES: &[Node] = &[Node::Seq(NESTED_GROUP), Node::Word(Word::keyword("x"))];
    static NESTED: Node = Node::Choice(NESTED_CHOICES);

    fn walk_nested(input: &str) -> NodeWalkResult {
        let mut ctx = WalkContext::new();
        let mut path = MatchedPath::new();
        let mut per_byte = Vec::new();
        let result = walk_node(input, 0, &NESTED, &mut ctx, &mut path, &mut per_byte);
        assert_eq!(
            ctx.subgrammar_depth, 0,
            "subgrammar_depth must be restored to 0 after the walk",
        );
        result
    }

    #[test]
    fn subgrammar_walks_a_recursive_grammar() {
        for input in ["x", "(x)", "((x))", "(((x)))"] {
            assert!(
                matches!(walk_nested(input), NodeWalkResult::Matched { .. }),
                "{input:?} should match the recursive Subgrammar grammar",
            );
        }
    }

    #[test]
    fn subgrammar_depth_cap_allows_exactly_the_limit() {
        let input = format!(
            "{}x{}",
            "(".repeat(MAX_SUBGRAMMAR_DEPTH),
            ")".repeat(MAX_SUBGRAMMAR_DEPTH),
        );
        assert!(
            matches!(walk_nested(&input), NodeWalkResult::Matched { .. }),
            "exactly MAX_SUBGRAMMAR_DEPTH nested groups should still walk",
        );
    }

    #[test]
    fn subgrammar_depth_cap_rejects_pathological_nesting() {
        let over = MAX_SUBGRAMMAR_DEPTH + 1;
        let input = format!("{}x{}", "(".repeat(over), ")".repeat(over));
        match walk_nested(&input) {
            NodeWalkResult::Failed {
                kind: FailureKind::Validation(err),
                ..
            } => assert_eq!(err.message_key, "parse.custom.expression_too_deep"),
            other => {
                panic!("expected an expression_too_deep failure, got {other:?}")
            }
        }
    }

    /// Trivial factory — ignores the context. The memo behaviour
    /// is keyed on the context, not the factory's output, so a
    /// constant factory is enough to exercise the cache.
    fn const_factory(_ctx: &WalkContext) -> Node {
        Node::Word(Word::keyword("memo_probe"))
    }

    #[test]
    fn resolve_dynamic_memoizes_identical_context() {
        let ctx = WalkContext::new();
        let first = resolve_dynamic(const_factory, &ctx);
        let second = resolve_dynamic(const_factory, &ctx);
        // Same factory + same context → the leaked Node is
        // reused (pointer identity), so repeated per-keystroke
        // walks against an unchanged schema leak only once.
        assert!(
            std::ptr::eq(first, second),
            "identical context should hit the memo cache",
        );
    }

    #[test]
    fn resolve_dynamic_distinct_context_does_not_share() {
        let ctx_a = WalkContext::new();
        let mut ctx_b = WalkContext::new();
        ctx_b.user_listed_columns = Some(vec!["Col".to_string()]);
        let a = resolve_dynamic(const_factory, &ctx_a);
        let b = resolve_dynamic(const_factory, &ctx_b);
        // Different context state → different cache key → a
        // separate entry (and a separate one-time leak).
        assert!(
            !std::ptr::eq(a, b),
            "distinct context must not collide in the memo cache",
        );
    }

    #[test]
    fn resolve_dynamic_cache_is_populated() {
        let ctx = WalkContext::new();
        let _ = resolve_dynamic(const_factory, &ctx);
        let populated = !DYNAMIC_CACHE.lock().expect("cache lock").is_empty();
        assert!(populated, "resolve_dynamic should populate the memo cache",);
    }

    // ---- ScopedSubgrammar (ADR-0032 §10.2) -----------------------

    // Recursive test grammar parallel to NESTED, but using
    // `ScopedSubgrammar` for the recursion — exercises the
    // push/pop discipline. Same shape (`x` | `( <self> )`),
    // different recursion variant.
    static SCOPED_NESTED_GROUP: &[Node] = &[
        Node::Punct('('),
        Node::ScopedSubgrammar(&SCOPED_NESTED),
        Node::Punct(')'),
    ];
    static SCOPED_NESTED_CHOICES: &[Node] = &[
        Node::Seq(SCOPED_NESTED_GROUP),
        Node::Word(Word::keyword("x")),
    ];
    static SCOPED_NESTED: Node = Node::Choice(SCOPED_NESTED_CHOICES);

    fn walk_scoped_nested(input: &str) -> (NodeWalkResult, usize) {
        let mut ctx = WalkContext::new();
        let mut path = MatchedPath::new();
        let mut per_byte = Vec::new();
        let baseline_frames = ctx.from_scope_stack.len();
        let result = walk_node(input, 0, &SCOPED_NESTED, &mut ctx, &mut path, &mut per_byte);
        assert_eq!(
            ctx.subgrammar_depth, 0,
            "subgrammar_depth must be restored to 0 after the walk",
        );
        assert_eq!(
            ctx.from_scope_stack.len(),
            baseline_frames,
            "from_scope_stack must be restored after the walk",
        );
        (result, baseline_frames)
    }

    #[test]
    fn scoped_subgrammar_walks_a_recursive_grammar() {
        for input in ["x", "(x)", "((x))", "(((x)))"] {
            let (result, _) = walk_scoped_nested(input);
            assert!(
                matches!(result, NodeWalkResult::Matched { .. }),
                "{input:?} should match the recursive ScopedSubgrammar grammar",
            );
        }
    }

    #[test]
    fn scoped_subgrammar_shares_depth_cap_with_subgrammar() {
        // Over the cap with ScopedSubgrammar recursion fails the
        // same way as the Subgrammar test above — both variants
        // share `WalkContext::subgrammar_depth`.
        let over = MAX_SUBGRAMMAR_DEPTH + 1;
        let input = format!("{}x{}", "(".repeat(over), ")".repeat(over));
        match walk_scoped_nested(&input).0 {
            NodeWalkResult::Failed {
                kind: FailureKind::Validation(err),
                ..
            } => assert_eq!(err.message_key, "parse.custom.expression_too_deep"),
            other => {
                panic!("expected expression_too_deep on pathological scoped nesting, got {other:?}",)
            }
        }
    }

    #[test]
    fn scoped_subgrammar_baseline_frame_is_always_present() {
        let ctx = WalkContext::new();
        assert_eq!(
            ctx.from_scope_stack.len(),
            1,
            "WalkContext::new should seed exactly one bottom frame",
        );
    }

    // ---- from_scope binding population (ADR-0032 §10.1) ----

    /// Walk a top-level SQL SELECT and return the bottom frame's
    /// `from_scope` after the walk completes. Used to verify that
    /// `writes_table` / `writes_table_alias` populate bindings.
    fn from_scope_after_walk(input: &str) -> Vec<crate::dsl::walker::context::TableBinding> {
        let mut ctx = WalkContext::new();
        let mut path = MatchedPath::new();
        let mut per_byte = Vec::new();
        let result = walk_node(
            input,
            0,
            &crate::dsl::grammar::sql_select::SQL_SELECT_STATEMENT,
            &mut ctx,
            &mut path,
            &mut per_byte,
        );
        assert!(
            matches!(result, NodeWalkResult::Matched { .. }),
            "{input:?} should match: got {result:?}"
        );
        // The bottom frame survives the walk; any ScopedSubgrammar
        // frames have been popped by now.
        ctx.from_scope_stack[0].from_scope.clone()
    }

    #[test]
    fn single_from_table_pushes_one_binding() {
        let bindings = from_scope_after_walk("select * from users");
        assert_eq!(bindings.len(), 1);
        assert_eq!(bindings[0].table, "users");
        assert_eq!(bindings[0].alias, None);
    }

    #[test]
    fn as_alias_on_from_table_is_captured() {
        let bindings = from_scope_after_walk("select * from users as u");
        assert_eq!(bindings.len(), 1);
        assert_eq!(bindings[0].table, "users");
        assert_eq!(bindings[0].alias, Some("u".to_string()));
    }

    #[test]
    fn bare_alias_on_from_table_is_captured() {
        let bindings = from_scope_after_walk("select * from users u");
        assert_eq!(bindings.len(), 1);
        assert_eq!(bindings[0].table, "users");
        assert_eq!(bindings[0].alias, Some("u".to_string()));
    }

    #[test]
    fn join_pushes_a_second_binding() {
        let bindings = from_scope_after_walk("select * from a join b on x = y");
        assert_eq!(bindings.len(), 2);
        assert_eq!(bindings[0].table, "a");
        assert_eq!(bindings[1].table, "b");
    }

    #[test]
    fn join_with_aliases() {
        let bindings = from_scope_after_walk("select * from a as x join b as y on x.id = y.id");
        assert_eq!(bindings.len(), 2);
        assert_eq!(bindings[0].table, "a");
        assert_eq!(bindings[0].alias, Some("x".to_string()));
        assert_eq!(bindings[1].table, "b");
        assert_eq!(bindings[1].alias, Some("y".to_string()));
    }

    #[test]
    fn three_way_join_pushes_three_bindings() {
        let bindings =
            from_scope_after_walk("select * from a join b on x = y left join c on y = z");
        assert_eq!(bindings.len(), 3);
        assert_eq!(bindings[0].table, "a");
        assert_eq!(bindings[1].table, "b");
        assert_eq!(bindings[2].table, "c");
    }

    #[test]
    fn subquery_bindings_do_not_leak_to_outer_scope() {
        // The inner `(SELECT id FROM inner_t)` pushes its
        // binding into the inner scope frame; on exit, the frame
        // pops and the inner binding is gone. The outer scope's
        // from_scope still contains only `outer_t`.
        let bindings =
            from_scope_after_walk("select * from outer_t where id in (select id from inner_t)");
        assert_eq!(bindings.len(), 1);
        assert_eq!(bindings[0].table, "outer_t");
    }

    #[test]
    fn cte_body_bindings_do_not_leak_to_outer_scope() {
        // The CTE body's `from base_table` pushes into the CTE
        // body's scope frame; on body-frame exit, the inner
        // binding goes away. The outer scope contains only
        // the CTE-name reference `cte_x`.
        let bindings =
            from_scope_after_walk("with cte_x as (select * from base_table) select * from cte_x");
        assert_eq!(bindings.len(), 1);
        assert_eq!(bindings[0].table, "cte_x");
    }

    #[test]
    fn from_scope_empty_for_select_without_from() {
        let bindings = from_scope_after_walk("select 1");
        assert!(bindings.is_empty());
    }

    // ---- cte_bindings & projection_aliases (ADR-0032 §10.3 / §10.4) ----

    /// Walk a top-level SELECT and return the bottom frame's
    /// `cte_bindings` and `projection_aliases` after the walk.
    fn frame_state_after_walk(
        input: &str,
    ) -> (Vec<crate::dsl::walker::context::CteBinding>, Vec<String>) {
        let mut ctx = WalkContext::new();
        let mut path = MatchedPath::new();
        let mut per_byte = Vec::new();
        let result = walk_node(
            input,
            0,
            &crate::dsl::grammar::sql_select::SQL_SELECT_STATEMENT,
            &mut ctx,
            &mut path,
            &mut per_byte,
        );
        assert!(
            matches!(result, NodeWalkResult::Matched { .. }),
            "{input:?} should match: got {result:?}"
        );
        let bottom = &ctx.from_scope_stack[0];
        (
            bottom.cte_bindings.clone(),
            bottom.projection_aliases.clone(),
        )
    }

    #[test]
    fn cte_name_pushes_placeholder_binding() {
        let (ctes, _) = frame_state_after_walk("with cte_x as (select 1) select * from cte_x");
        assert_eq!(ctes.len(), 1);
        assert_eq!(ctes[0].name, "cte_x");
        // §10.3 stage-2 harvest produces one CteColumn per
        // projection item. `SELECT 1` is a computed expression
        // without an alias → `CteColumn { name: None, type_:
        // None }`.
        assert_eq!(ctes[0].columns.len(), 1);
        assert!(ctes[0].columns[0].name.is_none());
        assert!(ctes[0].columns[0].type_.is_none());
    }

    #[test]
    fn multiple_ctes_push_in_order() {
        let (ctes, _) =
            frame_state_after_walk("with a as (select 1), b as (select 2) select * from b");
        assert_eq!(ctes.len(), 2);
        assert_eq!(ctes[0].name, "a");
        assert_eq!(ctes[1].name, "b");
    }

    #[test]
    fn recursive_cte_name_visible_in_body() {
        // The CTE name `r` is pushed BEFORE the body's
        // ScopedSubgrammar enters, so the body's `from r`
        // reference is structurally valid (parses).
        let (ctes, _) = frame_state_after_walk(
            "with recursive r as (select 1 union all select 2 from r) select * from r",
        );
        assert_eq!(ctes.len(), 1);
        assert_eq!(ctes[0].name, "r");
    }

    #[test]
    fn projection_aliases_captured_via_as_form() {
        let (_, aliases) = frame_state_after_walk("select a as alpha, b as beta from t");
        assert_eq!(aliases, vec!["alpha".to_string(), "beta".to_string()]);
    }

    #[test]
    fn projection_aliases_captured_via_bare_form() {
        let (_, aliases) = frame_state_after_walk("select a alpha, b beta from t");
        assert_eq!(aliases, vec!["alpha".to_string(), "beta".to_string()]);
    }

    #[test]
    fn projection_aliases_mixed_forms() {
        let (_, aliases) =
            frame_state_after_walk("select a as alpha, b beta, c, d as delta from t");
        assert_eq!(
            aliases,
            vec!["alpha".to_string(), "beta".to_string(), "delta".to_string()]
        );
    }

    #[test]
    fn projection_aliases_empty_when_no_aliases() {
        let (_, aliases) = frame_state_after_walk("select a, b from t");
        assert!(aliases.is_empty());
    }

    #[test]
    fn cte_body_aliases_do_not_leak_to_outer_scope() {
        // The body's projection_aliases live in the body's
        // scope frame, which pops on exit. The outer frame's
        // projection_aliases only carries the outer SELECT's
        // own aliases.
        let (_, aliases) = frame_state_after_walk(
            "with x as (select a as inner_a from t) select b as outer_b from x",
        );
        assert_eq!(aliases, vec!["outer_b".to_string()]);
    }

    // ---- §10.3 stage-2 CTE column-derivation harvest ----

    /// Schema-aware walk variant — returns the outer frame's
    /// `cte_bindings` after walking the input.
    fn cte_bindings_after_walk_with_schema(
        input: &str,
        schema: &crate::completion::SchemaCache,
    ) -> Vec<crate::dsl::walker::context::CteBinding> {
        let mut ctx = WalkContext::with_schema(schema);
        ctx.mode = crate::mode::Mode::Advanced;
        let mut path = MatchedPath::new();
        let mut per_byte = Vec::new();
        let result = walk_node(
            input,
            0,
            &crate::dsl::grammar::sql_select::SQL_SELECT_STATEMENT,
            &mut ctx,
            &mut path,
            &mut per_byte,
        );
        assert!(
            matches!(result, NodeWalkResult::Matched { .. }),
            "{input:?} should match: got {result:?}"
        );
        ctx.from_scope_stack[0].cte_bindings.clone()
    }

    fn schema_users() -> crate::completion::SchemaCache {
        use crate::completion::{SchemaCache, TableColumn};
        use crate::dsl::types::Type;
        let mut s = SchemaCache::default();
        s.tables.push("users".to_string());
        s.columns.push("id".to_string());
        s.columns.push("name".to_string());
        s.columns.push("age".to_string());
        s.table_columns.insert(
            "users".to_string(),
            vec![
                TableColumn {
                    name: "id".to_string(),
                    user_type: Type::Int,
                    not_null: false,
                    has_default: false,
                },
                TableColumn {
                    name: "name".to_string(),
                    user_type: Type::Text,
                    not_null: false,
                    has_default: false,
                },
                TableColumn {
                    name: "age".to_string(),
                    user_type: Type::Int,
                    not_null: false,
                    has_default: false,
                },
            ],
        );
        s
    }

    #[test]
    fn cte_harvest_star_expands_from_scope() {
        // Rule 1: `SELECT *` body — derived columns = every
        // column from the body frame's from_scope, with types.
        let schema = schema_users();
        let ctes = cte_bindings_after_walk_with_schema(
            "with x as (select * from users) select * from x",
            &schema,
        );
        assert_eq!(ctes.len(), 1);
        assert_eq!(ctes[0].columns.len(), 3);
        assert_eq!(ctes[0].columns[0].name.as_deref(), Some("id"));
        assert_eq!(ctes[0].columns[0].type_, Some(crate::dsl::types::Type::Int),);
        assert_eq!(ctes[0].columns[1].name.as_deref(), Some("name"));
        assert_eq!(
            ctes[0].columns[1].type_,
            Some(crate::dsl::types::Type::Text),
        );
        assert_eq!(ctes[0].columns[2].name.as_deref(), Some("age"));
    }

    #[test]
    fn cte_harvest_qualified_star_expands_one_binding() {
        // Rule 2: `t.*` — every column from binding `t`.
        let schema = schema_users();
        let ctes = cte_bindings_after_walk_with_schema(
            "with x as (select u.* from users u) select * from x",
            &schema,
        );
        assert_eq!(ctes.len(), 1);
        assert_eq!(ctes[0].columns.len(), 3);
        assert_eq!(ctes[0].columns[0].name.as_deref(), Some("id"));
    }

    #[test]
    fn cte_harvest_bare_ref_with_alias() {
        // Rule 5 variant: `col AS alias` — name = alias, type
        // preserved from the source column.
        let schema = schema_users();
        let ctes = cte_bindings_after_walk_with_schema(
            "with x as (select name as label from users) select * from x",
            &schema,
        );
        assert_eq!(ctes[0].columns.len(), 1);
        assert_eq!(ctes[0].columns[0].name.as_deref(), Some("label"));
        assert_eq!(
            ctes[0].columns[0].type_,
            Some(crate::dsl::types::Type::Text),
        );
    }

    #[test]
    fn cte_harvest_bare_ref_without_alias_uses_column_name() {
        // Rule 3: bare `col` — name = column name, type from
        // source column.
        let schema = schema_users();
        let ctes = cte_bindings_after_walk_with_schema(
            "with x as (select age from users) select * from x",
            &schema,
        );
        assert_eq!(ctes[0].columns.len(), 1);
        assert_eq!(ctes[0].columns[0].name.as_deref(), Some("age"));
        assert_eq!(ctes[0].columns[0].type_, Some(crate::dsl::types::Type::Int),);
    }

    #[test]
    fn cte_harvest_qualified_ref() {
        // Rule 4: `t.col` — name = column, type from binding.
        let schema = schema_users();
        let ctes = cte_bindings_after_walk_with_schema(
            "with x as (select u.name from users u) select * from x",
            &schema,
        );
        assert_eq!(ctes[0].columns.len(), 1);
        assert_eq!(ctes[0].columns[0].name.as_deref(), Some("name"));
        assert_eq!(
            ctes[0].columns[0].type_,
            Some(crate::dsl::types::Type::Text),
        );
    }

    #[test]
    fn cte_harvest_computed_no_alias_is_unnamed() {
        // Rule 6: computed expression without alias → name =
        // None, type = None.
        let schema = schema_users();
        let ctes = cte_bindings_after_walk_with_schema(
            "with x as (select age + 1 from users) select * from x",
            &schema,
        );
        assert_eq!(ctes[0].columns.len(), 1);
        assert!(ctes[0].columns[0].name.is_none());
        assert!(ctes[0].columns[0].type_.is_none());
    }

    #[test]
    fn cte_harvest_computed_with_alias() {
        // Rule 5: computed expression with alias → name =
        // alias, type = None (Amendment 1).
        let schema = schema_users();
        let ctes = cte_bindings_after_walk_with_schema(
            "with x as (select age + 1 as years from users) select * from x",
            &schema,
        );
        assert_eq!(ctes[0].columns.len(), 1);
        assert_eq!(ctes[0].columns[0].name.as_deref(), Some("years"));
        assert!(ctes[0].columns[0].type_.is_none());
    }

    #[test]
    fn cte_harvest_compound_takes_first_leg() {
        // For UNION / INTERSECT / EXCEPT bodies, columns come
        // from the first leg per ADR-0032 §10.3.
        let schema = schema_users();
        let ctes = cte_bindings_after_walk_with_schema(
            "with x as (select id from users union select age from users) select * from x",
            &schema,
        );
        // First leg: `select id from users` → one column `id`,
        // type Int. Second leg ignored.
        assert_eq!(ctes[0].columns.len(), 1);
        assert_eq!(ctes[0].columns[0].name.as_deref(), Some("id"));
    }

    #[test]
    fn cte_harvest_recursive_uses_non_recursive_leg() {
        // WITH RECURSIVE — the first (non-recursive) leg
        // dictates columns. The recursive leg self-references
        // the CTE name; we don't try to introspect.
        let schema = schema_users();
        let ctes = cte_bindings_after_walk_with_schema(
            "with recursive r as (select id from users union all select id from r) select * from r",
            &schema,
        );
        assert_eq!(ctes[0].columns.len(), 1);
        assert_eq!(ctes[0].columns[0].name.as_deref(), Some("id"));
    }

    #[test]
    fn cte_harvest_nested_with_in_cte_body() {
        // Nested WITH inside a CTE body now parses (ADR-0032
        // §10.3 — inner subqueries may declare their own CTEs).
        // The outer CTE's body has its own scope and its own
        // CTE inside it. The outer's `*` projects from its
        // body's FROM, which references the inner CTE; the
        // inner CTE's columns flow through `expand_binding`.
        let schema = schema_users();
        let ctes = cte_bindings_after_walk_with_schema(
            "with outer_cte as (with inner_cte as (select id, name from users) select * from inner_cte) select * from outer_cte",
            &schema,
        );
        let outer = ctes
            .iter()
            .find(|c| c.name == "outer_cte")
            .expect("outer_cte binding");
        assert_eq!(outer.columns.len(), 2);
        assert_eq!(outer.columns[0].name.as_deref(), Some("id"));
        assert_eq!(outer.columns[0].type_, Some(crate::dsl::types::Type::Int),);
        assert_eq!(outer.columns[1].name.as_deref(), Some("name"));
        assert_eq!(outer.columns[1].type_, Some(crate::dsl::types::Type::Text),);
    }

    #[test]
    fn cte_harvest_sibling_b_sees_a_columns() {
        // Sibling CTEs at the same level. When `b`'s body
        // walks, the outer scope's cte_bindings already
        // contains `a` (with harvested columns) and `b`'s
        // placeholder. `b`'s `FROM a` produces an empty-columns
        // TableBinding which `expand_binding` resolves through
        // the in-scope `a` CteBinding. So `*` in `b`'s body
        // expands to `a`'s columns.
        let schema = schema_users();
        let ctes = cte_bindings_after_walk_with_schema(
            "with a as (select id, name from users), b as (select * from a) select * from b",
            &schema,
        );
        let b = ctes.iter().find(|c| c.name == "b").expect("b binding");
        assert_eq!(b.columns.len(), 2);
        assert_eq!(b.columns[0].name.as_deref(), Some("id"));
        assert_eq!(b.columns[0].type_, Some(crate::dsl::types::Type::Int),);
        assert_eq!(b.columns[1].name.as_deref(), Some("name"));
        assert_eq!(b.columns[1].type_, Some(crate::dsl::types::Type::Text),);
    }

    #[test]
    fn cte_harvest_col_list_renames_positionally() {
        // `WITH x(a, b, c) AS (SELECT * FROM users)` —
        // positional rename overrides derived names; types
        // preserved.
        let schema = schema_users();
        let ctes = cte_bindings_after_walk_with_schema(
            "with x (a, b, c) as (select * from users) select * from x",
            &schema,
        );
        assert_eq!(ctes[0].columns.len(), 3);
        assert_eq!(ctes[0].columns[0].name.as_deref(), Some("a"));
        assert_eq!(ctes[0].columns[0].type_, Some(crate::dsl::types::Type::Int),);
        assert_eq!(ctes[0].columns[1].name.as_deref(), Some("b"));
        assert_eq!(
            ctes[0].columns[1].type_,
            Some(crate::dsl::types::Type::Text),
        );
        assert_eq!(ctes[0].columns[2].name.as_deref(), Some("c"));
    }
}