rdbms-playground/src/echo.rs

//! The DSL → SQL teaching echo renderer (ADR-0038).
//!
//! Maps a **DSL-form** `Command` to the equivalent advanced-mode SQL, so
//! a learner who typed the simple-mode form reads off how to spell it in
//! SQL (ADR-0030 §10). The output obeys the **copy-paste contract**
//! (ADR-0038 §1): it is runnable advanced-mode SQL in the playground's
//! own type vocabulary (`Type::keyword()`), so it round-trips through the
//! advanced walker. The standard-first dialect (ADR-0035 Amendment 2)
//! governs statement shape; the playground keywords fill the type slots.
//!
//! `None` means "no echo" — a command in Bucket C (ADR-0038 §7) or a form
//! not yet covered by the renderer. The caller (the runtime's `ExecuteDsl`
//! dispatch) only invokes this for DSL-form commands submitted in an
//! advanced effective mode (ADR-0037).

use crate::app::EffectiveMode;
use crate::dsl::ReferentialAction;
use crate::dsl::Command;
use crate::dsl::command::{
    ColumnSpec, CompareOp, Constraint, ConstraintKind, Expr, Operand, Predicate, RowFilter,
};
use crate::dsl::value::Value;

/// The dimmed `Executing SQL:` prefix on a teaching-echo line
/// (ADR-0038 §4 styled-runs polish).
///
/// The App appends the SQL to this prefix when building a
/// `OutputKind::TeachingEcho` line; `ui::render_output_line` strips
/// it back off, renders the prefix dimmed (`theme.muted`), and lexes
/// the rest in advanced mode for syntax highlighting — same treatment
/// the input echo gets. The trailing space is part of the constant so
/// the SQL position is predictable for the lexer's byte ranges.
pub const TEACHING_ECHO_LABEL: &str = "Executing SQL: ";

/// The teaching echo for a command submitted under `mode`, or `None`.
///
/// Fires only in an advanced effective mode (`AdvancedPersistent` /
/// `AdvancedOneShot`) — simple mode stays uncluttered (ADR-0030 §10) — and
/// only for a DSL-form command that has an echo (`command_to_sql`; a
/// `Sql*` / app command returns `None`). This is the runtime's gate;
/// replay never reaches it (it bypasses the spawn). (ADR-0037 + ADR-0038)
///
/// This pre-execution path covers every Bucket A row whose echo is a pure
/// function of the `Command`. The one exception is `show data`, whose
/// limited form orders by the table's primary key (not on the `Command`);
/// that is built post-execution by [`echo_for_query`] (ADR-0038 §4).
#[must_use]
pub fn echo_for(command: &Command, mode: EffectiveMode) -> Option<Vec<String>> {
    if mode.is_advanced() {
        command_to_sql(command).map(|sql| vec![sql])
    } else {
        None
    }
}

/// The teaching echo for a `Query`-outcome command (ADR-0038).
///
/// `show data` is the only DSL-form query that echoes — a `Command::Select`
/// is already SQL, so it has none. Fires only in an advanced effective mode.
///
/// `primary_key` is the table's primary-key columns, sourced from the
/// schema *after* execution: the limited form (`show data T limit n`)
/// orders by the primary key for a stable "first n", and that column list
/// is not carried on the `Command`. It is unused when the query is
/// unlimited. This is the one Bucket A row that needs schema info beyond
/// the `Command` (handoff §5 / ADR-0038 §4).
#[must_use]
pub fn echo_for_query(
    command: &Command,
    mode: EffectiveMode,
    primary_key: &[String],
) -> Option<Vec<String>> {
    if !mode.is_advanced() {
        return None;
    }
    match command {
        Command::ShowData {
            name,
            filter,
            limit,
        } => Some(vec![render_show_data(name, filter.as_ref(), *limit, primary_key)]),
        _ => None,
    }
}

/// Render the equivalent advanced-mode SQL for a DSL-form command, or
/// `None` when it has no echo (Bucket C, or a form covered elsewhere —
/// `show data` goes through [`echo_for_query`]).
#[must_use]
pub fn command_to_sql(command: &Command) -> Option<String> {
    match command {
        Command::CreateTable {
            name,
            columns,
            primary_key,
        } => Some(render_create_table(name, columns, primary_key)),
        Command::AddColumn {
            table,
            column,
            ty,
            not_null,
            unique,
            default,
            check,
        } => {
            let mut s = format!("ALTER TABLE {table} ADD COLUMN {column} {}", ty.keyword());
            append_constraints(&mut s, *not_null, *unique, default.as_ref(), check.as_ref());
            Some(s)
        }
        // `--cascade` also drops the column's covering indexes — a
        // multi-statement echo (Bucket B / category 2, Phase 2). The plain
        // drop is a single statement.
        Command::DropColumn {
            table,
            column,
            cascade,
        } => (!cascade).then(|| format!("ALTER TABLE {table} DROP COLUMN {column}")),
        Command::RenameColumn { table, old, new } => {
            Some(format!("ALTER TABLE {table} RENAME COLUMN {old} TO {new}"))
        }
        // The headline form (every conversion mode emits it); the
        // `--dont-convert` *caveat* line is category-3 (ADR-0038 §6, Phase 3).
        // `SET DATA TYPE` is the canonical ISO spelling (ADR-0035 Amendment 2).
        Command::ChangeColumnType {
            table, column, ty, ..
        } => Some(format!(
            "ALTER TABLE {table} ALTER COLUMN {column} SET DATA TYPE {}",
            ty.keyword()
        )),
        Command::AddConstraint {
            table,
            column,
            constraint,
        } => Some(match constraint {
            Constraint::NotNull => {
                format!("ALTER TABLE {table} ALTER COLUMN {column} SET NOT NULL")
            }
            Constraint::Default(v) => format!(
                "ALTER TABLE {table} ALTER COLUMN {column} SET DEFAULT {}",
                value_to_sql_literal(v)
            ),
            Constraint::Unique => format!("ALTER TABLE {table} ADD UNIQUE ({column})"),
            Constraint::Check(e) => {
                format!("ALTER TABLE {table} ADD CHECK ({})", expr_to_sql(e))
            }
        }),
        Command::DropConstraint {
            table,
            column,
            kind,
        } => match kind {
            ConstraintKind::NotNull => {
                Some(format!("ALTER TABLE {table} ALTER COLUMN {column} DROP NOT NULL"))
            }
            ConstraintKind::Default => {
                Some(format!("ALTER TABLE {table} ALTER COLUMN {column} DROP DEFAULT"))
            }
            // A column-level UNIQUE / CHECK is anonymous in our model —
            // no portable name to DROP CONSTRAINT by, so no echo (Bucket C,
            // ADR-0035 Amendment 2 residual gap / ADR-0038 §7).
            ConstraintKind::Unique | ConstraintKind::Check => None,
        },
        // Only the `--all-rows` fall-throughs echo: a WHERE-filtered
        // update / delete routes SQL-first in advanced mode (`Sql*`), so it
        // is already SQL and never reaches here as a DSL-form command
        // (ADR-0033 Amendment 3/4, ADR-0038 §7).
        Command::Update {
            table,
            assignments,
            filter: RowFilter::AllRows,
        } => Some(format!("UPDATE {table} SET {}", render_assignments(assignments))),
        Command::Delete {
            table,
            filter: RowFilter::AllRows,
        } => Some(format!("DELETE FROM {table}")),
        // Remaining forms: Bucket B (resolved names / multi-line — Phase 2),
        // category-3 prose (Phase 3), Bucket C (no echo), `show data`
        // (`echo_for_query`), and the `Sql*` / `App` variants.
        _ => None,
    }
}

