feat: ADR-0035 4a — SQL CREATE TABLE command, worker, and exit gate

Command + builder + worker for advanced-mode SQL CREATE TABLE
(sub-phase 4a), executed structurally through do_create_table:

- Command::SqlCreateTable + build_sql_create_table (ddl.rs): aliases via
  from_sql_name (incl. double precision), column- and table-level
  PRIMARY KEY, redundant-flag de-dup off a sole PK, IF NOT EXISTS.
  Advanced REGISTRY entry on the shared `create` word (SQL-first, DSL
  fallback); no-PK tables allowed (user-confirmed).
- Worker (db.rs): Request::SqlCreateTable + CreateOutcome + snapshot_then
  (one undo step); IF NOT EXISTS no-op (no snapshot, but journalled, like
  read-only commands). do_create_table inline-PK rule aligned with the
  rebuild generator schema_to_ddl — no round-trip DDL drift; serial
  autoincrement is independent of inline-PK (verified by round-trip
  tests).
- Runtime/App: dispatch + CommandOutcome::SchemaSkipped +
  AppEvent::DslCreateSkipped (structure + "already exists — skipped"
  note). Friendly catalog keys added (engine-neutral).

DEFAULT/CHECK/table-level UNIQUE are absent from the 4a grammar (parse
error with usage skeleton; friendly message + support land in the 4a.2
constraint slice) — user-confirmed.

Tests: type resolver, grammar shape, builder (incl. the PK
detection bug they caught), and tests/sql_create_table.rs (worker
round-trip, serial autoincrement first/non-first across rebuild, IF NOT
EXISTS no-op + journalling, no-PK table, one undo step) + a replay-as-
write test. 1739 pass / 0 fail / 1 ignored; clippy clean.

Exit gate: ADR-0035 Proposed -> Accepted (validated end-to-end by 4a);
README + requirements.md Q1 updated.

docs/adr/0035-advanced-mode-sql-ddl.md

@@ -2,11 +2,14 @@
 
 ## Status
 
-Proposed. Design agreed with the user (2026-05-24); implementation
+Accepted. Design agreed with the user (2026-05-24); the approach is
-is phased and pending (§13). This is **Phase 4** of the ADR-0030
+**validated end-to-end by sub-phase 4a** (`CREATE TABLE`, implemented
-roadmap (the advanced-mode SQL surface), the peer of ADR-0031
+2026-05-25 — plan `docs/plans/20260524-adr-0035-sql-ddl-4a.md`), so the
-(expression grammar), ADR-0032 (`SELECT`), and ADR-0033 (DML). It
+decision is accepted while the remaining sub-phases (**4a.2, 4b–4i**,
-**clarifies ADR-0030 §4** on how DDL is represented and executed.
+§13) continue. This is **Phase 4** of the ADR-0030 roadmap (the
+advanced-mode SQL surface), the peer of ADR-0031 (expression grammar),
+ADR-0032 (`SELECT`), and ADR-0033 (DML). It **clarifies ADR-0030 §4**
+on how DDL is represented and executed.
 
 **Refinements (2026-05-24, pre-implementation `/runda` round,
 user-confirmed).** Two open micro-calls were settled before 4a:
@@ -310,9 +313,12 @@ ADR-0033's structure:
   **clean-reuse column constraints only** — `NOT NULL` / `UNIQUE` /
   column-level `PRIMARY KEY` — + single/compound table-level
   `PRIMARY KEY`, plus `IF NOT EXISTS` (no-op-with-note, §4). Reuses
-  `do_create_table` (extended so a `serial` sole-PK inlines `PRIMARY
+  `do_create_table`, whose inline-PK rule is aligned with the rebuild
-  KEY` in a multi-column table, preserving autoincrement). **No FK**
+  generator `schema_to_ddl` (inline only a first-column single PK) so a
-  (4b); **no `DEFAULT`/`CHECK`/table-level `UNIQUE`** (4a.2).
+  created table and its rebuilt form have identical DDL; `serial`
+  autoincrement is independent of inline-vs-table-level PK (the insert
+  path computes the next value), verified by round-trip tests. **No
+  FK** (4b); **no `DEFAULT`/`CHECK`/table-level `UNIQUE`** (4a.2).
 - **4a.2 — The constraint slice.** Split out (2026-05-24,
   user-confirmed) for the constraints that are *not* a clean reuse:
   (1) **`CHECK`/`DEFAULT`** via the full `sql_expr` surface stored as

docs/plans/20260524-adr-0035-sql-ddl-4a.md

@@ -292,15 +292,22 @@ the ADR flags).
    Single-word aliases + an optional discarded `(len[,len])` cover the
    rest. Delivers the ADR §3 item without bending it.
 
-4. **`serial` PK inline emission (implementer call, step 3/worker).**
+4. **`do_create_table` inline-PK rule aligned with `schema_to_ddl`
-   `do_create_table` today inlines `PRIMARY KEY` (the rowid-alias that
+   (implementer call, step 3/worker; corrected by the `/runda` probe).**
-   makes `serial` autoincrement) **only when the table has one column**.
+   *Original premise (wrong):* a `serial` sole-PK in a multi-column SQL
-   SQL mode allows `CREATE TABLE t (id serial primary key, name text)` —
+   table needs an inline `PRIMARY KEY` (rowid-alias) or it loses
-   a `serial` sole-PK in a multi-column table — which would otherwise
+   autoincrement. **A round-trip probe disproved this:** after a rebuild
-   get a table-level PK and **lose autoincrement**. Step 3 extends the
+   the table gets a *table-level* PK (`schema_to_ddl` only inlines a
-   inline condition to "sole-PK column whose type is `serial` → inline
+   first-column PK), yet `serial` still autoincremented — the insert
-   `PRIMARY KEY` on that column," leaving the simple-mode paths
+   path computes the next value itself, independent of rowid-alias.
-   (single-column, or compound) unchanged. Covered by a worker test.
+   *Actual fix:* `do_create_table` now uses the **same inline rule as
+   `schema_to_ddl`** — inline only a single PK that is the **first
+   column** — so a freshly-created table and its rebuilt form have
+   identical DDL (no round-trip drift, the real latent issue a
+   multi-column SQL create would have exposed). Simple-mode paths
+   (always single-column create + `add column`) are unchanged. Covered
+   by two round-trip tests (serial PK as first **and** non-first
+   column survive a rebuild).
 
 5. **Redundant PK constraints (implementer call).** SQL mode is
    **lenient** like real engines: `id int primary key not null` /
@@ -311,6 +318,24 @@ the ADR flags).
    mode deliberately, matching the advanced-mode "trust the user like
    SQL" posture, ADR-0035 §7.)
 
