fix(state): repair legacy usage-profile migration stamp

codex-rs/state/src/migrations.rs

@@ -39,3 +39,25 @@ pub(crate) fn runtime_goals_migrator() -> Migrator {
 pub(crate) fn runtime_memories_migrator() -> Migrator {
     runtime_migrator(&MEMORIES_MIGRATOR)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::STATE_MIGRATOR;
+    use std::collections::BTreeMap;
+
+    #[test]
+    fn state_migration_versions_are_unique() {
+        let mut descriptions_by_version = BTreeMap::new();
+
+        for migration in STATE_MIGRATOR.iter() {
+            if let Some(first_description) =
+                descriptions_by_version.insert(migration.version, migration.description.as_ref())
+            {
+                panic!(
+                    "state migration version {} is duplicated by {:?} and {:?}",
+                    migration.version, first_description, migration.description
+                );
+            }
+        }
+    }
+}

codex-rs/state/src/runtime.rs

@@ -918,6 +918,9 @@ async fn open_sqlite(
     let pool = pool_result?;
     let started = Instant::now();
     let migrate_result = async {
+        if matches!(spec.kind, DbKind::State) {
+            repair_legacy_usage_profile_leases_migration_stamp(&pool, migrator).await?;
+        }
         if matches!(spec.kind, DbKind::Goals) {
             repair_legacy_goals_deferred_migration_stamp(&pool, migrator).await?;
         }
@@ -1033,10 +1036,20 @@ fn explain_migration_error(db_label: &str, err: sqlx::migrate::MigrateError) ->
 /// `legacy_0148_goals_deferred_checksum_matches_sqlx_checksum`.
 const LEGACY_0148_GOALS_DEFERRED_V5_CHECKSUM_HEX: &str = "3cfd6e6b956509f5cd9946b7d648daf1773baffa75d5ee6c472aa521987c2cf392dbdf39e6d9d8a8a64586f793331e6c";
 
+/// SHA-384 checksum of the state migration file
+/// `0068_usage_profile_leases.sql` exactly as shipped by PR #528 at
+/// `bae61b1418b9069145b82303c9f0c1d5929266f9`. Validated against sqlx's
+/// checksum algorithm and the archived migration bytes by
+/// `legacy_pr_528_usage_profile_leases_checksum_matches_sqlx_checksum`.
+const LEGACY_PR_528_USAGE_PROFILE_LEASES_V68_CHECKSUM_HEX: &str = "06b68a25316d8ff9f51895ed4aac0a66d13f72913c65ce059cc46398b20232fe175923cc178a9e4dfa164873a630624c";
+
 /// Version of the "thread goal deferred" migration in the current goals
 /// migration set (`goals_migrations/0008_thread_goal_deferred.sql`).
 const GOALS_DEFERRED_MIGRATION_VERSION: i64 = 8;
 
+/// Version of the renumbered usage-profile leases migration.
+const USAGE_PROFILE_LEASES_MIGRATION_VERSION: i64 = 69;
+
 fn decode_hex_checksum(hex: &str) -> anyhow::Result<Vec<u8>> {
     anyhow::ensure!(
         hex.len().is_multiple_of(2),
@@ -1111,6 +1124,57 @@ async fn repair_legacy_goals_deferred_migration_stamp(
     Ok(())
 }
 
+/// Re-stamp state databases initialized by the PR #528 artifact before
+/// validating the current state migration set.
+///
+/// PR #527 and #528 independently used version 68. The current migration set
+/// retains PR #527's thread-monitor migration at 68 and renumbers PR #528's
+/// usage-profile leases migration to 69. A database initialized by the PR #528
+/// artifact would otherwise fail SQLx checksum validation at 68 before the
+/// retained thread-monitor migration could apply.
+///
+/// The exact legacy checksum is the guard: a normal version-68 state row, a
+/// fresh database, and every unrelated migration record are left unchanged.
+/// The state-runtime startup lock is held while this runs, so the restamp
+/// cannot race another process's migrator.
+async fn repair_legacy_usage_profile_leases_migration_stamp(
+    pool: &SqlitePool,
+    migrator: &Migrator,
+) -> anyhow::Result<()> {
+    let has_migrations_table: Option<i64> = sqlx::query_scalar(
+        "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = '_sqlx_migrations'",
+    )
+    .fetch_optional(pool)
+    .await?;
+    if has_migrations_table.is_none() {
+        return Ok(());
+    }
+    let usage_profile_leases = migrator
+        .iter()
+        .find(|migration| migration.version == USAGE_PROFILE_LEASES_MIGRATION_VERSION)
+        .ok_or_else(|| {
+            anyhow::anyhow!(
+                "state migration set is missing version {USAGE_PROFILE_LEASES_MIGRATION_VERSION}"
+            )
+        })?;
+    let legacy_checksum = decode_hex_checksum(LEGACY_PR_528_USAGE_PROFILE_LEASES_V68_CHECKSUM_HEX)?;
+    let repaired = sqlx::query(
+        "UPDATE _sqlx_migrations SET version = ?, description = ?, checksum = ? WHERE version = 68 AND checksum = ?",
+    )
+    .bind(usage_profile_leases.version)
+    .bind(usage_profile_leases.description.as_ref())
+    .bind(usage_profile_leases.checksum.as_ref())
+    .bind(legacy_checksum)
+    .execute(pool)
+    .await?;
+    if repaired.rows_affected() > 0 {
+        warn!(
+            "re-stamped legacy PR #528 usage-profile leases migration version 68 as version {USAGE_PROFILE_LEASES_MIGRATION_VERSION}"
+        );
+    }
+    Ok(())
+}
+
 pub(super) async fn ensure_backfill_state_row_in_pool(
     pool: &sqlx::SqlitePool,
 ) -> anyhow::Result<()> {
@@ -1193,7 +1257,9 @@ pub async fn sqlite_integrity_check(path: &Path) -> anyhow::Result<Vec<String>>
 mod tests {
     use super::GOALS_DEFERRED_MIGRATION_VERSION;
     use super::LEGACY_0148_GOALS_DEFERRED_V5_CHECKSUM_HEX;
+    use super::LEGACY_PR_528_USAGE_PROFILE_LEASES_V68_CHECKSUM_HEX;
     use super::StateRuntime;
+    use super::USAGE_PROFILE_LEASES_MIGRATION_VERSION;
     use super::decode_hex_checksum;
     use super::explain_migration_error;
     use super::goals_db_path;
@@ -1335,6 +1401,62 @@ mod tests {
         }
     }
 
+    /// State migration `0068_usage_profile_leases.sql` exactly as shipped by
+    /// PR #528 at `bae61b1418b9069145b82303c9f0c1d5929266f9`. Kept outside
+    /// `migrations/` so the embedded migrator never picks it up.
+    const LEGACY_PR_528_USAGE_PROFILE_LEASES_V68_SQL: &str =
+        include_str!("runtime/fixtures/pr_528_0068_usage_profile_leases.sql");
+
+    fn legacy_pr_528_usage_profile_leases_migration() -> Migration {
+        Migration::new(
+            68,
+            Cow::Borrowed("usage profile leases"),
+            MigrationType::Simple,
+            LEGACY_PR_528_USAGE_PROFILE_LEASES_V68_SQL.into_sql_str(),
+            /*no_tx*/ false,
+        )
+    }
+
+    fn legacy_usage_profile_leases_migrator() -> Migrator {
+        let mut migrations = STATE_MIGRATOR
+            .iter()
+            .filter(|migration| migration.version < 68)
+            .cloned()
+            .collect::<Vec<_>>();
+        migrations.push(legacy_pr_528_usage_profile_leases_migration());
+        Migrator {
+            migrations: Cow::Owned(migrations),
+            ignore_missing: false,
+            locking: true,
+            no_tx: false,
+            table_name: STATE_MIGRATOR.table_name.clone(),
+            create_schemas: STATE_MIGRATOR.create_schemas.clone(),
+        }
+    }
+
+    #[test]
+    fn legacy_pr_528_usage_profile_leases_checksum_matches_sqlx_checksum() {
+        let expected = decode_hex_checksum(LEGACY_PR_528_USAGE_PROFILE_LEASES_V68_CHECKSUM_HEX)
+            .expect("legacy PR #528 checksum hex should decode");
+        assert_eq!(
+            expected.as_slice(),
+            legacy_pr_528_usage_profile_leases_migration()
+                .checksum
+                .as_ref(),
+            "hardcoded legacy checksum must match sqlx's checksum of the archived PR #528 migration bytes"
+        );
+        assert_eq!(
+            expected.as_slice(),
+            STATE_MIGRATOR
+                .iter()
+                .find(|migration| migration.version == USAGE_PROFILE_LEASES_MIGRATION_VERSION)
+                .expect("renumbered usage-profile leases migration")
+                .checksum
+                .as_ref(),
+            "renumbering the PR #528 migration to version 69 must preserve its bytes"
+        );
+    }
+
     async fn provider_credit_migration_stamps(pool: &SqlitePool) -> Vec<(i64, String, i64)> {
         sqlx::query_as(
             r#"
@@ -1614,6 +1736,118 @@ WHERE type = 'table' AND name = 'usage_profile_leases'
         let _ = tokio::fs::remove_dir_all(codex_home).await;
     }
 
+    #[tokio::test]
+    async fn state_db_stamped_by_pr_528_repairs_and_migrates() {
+        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 legacy usage-profile state db");
+
+        legacy_usage_profile_leases_migrator()
+            .run(&pool)
+            .await
+            .expect("apply PR #528 state migration set");
+        pool.close().await;
+
+        let strict_pool = open_db_pool(state_path.as_path()).await;
+        let strict_err = STATE_MIGRATOR
+            .run(&strict_pool)
+            .await
+            .expect_err("current migrator must reject the unrepaired PR #528 stamp");
+        assert!(matches!(strict_err, MigrateError::VersionMismatch(68)));
+        strict_pool.close().await;
+
+        let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string())
+            .await
+            .expect("state runtime should repair the PR #528 stamp");
+        assert_eq!(
+            vec![
+                (68, "thread monitor authorization".to_string(), 1),
+                (69, "usage profile leases".to_string(), 1),
+            ],
+            monitor_and_usage_profile_migration_stamps(runtime.pool.as_ref()).await
+        );
+        assert_eq!(
+            (true, true),
+            monitor_and_usage_profile_schema_presence(runtime.pool.as_ref()).await
+        );
+
+        drop(runtime);
+        let _ = tokio::fs::remove_dir_all(codex_home).await;
+    }
+
+    #[tokio::test]
+    async fn state_runtime_rejects_unknown_version_68_checksum_without_rewriting_it() {
+        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 mismatched-version state db");
+
+        migrator_through(&STATE_MIGRATOR, /*version*/ 67)
+            .run(&pool)
+            .await
+            .expect("apply state schema before version 68");
+        sqlx::query(
+            "INSERT INTO _sqlx_migrations (version, description, success, checksum, execution_time) VALUES (?, ?, ?, ?, ?)",
+        )
+        .bind(68_i64)
+        .bind("unknown version 68 migration")
+        .bind(true)
+        .bind(b"not-a-sha384-checksum".as_slice())
+        .bind(42_i64)
+        .execute(&pool)
+        .await
+        .expect("seed unknown version 68 migration stamp");
+        let before: (i64, String, bool, Vec<u8>, i64) = sqlx::query_as(
+            "SELECT version, description, success, checksum, execution_time FROM _sqlx_migrations WHERE version = 68",
+        )
+        .fetch_one(&pool)
+        .await
+        .expect("read seeded version 68 migration stamp");
+        pool.close().await;
+
+        let startup_err =
+            match StateRuntime::init(codex_home.clone(), "test-provider".to_string()).await {
+                Ok(_) => panic!("state runtime must reject an unknown version 68 checksum"),
+                Err(err) => err,
+            };
+        assert!(
+            startup_err.to_string().contains("state DB migration 68"),
+            "startup error should identify the mismatched state migration: {startup_err}"
+        );
+
+        let pool = open_db_pool(state_path.as_path()).await;
+        let after: (i64, String, bool, Vec<u8>, i64) = sqlx::query_as(
+            "SELECT version, description, success, checksum, execution_time FROM _sqlx_migrations WHERE version = 68",
+        )
+        .fetch_one(&pool)
+        .await
+        .expect("read rejected version 68 migration stamp");
+        assert_eq!(
+            before, after,
+            "startup must leave an unrecognized version 68 migration row untouched"
+        );
+        pool.close().await;
+
+        let _ = tokio::fs::remove_dir_all(codex_home).await;
+    }
+
     #[tokio::test]
     async fn thread_schedule_run_goal_migration_preserves_legacy_running_runs() {
         let codex_home = unique_temp_dir();

codex-rs/state/src/runtime/fixtures/pr_528_0068_usage_profile_leases.sql

@@ -0,0 +1,31 @@
+CREATE TABLE usage_profile_leases (
+    lease_id TEXT PRIMARY KEY CHECK(LENGTH(TRIM(lease_id)) > 0),
+    identity_sha256 TEXT NOT NULL CHECK(
+        LENGTH(identity_sha256) = 64
+        AND identity_sha256 = LOWER(identity_sha256)
+        AND identity_sha256 NOT GLOB '*[^0-9a-f]*'
+    ),
+    owner_id TEXT NOT NULL CHECK(LENGTH(TRIM(owner_id)) > 0),
+    profile_name TEXT NOT NULL CHECK(LENGTH(TRIM(profile_name)) > 0),
+    acquired_at_ms INTEGER NOT NULL,
+    heartbeat_at_ms INTEGER NOT NULL,
+    expires_at_ms INTEGER NOT NULL,
+    released_at_ms INTEGER,
+    release_reason TEXT CHECK(
+        release_reason IS NULL OR release_reason IN ('released', 'expired')
+    ),
+    CHECK(expires_at_ms > acquired_at_ms),
+    CHECK(
+        (released_at_ms IS NULL AND release_reason IS NULL)
+        OR
+        (released_at_ms IS NOT NULL AND release_reason IS NOT NULL)
+    )
+);
+
+CREATE UNIQUE INDEX idx_usage_profile_leases_active_identity
+    ON usage_profile_leases(identity_sha256)
+    WHERE released_at_ms IS NULL;
+
+CREATE INDEX idx_usage_profile_leases_active_expiry
+    ON usage_profile_leases(expires_at_ms)
+    WHERE released_at_ms IS NULL;