/// `CREATE TABLE <name> (<col defs>[, PRIMARY KEY (…)])` in the
/// playground's type vocabulary. A single first-column primary key is
/// rendered inline (`id serial PRIMARY KEY`), matching the rebuild
/// generator's rule (ADR-0035 §4a) and the round-trip surface; a compound
/// key becomes a table-level `PRIMARY KEY (a, b)`.
fn render_create_table(name: &str, columns: &[ColumnSpec], primary_key: &[String]) -> String {
    let inline_pk =
        primary_key.len() == 1 && columns.first().is_some_and(|c| c.name == primary_key[0]);
    let col_defs: Vec<String> = columns
        .iter()
        .map(|c| {
            let mut s = format!("{} {}", c.name, c.ty.keyword());
            if inline_pk && c.name == primary_key[0] {
                s.push_str(" PRIMARY KEY");
            }
            // The same column-constraint suffix `add column` emits (ADR-0029):
            // simple-mode `create table` can carry `default` / `check` too, so
            // the echo must render them or it is not equivalent (§1 contract).
            append_constraints(&mut s, c.not_null, c.unique, c.default.as_ref(), c.check.as_ref());
            s
        })
        .collect();
    if primary_key.len() > 1 {
        format!(
            "CREATE TABLE {name} ({}, PRIMARY KEY ({}))",
            col_defs.join(", "),
            primary_key.join(", "),
        )
    } else {
        format!("CREATE TABLE {name} ({})", col_defs.join(", "))
    }
}

/// `SELECT * FROM <name> [WHERE …] [ORDER BY <pk> LIMIT n]` — the `show
/// data` echo (ADR-0038 §7). When `limit` is set the worker orders by the
/// primary key for a stable "first n" (`build_query_data_sql`); the echo
/// reproduces that `ORDER BY` so it has the same effect (§1). The
/// `ORDER BY` is dropped when the table has no primary key, matching the
/// worker.
fn render_show_data(
    name: &str,
    filter: Option<&Expr>,
    limit: Option<u64>,
    primary_key: &[String],
) -> String {
    let mut s = format!("SELECT * FROM {name}");
    if let Some(expr) = filter {
        s.push_str(&format!(" WHERE {}", expr_to_sql(expr)));
    }
    if let Some(n) = limit {
        if !primary_key.is_empty() {
            s.push_str(&format!(" ORDER BY {}", primary_key.join(", ")));
        }
        s.push_str(&format!(" LIMIT {n}"));
    }
    s
}

/// `CREATE INDEX <name> ON <table> (col, …)` — the `add index` echo
/// (ADR-0038 §7 Bucket B). `name` is the resolved index name (the
/// user-given `as N` or the worker's auto-name `<table>_<cols>_idx`);
/// the runtime sources it from the post-execution table description.
pub(crate) fn render_create_index(name: &str, table: &str, columns: &[String]) -> String {
    format!("CREATE INDEX {name} ON {table} ({})", columns.join(", "))
}

/// `DROP INDEX <name>` — the positional-form `drop index` echo
/// (ADR-0038 §7 Bucket B). The runtime resolves the name **pre-execution**
/// (the index is gone post-exec) by describing the table and matching by
/// column set.
pub(crate) fn render_drop_index(name: &str) -> String {
    format!("DROP INDEX {name}")
}

/// `ALTER TABLE <C> ADD CONSTRAINT <name> FOREIGN KEY (<cc>) REFERENCES
/// <P> (<pc>) [ON DELETE …] [ON UPDATE …]` — the `add relationship` echo
/// (ADR-0038 §7 Bucket B), without `--create-fk`. Multi-line `--create-fk`
/// is a separate renderer (Slice 2b). The `ON DELETE` / `ON UPDATE`
/// clauses are emitted only when the action is non-default — the standard
/// (`NO ACTION`) is the implicit default in the SQL grammar, and emitting
/// it would clutter the echo without changing meaning.
pub(crate) fn render_add_relationship(
    name: &str,
    parent_table: &str,
    parent_column: &str,
    child_table: &str,
    child_column: &str,
    on_delete: ReferentialAction,
    on_update: ReferentialAction,
) -> String {
    let mut s = format!(
        "ALTER TABLE {child_table} ADD CONSTRAINT {name} FOREIGN KEY ({child_column}) REFERENCES {parent_table} ({parent_column})"
    );
    if on_delete != ReferentialAction::default_action() {
        s.push_str(&format!(" ON DELETE {}", on_delete.sql_clause()));
    }
    if on_update != ReferentialAction::default_action() {
        s.push_str(&format!(" ON UPDATE {}", on_update.sql_clause()));
    }
    s
}

