Fix state migration version collision

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<i64> = 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<i64> = 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();