db+grammar: 3d — shortid auto-fill for SQL INSERT (ADR-0033 §6)

When an INSERT's column list omits one or more shortid columns,
the worker now fills them. Command::SqlInsert gains listed_columns
and row_source, captured in build_sql_insert from the matched path
(the row source is located by the first values/select/with Word
token, so a string literal like 'select' can't be mistaken for the
keyword). do_sql_insert calls plan_shortid_autofill, which — per
the user-confirmed Option B — materialises the row source by
running it as a query, generates a distinct shortid per row via the
existing generate_shortid_batch (deduped against stored values),
and reconstructs a parameterised multi-row INSERT over the listed
columns plus the omitted shortid columns. Uniform for VALUES and
INSERT…SELECT, and handles multiple omitted shortids in one row
(each gets its own batch). No explicit list, no omitted shortid, or
a zero-row source → execute verbatim (the 3b path). serial stays
engine-filled via rowid. history.log keeps the original line, never
the rewrite (§11).

Tests: VALUES single/multi-row distinct; explicit override
honoured; INSERT…SELECT distinct fills; combined serial(engine) +
shortid(worker); two shortids (PK + non-PK) both fill; one provided
+ one omitted; compound-PK shortid member; mixed-case column name
(ADR-0009 DA gate); original-source-in-history on the rewrite path.
Still behind the dev `sqlinsert` entry word (3j). 1503 green,
clippy clean.

src/db.rs

@@ -591,6 +591,8 @@ enum Request {
         sql: String,
         source: Option<String>,
         target_table: String,
+        listed_columns: Vec<String>,
+        row_source: String,
         reply: oneshot::Sender<Result<InsertResult, DbError>>,
     },
     /// Capture the query plan for an explainable command via