/// `ALTER TABLE <C> DROP CONSTRAINT <name>` — the `drop relationship`
/// echo (ADR-0038 §7 Bucket B). The runtime resolves both `name` (for an
/// `Endpoints` selector) and `child_table` (for a `Named` selector) **pre-
/// execution** via a describe — for a `Named` drop the worker resolves
/// the child table from metadata, which is gone after the drop.
pub(crate) fn render_drop_relationship(name: &str, child_table: &str) -> String {
    format!("ALTER TABLE {child_table} DROP CONSTRAINT {name}")
}

/// Multi-line echo for `drop column T.c --cascade` (ADR-0038 §7 Bucket B,
/// category 2). Emits one `DROP INDEX <name>` line per covering index
/// (ADR-0025) followed by the final `ALTER TABLE T DROP COLUMN c`. The
/// SQL `DROP COLUMN` refuses an indexed column, so the indexes must come
/// first — the lines *are* the explanation, no prose (§6 category 2).
/// With zero dropped indexes (`--cascade` set on an unindexed column) the
/// result is a single line, still correct.
pub(crate) fn render_drop_column_cascade(
    table: &str,
    column: &str,
    dropped_indexes: &[String],
) -> Vec<String> {
    let mut lines: Vec<String> = dropped_indexes
        .iter()
        .map(|name| format!("DROP INDEX {name}"))
        .collect();
    lines.push(format!("ALTER TABLE {table} DROP COLUMN {column}"));
    lines
}

/// Multi-line echo for `add 1:n relationship … --create-fk` when the
/// child column was *newly created* (ADR-0038 §7 Bucket B, category 2).
/// Emits the `ALTER TABLE … ADD COLUMN …` line first (with the FK
/// child-side type — `Type::fk_target_type` of the parent's PK type),
/// then the `ALTER TABLE … ADD CONSTRAINT … FOREIGN KEY …` line. When
/// the column already existed, the runtime instead uses
/// [`render_add_relationship`] for a single-line echo (the `ADD COLUMN`
/// line would be a no-op-with-error in advanced SQL — "column already
/// exists" — and the catalogue specifies "one line if the column already
/// existed").
#[allow(clippy::too_many_arguments)] // the SQL FK has many slots — all inherent.
pub(crate) fn render_add_relationship_create_fk(
    name: &str,
    parent_table: &str,
    parent_column: &str,
    child_table: &str,
    child_column: &str,
    on_delete: ReferentialAction,
    on_update: ReferentialAction,
    new_child_column_type: crate::dsl::types::Type,
) -> Vec<String> {
    vec![
        format!(
            "ALTER TABLE {child_table} ADD COLUMN {child_column} {}",
            new_child_column_type.keyword()
        ),
        render_add_relationship(
            name,
            parent_table,
            parent_column,
            child_table,
            child_column,
            on_delete,
            on_update,
        ),
    ]
}

/// Append the `NOT NULL` / `UNIQUE` / `DEFAULT` / `CHECK` column-constraint
/// suffix (ADR-0029). The advanced-mode column-constraint grammar is
/// order-independent (`Repeated(Choice…)`, ADR-0035 §4a), so this fixed
/// order round-trips. Used by both `create table` and `add column`.
fn append_constraints(
    s: &mut String,
    not_null: bool,
    unique: bool,
    default: Option<&Value>,
    check: Option<&Expr>,
) {
    if not_null {
        s.push_str(" NOT NULL");
    }
    if unique {
        s.push_str(" UNIQUE");
    }
    if let Some(v) = default {
        s.push_str(&format!(" DEFAULT {}", value_to_sql_literal(v)));
    }
    if let Some(e) = check {
        s.push_str(&format!(" CHECK ({})", expr_to_sql(e)));
    }
}

/// `<col> = <literal>, …` — an `UPDATE … SET` assignment list.
fn render_assignments(assignments: &[(String, Value)]) -> String {
    assignments
        .iter()
        .map(|(col, val)| format!("{col} = {}", value_to_sql_literal(val)))
        .collect::<Vec<_>>()
        .join(", ")
}

/// A `Value` as a runnable SQL literal (ADR-0038 §5). Most forms reuse
/// `Value`'s own `Display` (`'O''Hara'`, bare numbers, `true`/`false`,
/// quoted ISO dates); only `NULL` differs — `Display` writes lowercase
/// `null`, but §5 (and the advanced grammar's canonical form) want `NULL`.
fn value_to_sql_literal(value: &Value) -> String {
    match value {
        Value::Null => "NULL".to_string(),
        other => other.to_string(),
    }
}

/// A WHERE / CHECK [`Expr`] as advanced-mode SQL. Mirrors the worker's
/// `compile_expr` operator spellings (`<>`, `LIKE`, `BETWEEN`, `IN`,
/// `IS NULL`, parenthesised `AND` / `OR` / `NOT`) but emits **bare column
/// identifiers** (the echo's unquoted style, matching `render_create_table`)
/// and **inline literals** instead of `?` placeholders, so the line is
/// runnable (§1).
fn expr_to_sql(expr: &Expr) -> String {
    match expr {
        Expr::Or(terms) => join_terms(terms, "OR"),
        Expr::And(terms) => join_terms(terms, "AND"),
        Expr::Not(inner) => format!("(NOT {})", expr_to_sql(inner)),
        Expr::Predicate(p) => predicate_to_sql(p),
    }
}

fn join_terms(terms: &[Expr], op: &str) -> String {
    let parts: Vec<String> = terms.iter().map(expr_to_sql).collect();
    format!("({})", parts.join(&format!(" {op} ")))
}

