From cc711be083a5a61fe944242033ba3bfd02f02773 Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Mon, 10 Aug 2026 01:36:26 +0300 Subject: [PATCH] Fix state migration version collision Move usage profile leases to migration 0069 and cover fresh and monitor-0068-stamped StateRuntime initialization. Agent: quintilianus --- ...ases.sql => 0069_usage_profile_leases.sql} | 0 codex-rs/state/src/runtime.rs | 135 ++++++++++++++++++ 2 files changed, 135 insertions(+) rename codex-rs/state/migrations/{0068_usage_profile_leases.sql => 0069_usage_profile_leases.sql} (100%) diff --git a/codex-rs/state/migrations/0068_usage_profile_leases.sql b/codex-rs/state/migrations/0069_usage_profile_leases.sql similarity index 100% rename from codex-rs/state/migrations/0068_usage_profile_leases.sql rename to codex-rs/state/migrations/0069_usage_profile_leases.sql diff --git a/codex-rs/state/src/runtime.rs b/codex-rs/state/src/runtime.rs index 5a70139fc..4edfe4276 100644 --- a/codex-rs/state/src/runtime.rs +++ b/codex-rs/state/src/runtime.rs @@ -1374,6 +1374,70 @@ WHERE type = 'table' AND name = 'workflow_provider_credit_reservations' (bootstrap_column.is_some(), provider_table.is_some()) } + fn migrator_through_thread_monitor_authorization() -> Migrator { + Migrator { + migrations: Cow::Owned( + STATE_MIGRATOR + .migrations + .iter() + .filter(|migration| { + migration.version < 68 + || (migration.version == 68 + && migration.description.as_ref() == "thread monitor authorization") + }) + .cloned() + .collect(), + ), + ignore_missing: false, + locking: true, + no_tx: false, + table_name: STATE_MIGRATOR.table_name.clone(), + create_schemas: STATE_MIGRATOR.create_schemas.clone(), + } + } + + async fn usage_profile_lease_migration_stamps(pool: &SqlitePool) -> Vec<(i64, String, i64)> { + sqlx::query_as( + r#" +SELECT version, description, COUNT(*) +FROM _sqlx_migrations +WHERE version IN (68, 69) +GROUP BY version, description +ORDER BY version + "#, + ) + .fetch_all(pool) + .await + .expect("monitor and lease migration stamps should query") + } + + async fn usage_profile_lease_schema_presence(pool: &SqlitePool) -> (bool, bool) { + let monitor_authorization_column: Option = sqlx::query_scalar( + r#" +SELECT 1 +FROM pragma_table_info('thread_monitors') +WHERE name = 'authorization_json' + "#, + ) + .fetch_optional(pool) + .await + .expect("thread monitor authorization column should query"); + let lease_table: Option = sqlx::query_scalar( + r#" +SELECT 1 +FROM sqlite_master +WHERE type = 'table' AND name = 'usage_profile_leases' + "#, + ) + .fetch_optional(pool) + .await + .expect("usage profile lease table should query"); + ( + monitor_authorization_column.is_some(), + lease_table.is_some(), + ) + } + #[test] fn provider_credit_migration_follows_branch_bootstrap_in_inventory() { let migrations = STATE_MIGRATOR @@ -1390,6 +1454,77 @@ WHERE type = 'table' AND name = 'workflow_provider_credit_reservations' ); } + #[tokio::test] + async fn fresh_state_runtime_applies_monitor_0068_then_usage_profile_leases_0069() { + let codex_home = unique_temp_dir(); + let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string()) + .await + .expect("fresh state runtime should initialize"); + + assert_eq!( + vec![ + (68, "thread monitor authorization".to_string(), 1), + (69, "usage profile leases".to_string(), 1), + ], + usage_profile_lease_migration_stamps(runtime.pool.as_ref()).await + ); + assert_eq!( + (true, true), + usage_profile_lease_schema_presence(runtime.pool.as_ref()).await + ); + + drop(runtime); + let _ = tokio::fs::remove_dir_all(codex_home).await; + } + + #[tokio::test] + async fn state_runtime_upgrades_database_stamped_with_monitor_0068() { + let codex_home = unique_temp_dir(); + tokio::fs::create_dir_all(&codex_home) + .await + .expect("create codex home"); + let state_path = state_db_path(codex_home.as_path()); + let pool = SqlitePool::connect_with( + SqliteConnectOptions::new() + .filename(&state_path) + .create_if_missing(true), + ) + .await + .expect("open monitor-era state db"); + + migrator_through_thread_monitor_authorization() + .run(&pool) + .await + .expect("apply state schema through monitor authorization 0068"); + assert_eq!( + vec![(68, "thread monitor authorization".to_string(), 1)], + usage_profile_lease_migration_stamps(&pool).await + ); + assert_eq!( + (true, false), + usage_profile_lease_schema_presence(&pool).await + ); + pool.close().await; + + let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string()) + .await + .expect("state runtime should upgrade monitor-0068 database"); + assert_eq!( + vec![ + (68, "thread monitor authorization".to_string(), 1), + (69, "usage profile leases".to_string(), 1), + ], + usage_profile_lease_migration_stamps(runtime.pool.as_ref()).await + ); + assert_eq!( + (true, true), + usage_profile_lease_schema_presence(runtime.pool.as_ref()).await + ); + + drop(runtime); + let _ = tokio::fs::remove_dir_all(codex_home).await; + } + #[tokio::test] async fn fresh_state_store_applies_0066_then_0067_exactly_once() { let codex_home = unique_temp_dir();