fix(session): serialize concurrent schema migrations

crates/tinyagents-session/src/README.md

@@ -36,8 +36,11 @@ host chooses only where its workspace lives:
 Re-exported from the crate root (see `src/lib.rs`); the full surface stays
 reachable under `session::` and `session::run_ledger::`.
 
-- **Recording** — `record_session_start`, `record_message`, `record_tool_call`,
-  `record_session_end`
+- **Recording** — `record_session_start`, `record_message`,
+  `record_message_with_reasoning`, `record_tool_call`, `record_session_end`.
+  Use `record_message_with_reasoning` when an assistant response has hidden
+  reasoning that must remain separate from its visible content;
+  `record_message` remains the compatibility wrapper for content-only callers.
 - **Querying** — `get_session`, `list_sessions`, `search_sessions`,
   `list_messages`, `list_tool_calls`, `list_children`
 - **Recovery** — `mark_interrupted`
@@ -53,7 +56,7 @@ Six tables plus one FTS5 virtual table, created on demand and idempotently:
 | Table | Holds |
 | --- | --- |
 | `sessions` | one row per session; lineage via `parent_session_id` |
-| `session_messages` | per-message content, model, tokens, cost |
+| `session_messages` | per-message visible content, optional assistant reasoning, model, tokens, cost |
 | `session_tool_calls` | tool name, input, bounded output, status, duration |
 | `sessions_fts` | FTS5 index over session name, message content, tool name |
 | `agent_runs` / `workflow_runs` | background execution state |

crates/tinyagents-session/src/migrations.rs

@@ -14,7 +14,8 @@
 //! `schema_version` table records the highest index applied. On connection open
 //! every migration with an index greater than the recorded version runs, in
 //! order, each inside its own transaction, and the version is bumped after each
-//! one.
+//! one. The version is re-read after taking the write lock so two connections
+//! opening the same newly-upgraded workspace cannot repeat non-idempotent DDL.
 //!
 //! Two rules keep this sound:
 //!
@@ -284,36 +285,58 @@ pub(super) fn apply(conn: &Connection) -> Result<()> {
         if version <= current {
             continue;
         }
-        conn.execute_batch("BEGIN IMMEDIATE")
-            .storage_context("begin migration transaction")?;
-        let applied = (|| -> Result<()> {
-            conn.execute_batch(sql)
-                .storage_context(&format!("failed to apply session DB migration {version}"))?;
-            conn.execute(
-                "INSERT INTO schema_version (id, version) VALUES (1, ?1)
-                 ON CONFLICT(id) DO UPDATE SET version = excluded.version",
-                params![version],
+        apply_one(conn, version, sql)?;
+    }
+    Ok(())
+}
+
+/// Apply one migration after acquiring the database write lock.
+///
+/// The schema version is deliberately re-read *after* `BEGIN IMMEDIATE`:
+/// another connection may have completed this migration after our optimistic
+/// read in [`apply`] but before this connection acquired the lock.
+pub(super) fn apply_one(conn: &Connection, version: i64, sql: &str) -> Result<bool> {
+    conn.execute_batch("BEGIN IMMEDIATE")
+        .storage_context("begin migration transaction")?;
+    let applied = (|| -> Result<bool> {
+        let locked_current: i64 = conn
+            .query_row(
+                "SELECT COALESCE((SELECT version FROM schema_version WHERE id = 1), -1)",
+                [],
+                |row| row.get(0),
             )
-            .storage_context("failed to record schema version")?;
-            Ok(())
-        })();
-        match applied {
-            Ok(()) => {
-                conn.execute_batch("COMMIT")
-                    .storage_context("commit migration transaction")?;
+            .storage_context("re-read schema_version under migration lock")?;
+        if version <= locked_current {
+            return Ok(false);
+        }
+        conn.execute_batch(sql)
+            .storage_context(&format!("failed to apply session DB migration {version}"))?;
+        conn.execute(
+            "INSERT INTO schema_version (id, version) VALUES (1, ?1)
+             ON CONFLICT(id) DO UPDATE SET version = excluded.version",
+            params![version],
+        )
+        .storage_context("failed to record schema version")?;
+        Ok(true)
+    })();
+    match applied {
+        Ok(did_apply) => {
+            conn.execute_batch("COMMIT")
+                .storage_context("commit migration transaction")?;
+            if did_apply {
                 tinyagents_tracing::debug!("{LOG_PREFIX} applied migration {version}");
             }
-            Err(err) => {
-                if let Err(rollback) = conn.execute_batch("ROLLBACK") {
-                    tinyagents_tracing::warn!(
-                        "{LOG_PREFIX} rollback of migration {version} failed: {rollback} (original: {err})"
-                    );
-                }
-                return Err(err);
+            Ok(did_apply)
+        }
+        Err(err) => {
+            if let Err(rollback) = conn.execute_batch("ROLLBACK") {
+                tinyagents_tracing::warn!(
+                    "{LOG_PREFIX} rollback of migration {version} failed: {rollback} (original: {err})"
+                );
             }
+            Err(err)
         }
     }
-    Ok(())
 }
 
 #[cfg(test)]

crates/tinyagents-session/src/test.rs

@@ -4,7 +4,7 @@
 //! than per-file inline `mod tests` blocks. Sections mirror the source files.
 
 use super::context::StorageContext;
-use super::migrations::apply as init_schema;
+use super::migrations::{MIGRATIONS, apply as init_schema, apply_one};
 use super::ops::*;
 use super::store::with_memory_connection;
 use super::types::*;
@@ -727,6 +727,35 @@ fn schema_is_idempotent() {
     assert_eq!(count, 0);
 }
 
+#[test]
+fn stale_migration_plan_rechecks_version_after_lock() {
+    let dir = tempfile::tempdir().expect("tempdir");
+    let path = dir.path().join("sessions.db");
+    let first = Connection::open(&path).expect("open first");
+    let stale = Connection::open(&path).expect("open stale");
+
+    init_schema(&first).expect("first connection migrates");
+    let latest = MIGRATIONS.len() as i64 - 1;
+    let did_apply = apply_one(&stale, latest, MIGRATIONS[latest as usize])
+        .expect("stale plan is skipped rather than repeating ALTER TABLE");
+
+    assert!(!did_apply);
+    let columns: Vec<String> = stale
+        .prepare("PRAGMA table_info(session_messages)")
+        .expect("prepare columns")
+        .query_map([], |row| row.get(1))
+        .expect("query columns")
+        .collect::<rusqlite::Result<_>>()
+        .expect("collect columns");
+    assert_eq!(
+        columns
+            .iter()
+            .filter(|column| column.as_str() == "reasoning_content")
+            .count(),
+        1
+    );
+}
+
 #[test]
 fn wal_mode_is_set() {
     with_memory_connection(|conn| {