fn predicate_to_sql(predicate: &Predicate) -> String {
    match predicate {
        Predicate::Compare { left, op, right } => format!(
            "{} {} {}",
            operand_to_sql(left),
            compare_op_sql(*op),
            operand_to_sql(right)
        ),
        Predicate::Like {
            target,
            pattern,
            negated,
        } => {
            let not = if *negated { "NOT " } else { "" };
            format!("{} {not}LIKE {}", operand_to_sql(target), operand_to_sql(pattern))
        }
        Predicate::Between {
            target,
            low,
            high,
            negated,
        } => {
            let not = if *negated { "NOT " } else { "" };
            format!(
                "{} {not}BETWEEN {} AND {}",
                operand_to_sql(target),
                operand_to_sql(low),
                operand_to_sql(high)
            )
        }
        Predicate::In {
            target,
            items,
            negated,
        } => {
            let not = if *negated { "NOT " } else { "" };
            let rendered: Vec<String> = items.iter().map(operand_to_sql).collect();
            format!("{} {not}IN ({})", operand_to_sql(target), rendered.join(", "))
        }
        Predicate::IsNull { target, negated } => {
            let not = if *negated { "NOT " } else { "" };
            format!("{} IS {not}NULL", operand_to_sql(target))
        }
    }
}

fn operand_to_sql(operand: &Operand) -> String {
    match operand {
        Operand::Column { name, .. } => name.clone(),
        Operand::Literal { value, .. } => value_to_sql_literal(value),
    }
}