+6. **Deferred constraints surface as a parse error (user-confirmed
+   2026-05-25).** `DEFAULT` / `CHECK` / table-level `UNIQUE` are absent
+   from the 4a grammar, so typing them is an ordinary parse error — the
+   ADR-0021 usage skeleton lists the supported `CREATE TABLE` form,
+   implicitly communicating what is not yet available. A bespoke
+   "not yet supported" message needs the deferred expression-parsing
+   work and so lands with 4a.2, when those shapes are added.
+
+7. **No-PK tables allowed in advanced mode (user-confirmed
+   2026-05-25).** `create table t (id int)` with no primary key is
+   accepted (standard SQL; the §7 trust posture), unlike simple mode
+   which requires/defaults a PK. Pinned by a worker test.
+
+8. **No-op skip is journalled (`/runda` fix).** A successful
+   `CREATE TABLE IF NOT EXISTS` no-op appends its line to `history.log`
+   like other read-only/no-op commands (`show table`) — the complete
+   journal (ADR-0034) — while taking no undo snapshot.
+
 ## 7. Devil's Advocate review of this plan
 
 - **Does it reuse rather than fork execution?** Yes — `do_create_table`

docs/requirements.md

@@ -209,14 +209,17 @@ handoff-14 cleanup; 449 after B2/C2.)
 
 - [ ] **Q1** SQL parsed via `sqlparser-rs`; supported subset is
   defined (specifics deferred to a future ADR).
