Runtime: schema-aware replay parsing

run_replay parsed each line with the schemaless parse_command, so
Phase D typed-slot rejections (wrong-count value lists, wrong-type
column values) fired only at bind time during replay — inconsistent
with the interactive path (handoff-12 §2.1).

run_replay now re-snapshots the schema per line (the schema mutates
as replayed create-table / add-column commands run) and parses with
parse_command_with_schema. Extracted build_schema_cache, shared with
the interactive refresh_schema_cache.

Added a replay integration test asserting a typed-slot violation is
caught at parse time (through the replay.error_parse wrapper).

src/runtime.rs

@@ -838,6 +838,19 @@ async fn refresh_schema_cache(
     database: &Database,
     event_tx: &mpsc::Sender<AppEvent>,
 ) {
+    let cache = build_schema_cache(database).await;
+    let _ = event_tx.send(AppEvent::SchemaCacheRefreshed(cache)).await;
+}
+
+/// Build a `SchemaCache` snapshot from the live database.
+///
+/// Shared by `refresh_schema_cache` (interactive path — wraps
+/// the result in a `SchemaCacheRefreshed` event) and the replay
+/// path (which re-snapshots per line because the schema mutates
+/// as replayed `create table` / `add column` commands run).
+/// Best-effort: a failed query leaves that list empty and the
+/// walker falls back to schemaless behaviour.
+async fn build_schema_cache(database: &Database) -> crate::completion::SchemaCache {
     use crate::completion::{SchemaCache, TableColumn};
     use crate::dsl::grammar::IdentSource;
     let mut cache = SchemaCache::default();
@@ -872,7 +885,7 @@ async fn refresh_schema_cache(
             cache.table_columns.insert(name, cols);
         }
     }
-    let _ = event_tx.send(AppEvent::SchemaCacheRefreshed(cache)).await;
+    cache
 }
 
 /// Read `project.yaml` + `data/` to compute the rebuild
@@ -1440,9 +1453,19 @@ pub async fn run_replay(
             continue;
         }
         // Parse the line through the same DSL parser the
-        // interactive path uses. A failure here is structural
-        // (bad syntax) — report and stop without dispatching.
-        let command = match crate::dsl::parse_command(trimmed) {
+        // interactive path uses. The schema is re-snapshotted
+        // every line because earlier replayed commands
+        // (`create table`, `add column`, …) mutate it — so
+        // Phase D typed-slot rejections (wrong-count value
+        // lists, wrong-type column values) fire at replay
+        // parse time, matching the interactive path, rather
+        // than only at bind time. A failure here is structural
+        // (bad syntax / typed-slot reject) — report and stop
+        // without dispatching.
+        let schema = build_schema_cache(database).await;
+        let command = match crate::dsl::parser::parse_command_with_schema(
+            trimmed, &schema,
+        ) {
             Ok(c) => c,
             Err(e) => {
                 events.push(AppEvent::ReplayFailed {

tests/replay_command.rs

@@ -227,6 +227,55 @@ fn replay_aborts_on_first_parse_failure_and_reports_line() {
     );
 }
 
+#[test]
+fn replay_rejects_typed_slot_violation_at_parse_time() {
+    // Schema-aware replay (handoff-13 §2.1 fix): run_replay
+    // re-snapshots the schema per line and parses with
+    // parse_command_with_schema. So a wrong-type value in a
+    // value list is caught at *parse* time during replay —
+    // surfaced through the `replay.error_parse` wrapper ("parse
+    // error …") — exactly as the interactive path would, rather
+    // than only at bind time.
+    //
+    // `'not a number'` (a string) lands in the int `count`
+    // slot. The schemaless parser would accept it (a string is
+    // a value literal) and only bind-time would reject; the
+    // schema-aware parser rejects it at parse time. Asserting
+    // the error went through the parse wrapper proves the
+    // schema was threaded.
+    let data = tempdir();
+    let (project, db) = open_project_db(data.path());
+    write_script(
+        project.path(),
+        "typed.commands",
+        "create table T with pk id:int\n\
+         add column T: count (int)\n\
+         insert into T values (1, 'not a number')\n",
+    );
+
+    let events = rt().block_on(async {
+        run_replay(&db, project.path(), "typed.commands").await
+    });
+    let failed = assert_failed_at(&events, 3);
+    let AppEvent::ReplayFailed { error, .. } = failed else {
+        unreachable!()
+    };
+    assert!(
+        error.contains("parse error"),
+        "typed-slot violation should be caught at parse time, got: {error}",
+    );
+
+    // The earlier two lines stayed applied; the failing insert
+    // did not run.
+    let data_result = rt()
+        .block_on(async { db.query_data("T".to_string(), None).await })
+        .expect("query_data");
+    assert!(
+        data_result.rows.is_empty(),
+        "the rejected insert must not have dispatched",
+    );
+}
+
 #[test]
 fn replay_aborts_on_first_runtime_failure_and_reports_line() {
     let data = tempdir();