Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
137 changes: 137 additions & 0 deletions codex-rs/state/src/runtime.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1374,6 +1374,72 @@ 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 monitor_and_usage_profile_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 usage profile migration stamps should query")
}

async fn monitor_and_usage_profile_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 usage_profile_leases_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 leases table should query");
(
monitor_authorization_column.is_some(),
usage_profile_leases_table.is_some(),
)
}

#[test]
fn provider_credit_migration_follows_branch_bootstrap_in_inventory() {
let migrations = STATE_MIGRATOR
Expand Down Expand Up @@ -1477,6 +1543,77 @@ WHERE type = 'table' AND name = 'workflow_provider_credit_reservations'
let _ = tokio::fs::remove_dir_all(codex_home).await;
}

#[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),
],
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_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-0068 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)],
monitor_and_usage_profile_migration_stamps(&pool).await
);
assert_eq!(
(true, false),
monitor_and_usage_profile_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),
],
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 thread_schedule_run_goal_migration_preserves_legacy_running_runs() {
let codex_home = unique_temp_dir();
Expand Down
Loading