-  *(Progress: SQL handling in advanced mode is still a
+  *(Progress: the advanced-mode SQL surface is authored as grammar
-  placeholder echo. The architecture is now decided — ADR-0030:
+  within the unified grammar tree (ADR-0030 / ADR-0024) and parsed by
-  SQL is authored as grammar within the unified grammar tree
+  the existing walker — **not** a separate batch parser — so SQL gets
-  (ADR-0024) and parsed by the existing walker, **not** a
+  the same completion / highlighting / hints as the DSL (ADR-0001's
-  separate batch parser — so SQL gets the same completion /
+  `sqlparser-rs` reservation is superseded). Implemented so far: full
-  highlighting / hints as the DSL. ADR-0001's `sqlparser-rs`
+  `SELECT` (ADR-0032), `INSERT` / `UPDATE` / `DELETE` (ADR-0033), and
-  reservation is superseded. Implementation is phased and
+  `CREATE TABLE` (ADR-0035 sub-phase 4a, 2026-05-25 — columns + types +
-  pending.)*
+  `NOT NULL` / `UNIQUE` / `PRIMARY KEY` + `IF NOT EXISTS`, executed
+  structurally). Remaining DDL — `CREATE TABLE` constraints (4a.2),
+  FK (4b), `DROP TABLE` (4c), indexes (4d), `ALTER TABLE` (4e–4h) — is
+  phased per ADR-0035 §13.)*
 - [ ] **Q2** Non-standard syntax rejected with a clear message
   pointing at the supported subset.
   *(Design done — ADR-0030 §8: out-of-subset statements are

src/app.rs

@@ -442,6 +442,23 @@ impl App {
                 self.handle_dsl_success(&command, description);
                 Vec::new()
             }
+            AppEvent::DslCreateSkipped {
+                command,
+                description,
+            } => {
+                // No-op (CREATE TABLE IF NOT EXISTS on an existing
+                // table, ADR-0035 §4): the skip note, then the existing
+                // structure — no misleading "[ok] create table" line.
+                self.note_system(crate::t!(
+                    "ddl.create_skipped_exists",
+                    name = command.target_table()
+                ));
+                for line in crate::output_render::render_structure(&description) {
+                    self.note_system(line);
+                }
+                self.current_table = Some(description);
+                Vec::new()
+            }
             AppEvent::DslDataSucceeded { command, data } => {
                 self.handle_dsl_query_success(&command, &data);
                 Vec::new()
@@ -1533,6 +1550,9 @@ impl App {
         use crate::friendly::{Operation, TranslateContext};
         let (operation, fallback_table, fallback_column) = match command {
             C::CreateTable { name, .. } => (Operation::CreateTable, Some(name.as_str()), None),
+            C::SqlCreateTable { name, .. } => {
+                (Operation::CreateTable, Some(name.as_str()), None)
+            }
             C::DropTable { name } => (Operation::DropTable, Some(name.as_str()), None),
             C::AddColumn { table, column, .. } => (
                 Operation::AddColumn,

src/completion.rs

@@ -2073,9 +2073,12 @@ mod tests {
 
     #[test]
     fn new_name_slot_offers_no_candidates_even_with_populated_cache() {
-        // `create table ` — the table-name slot is NewName.
+        // `create table ` — the table-name slot is NewName, so even
-        // Even if the cache has table/column entries, no
+        // with a populated cache no *schema* candidates are offered
-        // candidates are offered (the user invents the name).
+        // (the user invents the name). In advanced mode the sole
+        // candidate here is the optional `if` keyword (the
+        // `IF NOT EXISTS` prefix, ADR-0035 §4) — never a cached
+        // table/column name.
         let cache = SchemaCache {
             tables: vec!["Existing".to_string()],
             columns: vec!["AlsoExisting".to_string()],
@@ -2083,7 +2086,11 @@ mod tests {
             ..SchemaCache::default()
         };
         let cs = cands_with("create table ", 13, &cache);
-        assert!(cs.is_empty(), "got {cs:?}");
+        assert!(
+            !cs.iter().any(|c| c == "Existing" || c == "AlsoExisting"),
+            "NewName slot must not surface schema candidates; got {cs:?}"
+        );
+        assert_eq!(cs, vec!["if".to_string()], "only the advanced IF NOT EXISTS keyword");
     }
 
     fn keyword_cand(text: &str) -> Candidate {

src/db.rs

@@ -457,6 +457,18 @@ enum Request {
         source: Option<String>,
         reply: oneshot::Sender<Result<TableDescription, DbError>>,
     },
+    /// Advanced-mode SQL `CREATE TABLE` (ADR-0035 §1, 4a). Executes
+    /// structurally through `do_create_table`; `if_not_exists` turns
+    /// an existing table into a no-op (`CreateOutcome::Skipped`, no
+    /// snapshot) instead of an error (ADR-0035 §4).
+    SqlCreateTable {
+        name: String,
+        columns: Vec<ColumnSpec>,
+        primary_key: Vec<String>,
+        if_not_exists: bool,
+        source: Option<String>,
+        reply: oneshot::Sender<Result<CreateOutcome, DbError>>,
+    },
     DropTable {
         name: String,
         source: Option<String>,
@@ -813,6 +825,30 @@ impl Database {
         recv.await.map_err(|_| DbError::WorkerGone)?
     }
 
+    /// Advanced-mode SQL `CREATE TABLE` (ADR-0035 §1, 4a). Executes
+    /// structurally; returns whether the table was created or skipped
+    /// (the `IF NOT EXISTS` no-op, ADR-0035 §4).
+    pub async fn sql_create_table(
+        &self,
+        name: String,
+        columns: Vec<ColumnSpec>,
+        primary_key: Vec<String>,
+        if_not_exists: bool,
+        source: Option<String>,
+    ) -> Result<CreateOutcome, DbError> {
+        let (reply, recv) = oneshot::channel();
+        self.send(Request::SqlCreateTable {
+            name,
+            columns,
+            primary_key,
+            if_not_exists,
+            source,
+            reply,
+        })
+        .await?;
+        recv.await.map_err(|_| DbError::WorkerGone)?
+    }
+
     pub async fn drop_table(&self, name: String, source: Option<String>) -> Result<(), DbError> {
         let (reply, recv) = oneshot::channel();
         self.send(Request::DropTable { name, source, reply }).await?;
@@ -1653,6 +1689,42 @@ fn handle_request(
                 &primary_key,
             ));
         }
+        Request::SqlCreateTable {
+            name,
+            columns,
+            primary_key,
+            if_not_exists,
+            source,
+            reply,
+        } => {
+            // `IF NOT EXISTS` on an existing table is a no-op: reply
+            // `Skipped` with the existing structure and take **no**
+            // snapshot (there is nothing to undo). The submitted line is
+            // still journalled — like other read-only / no-op commands
+            // (`show table`), it belongs in the complete journal
+            // (ADR-0034). ADR-0035 §4.
+            if if_not_exists && user_table_exists(conn, &name).unwrap_or(false) {
+                let result = do_describe_table(conn, &name).and_then(|desc| {
+                    if let (Some(p), Some(text)) = (persistence, source.as_deref()) {
+                        p.append_history(text).map_err(DbError::from_persistence)?;
+                    }
+                    Ok(CreateOutcome::Skipped(desc))
+                });
+                let _ = reply.send(result);
+            } else {
+                snapshot_then(snap, batch, conn, source.as_deref(), reply, || {
+                    do_create_table(
+                        conn,
+                        persistence,
+                        source.as_deref(),
+                        &name,
+                        &columns,
+                        &primary_key,
+                    )
+                    .map(CreateOutcome::Created)
+                });
+            }
+        }
         Request::DropTable {
             name,
             source,
@@ -2507,6 +2579,18 @@ fn insert_column_metadata(
     Ok(())
 }
 
+/// The result of an advanced-mode SQL `CREATE TABLE` (ADR-0035 §4).
+///
+/// Either the table was created, or `IF NOT EXISTS` matched an
+/// existing table and the statement was a no-op. Both carry the
+/// table's structure so the runtime can render it; `Skipped` also
+/// drives the "already exists — skipped" note.
+#[derive(Debug)]
+pub enum CreateOutcome {
+    Created(TableDescription),
+    Skipped(TableDescription),
+}
+
 fn do_create_table(
     conn: &Connection,
     persistence: Option<&Persistence>,
@@ -2525,14 +2609,21 @@ fn do_create_table(
         ));
     }
 
-    // Generate the column list. For a single-column PK we inline
+    // Inline `PRIMARY KEY` on the column when the table has a single
-    // `PRIMARY KEY` on the column itself, which is required for
+    // primary-key column and it is the **first** column — the exact
-    // SQLite STRICT tables to give an `INTEGER PRIMARY KEY`
+    // rule [`schema_to_ddl`] uses on rebuild, so a table's DDL is
-    // column its rowid-alias semantics. For compound PKs (or
+    // identical whether freshly created or reconstructed (no
-    // when the single PK is on a non-first column) we emit a
+    // round-trip drift). SQLite grants an inline single-column PK
-    // table-level constraint.
+    // rowid-alias semantics; a compound PK, or a single PK that is not
-    let single_inline_pk = primary_key.len() == 1 && columns.len() == 1
+    // the first column, gets a table-level constraint. `serial`
-        && primary_key[0] == columns[0].name;
+    // autoincrement does **not** depend on this — the insert path
+    // computes the next value itself (verified by the multi-column /
+    // rebuild round-trip tests) — so the choice is purely about
+    // matching the rebuild generator (ADR-0035 §6.4).
+    let inline_pk_col: Option<&str> = (primary_key.len() == 1
+        && !columns.is_empty()
+        && primary_key[0] == columns[0].name)
+        .then(|| primary_key[0].as_str());
 
     // Compile each column's CHECK once (ADR-0029 §4) — reused
     // by the DDL clause and the metadata insert below. The
@@ -2550,13 +2641,14 @@ fn do_create_table(
             ident = quote_ident(&col.name),
             sqlite_type = col.ty.sqlite_strict_type(),
         );
-        if single_inline_pk {
+        if inline_pk_col == Some(col.name.as_str()) {
             clause.push_str(" PRIMARY KEY");
         }
         // ADR-0029 column constraints. A single-column PK is
-        // already NOT NULL + UNIQUE; the grammar rejects
+        // already NOT NULL + UNIQUE; the simple-mode grammar
-        // redundant declarations (ADR-0029 §9) so a PK column
+        // rejects redundant declarations (ADR-0029 §9) and the
-        // never carries them here.
+        // SQL builder de-dups them (ADR-0035 §6.5), so an
+        // inline-PK column never carries them here.
         clause.push_str(&column_constraints_sql(col)?);
         if let Some(cs) = check_sql {
             clause.push_str(&format!(" CHECK ({cs})"));
@@ -2570,7 +2662,7 @@ fn do_create_table(
         columns = column_clauses.join(", "),
     );
 
-    if !single_inline_pk && !primary_key.is_empty() {
+    if inline_pk_col.is_none() && !primary_key.is_empty() {
         let pk_idents: Vec<String> = primary_key.iter().map(|n| quote_ident(n)).collect();
         ddl.push_str(", PRIMARY KEY (");
         ddl.push_str(&pk_idents.join(", "));

src/dsl/command.rs

@@ -122,6 +122,19 @@ pub enum Command {
     DropTable {
         name: String,
     },
+    /// Advanced-mode SQL `CREATE TABLE` (ADR-0035 §1, sub-phase 4a).
+    /// Its own command, but executed **structurally** through the
+    /// same `do_create_table` machinery as [`Self::CreateTable`] —
+    /// the columns / PK reuse `ColumnSpec`. `if_not_exists` makes an
+    /// already-existing table a no-op-with-note rather than an error
+    /// (ADR-0035 §4). 4a carries only `NOT NULL` / `UNIQUE` /
+    /// `PRIMARY KEY`; `DEFAULT` / `CHECK` are the 4a.2 slice.
+    SqlCreateTable {
+        name: String,
+        columns: Vec<ColumnSpec>,
+        primary_key: Vec<String>,
+        if_not_exists: bool,
+    },
     /// Add a column to an existing table. The column carries
     /// its constraints from the same suffix grammar as
     /// `create table` (ADR-0029); `check` is `None` until the
@@ -625,6 +638,7 @@ impl Command {
     pub const fn verb(&self) -> &'static str {
         match self {
             Self::CreateTable { .. } => "create table",
+            Self::SqlCreateTable { .. } => "create table",
             Self::DropTable { .. } => "drop table",
             Self::AddColumn { .. } => "add column",
             Self::DropColumn { .. } => "drop column",
@@ -673,6 +687,7 @@ impl Command {
     pub fn target_table(&self) -> &str {
         match self {
             Self::CreateTable { name, .. }
+            | Self::SqlCreateTable { name, .. }
             | Self::DropTable { name }
             | Self::ShowTable { name }
             | Self::ShowData { name, .. } => name,

src/dsl/grammar/ddl.rs

@@ -1288,6 +1288,135 @@ pub static CREATE: CommandNode = CommandNode {
     help_id: Some("ddl.create"),
     usage_ids: &["parse.usage.create_table"],};
 
+/// Build a `Command::SqlCreateTable` from the advanced-mode SQL
+/// `CREATE TABLE` shape (ADR-0035 §1, sub-phase 4a). Executes
+/// structurally — extracts the same `ColumnSpec`/`primary_key` the
+/// simple-mode builder produces so the worker reuses `do_create_table`.
+///
+/// 4a surface: columns + types (the §3 alias map incl. `double
+/// precision`) + `NOT NULL` / `UNIQUE` / column- and table-level
+/// `PRIMARY KEY` + `IF NOT EXISTS`. `DEFAULT` / `CHECK` /
+/// table-level `UNIQUE` are absent from the grammar (4a.2), so they
+/// never reach this builder.
+fn build_sql_create_table(path: &MatchedPath, _source: &str) -> Result<Command, ValidationError> {
+    let name = require_ident(path, "table_name")?;
+    // `if` only appears in the `IF NOT EXISTS` prefix (the `not` of
+    // `NOT NULL` never carries an `if`), so its presence is the flag.
+    let if_not_exists = path
+        .items
+        .iter()
+        .any(|i| matches!(i.kind, MatchedKind::Word("if")));
+
+    let mut columns: Vec<ColumnSpec> = Vec::new();
+    let mut primary_key: Vec<String> = Vec::new();
+    let mut pending_name: Option<String> = None;
+    let mut items = path.items.iter().peekable();
+    while let Some(item) = items.next() {
+        match &item.kind {
+            // A column name stashes until its type finalises the spec.
+            MatchedKind::Ident { role: "col_name", .. } => {
+                pending_name = Some(item.text.clone());
+            }
+            // Single-word type — resolve through the SQL alias map.
+            MatchedKind::Ident { role: "col_type", .. } => {
+                let ty = Type::from_sql_name(&item.text).ok_or_else(|| ValidationError {
+                    message_key: "parse.error_wrapper",
+                    args: vec![("detail", "unknown type".to_string())],
+                })?;
+                let col_name = pending_name.take().ok_or_else(|| ValidationError {
+                    message_key: "parse.error_wrapper",
+                    args: vec![("detail", "column type without a name".to_string())],
+                })?;
+                columns.push(ColumnSpec::new(col_name, ty));
+            }
+            // `double precision` — the two-word alias maps to `real`.
+            // The grammar guarantees `precision` follows `double`.
+            MatchedKind::Word("double") => {
+                if matches!(
+                    items.peek().map(|i| &i.kind),
+                    Some(MatchedKind::Word("precision"))
+                ) {
+                    items.next();
+                }
+                let col_name = pending_name.take().ok_or_else(|| ValidationError {
+                    message_key: "parse.error_wrapper",
+                    args: vec![("detail", "column type without a name".to_string())],
+                })?;
+                columns.push(ColumnSpec::new(col_name, Type::Real));
+            }
+            // A table-level `PRIMARY KEY (col, …)` column reference.
+            MatchedKind::Ident { role: "pk_column", .. } => {
+                primary_key.push(item.text.clone());
+            }
+            // `not null` column constraint (only once a column exists;
+            // the `IF NOT EXISTS` `not` precedes every column).
+            MatchedKind::Word("not") => {
+                if matches!(
+                    items.peek().map(|i| &i.kind),
+                    Some(MatchedKind::Word("null"))
+                ) {
+                    items.next();
+                    if let Some(last) = columns.last_mut() {
+                        last.not_null = true;
+                    }
+                }
+            }
+            MatchedKind::Word("unique") => {
+                if let Some(last) = columns.last_mut() {
+                    last.unique = true;
+                }
+            }
+            // `primary key` — either a column-level constraint (mark
+            // the most recent column) or the table-level clause (whose
+            // `pk_column` idents follow and are collected above).
+            MatchedKind::Word("primary") => {
+                if matches!(items.peek().map(|i| &i.kind), Some(MatchedKind::Word("key"))) {
+                    items.next();
+                    // Table-level `PRIMARY KEY (…)` is followed by `(`
+                    // (then `pk_column` idents, collected above);
+                    // column-level `PRIMARY KEY` is not, and marks the
+                    // most-recent column.
+                    let table_level = matches!(
+                        items.peek().map(|i| &i.kind),
+                        Some(MatchedKind::Punct('('))
+                    );
+                    if !table_level && let Some(last) = columns.last() {
+                        primary_key.push(last.name.clone());
+                    }
+                }
+            }
+            _ => {}
+        }
+    }
+
+    // De-dup redundant flags off a sole primary-key column (ADR-0035
+    // §6.5): a single-column PK is already NOT NULL + UNIQUE, so
+    // emitting them again would create a spurious index. Advanced mode
+    // accepts the redundant spelling (real SQL does) rather than
+    // rejecting it like simple mode (ADR-0029 §9).
+    if primary_key.len() == 1
+        && let Some(c) = columns.iter_mut().find(|c| c.name == primary_key[0])
+    {
+        c.not_null = false;
+        c.unique = false;
+    }
+
+    Ok(Command::SqlCreateTable {
+        name,
+        columns,
+        primary_key,
+        if_not_exists,
+    })
+}
+
+pub static SQL_CREATE_TABLE: CommandNode = CommandNode {
+    entry: Word::keyword("create"),
+    shape: Node::Subgrammar(&super::sql_create_table::SQL_CREATE_TABLE_SHAPE),
+    ast_builder: build_sql_create_table,
+    help_id: Some("ddl.sql_create_table"),
+    usage_ids: &["parse.usage.sql_create_table"],
+};
+
 // =================================================================
 // Tests — `create table` column constraints (ADR-0029 §2.1, §9)
 // =================================================================

src/dsl/grammar/mod.rs

@@ -584,6 +584,12 @@ pub static REGISTRY: &[(&CommandNode, CommandCategory)] = &[
     (&data::SQL_INSERT, CommandCategory::Advanced),
     (&data::SQL_UPDATE, CommandCategory::Advanced),
     (&data::SQL_DELETE, CommandCategory::Advanced),
+    // Shared entry word `create` (ADR-0035 §2): the simple
+    // `ddl::CREATE` (above) and this advanced SQL node. The
+    // dispatcher tries SQL first in advanced mode and falls back to
+    // the `create table … with pk …` DSL node when the SQL shape
+    // does not match — the `insert` precedent.
+    (&ddl::SQL_CREATE_TABLE, CommandCategory::Advanced),
 ];
 
 /// Whether `entry` names an advanced-mode-only command (ADR-0030

src/dsl/grammar/sql_create_table.rs

@@ -367,10 +367,177 @@ mod tests {
     fn deferred_constraints_are_not_accepted_in_4a() {
         // DEFAULT / CHECK / table-level UNIQUE belong to the 4a.2
         // constraint slice; their shapes are absent here, so they do
-        // not walk (the builder turns this into a friendly
+        // not walk (surfacing as a parse error with the usage
-        // "not yet supported" — tested there).
+        // skeleton, which lists the supported surface).
         bad("table t (id int default 0)");
         bad("table t (id int check (id > 0))");
         bad("table t (a int, b int, unique (a, b))");
     }
 }
+
+// =================================================================
+// Builder tests — `parse_command` (advanced mode) lowers the SQL
+// `CREATE TABLE` to `Command::SqlCreateTable` with the right fields
+// (ADR-0035 §1/§3, sub-phase 4a).
+// =================================================================
+
+#[cfg(test)]
+mod builder_tests {
+    use crate::dsl::command::Command;
+    use crate::dsl::parser::{parse_command, parse_command_in_mode};
+    use crate::dsl::types::Type;
+    use crate::mode::Mode;
+
+    /// Parse in advanced mode and unwrap the `SqlCreateTable` fields.
+    fn sct(input: &str) -> (String, Vec<(String, Type)>, Vec<String>, bool) {
+        match parse_command(input).expect("should parse as SQL CREATE TABLE") {
+            Command::SqlCreateTable {
+                name,
+                columns,
+                primary_key,
+                if_not_exists,
+            } => (
+                name,
+                columns.into_iter().map(|c| (c.name, c.ty)).collect(),
+                primary_key,
+                if_not_exists,
+            ),
+            other => panic!("expected SqlCreateTable, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn minimal_columns_and_types() {
+        let (name, cols, pk, ine) = sct("create table t (id int, name text)");
+        assert_eq!(name, "t");
+        assert_eq!(
+            cols,
+            vec![("id".to_string(), Type::Int), ("name".to_string(), Type::Text)]
+        );
+        assert!(pk.is_empty(), "no PK declared");
+        assert!(!ine);
+    }
+
+    #[test]
+    fn integer_primary_key_is_plain_int_not_serial() {
+        // ADR-0035 §3: the type map is lexical; INTEGER PRIMARY KEY is
+        // a plain int PK, NOT auto-increment (that is `serial`).
+        let (_, cols, pk, _) = sct("create table t (id integer primary key)");
+        assert_eq!(cols, vec![("id".to_string(), Type::Int)]);
+        assert_eq!(pk, vec!["id".to_string()]);
+    }
+
+    #[test]
+    fn standard_sql_aliases_map_to_playground_types() {
+        let (_, cols, _, _) = sct(
+            "create table t (a bigint, b varchar, c boolean, d timestamp, \
+             e numeric, f float, g binary)",
+        );
+        assert_eq!(
+            cols,
+            vec![
+                ("a".to_string(), Type::Int),
+                ("b".to_string(), Type::Text),
+                ("c".to_string(), Type::Bool),
+                ("d".to_string(), Type::DateTime),
+                ("e".to_string(), Type::Decimal),
+                ("f".to_string(), Type::Real),
+                ("g".to_string(), Type::Blob),
+            ]
+        );
+    }
+
+    #[test]
+    fn double_precision_maps_to_real() {
+        let (_, cols, _, _) = sct("create table t (id int, x double precision)");
+        assert_eq!(
+            cols,
+            vec![("id".to_string(), Type::Int), ("x".to_string(), Type::Real)]
+        );
+    }
+
+    #[test]
+    fn length_args_are_ignored() {
+        let (_, cols, _, _) = sct("create table t (a varchar(255), b numeric(10, 2))");
+        assert_eq!(
+            cols,
+            vec![("a".to_string(), Type::Text), ("b".to_string(), Type::Decimal)]
+        );
+    }
+
+    #[test]
+    fn column_level_primary_key_populates_pk() {
+        let (_, _, pk, _) = sct("create table t (id serial primary key, name text)");
+        assert_eq!(pk, vec!["id".to_string()]);
+    }
+
+    #[test]
+    fn table_level_compound_primary_key() {
+        let (_, _, pk, _) = sct("create table t (a int, b int, primary key (a, b))");
+        assert_eq!(pk, vec!["a".to_string(), "b".to_string()]);
+    }
+
+    #[test]
+    fn if_not_exists_sets_the_flag() {
+        let (name, _, _, ine) = sct("create table if not exists t (id int)");
+        assert_eq!(name, "t");
+        assert!(ine);
+    }
+
+    #[test]
+    fn not_null_and_unique_attach_to_their_column() {
+        match parse_command("create table t (id int primary key, code text not null unique)")
+            .expect("parses")
+        {
+            Command::SqlCreateTable { columns, .. } => {
+                let code = columns.iter().find(|c| c.name == "code").expect("code col");
+                assert!(code.not_null && code.unique);
+            }
+            other => panic!("expected SqlCreateTable, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn redundant_constraints_deduped_off_sole_pk_column() {
+        // ADR-0035 §6.5: advanced mode accepts the redundant spelling
+        // and silently drops the flags off the sole PK column.
+        match parse_command("create table t (id int primary key not null unique)")
+            .expect("parses")
+        {
+            Command::SqlCreateTable {
+                columns,
+                primary_key,
+                ..
+            } => {
+                assert_eq!(primary_key, vec!["id".to_string()]);
+                let id = &columns[0];
+                assert!(!id.not_null && !id.unique, "redundant flags deduped off PK");
+            }
+            other => panic!("expected SqlCreateTable, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn simple_create_still_parses_as_dsl_in_both_modes() {
+        // The shared `create` entry word: the DSL `with pk` form falls
+        // back to `Command::CreateTable` even in advanced mode, and is
+        // the only shape in simple mode (ADR-0035 §2 dispatch).
+        for mode in [Mode::Simple, Mode::Advanced] {
+            let cmd = parse_command_in_mode("create table T with pk id(serial)", mode)
+                .unwrap_or_else(|e| panic!("{mode:?} should parse the DSL form: {e:?}"));
+            assert!(
+                matches!(cmd, Command::CreateTable { .. }),
+                "{mode:?}: expected DSL CreateTable, got {cmd:?}"
+            );
+        }
+    }
+
+    #[test]
+    fn sql_create_is_advanced_only() {
+        // The SQL `( … )` form is not available in simple mode.
+        assert!(
+            parse_command_in_mode("create table t (id int)", Mode::Simple).is_err(),
+            "SQL CREATE TABLE must not parse in simple mode"
+        );
+    }
+}

src/event.rs

@@ -28,6 +28,13 @@ pub enum AppEvent {
         command: Command,
         description: Option<TableDescription>,
     },
+    /// A SQL `CREATE TABLE IF NOT EXISTS` matched an existing table —
+    /// a no-op (ADR-0035 §4). Renders the existing structure plus an
+    /// "already exists — skipped" note.
+    DslCreateSkipped {
+        command: Command,
+        description: TableDescription,
+    },
     /// A `show data` query succeeded.
     DslDataSucceeded { command: Command, data: DataResult },
     /// An `explain …` command succeeded (ADR-0028). `plan`

src/friendly/keys.rs

@@ -171,6 +171,9 @@ pub const KEYS_AND_PLACEHOLDERS: &[(&str, &[&str])] = &[
     ("help.app.undo", &[]),
     ("help.app.redo", &[]),
     ("help.ddl.create", &[]),
+    ("help.ddl.sql_create_table", &[]),
+    // Advanced-mode SQL CREATE TABLE no-op note (ADR-0035 §4).
+    ("ddl.create_skipped_exists", &["name"]),
     ("help.ddl.drop", &[]),
     ("help.ddl.add", &[]),
     ("help.ddl.rename", &[]),
@@ -241,6 +244,7 @@ pub const KEYS_AND_PLACEHOLDERS: &[(&str, &[&str])] = &[
     ("parse.usage.add_relationship", &[]),
     ("parse.usage.change_column", &[]),
     ("parse.usage.create_table", &[]),
+    ("parse.usage.sql_create_table", &[]),
     ("parse.usage.delete", &[]),
     ("parse.usage.drop_column", &[]),
     ("parse.usage.drop_constraint", &[]),

src/friendly/strings/en-US.yaml

@@ -260,6 +260,9 @@ help:
   ddl:
     create: |-
         create table <T> with pk [<col>(<type>), ...]    — create a table
+    sql_create_table: |-
+        create table [if not exists] <T> (<col> <type> [not null] [unique] [primary key], ...
+          [, primary key (<col>, ...)])    — create a table (advanced SQL)
     drop: |-
         drop table <T>                                  — remove a table
         drop column [from] [table] <T>: <col> [--cascade]   — remove a column
@@ -368,6 +371,13 @@ hint:
   # explicit-column form so the skipped column is discoverable.
   value_slot_autogen_skipped: "({columns} auto-generated — skipped here; list columns explicitly, e.g. `insert into T (...) values (...)`, to set it.)"
 
+# Advanced-mode SQL DDL execution notes (ADR-0035).
+ddl:
+  # `create table if not exists <T>` where the table is already
+  # present: a no-op that succeeds with this note instead of an
+  # "already exists" error.
+  create_skipped_exists: "table '{name}' already exists — skipped (no changes made)"
+
 parse:
   # Wrapper around chumsky's structural error message. The
   # caret pointer (visualising the failure column) is printed
@@ -435,6 +445,7 @@ parse:
   # placeholders. ADR-0009's surface conventions apply.
   usage:
     create_table: "create table <Name> with pk [<col>(<type>)[, ...]]"
+    sql_create_table: "create table [if not exists] <Name> (<col> <type> [not null] [unique] [primary key], ... [, primary key (<col>, ...)])"
     drop_table: "drop table <Name>"
     drop_column: "drop column [from] [table] <Table>: <Name>"
     drop_relationship: |-

src/runtime.rs

@@ -29,8 +29,8 @@ use crate::action::Action;
 use crate::app::App;
 use crate::cli::Args;
 use crate::db::{
-    AddColumnResult, ChangeColumnTypeResult, DataResult, Database, DbError, DeleteResult,
+    AddColumnResult, ChangeColumnTypeResult, CreateOutcome, DataResult, Database, DbError,
-    DropColumnResult, InsertResult, QueryPlan, TableDescription, UpdateResult,
+    DeleteResult, DropColumnResult, InsertResult, QueryPlan, TableDescription, UpdateResult,
 };
 use crate::dsl::{Command, ColumnSpec};
 use crate::dsl::walker::Severity;
@@ -1253,6 +1253,10 @@ fn spawn_dsl_dispatch(
                 command: command.clone(),
                 description,
             },
+            Ok(CommandOutcome::SchemaSkipped(description)) => AppEvent::DslCreateSkipped {
+                command: command.clone(),
+                description,
+            },
             Ok(CommandOutcome::Query(data)) => AppEvent::DslDataSucceeded {
                 command: command.clone(),
                 data,
@@ -1645,6 +1649,11 @@ fn parse_qualified_target(message: &str) -> Option<(String, String)> {
 
 enum CommandOutcome {
     Schema(Option<TableDescription>),
+    /// A SQL `CREATE TABLE IF NOT EXISTS` that matched an existing
+    /// table — a no-op (ADR-0035 §4). Carries the existing structure
+    /// so the App can render it alongside the "already exists —
+    /// skipped" note.
+    SchemaSkipped(TableDescription),
     Query(DataResult),
     QueryPlan(QueryPlan),
     Insert(InsertResult),
@@ -1911,6 +1920,18 @@ async fn execute_command_typed(
             .create_table(name, columns, primary_key, src)
             .await
             .map(|d| CommandOutcome::Schema(Some(d))),
+        Command::SqlCreateTable {
+            name,
+            columns,
+            primary_key,
+            if_not_exists,
+        } => database
+            .sql_create_table(name, columns, primary_key, if_not_exists, src)
+            .await
+            .map(|outcome| match outcome {
+                CreateOutcome::Created(d) => CommandOutcome::Schema(Some(d)),
+                CreateOutcome::Skipped(d) => CommandOutcome::SchemaSkipped(d),
+            }),
         Command::DropTable { name } => database
             .drop_table(name, src)
             .await

tests/replay_command.rs

@@ -88,6 +88,38 @@ fn assert_failed_at(events: &[AppEvent], expected_line: usize) -> &AppEvent {
     }
 }
 
+#[test]
+fn replay_runs_advanced_sql_create_table_as_a_write() {
+    // ADR-0035 §10: `create` is a schema-write entry word (not in the
+    // ADR-0034 app-lifecycle skip set), so an advanced-mode SQL
+    // `CREATE TABLE` line replays as a write — re-applied, not skipped
+    // — and executes structurally (the table is rebuilt from the line).
+    let data = tempdir();
+    let (project, db) = open_project_db(data.path());
+    write_script(
+        project.path(),
+        "ddl.commands",
+        "create table Widget (id serial primary key, name text)\n\
+         insert into Widget (name) values ('gadget')\n",
+    );
+
+    let events = rt().block_on(async { run_replay(&db, project.path(), "ddl.commands").await });
+    assert_completed(&events, 2);
+
+    // The SQL DDL line actually created the structural table…
+    let desc = rt()
+        .block_on(async { db.describe_table("Widget".to_string(), None).await })
+        .expect("describe");
+    let names: Vec<String> = desc.columns.iter().map(|c| c.name.clone()).collect();
+    assert_eq!(names, vec!["id".to_string(), "name".to_string()]);
+    // …and the following insert (serial id auto-filled) ran against it.
+    let rows = rt()
+        .block_on(async { db.query_data("Widget".to_string(), None, None, None).await })
+        .expect("query")
+        .rows;
+    assert_eq!(rows.len(), 1);
+}
+
 #[test]
 fn replay_three_lines_dispatches_three_commands() {
     let data = tempdir();

tests/sql_create_table.rs

@@ -0,0 +1,368 @@
+//! Sub-phase 4a integration tests for advanced-mode SQL
+//! `CREATE TABLE` (ADR-0035 §1/§4).
+//!
+//! Worker round-trip: a `Command::SqlCreateTable` executes
+//! **structurally** through the existing `do_create_table` machinery,
+//! so an advanced-mode-created table is a first-class playground
+//! object (metadata + the ten-type vocabulary). Covers:
+//! - Created tables appear in `list_tables` and `describe_table`
+//!   reports the playground `user_type` per column.
+//! - A `serial` sole-PK autoincrements even in a multi-column table
+//!   (the §6.4 inline-`PRIMARY KEY` extension).
+//! - `IF NOT EXISTS` on an existing table is a no-op (`Skipped`); the
+//!   plain form errors when the table exists (§4).
+//! - One SQL `CREATE TABLE` is exactly one undo step (ADR-0006).
+//!
+//! Parsing (text → `Command::SqlCreateTable`) is covered by the
+//! `builder_tests` in `src/dsl/grammar/sql_create_table.rs`; these
+//! tests drive the worker directly, mirroring `tests/sql_insert.rs`.
+
+use rdbms_playground::db::{CreateOutcome, Database};
+use rdbms_playground::dsl::{ColumnSpec, Type, Value};
+use rdbms_playground::persistence::Persistence;
+use rdbms_playground::project;
+
+fn rt() -> tokio::runtime::Runtime {
+    tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .expect("tokio rt")
+}
+
+fn open(undo: bool) -> (project::Project, Database, tempfile::TempDir) {
+    let dir = tempfile::tempdir().expect("create tempdir");
+    let project =
+        project::open_or_create(None, Some(dir.path())).expect("open or create project");
+    let persistence = Persistence::new(project.path().to_path_buf());
+    let db = Database::open_with_persistence_and_undo(project.db_path(), persistence, undo)
+        .expect("open db with persistence");
+    (project, db, dir)
+}
+
+#[test]
+fn created_table_appears_with_playground_types() {
+    let (_p, db, _d) = open(false);
+    let r = rt();
+    let out = r
+        .block_on(db.sql_create_table(
+            "Widget".to_string(),
+            vec![
+                ColumnSpec::new("id", Type::Int),
+                ColumnSpec::new("name", Type::Text),
+            ],
+            vec!["id".to_string()],
+            false,
+            Some("create table Widget (id int primary key, name text)".to_string()),
+        ))
+        .expect("create should succeed");
+    assert!(matches!(out, CreateOutcome::Created(_)));
+
+    let tables = r.block_on(db.list_tables()).expect("list");
+    assert!(tables.contains(&"Widget".to_string()));
+
+    let desc = r
+        .block_on(db.describe_table("Widget".to_string(), None))
+        .expect("describe");
+    let types: Vec<(String, Option<Type>)> = desc
+        .columns
+        .iter()
+        .map(|c| (c.name.clone(), c.user_type))
+        .collect();
+    assert_eq!(
+        types,
+        vec![
+            ("id".to_string(), Some(Type::Int)),
+            ("name".to_string(), Some(Type::Text)),
+        ]
+    );
+}
+
+#[test]
+fn integer_primary_key_is_plain_int() {
+    // ADR-0035 §3: INTEGER PRIMARY KEY maps to plain `int`, not
+    // `serial`. The structural object reports `int`.
+    let (_p, db, _d) = open(false);
+    let r = rt();
+    r.block_on(db.sql_create_table(
+        "T".to_string(),
+        vec![ColumnSpec::new("id", Type::Int)],
+        vec!["id".to_string()],
+        false,
+        Some("create table T (id integer primary key)".to_string()),
+    ))
+    .expect("create");
+    let desc = r
+        .block_on(db.describe_table("T".to_string(), None))
+        .expect("describe");
+    assert_eq!(desc.columns[0].user_type, Some(Type::Int));
+}
+
+#[test]
+fn serial_pk_autoincrements_in_multi_column_table() {
+    // §6.4: a `serial` sole-PK in a *multi-column* table must inline
+    // `PRIMARY KEY` so it keeps autoincrement (rowid-alias) semantics
+    // — the case simple mode never produces in one statement.
+    let (_p, db, _d) = open(false);
+    let r = rt();
+    r.block_on(db.sql_create_table(
+        "T".to_string(),
+        vec![
+            ColumnSpec::new("id", Type::Serial),
+            ColumnSpec::new("name", Type::Text),
+        ],
+        vec!["id".to_string()],
+        false,
+        Some("create table T (id serial primary key, name text)".to_string()),
+    ))
+    .expect("create");
+
+    // Form B inserts (no column list): the serial id is auto-filled.
+    for name in ["a", "b"] {
+        r.block_on(db.insert(
+            "T".to_string(),
+            None,
+            vec![Value::Text(name.to_string())],
+            Some(format!("insert into T (name) values ('{name}')")),
+        ))
+        .expect("insert");
+    }
+
+    let data = r
+        .block_on(db.query_data("T".to_string(), None, None, None))
+        .expect("query");
+    let id_idx = data
+        .columns
+        .iter()
+        .position(|c| c == "id")
+        .expect("id column");
+    let mut ids: Vec<Option<String>> = data.rows.iter().map(|row| row[id_idx].clone()).collect();
+    ids.sort();
+    assert_eq!(
+        ids,
+        vec![Some("1".to_string()), Some("2".to_string())],
+        "serial PK autoincremented 1, 2"
+    );
+}
+
+#[test]
+fn if_not_exists_is_a_noop_when_table_exists() {
+    let (_p, db, _d) = open(false);
+    let r = rt();
+    let specs = || vec![ColumnSpec::new("id", Type::Int)];
+    r.block_on(db.sql_create_table(
+        "T".to_string(),
+        specs(),
+        vec!["id".to_string()],
+        false,
+        Some("create table T (id int)".to_string()),
+    ))
+    .expect("first create");
+
+    let out = r
+        .block_on(db.sql_create_table(
+            "T".to_string(),
+            specs(),
+            vec!["id".to_string()],
+            true, // IF NOT EXISTS
+            Some("create table if not exists T (id int)".to_string()),
+        ))
+        .expect("second create should succeed as a no-op");
+    assert!(
+        matches!(out, CreateOutcome::Skipped(_)),
+        "IF NOT EXISTS on an existing table is a no-op"
+    );
+
+    let tables = r.block_on(db.list_tables()).expect("list");
+    assert_eq!(tables.iter().filter(|t| t.as_str() == "T").count(), 1);
+}
+
+#[test]
+fn table_without_primary_key_is_allowed() {
+    // Advanced mode allows a PK-less table (standard SQL; the
+    // "trust the user like SQL" posture, ADR-0035 §7) — unlike simple
+    // mode, which requires/defaults a PK. User-confirmed 2026-05-25.
+    let (_p, db, _d) = open(false);
+    let r = rt();
+    let out = r
+        .block_on(db.sql_create_table(
+            "Notes".to_string(),
+            vec![ColumnSpec::new("body", Type::Text)],
+            vec![], // no primary key
+            false,
+            Some("create table Notes (body text)".to_string()),
+        ))
+        .expect("a PK-less table should create");
+    assert!(matches!(out, CreateOutcome::Created(_)));
+    // And it is usable: a row inserts and reads back.
+    r.block_on(db.insert(
+        "Notes".to_string(),
+        None,
+        vec![Value::Text("hello".to_string())],
+        Some("insert into Notes (body) values ('hello')".to_string()),
+    ))
+    .expect("insert into PK-less table");
+    let data = r
+        .block_on(db.query_data("Notes".to_string(), None, None, None))
+        .expect("query");
+    assert_eq!(data.rows.len(), 1);
+}
+
+#[test]
+fn if_not_exists_noop_is_journalled() {
+    // A successful no-op is still a submission and belongs in the
+    // complete journal (ADR-0034) — like read-only `show table`, and
+    // unlike a *failed* duplicate-create (journalled `err`).
+    let (p, db, _d) = open(false);
+    let r = rt();
+    r.block_on(db.sql_create_table(
+        "T".to_string(),
+        vec![ColumnSpec::new("id", Type::Int)],
+        vec!["id".to_string()],
+        false,
+        Some("create table T (id int)".to_string()),
+    ))
+    .expect("first create");
+    let noop = "create table if not exists T (id int)";
+    let out = r
+        .block_on(db.sql_create_table(
+            "T".to_string(),
+            vec![ColumnSpec::new("id", Type::Int)],
+            vec!["id".to_string()],
+            true,
+            Some(noop.to_string()),
+        ))
+        .expect("no-op");
+    assert!(matches!(out, CreateOutcome::Skipped(_)));
+    let log = std::fs::read_to_string(p.path().join("history.log")).expect("read history.log");
+    assert!(log.contains(noop), "the no-op skip should be journalled; log:\n{log}");
+}
+
+#[test]
+fn plain_create_errors_when_table_exists() {
+    let (_p, db, _d) = open(false);
+    let r = rt();
+    let specs = || vec![ColumnSpec::new("id", Type::Int)];
+    r.block_on(db.sql_create_table(
+        "T".to_string(),
+        specs(),
+        vec!["id".to_string()],
+        false,
+        Some("create table T (id int)".to_string()),
+    ))
+    .expect("first create");
+
+    let err = r.block_on(db.sql_create_table(
+        "T".to_string(),
+        specs(),
+        vec!["id".to_string()],
+        false, // no IF NOT EXISTS
+        Some("create table T (id int)".to_string()),
+    ));
+    assert!(err.is_err(), "re-creating an existing table without IF NOT EXISTS errors");
+}
+
+#[test]
+fn sql_create_table_is_one_undo_step() {
+    let (_p, db, _d) = open(true); // undo enabled
+    let r = rt();
+    r.block_on(db.sql_create_table(
+        "T".to_string(),
+        vec![ColumnSpec::new("id", Type::Int)],
+        vec!["id".to_string()],
+        false,
+        Some("create table T (id int)".to_string()),
+    ))
+    .expect("create");
+    assert!(r.block_on(db.list_tables()).unwrap().contains(&"T".to_string()));
+
+    let undone = r.block_on(db.undo()).expect("undo call");
+    assert!(undone.is_some(), "the CREATE TABLE recorded one undo step");
+    assert!(
+        !r.block_on(db.list_tables()).unwrap().contains(&"T".to_string()),
+        "table is gone after a single undo"
+    );
+}
+
+/// Sorted `id` column values of table `T`.
+fn ids(db: &Database, r: &tokio::runtime::Runtime) -> Vec<Option<String>> {
+    let d = r
+        .block_on(db.query_data("T".to_string(), None, None, None))
+        .expect("query");
+    let idx = d.columns.iter().position(|c| c == "id").expect("id column");
+    let mut v: Vec<Option<String>> = d.rows.iter().map(|row| row[idx].clone()).collect();
+    v.sort();
+    v
+}
+
+fn insert_row(db: &Database, r: &tokio::runtime::Runtime, name: &str) {
+    r.block_on(db.insert(
+        "T".to_string(),
+        None,
+        vec![Value::Text(name.to_string())],
+        Some(format!("insert into T (name) values ('{name}')")),
+    ))
+    .expect("insert");
+}
+
+/// `serial` PK as the **first** column must keep autoincrement across a
+/// rebuild: the structural create and the `schema_to_ddl` rebuild both
+/// inline `PRIMARY KEY` on a first-column single PK, so the DDL is
+/// identical and the sequence continues (id 3 after rebuild).
+#[test]
+fn serial_pk_first_column_autoincrements_after_rebuild() {
+    let (p, db, _d) = open(false);
+    let r = rt();
+    r.block_on(db.sql_create_table(
+        "T".to_string(),
+        vec![
+            ColumnSpec::new("id", Type::Serial),
+            ColumnSpec::new("name", Type::Text),
+        ],
+        vec!["id".to_string()],
+        false,
+        Some("create table T (id serial primary key, name text)".to_string()),
+    ))
+    .expect("create");
+    insert_row(&db, &r, "a");
+    insert_row(&db, &r, "b");
+    r.block_on(db.rebuild_from_text(p.path().to_path_buf(), None))
+        .expect("rebuild");
+    insert_row(&db, &r, "c");
+    assert_eq!(
+        ids(&db, &r),
+        vec![Some("1".to_string()), Some("2".to_string()), Some("3".to_string())]
+    );
+}
+
+/// `serial` PK as a **non-first** column must also keep autoincrement
+/// across a rebuild. Here the rebuild emits a *table-level* PK (the PK
+/// is not column 0), proving autoincrement does not rely on the
+/// rowid-alias / inline-PK form — the insert path computes the next
+/// value itself (ADR-0035 §6.4). Guards against silent round-trip loss.
+#[test]
+fn serial_pk_non_first_column_autoincrements_after_rebuild() {
+    let (p, db, _d) = open(false);
+    let r = rt();
+    r.block_on(db.sql_create_table(
+        "T".to_string(),
+        vec![
+            ColumnSpec::new("name", Type::Text),
+            ColumnSpec::new("id", Type::Serial),
+        ],
+        vec!["id".to_string()],
+        false,
+        Some("create table T (name text, id serial primary key)".to_string()),
+    ))
+    .expect("create");
+    insert_row(&db, &r, "a");
+    insert_row(&db, &r, "b");
+    assert_eq!(ids(&db, &r), vec![Some("1".to_string()), Some("2".to_string())]);
+    r.block_on(db.rebuild_from_text(p.path().to_path_buf(), None))
+        .expect("rebuild");
+    insert_row(&db, &r, "c");
+    assert_eq!(
+        ids(&db, &r),
+        vec![Some("1".to_string()), Some("2".to_string()), Some("3".to_string())],
+        "serial keeps autoincrement after a rebuild even as a non-first column"
+    );
+}

tests/typing_surface/mod.rs

@@ -216,6 +216,7 @@ fn command_kind_label(cmd: &rdbms_playground::dsl::Command) -> String {
     use rdbms_playground::dsl::Command::*;
     match cmd {
         CreateTable { .. } => "CreateTable".into(),
+        SqlCreateTable { .. } => "SqlCreateTable".into(),
         DropTable { .. } => "DropTable".into(),
         AddColumn { .. } => "AddColumn".into(),
         DropColumn { .. } => "DropColumn".into(),