From 872a5208807c4273fbadf0eae7f977bf7c0af3a0 Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Mon, 10 Aug 2026 02:26:09 +0300 Subject: [PATCH 1/3] test(state): reject duplicate migration versions Agent: iapp-factory-coordinator --- codex-rs/state/src/migrations.rs | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/codex-rs/state/src/migrations.rs b/codex-rs/state/src/migrations.rs index 9963b21ad1..17a83be78a 100644 --- a/codex-rs/state/src/migrations.rs +++ b/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 + ); + } + } + } +} From a8e3df1766119d7656d723806244151fc6486bbd Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Mon, 10 Aug 2026 02:39:12 +0300 Subject: [PATCH 2/3] fix(state): repair duplicate migration 0068 Preserve the thread-monitor migration at version 68, restamp only the legacy PR #528 usage-profile lease checksum as version 69, and reject future duplicate state migration versions. Agent: iapp-factory-coordinator --- ...ases.sql => 0069_usage_profile_leases.sql} | 0 codex-rs/state/src/migrations.rs | 4 +- codex-rs/state/src/runtime.rs | 185 ++++++++++++++++++ 3 files changed, 187 insertions(+), 2 deletions(-) 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/migrations.rs b/codex-rs/state/src/migrations.rs index 17a83be78a..7d4507b2da 100644 --- a/codex-rs/state/src/migrations.rs +++ b/codex-rs/state/src/migrations.rs @@ -50,8 +50,8 @@ mod tests { 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()) + if let Some(first_description) = + descriptions_by_version.insert(migration.version, migration.description.as_ref()) { panic!( "state migration version {} is duplicated by {:?} and {:?}", diff --git a/codex-rs/state/src/runtime.rs b/codex-rs/state/src/runtime.rs index 5a70139fc7..5d2e81133c 100644 --- a/codex-rs/state/src/runtime.rs +++ b/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?; } @@ -1037,6 +1040,9 @@ const LEGACY_0148_GOALS_DEFERRED_V5_CHECKSUM_HEX: &str = "3cfd6e6b956509f5cd9946 /// 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> { anyhow::ensure!( hex.len().is_multiple_of(2), @@ -1111,6 +1117,56 @@ 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 = 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 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(usage_profile_leases.checksum.as_ref()) + .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<()> { @@ -1194,6 +1250,7 @@ mod tests { use super::GOALS_DEFERRED_MIGRATION_VERSION; use super::LEGACY_0148_GOALS_DEFERRED_V5_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 +1392,29 @@ mod tests { } } + fn legacy_usage_profile_leases_migrator() -> Migrator { + let mut migrations = STATE_MIGRATOR + .iter() + .filter(|migration| migration.version < 68) + .cloned() + .collect::>(); + let mut legacy_usage_profile_leases = STATE_MIGRATOR + .iter() + .find(|migration| migration.version == USAGE_PROFILE_LEASES_MIGRATION_VERSION) + .expect("renumbered usage-profile leases migration") + .clone(); + legacy_usage_profile_leases.version = 68; + migrations.push(legacy_usage_profile_leases); + 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(), + } + } + async fn provider_credit_migration_stamps(pool: &SqlitePool) -> Vec<(i64, String, i64)> { sqlx::query_as( r#" @@ -1477,6 +1557,111 @@ WHERE type = 'table' AND name = 'workflow_provider_credit_reservations' 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 state db"); + legacy_usage_profile_leases_migrator() + .run(&pool) + .await + .expect("apply PR #528 state migration set"); + let stamped: Vec<(i64, String, Vec)> = sqlx::query_as( + "SELECT version, description, checksum FROM _sqlx_migrations ORDER BY version", + ) + .fetch_all(&pool) + .await + .expect("legacy stamps should query"); + assert_eq!(68, stamped.last().expect("legacy version 68 stamp").0); + assert_eq!( + "usage profile leases", + stamped.last().expect("legacy version 68 stamp").1 + ); + 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"); + runtime.pool.close().await; + runtime.logs_pool.close().await; + runtime.goals_pool.close().await; + runtime.memories_pool.close().await; + drop(runtime); + + let query_pool = open_db_pool(state_path.as_path()).await; + let stamped: Vec<(i64, String, Vec)> = sqlx::query_as( + "SELECT version, description, checksum FROM _sqlx_migrations ORDER BY version", + ) + .fetch_all(&query_pool) + .await + .expect("repaired stamps should query"); + assert_eq!( + STATE_MIGRATOR + .iter() + .map(|migration| migration.version) + .collect::>(), + stamped + .iter() + .map(|(version, _, _)| *version) + .collect::>() + ); + let usage_profile_leases = STATE_MIGRATOR + .iter() + .find(|migration| migration.version == USAGE_PROFILE_LEASES_MIGRATION_VERSION) + .expect("renumbered usage-profile leases migration"); + let usage_profile_leases_stamp = stamped + .iter() + .find(|(version, _, _)| *version == USAGE_PROFILE_LEASES_MIGRATION_VERSION) + .expect("version 69 stamp"); + assert_eq!( + ( + "usage profile leases", + usage_profile_leases.checksum.as_ref() + ), + ( + usage_profile_leases_stamp.1.as_str(), + usage_profile_leases_stamp.2.as_slice() + ) + ); + let authorization_column: Option = sqlx::query_scalar( + "SELECT name FROM pragma_table_info('thread_monitors') WHERE name = 'authorization_json'", + ) + .fetch_optional(&query_pool) + .await + .expect("thread monitor authorization column should query"); + assert_eq!(Some("authorization_json".to_string()), authorization_column); + let usage_profile_leases_table: Option = sqlx::query_scalar( + "SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'usage_profile_leases'", + ) + .fetch_optional(&query_pool) + .await + .expect("usage-profile leases table should query"); + assert_eq!( + Some("usage_profile_leases".to_string()), + usage_profile_leases_table + ); + query_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(); From 4485ffb4c917ff3d11f5581a45bb113a84e65b09 Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Mon, 10 Aug 2026 03:21:14 +0300 Subject: [PATCH 3/3] test(state): pin migration repair guards Archive the exact PR #528 migration bytes, pin their checksum independently, and prove unknown version-68 stamps fail closed without mutation. Agent: iapp-factory-coordinator --- codex-rs/state/src/runtime.rs | 122 ++++++++++++++++-- .../pr_528_0068_usage_profile_leases.sql | 31 +++++ 2 files changed, 145 insertions(+), 8 deletions(-) create mode 100644 codex-rs/state/src/runtime/fixtures/pr_528_0068_usage_profile_leases.sql diff --git a/codex-rs/state/src/runtime.rs b/codex-rs/state/src/runtime.rs index dfd40cdb7b..8455a0574d 100644 --- a/codex-rs/state/src/runtime.rs +++ b/codex-rs/state/src/runtime.rs @@ -1036,6 +1036,13 @@ 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; @@ -1150,13 +1157,14 @@ async fn repair_legacy_usage_profile_leases_migration_stamp( "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(usage_profile_leases.checksum.as_ref()) + .bind(legacy_checksum) .execute(pool) .await?; if repaired.rows_affected() > 0 { @@ -1249,6 +1257,7 @@ pub async fn sqlite_integrity_check(path: &Path) -> anyhow::Result> 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; @@ -1392,19 +1401,29 @@ 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::>(); - let mut legacy_usage_profile_leases = STATE_MIGRATOR - .iter() - .find(|migration| migration.version == USAGE_PROFILE_LEASES_MIGRATION_VERSION) - .expect("renumbered usage-profile leases migration") - .clone(); - legacy_usage_profile_leases.version = 68; - migrations.push(legacy_usage_profile_leases); + migrations.push(legacy_pr_528_usage_profile_leases_migration()); Migrator { migrations: Cow::Owned(migrations), ignore_missing: false, @@ -1415,6 +1434,29 @@ mod tests { } } + #[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#" @@ -1742,6 +1784,70 @@ WHERE type = 'table' AND name = 'usage_profile_leases' 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, 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, 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(); diff --git a/codex-rs/state/src/runtime/fixtures/pr_528_0068_usage_profile_leases.sql b/codex-rs/state/src/runtime/fixtures/pr_528_0068_usage_profile_leases.sql new file mode 100644 index 0000000000..9911c8fa0a --- /dev/null +++ b/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;