@@ -1060,12 +1062,16 @@ impl Database {
         sql: String,
         source: Option<String>,
         target_table: String,
+        listed_columns: Vec<String>,
+        row_source: String,
     ) -> Result<InsertResult, DbError> {
         let (reply, recv) = oneshot::channel();
         self.send(Request::RunSqlInsert {
             sql,
             source,
             target_table,
+            listed_columns,
+            row_source,
             reply,
         })
         .await?;
@@ -1521,6 +1527,8 @@ fn handle_request(conn: &Connection, persistence: Option<&Persistence>, req: Req
             sql,
             source,
             target_table,
+            listed_columns,
+            row_source,
             reply,
         } => {
             let _ = reply.send(do_sql_insert(
@@ -1529,6 +1537,8 @@ fn handle_request(conn: &Connection, persistence: Option<&Persistence>, req: Req
                 source.as_deref(),
                 &sql,
                 &target_table,
+                &listed_columns,
+                &row_source,
             ));
         }
         Request::RebuildFromText {
@@ -5747,6 +5757,149 @@ fn do_run_select_request(
     Ok(data)
 }
 
+/// Currently-stored non-NULL values of one column, for shortid
+/// collision-avoidance (passed to `generate_shortid_batch`).
+fn existing_shortids(
+    conn: &Connection,
+    table: &str,
+    column: &str,
+) -> Result<Vec<String>, DbError> {
+    let mut stmt = conn
+        .prepare(&format!(
+            "SELECT {col} FROM {tbl} WHERE {col} IS NOT NULL;",
+            col = quote_ident(column),
+            tbl = quote_ident(table),
+        ))
+        .map_err(DbError::from_rusqlite)?;
+    let rows = stmt
+        .query_map([], |r| r.get::<_, String>(0))
+        .map_err(DbError::from_rusqlite)?;
+    let mut out = Vec::new();
+    for r in rows {
+        out.push(r.map_err(DbError::from_rusqlite)?);
+    }
+    Ok(out)
+}
+
+/// Plan `shortid` auto-fill for a SQL `INSERT` (ADR-0033 §6,
+/// sub-phase 3d).
+///
+/// Returns the SQL the worker should execute plus its bound
+/// params. When the user's `(column_list)` omits one or more
+/// `shortid` columns, this materialises the row source (Option B:
+/// run it as a query), synthesises a fresh distinct id per row via
+/// `generate_shortid_batch`, and reconstructs a parameterised
+/// multi-row `INSERT` over the listed columns plus the omitted
+/// shortid columns. Otherwise it returns the original `sql`
+/// verbatim with no params (the 3b path):
+///
+/// - no explicit column list → the row source supplies every
+///   column positionally (a listed shortid is the user's value);
+/// - no omitted shortid column → nothing to fill;
+/// - the row source yields zero rows → nothing to fill (the
+///   verbatim INSERT inserts nothing without a NOT-NULL violation).
+///
+/// `serial` columns are not handled here — an omitted `serial`
+/// primary key is filled by the engine's rowid (ADR-0033 §6).
+fn plan_shortid_autofill(
+    conn: &Connection,
+    target_table: &str,
+    sql: &str,
+    listed_columns: &[String],
+    row_source: &str,
+) -> Result<(String, Vec<rusqlite::types::Value>), DbError> {
+    if listed_columns.is_empty() {
+        return Ok((sql.to_string(), Vec::new()));
+    }
+    let schema = read_schema(conn, target_table)?;
+    // Identifiers are case-preserving but matched case-insensitively
+    // (ADR-0009): a shortid column counts as omitted unless the user
+    // listed a name equal to it ignoring ASCII case.
+    let listed_ci: Vec<String> =
+        listed_columns.iter().map(|c| c.to_ascii_lowercase()).collect();
+    let omitted_shortids: Vec<String> = schema
+        .columns
+        .iter()
+        .filter(|c| c.user_type == Some(Type::ShortId))
+        .filter(|c| !listed_ci.contains(&c.name.to_ascii_lowercase()))
+        .map(|c| c.name.clone())
+        .collect();
+    if omitted_shortids.is_empty() {
+        return Ok((sql.to_string(), Vec::new()));
+    }
+
+    // Materialise the row source (VALUES / SELECT / WITH … SELECT)
+    // as concrete rows for the listed columns.
+    let listed_count = listed_columns.len();
+    let mut stmt = conn.prepare(row_source).map_err(DbError::from_rusqlite)?;
+    let mut rows: Vec<Vec<rusqlite::types::Value>> = Vec::new();
+    {
+        let mut q = stmt.query([]).map_err(DbError::from_rusqlite)?;
+        while let Some(r) = q.next().map_err(DbError::from_rusqlite)? {
+            let mut cells = Vec::with_capacity(listed_count);
+            for i in 0..listed_count {
+                cells.push(
+                    r.get::<_, rusqlite::types::Value>(i)
+                        .map_err(DbError::from_rusqlite)?,
+                );
+            }
+            rows.push(cells);
+        }
+    }
+    let n = rows.len();
+    if n == 0 {
+        // Nothing to insert — the verbatim statement inserts zero
+        // rows without touching the omitted shortid column.
+        return Ok((sql.to_string(), Vec::new()));
+    }
+
+    // A fresh, distinct shortid per row for each omitted column,
+    // avoiding collision with values already stored in that column.
+    let mut id_batches: Vec<Vec<rusqlite::types::Value>> =
+        Vec::with_capacity(omitted_shortids.len());
+    for col in &omitted_shortids {
+        let existing = existing_shortids(conn, target_table, col)?;
+        id_batches.push(generate_shortid_batch(n, &existing)?);
+    }
+
+    // Reconstruct: listed columns followed by the omitted shortid
+    // columns; one parameterised tuple per materialised row.
+    let all_cols: Vec<&String> =
+        listed_columns.iter().chain(omitted_shortids.iter()).collect();
+    let cols_csv = all_cols
+        .iter()
+        .map(|c| quote_ident(c))
+        .collect::<Vec<_>>()
+        .join(", ");
+    let per_tuple = all_cols.len();
+    let mut params: Vec<rusqlite::types::Value> = Vec::with_capacity(n * per_tuple);
+    let mut tuples: Vec<String> = Vec::with_capacity(n);
+    let mut ph = 1;
+    for (row_idx, row) in rows.into_iter().enumerate() {
+        for cell in row {
+            params.push(cell);
+        }
+        for batch in &id_batches {
+            params.push(batch[row_idx].clone());
+        }
+        let placeholders = (0..per_tuple)
+            .map(|_| {
+                let s = format!("?{ph}");
+                ph += 1;
+                s
+            })
+            .collect::<Vec<_>>()
+            .join(", ");
+        tuples.push(format!("({placeholders})"));
+    }
+    let exec_sql = format!(
+        "INSERT INTO {tbl} ({cols_csv}) VALUES {vals};",
+        tbl = quote_ident(target_table),
+        vals = tuples.join(", "),
+    );
+    Ok((exec_sql, params))
+}
+
 /// Worker handler for `Request::RunSqlInsert` (ADR-0033 §1,
 /// sub-phase 3b). Mirrors `do_insert`'s persistence discipline:
 /// run the validated SQL inside a transaction, re-persist the
@@ -5754,8 +5907,12 @@ fn do_run_select_request(
 /// `finalize_persistence` *before* `tx.commit()` (so a
 /// persistence failure rolls the insert back), then commit.
 ///
-/// Grammar-as-text (ADR-0030 §4): the values are literals in
-/// `sql`, so no parameters are bound. FK / UNIQUE / NOT NULL
+/// Grammar-as-text (ADR-0030 §4): normally the values are literals
+/// in `sql` and no parameters are bound. The sub-phase 3d shortid
+/// auto-fill path is the exception — it reconstructs a
+/// parameterised `INSERT` (see `plan_shortid_autofill`); either
+/// way `history.log` records the original `source`, never the
+/// rewritten statement (ADR-0030 §11). FK / UNIQUE / NOT NULL
 /// engine errors surface enriched via `execute_with_fk_enrichment`
 /// + the friendly-error layer.
 ///
@@ -5771,12 +5928,23 @@ fn do_sql_insert(
     source: Option<&str>,
     sql: &str,
     target_table: &str,
+    listed_columns: &[String],
+    row_source: &str,
 ) -> Result<InsertResult, DbError> {
     debug!(sql = %sql, table = %target_table, "sql_insert");
+    // Sub-phase 3d: when the user's column list omits one or more
+    // `shortid` columns, the worker materialises the row source,
+    // synthesises fresh distinct ids, and reinserts the augmented
+    // rows. Returns the executable SQL + bound params; an empty
+    // params vec with the original `sql` means "no auto-fill —
+    // execute verbatim" (the 3b path).
+    let (exec_sql, params) =
+        plan_shortid_autofill(conn, target_table, sql, listed_columns, row_source)?;
     let tx = conn
         .unchecked_transaction()
         .map_err(DbError::from_rusqlite)?;
-    let rows_affected = execute_with_fk_enrichment(conn, target_table, sql, &[])?;
+    let rows_affected =
+        execute_with_fk_enrichment(conn, target_table, &exec_sql, &params)?;
     let last = conn.last_insert_rowid();
     let rowids: Vec<i64> = if rows_affected == 0 {
         Vec::new()

src/dsl/command.rs

@@ -297,12 +297,20 @@ pub enum Command {
     /// the validated statement the worker executes verbatim;
     /// `target_table` is extracted from the parse so the worker can
     /// re-persist that table's CSV after a successful insert
-    /// (ADR-0030 §11) without re-parsing the SQL. `listed_columns`
-    /// (3d, `shortid` auto-fill) and `returning` (3g) are added by
-    /// the sub-phases that read them.
+    /// (ADR-0030 §11) without re-parsing the SQL.
+    ///
+    /// `listed_columns` is the user's explicit `(col, …)` list
+    /// (empty when the form omits it); `row_source` is the
+    /// `VALUES …` / `SELECT …` / `WITH … SELECT …` text. Both are
+    /// captured for sub-phase 3d's `shortid` auto-fill: when the
+    /// list omits a `shortid` column, the worker materialises the
+    /// row source, generates fresh ids, and reinserts. `returning`
+    /// (3g) is added by the sub-phase that reads it.
     SqlInsert {
         sql: String,
         target_table: String,
+        listed_columns: Vec<String>,
+        row_source: String,
     },
     /// App-lifecycle command (per ADR-0003). These work in both
     /// simple and advanced modes; the dispatcher branches on the

src/dsl/grammar/data.rs

@@ -876,6 +876,38 @@ fn build_sql_insert(path: &MatchedPath, source: &str) -> Result<Command, Validat
             _ => None,
         })
         .unwrap_or_default();
+    // The user's explicit `(col, …)` list, in order (empty when the
+    // form omits it). Sub-phase 3d reads this to decide which
+    // `shortid` columns were left for the worker to auto-fill.
+    let listed_columns: Vec<String> = path
+        .items
+        .iter()
+        .filter_map(|item| match item.kind {
+            MatchedKind::Ident {
+                role: "insert_column",
+                ..
+            } => Some(item.text.clone()),
+            _ => None,
+        })
+        .collect();
+    // The row source is everything from the `VALUES` / `SELECT` /
+    // `WITH` keyword onward. Located by the first matching *Word
+    // token* in the path (not a text scan), so a string literal
+    // like `values ('select')` can't be mistaken for the keyword.
+    let row_source = path
+        .items
+        .iter()
+        .find(|item| {
+            matches!(item.kind, MatchedKind::Word("values" | "select" | "with"))
+        })
+        .map(|item| {
+            source[item.span.0..]
+                .trim()
+                .trim_end_matches(';')
+                .trim()
+                .to_string()
+        })
+        .unwrap_or_default();
     // Everything after the entry word is the `INTO …` tail; prefix
     // the real `insert` keyword for the engine.
     let tail = path
@@ -883,7 +915,12 @@ fn build_sql_insert(path: &MatchedPath, source: &str) -> Result<Command, Validat
         .first()
         .map_or(source, |entry| &source[entry.span.1..]);
     let sql = format!("insert {}", tail.trim());
-    Ok(Command::SqlInsert { sql, target_table })
+    Ok(Command::SqlInsert {
+        sql,
+        target_table,
+        listed_columns,
+        row_source,
+    })
 }
 
 // =================================================================

src/runtime.rs

@@ -1885,8 +1885,13 @@ async fn execute_command_typed(
         // text: the worker runs the validated `sql` and re-persists
         // the parsed `target_table`'s CSV. Reuses the DSL insert
         // outcome (affected-row count + auto-show).
-        Command::SqlInsert { sql, target_table } => database
-            .run_sql_insert(sql, src, target_table)
+        Command::SqlInsert {
+            sql,
+            target_table,
+            listed_columns,
+            row_source,
+        } => database
+            .run_sql_insert(sql, src, target_table, listed_columns, row_source)
             .await
             .map(CommandOutcome::Insert),
         // `EXPLAIN QUERY PLAN` never executes the wrapped