Files
codex/codex-rs/core/tests/suite/personality_migration.rs
T
Rasmus RygaardandGitHub 7e310bc7f3 Inject state DB, agent graph store (#20689)
## Why

We want the agent graph store to be passed down the stack as a real
dependency, the same way we already treat the thread store.

This will let us inject the agent graph store as a real dependency and
support implementations other than the local SQLite-backed one. Right
now most code instantiates a state DB and an agent graph store
just-in-time. Ideally, we would not depend on the state DB directly but
only read through the higher-level interfaces.

This change makes the dependency boundaries explicit and moves state DB
initialization to process bootstrap instead of hiding it inside local
store implementations.

## What changed

- `ThreadManager` now requires a `StateDbHandle` and an
`AgentGraphStore` at construction time instead of treating them as
optional internals.
- The local store constructors no longer lazily initialize SQLite.
Callers now initialize the state DB once per process and use that shared
handle to build:
  - `LocalThreadStore`
  - `LocalAgentGraphStore`
- App bootstraps (`app-server`, `mcp-server`, `prompt_debug`, and the
thread-manager sample) now initialize the state DB up front and inject
the resulting handle down the stack.
- `app-server` now consistently uses its process-scoped state DB handle
instead of reopening SQLite or trying to recover it from loaded threads.
- Device-key storage now reuses the shared state DB handle instead of
maintaining its own lazy opener.
- The thread archive / descendant traversal paths now use the injected
`AgentGraphStore` instead of reaching through local
thread-store-specific state.

## Verification

- `cargo check -p codex-core -p codex-thread-store -p codex-app-server
-p codex-mcp-server -p codex-thread-manager-sample --tests`
- `cargo test -p codex-thread-store`
- `cargo test -p codex-core
thread_manager_accepts_separate_agent_graph_store_and_thread_store --
--nocapture`
- `cargo test -p codex-app-server
thread_archive_archives_spawned_descendants -- --nocapture`
2026-05-05 21:45:29 +00:00

362 lines
12 KiB
Rust

use codex_config::config_toml::ConfigToml;
use codex_core::ARCHIVED_SESSIONS_SUBDIR;
use codex_core::SESSIONS_SUBDIR;
use codex_core::personality_migration::PERSONALITY_MIGRATION_FILENAME;
use codex_core::personality_migration::PersonalityMigrationStatus;
use codex_core::personality_migration::maybe_migrate_personality;
use codex_protocol::ThreadId;
use codex_protocol::config_types::Personality;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::RolloutLine;
use codex_protocol::protocol::SessionMeta;
use codex_protocol::protocol::SessionMetaLine;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::UserMessageEvent;
use codex_rollout::RolloutConfig;
use codex_rollout::state_db::StateDbHandle;
use pretty_assertions::assert_eq;
use std::io;
use std::path::Path;
use tempfile::TempDir;
use tokio::io::AsyncWriteExt;
const TEST_TIMESTAMP: &str = "2025-01-01T00-00-00";
async fn read_config_toml(codex_home: &Path) -> io::Result<ConfigToml> {
let contents = tokio::fs::read_to_string(codex_home.join("config.toml")).await?;
toml::from_str(&contents).map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err))
}
async fn state_db_for_test(codex_home: &Path) -> io::Result<StateDbHandle> {
let config = RolloutConfig {
codex_home: codex_home.to_path_buf(),
sqlite_home: codex_home.to_path_buf(),
cwd: codex_home.to_path_buf(),
model_provider_id: "openai".to_string(),
generate_memories: false,
};
codex_rollout::state_db::try_init(&config)
.await
.map_err(io::Error::other)
}
async fn run_migration(
codex_home: &Path,
config_toml: &ConfigToml,
) -> io::Result<PersonalityMigrationStatus> {
let state_db = state_db_for_test(codex_home).await?;
maybe_migrate_personality(codex_home, config_toml, state_db).await
}
async fn write_session_with_user_event(codex_home: &Path) -> io::Result<()> {
let thread_id = ThreadId::new();
let dir = codex_home
.join(SESSIONS_SUBDIR)
.join("2025")
.join("01")
.join("01");
write_rollout_with_user_event(&dir, thread_id).await
}
async fn write_archived_session_with_user_event(codex_home: &Path) -> io::Result<()> {
let thread_id = ThreadId::new();
let dir = codex_home.join(ARCHIVED_SESSIONS_SUBDIR);
write_rollout_with_user_event(&dir, thread_id).await
}
async fn write_session_with_meta_only(codex_home: &Path) -> io::Result<()> {
let thread_id = ThreadId::new();
let dir = codex_home
.join(SESSIONS_SUBDIR)
.join("2025")
.join("01")
.join("01");
write_rollout_with_meta_only(&dir, thread_id).await
}
async fn write_rollout_with_user_event(dir: &Path, thread_id: ThreadId) -> io::Result<()> {
tokio::fs::create_dir_all(&dir).await?;
let file_path = dir.join(format!("rollout-{TEST_TIMESTAMP}-{thread_id}.jsonl"));
let mut file = tokio::fs::File::create(&file_path).await?;
let session_meta = SessionMetaLine {
meta: SessionMeta {
id: thread_id,
forked_from_id: None,
timestamp: TEST_TIMESTAMP.to_string(),
cwd: std::path::PathBuf::from("."),
originator: "test_originator".to_string(),
cli_version: "test_version".to_string(),
source: SessionSource::Cli,
agent_path: None,
agent_nickname: None,
agent_role: None,
model_provider: None,
base_instructions: None,
dynamic_tools: None,
memory_mode: None,
},
git: None,
};
let meta_line = RolloutLine {
timestamp: TEST_TIMESTAMP.to_string(),
item: RolloutItem::SessionMeta(session_meta),
};
let user_event = RolloutLine {
timestamp: TEST_TIMESTAMP.to_string(),
item: RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
message: "hello".to_string(),
images: None,
local_images: Vec::new(),
text_elements: Vec::new(),
})),
};
let meta_json = serde_json::to_string(&meta_line)?;
file.write_all(format!("{meta_json}\n").as_bytes()).await?;
let user_json = serde_json::to_string(&user_event)?;
file.write_all(format!("{user_json}\n").as_bytes()).await?;
Ok(())
}
async fn write_rollout_with_meta_only(dir: &Path, thread_id: ThreadId) -> io::Result<()> {
tokio::fs::create_dir_all(&dir).await?;
let file_path = dir.join(format!("rollout-{TEST_TIMESTAMP}-{thread_id}.jsonl"));
let mut file = tokio::fs::File::create(&file_path).await?;
let session_meta = SessionMetaLine {
meta: SessionMeta {
id: thread_id,
forked_from_id: None,
timestamp: TEST_TIMESTAMP.to_string(),
cwd: std::path::PathBuf::from("."),
originator: "test_originator".to_string(),
cli_version: "test_version".to_string(),
source: SessionSource::Cli,
agent_path: None,
agent_nickname: None,
agent_role: None,
model_provider: None,
base_instructions: None,
dynamic_tools: None,
memory_mode: None,
},
git: None,
};
let meta_line = RolloutLine {
timestamp: TEST_TIMESTAMP.to_string(),
item: RolloutItem::SessionMeta(session_meta),
};
let meta_json = serde_json::to_string(&meta_line)?;
file.write_all(format!("{meta_json}\n").as_bytes()).await?;
Ok(())
}
fn parse_config_toml(contents: &str) -> io::Result<ConfigToml> {
toml::from_str(contents).map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err))
}
#[tokio::test]
async fn migration_marker_exists_no_sessions_no_change() -> io::Result<()> {
let temp = TempDir::new()?;
let marker_path = temp.path().join(PERSONALITY_MIGRATION_FILENAME);
tokio::fs::write(&marker_path, "v1\n").await?;
let status = run_migration(temp.path(), &ConfigToml::default()).await?;
assert_eq!(status, PersonalityMigrationStatus::SkippedMarker);
assert_eq!(
tokio::fs::try_exists(temp.path().join("config.toml")).await?,
false
);
Ok(())
}
#[tokio::test]
async fn no_marker_no_sessions_no_change() -> io::Result<()> {
let temp = TempDir::new()?;
let status = run_migration(temp.path(), &ConfigToml::default()).await?;
assert_eq!(status, PersonalityMigrationStatus::SkippedNoSessions);
assert_eq!(
tokio::fs::try_exists(temp.path().join(PERSONALITY_MIGRATION_FILENAME)).await?,
true
);
assert_eq!(
tokio::fs::try_exists(temp.path().join("config.toml")).await?,
false
);
Ok(())
}
#[tokio::test]
async fn no_marker_sessions_sets_personality() -> io::Result<()> {
let temp = TempDir::new()?;
write_session_with_user_event(temp.path()).await?;
let status = run_migration(temp.path(), &ConfigToml::default()).await?;
assert_eq!(status, PersonalityMigrationStatus::Applied);
assert_eq!(
tokio::fs::try_exists(temp.path().join(PERSONALITY_MIGRATION_FILENAME)).await?,
true
);
let persisted = read_config_toml(temp.path()).await?;
assert_eq!(persisted.personality, Some(Personality::Pragmatic));
Ok(())
}
#[tokio::test]
async fn no_marker_sessions_preserves_existing_config_fields() -> io::Result<()> {
let temp = TempDir::new()?;
write_session_with_user_event(temp.path()).await?;
tokio::fs::write(temp.path().join("config.toml"), "model = \"gpt-5.4\"\n").await?;
let config_toml = read_config_toml(temp.path()).await?;
let status = run_migration(temp.path(), &config_toml).await?;
assert_eq!(status, PersonalityMigrationStatus::Applied);
let persisted = read_config_toml(temp.path()).await?;
assert_eq!(persisted.model, Some("gpt-5.4".to_string()));
assert_eq!(persisted.personality, Some(Personality::Pragmatic));
Ok(())
}
#[tokio::test]
async fn no_marker_meta_only_rollout_is_treated_as_no_sessions() -> io::Result<()> {
let temp = TempDir::new()?;
write_session_with_meta_only(temp.path()).await?;
let status = run_migration(temp.path(), &ConfigToml::default()).await?;
assert_eq!(status, PersonalityMigrationStatus::SkippedNoSessions);
assert_eq!(
tokio::fs::try_exists(temp.path().join(PERSONALITY_MIGRATION_FILENAME)).await?,
true
);
assert_eq!(
tokio::fs::try_exists(temp.path().join("config.toml")).await?,
false
);
Ok(())
}
#[tokio::test]
async fn no_marker_explicit_global_personality_skips_migration() -> io::Result<()> {
let temp = TempDir::new()?;
write_session_with_user_event(temp.path()).await?;
let config_toml = parse_config_toml("personality = \"friendly\"\n")?;
let status = run_migration(temp.path(), &config_toml).await?;
assert_eq!(
status,
PersonalityMigrationStatus::SkippedExplicitPersonality
);
assert_eq!(
tokio::fs::try_exists(temp.path().join(PERSONALITY_MIGRATION_FILENAME)).await?,
true
);
assert_eq!(
tokio::fs::try_exists(temp.path().join("config.toml")).await?,
false
);
Ok(())
}
#[tokio::test]
async fn no_marker_profile_personality_skips_migration() -> io::Result<()> {
let temp = TempDir::new()?;
write_session_with_user_event(temp.path()).await?;
let config_toml = parse_config_toml(
r#"
profile = "work"
[profiles.work]
personality = "friendly"
"#,
)?;
let status = run_migration(temp.path(), &config_toml).await?;
assert_eq!(
status,
PersonalityMigrationStatus::SkippedExplicitPersonality
);
assert_eq!(
tokio::fs::try_exists(temp.path().join(PERSONALITY_MIGRATION_FILENAME)).await?,
true
);
assert_eq!(
tokio::fs::try_exists(temp.path().join("config.toml")).await?,
false
);
Ok(())
}
#[tokio::test]
async fn marker_short_circuits_invalid_profile_resolution() -> io::Result<()> {
let temp = TempDir::new()?;
tokio::fs::write(temp.path().join(PERSONALITY_MIGRATION_FILENAME), "v1\n").await?;
let config_toml = parse_config_toml("profile = \"missing\"\n")?;
let status = run_migration(temp.path(), &config_toml).await?;
assert_eq!(status, PersonalityMigrationStatus::SkippedMarker);
Ok(())
}
#[tokio::test]
async fn invalid_selected_profile_returns_error_and_does_not_write_marker() -> io::Result<()> {
let temp = TempDir::new()?;
let config_toml = parse_config_toml("profile = \"missing\"\n")?;
let err = run_migration(temp.path(), &config_toml)
.await
.expect_err("missing profile should fail");
assert_eq!(err.kind(), io::ErrorKind::InvalidData);
assert_eq!(
tokio::fs::try_exists(temp.path().join(PERSONALITY_MIGRATION_FILENAME)).await?,
false
);
Ok(())
}
#[tokio::test]
async fn applied_migration_is_idempotent_on_second_run() -> io::Result<()> {
let temp = TempDir::new()?;
write_session_with_user_event(temp.path()).await?;
let first_status = run_migration(temp.path(), &ConfigToml::default()).await?;
let second_status = run_migration(temp.path(), &ConfigToml::default()).await?;
assert_eq!(first_status, PersonalityMigrationStatus::Applied);
assert_eq!(second_status, PersonalityMigrationStatus::SkippedMarker);
let persisted = read_config_toml(temp.path()).await?;
assert_eq!(persisted.personality, Some(Personality::Pragmatic));
Ok(())
}
#[tokio::test]
async fn no_marker_archived_sessions_sets_personality() -> io::Result<()> {
let temp = TempDir::new()?;
write_archived_session_with_user_event(temp.path()).await?;
let status = run_migration(temp.path(), &ConfigToml::default()).await?;
assert_eq!(status, PersonalityMigrationStatus::Applied);
assert_eq!(
tokio::fs::try_exists(temp.path().join(PERSONALITY_MIGRATION_FILENAME)).await?,
true
);
let persisted = read_config_toml(temp.path()).await?;
assert_eq!(persisted.personality, Some(Personality::Pragmatic));
Ok(())
}