/// The SQL spelling of a comparison operator — `<>` for inequality, the
/// standard form (ADR-0026 §6), matching the worker's `compare_op_sql`.
const fn compare_op_sql(op: CompareOp) -> &'static str {
    match op {
        CompareOp::Eq => "=",
        CompareOp::NotEq => "<>",
        CompareOp::Lt => "<",
        CompareOp::LtEq => "<=",
        CompareOp::Gt => ">",
        CompareOp::GtEq => ">=",
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::dsl::command::{ChangeColumnMode, ConstraintKind};
    use crate::dsl::types::Type;

    fn create_table(name: &str, cols: Vec<ColumnSpec>, pk: &[&str]) -> Command {
        Command::CreateTable {
            name: name.to_string(),
            columns: cols,
            primary_key: pk.iter().map(|s| (*s).to_string()).collect(),
        }
    }

    /// A `column = value` equality predicate, the most common WHERE leaf.
    fn eq(column: &str, value: Value) -> Expr {
        Expr::Predicate(Predicate::Compare {
            left: Operand::Column {
                name: column.to_string(),
                span: Operand::NO_SPAN,
            },
            op: CompareOp::Eq,
            right: Operand::Literal {
                value,
                span: Operand::NO_SPAN,
            },
        })
    }

    /// Parse `sql` through the advanced-mode walker (the round-trip target).
    fn reparse(sql: &str) -> Result<Command, crate::dsl::parser::ParseError> {
        crate::dsl::parser::parse_command_in_mode(sql, crate::mode::Mode::Advanced)
    }

    // --- create table (ADR-0038 §7 Bucket A) -------------------------

    #[test]
    fn create_table_single_serial_pk_renders_inline() {
        let cmd = create_table("Other", vec![ColumnSpec::new("id", Type::Serial)], &["id"]);
        assert_eq!(
            command_to_sql(&cmd).as_deref(),
            Some("CREATE TABLE Other (id serial PRIMARY KEY)")
        );
    }

    #[test]
    fn create_table_compound_pk_renders_table_level() {
        let cmd = create_table(
            "T",
            vec![ColumnSpec::new("a", Type::Int), ColumnSpec::new("b", Type::Int)],
            &["a", "b"],
        );
        assert_eq!(
            command_to_sql(&cmd).as_deref(),
            Some("CREATE TABLE T (a int, b int, PRIMARY KEY (a, b))")
        );
    }

    #[test]
    fn create_table_renders_default_and_check_constraints() {
        // §1 copy-paste contract: simple-mode `create table` can carry
        // per-column `default` / `check` (ADR-0029), so the echo must too,
        // or it is not equivalent.
        let age = ColumnSpec {
            check: Some(Expr::Predicate(Predicate::Compare {
                left: Operand::Column {
                    name: "age".to_string(),
                    span: Operand::NO_SPAN,
                },
                op: CompareOp::GtEq,
                right: Operand::Literal {
                    value: Value::Number("0".to_string()),
                    span: Operand::NO_SPAN,
                },
            })),
            ..ColumnSpec::new("age", Type::Int)
        };
        let grade = ColumnSpec {
            default: Some(Value::Text("A".to_string())),
            ..ColumnSpec::new("grade", Type::Text)
        };
        let cmd = create_table("T", vec![ColumnSpec::new("id", Type::Serial), age, grade], &["id"]);
        let sql = command_to_sql(&cmd).expect("echo");
        assert_eq!(
            sql,
            "CREATE TABLE T (id serial PRIMARY KEY, age int CHECK (age >= 0), grade text DEFAULT 'A')"
        );
        assert!(matches!(reparse(&sql), Ok(Command::SqlCreateTable { .. })));
    }

    #[test]
    fn create_table_echo_round_trips_in_advanced_mode() {
        // ADR-0038 §1 copy-paste contract: the echo is runnable advanced SQL.
        let cmd = create_table("Other", vec![ColumnSpec::new("id", Type::Serial)], &["id"]);
        let sql = command_to_sql(&cmd).expect("echo");
        assert!(matches!(reparse(&sql), Ok(Command::SqlCreateTable { .. })));
    }

    // --- add column --------------------------------------------------

    #[test]
    fn add_column_renders_type_and_constraints_and_round_trips() {
        let cmd = Command::AddColumn {
            table: "T".to_string(),
            column: "note".to_string(),
            ty: Type::Text,
            not_null: true,
            unique: false,
            default: Some(Value::Text("n/a".to_string())),
            check: None,
        };
        let sql = command_to_sql(&cmd).expect("echo");
        assert_eq!(sql, "ALTER TABLE T ADD COLUMN note text NOT NULL DEFAULT 'n/a'");
        assert!(matches!(
            reparse(&sql),
            Ok(Command::SqlAlterTable { .. })
        ));
    }

    #[test]
    fn add_column_with_unique_and_check_round_trips() {
        // ADD COLUMN's constraint grammar is its own production — confirm
        // the full UNIQUE / CHECK suffix round-trips there too, not just on
        // CREATE TABLE.
        let cmd = Command::AddColumn {
            table: "T".to_string(),
            column: "score".to_string(),
            ty: Type::Int,
            not_null: false,
            unique: true,
            default: None,
            check: Some(Expr::Predicate(Predicate::Compare {
                left: Operand::Column {
                    name: "score".to_string(),
                    span: Operand::NO_SPAN,
                },
                op: CompareOp::GtEq,
                right: Operand::Literal {
                    value: Value::Number("0".to_string()),
                    span: Operand::NO_SPAN,
                },
            })),
        };
        let sql = command_to_sql(&cmd).expect("echo");
        assert_eq!(sql, "ALTER TABLE T ADD COLUMN score int UNIQUE CHECK (score >= 0)");
        assert!(matches!(reparse(&sql), Ok(Command::SqlAlterTable { .. })));
    }

    // --- drop column -------------------------------------------------

    #[test]
    fn drop_column_plain_round_trips() {
        let cmd = Command::DropColumn {
            table: "T".to_string(),
            column: "c".to_string(),
            cascade: false,
        };
        let sql = command_to_sql(&cmd).expect("echo");
        assert_eq!(sql, "ALTER TABLE T DROP COLUMN c");
        assert!(matches!(reparse(&sql), Ok(Command::SqlAlterTable { .. })));
    }

    #[test]
    fn drop_column_cascade_has_no_echo_yet() {
        // Multi-statement (drops covering indexes too) — Bucket B, Phase 2.
        let cmd = Command::DropColumn {
            table: "T".to_string(),
            column: "c".to_string(),
            cascade: true,
        };
        assert!(command_to_sql(&cmd).is_none());
    }

    // --- rename column -----------------------------------------------

    #[test]
    fn rename_column_round_trips() {
        let cmd = Command::RenameColumn {
            table: "T".to_string(),
            old: "a".to_string(),
            new: "b".to_string(),
        };
        let sql = command_to_sql(&cmd).expect("echo");
        assert_eq!(sql, "ALTER TABLE T RENAME COLUMN a TO b");
        assert!(matches!(reparse(&sql), Ok(Command::SqlAlterTable { .. })));
    }

    // --- change column type ------------------------------------------

    #[test]
    fn change_column_renders_set_data_type_for_every_mode() {
        for mode in [
            ChangeColumnMode::Default,
            ChangeColumnMode::ForceConversion,
            ChangeColumnMode::DontConvert,
        ] {
            let cmd = Command::ChangeColumnType {
                table: "T".to_string(),
                column: "c".to_string(),
                ty: Type::Text,
                mode,
            };
            let sql = command_to_sql(&cmd).expect("echo");
            assert_eq!(sql, "ALTER TABLE T ALTER COLUMN c SET DATA TYPE text");
            assert!(matches!(reparse(&sql), Ok(Command::SqlAlterTable { .. })));
        }
    }

    #[test]
    fn change_column_to_playground_type_keyword_round_trips() {
        let cmd = Command::ChangeColumnType {
            table: "T".to_string(),
            column: "code".to_string(),
            ty: Type::ShortId,
            mode: ChangeColumnMode::Default,
        };
        let sql = command_to_sql(&cmd).expect("echo");
        assert_eq!(sql, "ALTER TABLE T ALTER COLUMN code SET DATA TYPE shortid");
        assert!(matches!(reparse(&sql), Ok(Command::SqlAlterTable { .. })));
    }

    // --- add constraint ----------------------------------------------

    fn add_constraint(constraint: Constraint) -> Command {
        Command::AddConstraint {
            table: "T".to_string(),
            column: "c".to_string(),
            constraint,
        }
    }

    #[test]
    fn add_constraint_not_null_round_trips() {
        let sql = command_to_sql(&add_constraint(Constraint::NotNull)).expect("echo");
        assert_eq!(sql, "ALTER TABLE T ALTER COLUMN c SET NOT NULL");
        assert!(matches!(reparse(&sql), Ok(Command::SqlAlterTable { .. })));
    }

    #[test]
    fn add_constraint_default_round_trips() {
        let sql = command_to_sql(&add_constraint(Constraint::Default(Value::Number(
            "0".to_string(),
        ))))
        .expect("echo");
        assert_eq!(sql, "ALTER TABLE T ALTER COLUMN c SET DEFAULT 0");
        assert!(matches!(reparse(&sql), Ok(Command::SqlAlterTable { .. })));
    }

    #[test]
    fn add_constraint_unique_round_trips() {
        let sql = command_to_sql(&add_constraint(Constraint::Unique)).expect("echo");
        assert_eq!(sql, "ALTER TABLE T ADD UNIQUE (c)");
        assert!(matches!(reparse(&sql), Ok(Command::SqlAlterTable { .. })));
    }

    #[test]
    fn add_constraint_check_round_trips() {
        let expr = Expr::Predicate(Predicate::Compare {
            left: Operand::Column {
                name: "c".to_string(),
                span: Operand::NO_SPAN,
            },
            op: CompareOp::Gt,
            right: Operand::Literal {
                value: Value::Number("0".to_string()),
                span: Operand::NO_SPAN,
            },
        });
        let sql = command_to_sql(&add_constraint(Constraint::Check(expr))).expect("echo");
        assert_eq!(sql, "ALTER TABLE T ADD CHECK (c > 0)");
        assert!(matches!(reparse(&sql), Ok(Command::SqlAlterTable { .. })));
    }

    // --- drop constraint ---------------------------------------------

    fn drop_constraint(kind: ConstraintKind) -> Command {
        Command::DropConstraint {
            table: "T".to_string(),
            column: "c".to_string(),
            kind,
        }
    }

    #[test]
    fn drop_constraint_not_null_round_trips() {
        let sql = command_to_sql(&drop_constraint(ConstraintKind::NotNull)).expect("echo");
        assert_eq!(sql, "ALTER TABLE T ALTER COLUMN c DROP NOT NULL");
        assert!(matches!(reparse(&sql), Ok(Command::SqlAlterTable { .. })));
    }

    #[test]
    fn drop_constraint_default_round_trips() {
        let sql = command_to_sql(&drop_constraint(ConstraintKind::Default)).expect("echo");
        assert_eq!(sql, "ALTER TABLE T ALTER COLUMN c DROP DEFAULT");
        assert!(matches!(reparse(&sql), Ok(Command::SqlAlterTable { .. })));
    }

    #[test]
    fn drop_constraint_unique_and_check_have_no_echo() {
        // Column-level UNIQUE / CHECK is anonymous — Bucket C (ADR-0038 §7).
        assert!(command_to_sql(&drop_constraint(ConstraintKind::Unique)).is_none());
        assert!(command_to_sql(&drop_constraint(ConstraintKind::Check)).is_none());
    }

    // --- update / delete --all-rows ----------------------------------

    #[test]
    fn update_all_rows_round_trips() {
        let cmd = Command::Update {
            table: "T".to_string(),
            assignments: vec![
                ("status".to_string(), Value::Text("done".to_string())),
                ("score".to_string(), Value::Number("10".to_string())),
            ],
            filter: RowFilter::AllRows,
        };
        let sql = command_to_sql(&cmd).expect("echo");
        assert_eq!(sql, "UPDATE T SET status = 'done', score = 10");
        assert!(matches!(reparse(&sql), Ok(Command::SqlUpdate { .. })));
    }

    #[test]
    fn update_with_where_has_no_echo() {
        // A WHERE-filtered update is SQL-first in advanced mode (SqlUpdate).
        let cmd = Command::Update {
            table: "T".to_string(),
            assignments: vec![("a".to_string(), Value::Number("1".to_string()))],
            filter: RowFilter::Where(eq("id", Value::Number("1".to_string()))),
        };
        assert!(command_to_sql(&cmd).is_none());
    }

    #[test]
    fn delete_all_rows_round_trips() {
        let cmd = Command::Delete {
            table: "T".to_string(),
            filter: RowFilter::AllRows,
        };
        let sql = command_to_sql(&cmd).expect("echo");
        assert_eq!(sql, "DELETE FROM T");
        assert!(matches!(reparse(&sql), Ok(Command::SqlDelete { .. })));
    }

    #[test]
    fn delete_with_where_has_no_echo() {
        let cmd = Command::Delete {
            table: "T".to_string(),
            filter: RowFilter::Where(eq("id", Value::Number("1".to_string()))),
        };
        assert!(command_to_sql(&cmd).is_none());
    }

    // --- show data (echo_for_query) ----------------------------------

    #[test]
    fn show_data_plain_round_trips() {
        let cmd = Command::ShowData {
            name: "T".to_string(),
            filter: None,
            limit: None,
        };
        let lines = echo_for_query(&cmd, EffectiveMode::AdvancedPersistent, &[]).expect("echo");
        assert_eq!(lines.as_slice(), &["SELECT * FROM T"]);
        assert!(matches!(reparse(&lines[0]), Ok(Command::Select { .. })));
    }

    #[test]
    fn show_data_with_where_round_trips() {
        let cmd = Command::ShowData {
            name: "T".to_string(),
            filter: Some(eq("name", Value::Text("Bob".to_string()))),
            limit: None,
        };
        let lines = echo_for_query(&cmd, EffectiveMode::AdvancedPersistent, &[]).expect("echo");
        assert_eq!(lines.as_slice(), &["SELECT * FROM T WHERE name = 'Bob'"]);
        assert!(matches!(reparse(&lines[0]), Ok(Command::Select { .. })));
    }

    #[test]
    fn show_data_with_limit_orders_by_primary_key() {
        let cmd = Command::ShowData {
            name: "T".to_string(),
            filter: None,
            limit: Some(5),
        };
        let pk = vec!["id".to_string()];
        let lines = echo_for_query(&cmd, EffectiveMode::AdvancedPersistent, &pk).expect("echo");
        assert_eq!(lines.as_slice(), &["SELECT * FROM T ORDER BY id LIMIT 5"]);
        assert!(matches!(reparse(&lines[0]), Ok(Command::Select { .. })));
    }

    #[test]
    fn show_data_with_limit_and_compound_pk_orders_by_all_pk_columns() {
        let cmd = Command::ShowData {
            name: "T".to_string(),
            filter: Some(eq("active", Value::Bool(true))),
            limit: Some(3),
        };
        let pk = vec!["a".to_string(), "b".to_string()];
        let lines = echo_for_query(&cmd, EffectiveMode::AdvancedPersistent, &pk).expect("echo");
        assert_eq!(
            lines.as_slice(),
            &["SELECT * FROM T WHERE active = true ORDER BY a, b LIMIT 3"]
        );
        assert!(matches!(reparse(&lines[0]), Ok(Command::Select { .. })));
    }

    #[test]
    fn show_data_with_limit_but_no_primary_key_omits_order_by() {
        // Matches the worker: no PK → no ORDER BY (build_query_data_sql).
        let cmd = Command::ShowData {
            name: "T".to_string(),
            filter: None,
            limit: Some(2),
        };
        let lines = echo_for_query(&cmd, EffectiveMode::AdvancedPersistent, &[]).expect("echo");
        assert_eq!(lines.as_slice(), &["SELECT * FROM T LIMIT 2"]);
    }

    #[test]
    fn show_data_is_silent_in_simple_mode() {
        let cmd = Command::ShowData {
            name: "T".to_string(),
            filter: None,
            limit: None,
        };
        assert!(echo_for_query(&cmd, EffectiveMode::Simple, &[]).is_none());
    }

    #[test]
    fn select_is_not_echoed_as_a_query() {
        // A `Command::Select` is already SQL — no echo (ADR-0038 §7).
        let cmd = Command::Select {
            sql: "select * from T".to_string(),
        };
        assert!(echo_for_query(&cmd, EffectiveMode::AdvancedPersistent, &[]).is_none());
    }

    // --- Bucket B single-statement renderers (Phase 2, Slice 2a) -----

    #[test]
    fn add_index_renders_and_round_trips() {
        // Named form — the name is what was passed in.
        let sql = render_create_index("MyIdx", "T", &["a".to_string(), "b".to_string()]);
        assert_eq!(sql, "CREATE INDEX MyIdx ON T (a, b)");
        assert!(matches!(reparse(&sql), Ok(Command::SqlCreateIndex { .. })));
    }

    #[test]
    fn add_index_auto_name_format_matches_worker() {
        // Mirrors the worker's `resolve_index_name` (`{table}_{cols}_idx`) —
        // not directly used by the renderer (the runtime sources the resolved
        // name from the description), but pins the expected auto-name shape.
        let sql = render_create_index("Customers_Email_idx", "Customers", &["Email".to_string()]);
        assert_eq!(sql, "CREATE INDEX Customers_Email_idx ON Customers (Email)");
        assert!(matches!(reparse(&sql), Ok(Command::SqlCreateIndex { .. })));
    }

    #[test]
    fn drop_index_round_trips() {
        let sql = render_drop_index("Customers_Email_idx");
        assert_eq!(sql, "DROP INDEX Customers_Email_idx");
        assert!(matches!(reparse(&sql), Ok(Command::SqlDropIndex { .. })));
    }

    #[test]
    fn add_relationship_no_referential_actions_round_trips() {
        // Default `NoAction` / `NoAction` → no `ON DELETE` / `ON UPDATE`
        // clauses (the implicit standard default — emitting them would
        // clutter the echo without changing meaning).
        let sql = render_add_relationship(
            "Orders_CustId_to_Customers_id",
            "Customers",
            "id",
            "Orders",
            "CustId",
            ReferentialAction::NoAction,
            ReferentialAction::NoAction,
        );
        assert_eq!(
            sql,
            "ALTER TABLE Orders ADD CONSTRAINT Orders_CustId_to_Customers_id FOREIGN KEY (CustId) REFERENCES Customers (id)"
        );
        assert!(matches!(reparse(&sql), Ok(Command::SqlAlterTable { .. })));
    }

    #[test]
    fn add_relationship_with_cascade_and_set_null_round_trips() {
        let sql = render_add_relationship(
            "places",
            "Customers",
            "id",
            "Orders",
            "CustId",
            ReferentialAction::Cascade,
            ReferentialAction::SetNull,
        );
        assert_eq!(
            sql,
            "ALTER TABLE Orders ADD CONSTRAINT places FOREIGN KEY (CustId) REFERENCES Customers (id) ON DELETE CASCADE ON UPDATE SET NULL"
        );
        assert!(matches!(reparse(&sql), Ok(Command::SqlAlterTable { .. })));
    }

    #[test]
    fn drop_relationship_round_trips() {
        let sql = render_drop_relationship("places", "Orders");
        assert_eq!(sql, "ALTER TABLE Orders DROP CONSTRAINT places");
        assert!(matches!(reparse(&sql), Ok(Command::SqlAlterTable { .. })));
    }

    // --- Bucket B multi-statement renderers (Phase 2, Slice 2b) ------

    #[test]
    fn drop_column_cascade_emits_drop_indexes_then_drop_column_and_each_round_trips() {
        let lines = render_drop_column_cascade(
            "Orders",
            "CustId",
            &["Orders_CustId_idx".to_string(), "Orders_CustId_Day_idx".to_string()],
        );
        assert_eq!(
            lines.as_slice(),
            &[
                "DROP INDEX Orders_CustId_idx",
                "DROP INDEX Orders_CustId_Day_idx",
                "ALTER TABLE Orders DROP COLUMN CustId",
            ]
        );
        // Each line is itself runnable advanced-mode SQL (the §1 contract
        // holds per line for category 2).
        assert!(matches!(reparse(&lines[0]), Ok(Command::SqlDropIndex { .. })));
        assert!(matches!(reparse(&lines[1]), Ok(Command::SqlDropIndex { .. })));
        assert!(matches!(reparse(&lines[2]), Ok(Command::SqlAlterTable { .. })));
    }

    #[test]
    fn drop_column_cascade_with_no_covering_indexes_is_single_line() {
        // `--cascade` flagged on an unindexed column collapses to the
        // plain `DROP COLUMN` — still semantically equivalent.
        let lines = render_drop_column_cascade("T", "c", &[]);
        assert_eq!(lines.as_slice(), &["ALTER TABLE T DROP COLUMN c"]);
        assert!(matches!(reparse(&lines[0]), Ok(Command::SqlAlterTable { .. })));
    }

    #[test]
    fn add_relationship_create_fk_emits_add_column_then_fk_and_each_round_trips() {
        let lines = render_add_relationship_create_fk(
            "Customers_id_to_Orders_CustId",
            "Customers",
            "id",
            "Orders",
            "CustId",
            ReferentialAction::Cascade,
            ReferentialAction::NoAction,
            // Parent PK is `serial` → child FK column is `int`
            // (`Type::fk_target_type` strips auto-gen semantics; ADR-0011).
            crate::dsl::types::Type::Int,
        );
        assert_eq!(
            lines.as_slice(),
            &[
                "ALTER TABLE Orders ADD COLUMN CustId int",
                "ALTER TABLE Orders ADD CONSTRAINT Customers_id_to_Orders_CustId FOREIGN KEY (CustId) REFERENCES Customers (id) ON DELETE CASCADE",
            ]
        );
        assert!(matches!(reparse(&lines[0]), Ok(Command::SqlAlterTable { .. })));
        assert!(matches!(reparse(&lines[1]), Ok(Command::SqlAlterTable { .. })));
    }

    #[test]
    fn add_relationship_create_fk_with_shortid_parent_targets_text_column() {
        // `Type::fk_target_type(ShortId)` → Text (ADR-0011).
        let lines = render_add_relationship_create_fk(
            "Items_code_to_Lines_code",
            "Items",
            "code",
            "Lines",
            "code",
            ReferentialAction::NoAction,
            ReferentialAction::NoAction,
            crate::dsl::types::Type::Text,
        );
        assert_eq!(lines[0], "ALTER TABLE Lines ADD COLUMN code text");
        // No referential clauses when both default.
        assert_eq!(
            lines[1],
            "ALTER TABLE Lines ADD CONSTRAINT Items_code_to_Lines_code FOREIGN KEY (code) REFERENCES Items (code)"
        );
    }

    // --- expr / literal rendering ------------------------------------

    #[test]
    fn expr_renders_boolean_combinators_and_operators() {
        // age >= 18 AND (name <> 'x' OR active = true)
        let expr = Expr::And(vec![
            Expr::Predicate(Predicate::Compare {
                left: Operand::Column {
                    name: "age".to_string(),
                    span: Operand::NO_SPAN,
                },
                op: CompareOp::GtEq,
                right: Operand::Literal {
                    value: Value::Number("18".to_string()),
                    span: Operand::NO_SPAN,
                },
            }),
            Expr::Or(vec![
                Expr::Predicate(Predicate::Compare {
                    left: Operand::Column {
                        name: "name".to_string(),
                        span: Operand::NO_SPAN,
                    },
                    op: CompareOp::NotEq,
                    right: Operand::Literal {
                        value: Value::Text("x".to_string()),
                        span: Operand::NO_SPAN,
                    },
                }),
                eq("active", Value::Bool(true)),
            ]),
        ]);
        assert_eq!(
            expr_to_sql(&expr),
            "(age >= 18 AND (name <> 'x' OR active = true))"
        );
    }

    #[test]
    fn value_literal_renders_null_uppercase_and_quotes_text() {
        assert_eq!(value_to_sql_literal(&Value::Null), "NULL");
        assert_eq!(value_to_sql_literal(&Value::Text("O'Hara".to_string())), "'O''Hara'");
        assert_eq!(value_to_sql_literal(&Value::Number("3.14".to_string())), "3.14");
        assert_eq!(value_to_sql_literal(&Value::Bool(false)), "false");
    }

    // --- gating ------------------------------------------------------

    #[test]
    fn echo_for_gates_on_advanced_mode() {
        let cmd = create_table("Other", vec![ColumnSpec::new("id", Type::Serial)], &["id"]);
        assert!(echo_for(&cmd, EffectiveMode::AdvancedPersistent).is_some());
        assert!(echo_for(&cmd, EffectiveMode::AdvancedOneShot).is_some());
        assert!(
            echo_for(&cmd, EffectiveMode::Simple).is_none(),
            "simple mode stays uncluttered (ADR-0030 §10)"
        );
    }

    #[test]
    fn bucket_c_no_echo_commands_all_return_none() {
        // ADR-0038 §7 Bucket C: pin the no-echo cases that flow through
        // the `command_to_sql` catch-all (alongside the already-tested
        // `Sql*`/`Select` and column-level UNIQUE/CHECK drop). A drift
        // here — say, a renderer arm grown into a Bucket C command —
        // would silently leak echoes the §10 / §13 contract forbids.
        use crate::dsl::command::AppCommand;

        // `show table T` — structure display, no SQL spelling in the
        // surface (ADR-0038 §7 Bucket C).
        assert!(
            command_to_sql(&Command::ShowTable {
                name: "T".to_string(),
            })
            .is_none(),
            "show table is Bucket C — no echo"
        );

        // `explain …` — EXPLAIN of advanced SQL is the deferred OOS-2
        // follow-up (ADR-0039); the DSL `explain` wrapper itself echoes
        // nothing (ADR-0038 §7).
        assert!(
            command_to_sql(&Command::Explain {
                query: Box::new(Command::ShowData {
                    name: "T".to_string(),
                    filter: None,
                    limit: None,
                }),
            })
            .is_none(),
            "explain is Bucket C — no echo"
        );

        // `replay <path>` — app-lifecycle, no SQL form (ADR-0038 §7).
        // (Also: replay bypasses the echo-bearing spawn entirely, so it
        // never reaches command_to_sql in practice — but pinning the
        // catch-all here guards against a future arm sneaking in.)
        assert!(
            command_to_sql(&Command::Replay {
                path: "history.log".to_string(),
            })
            .is_none(),
            "replay is Bucket C — no echo"
        );

        // `Command::App(_)` — every app-lifecycle command, regardless of
        // verb. ADR-0030 §10: "app-level commands have no SQL form and
        // are not echoed." Sample two: `quit` (verb-only) and `mode
        // advanced` (verb + payload).
        for app in [
            AppCommand::Quit,
            AppCommand::Help,
            AppCommand::Rebuild,
            AppCommand::Save,
            AppCommand::New,
            AppCommand::Load,
            AppCommand::Undo,
            AppCommand::Redo,
            AppCommand::Mode {
                value: crate::dsl::command::ModeValue::Advanced,
            },
        ] {
            assert!(
                command_to_sql(&Command::App(app.clone())).is_none(),
                "Command::App({app:?}) is Bucket C — no echo"
            );
            // Also confirm echo_for gates the same in advanced mode.
            assert!(
                echo_for(&Command::App(app), EffectiveMode::AdvancedPersistent).is_none(),
            );
        }
    }

    #[test]
    fn sql_entered_command_is_not_echoed() {
        // A command the user typed as SQL (SqlCreateTable) is not echoed
        // back (ADR-0030 §10) — command_to_sql covers only DSL-form variants.
        let cmd = Command::SqlCreateTable {
            name: "T".to_string(),
            columns: vec![ColumnSpec::new("id", Type::Serial)],
            primary_key: vec!["id".to_string()],
            unique_constraints: Vec::new(),
            check_constraints: Vec::new(),
            foreign_keys: Vec::new(),
            if_not_exists: false,
        };
        assert!(command_to_sql(&cmd).is_none());
        assert!(echo_for(&cmd, EffectiveMode::AdvancedPersistent).is_none());
    }
}