diff --git a/crates/tinyagents-session/src/README.md b/crates/tinyagents-session/src/README.md index c045196..ac485a3 100644 --- a/crates/tinyagents-session/src/README.md +++ b/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 | diff --git a/crates/tinyagents-session/src/migrations.rs b/crates/tinyagents-session/src/migrations.rs index adb93a8..b12ff96 100644 --- a/crates/tinyagents-session/src/migrations.rs +++ b/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 { + conn.execute_batch("BEGIN IMMEDIATE") + .storage_context("begin migration transaction")?; + let applied = (|| -> Result { + 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)] diff --git a/crates/tinyagents-session/src/test.rs b/crates/tinyagents-session/src/test.rs index f4f984f..7257139 100644 --- a/crates/tinyagents-session/src/test.rs +++ b/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 = stale + .prepare("PRAGMA table_info(session_messages)") + .expect("prepare columns") + .query_map([], |row| row.get(1)) + .expect("query columns") + .collect::>() + .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| {