use std::time::Duration; use anyhow::Result; use app_test_support::ChatGptAuthFixture; use app_test_support::TestAppServer; use app_test_support::create_mock_responses_server_repeating_assistant; use app_test_support::start_analytics_events_server; use app_test_support::to_response; use app_test_support::write_chatgpt_auth; use app_test_support::write_mock_responses_config_toml; use codex_app_server_protocol::ExternalAgentConfigDetectResponse; use codex_app_server_protocol::ExternalAgentConfigImportCompletedNotification; use codex_app_server_protocol::ExternalAgentConfigImportHistoriesReadResponse; use codex_app_server_protocol::ExternalAgentConfigImportProgressNotification; use codex_app_server_protocol::ExternalAgentConfigImportResponse; use codex_app_server_protocol::ExternalAgentConfigMigrationItemType; use codex_app_server_protocol::JSONRPCResponse; use codex_app_server_protocol::PluginListParams; use codex_app_server_protocol::PluginListResponse; use codex_app_server_protocol::RequestId; use codex_app_server_protocol::ThreadItem; use codex_app_server_protocol::ThreadListParams; use codex_app_server_protocol::ThreadListResponse; use codex_app_server_protocol::ThreadReadParams; use codex_app_server_protocol::ThreadReadResponse; use codex_app_server_protocol::ThreadResumeParams; use codex_app_server_protocol::ThreadResumeResponse; use codex_app_server_protocol::TurnStartParams; use codex_app_server_protocol::UserInput; use codex_config::types::AuthCredentialsStoreMode; use core_test_support::responses; use pretty_assertions::assert_eq; use std::collections::BTreeMap; use std::path::Path; use std::path::PathBuf; use tempfile::TempDir; #[cfg(unix)] use tokio::io::AsyncWriteExt; use tokio::time::timeout; use super::analytics::wait_for_analytics_event; const DEFAULT_TIMEOUT: Duration = Duration::from_secs(60); fn external_agent_home(codex_home: &Path) -> PathBuf { codex_home.join(concat!(".", "cl", "aude")) } fn assert_import_response(response: ExternalAgentConfigImportResponse) -> String { assert!(!response.import_id.is_empty()); response.import_id } #[tokio::test] async fn external_agent_config_import_sends_completion_notification_for_sync_only_import() -> Result<()> { let codex_home = TempDir::new()?; let sqlite_home = TempDir::new()?; let home_dir = codex_home.path().display().to_string(); let sqlite_home_dir = sqlite_home.path().display().to_string(); let mut mcp = TestAppServer::new_with_env( codex_home.path(), &[ ("HOME", Some(home_dir.as_str())), ("CODEX_SQLITE_HOME", Some(sqlite_home_dir.as_str())), ], ) .await?; timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??; let request_id = mcp .send_raw_request( "externalAgentConfig/import", Some(serde_json::json!({ "migrationItems": [{ "itemType": "CONFIG", "description": "Import config", "cwd": null }] })), ) .await?; let response: JSONRPCResponse = timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; let response: ExternalAgentConfigImportResponse = to_response(response)?; let import_id = assert_import_response(response); let progress = timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_notification_message("externalAgentConfig/import/progress"), ) .await??; assert_eq!(progress.method, "externalAgentConfig/import/progress"); let progress: ExternalAgentConfigImportProgressNotification = serde_json::from_value(progress.params.expect("progress params"))?; assert_eq!(progress.import_id, import_id); assert_eq!(progress.item_type_results.len(), 1); assert_eq!( progress.item_type_results[0].item_type, ExternalAgentConfigMigrationItemType::Config ); let notification = timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_notification_message("externalAgentConfig/import/completed"), ) .await??; assert_eq!(notification.method, "externalAgentConfig/import/completed"); let completed: ExternalAgentConfigImportCompletedNotification = serde_json::from_value(notification.params.expect("completed params"))?; assert_eq!(completed.import_id, import_id); let state_db = codex_state::StateRuntime::init(sqlite_home.path().to_path_buf(), "mock_provider".into()) .await?; let details_record = state_db .external_agent_config_import_details_record(&import_id) .await? .expect("completed import details should be recorded by import id"); let expected_successes = completed .item_type_results .iter() .flat_map(|type_result| type_result.successes.iter()) .collect::>(); let expected_failures = completed .item_type_results .iter() .flat_map(|type_result| type_result.failures.iter()) .collect::>(); assert_eq!( serde_json::to_value(&details_record.successes)?, serde_json::to_value(&expected_successes)? ); assert_eq!( serde_json::to_value(&details_record.failures)?, serde_json::to_value(&expected_failures)? ); let request_id = mcp .send_raw_request( "externalAgentConfig/import/readHistories", /*params*/ None, ) .await?; let response: JSONRPCResponse = timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; let response: ExternalAgentConfigImportHistoriesReadResponse = to_response(response)?; let entry = response .data .iter() .find(|entry| entry.import_id == import_id) .expect("import history entry should be available"); assert!(entry.completed_at_ms > 0); assert_eq!( serde_json::to_value(&entry.successes)?, serde_json::to_value(&expected_successes)? ); assert_eq!( serde_json::to_value(&entry.failures)?, serde_json::to_value(&expected_failures)? ); Ok(()) } #[tokio::test] async fn external_agent_config_import_reports_failed_sync_import_in_completion() -> Result<()> { let codex_home = TempDir::new()?; write_chatgpt_auth( codex_home.path(), ChatGptAuthFixture::new("chatgpt-token") .account_id("account-123") .chatgpt_user_id("user-123") .chatgpt_account_id("account-123"), AuthCredentialsStoreMode::File, )?; let source_home = external_agent_home(codex_home.path()); std::fs::create_dir_all(&source_home)?; std::fs::write( source_home.join("settings.json"), r#"{"env":{"FOO":"bar"}}"#, )?; std::fs::write(codex_home.path().join("config.toml"), "invalid = [")?; let home_dir = codex_home.path().display().to_string(); let analytics_capture_file = codex_home.path().join("analytics-events.jsonl"); let analytics_capture_file = analytics_capture_file.display().to_string(); let mut mcp = TestAppServer::new_with_env( codex_home.path(), &[ ("HOME", Some(home_dir.as_str())), ( "CODEX_ANALYTICS_EVENTS_CAPTURE_FILE", Some(analytics_capture_file.as_str()), ), ], ) .await?; timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??; let request_id = mcp .send_raw_request( "externalAgentConfig/import", Some(serde_json::json!({ "source": "test_import", "migrationItems": [ { "itemType": "CONFIG", "description": "Import config", "cwd": null }, { "itemType": "COMMANDS", "description": "Import commands", "cwd": null } ] })), ) .await?; let response: JSONRPCResponse = timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; let response: ExternalAgentConfigImportResponse = to_response(response)?; let import_id = assert_import_response(response); let notification = timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_notification_message("externalAgentConfig/import/completed"), ) .await??; let completed: ExternalAgentConfigImportCompletedNotification = serde_json::from_value(notification.params.expect("completed params"))?; assert_eq!(completed.import_id, import_id); let config_result = completed .item_type_results .iter() .find(|result| result.item_type == ExternalAgentConfigMigrationItemType::Config) .expect("config result"); assert!(config_result.successes.is_empty()); assert_eq!(config_result.failures.len(), 1); let config_failure = &config_result.failures[0]; assert_eq!( config_failure.error_type.as_deref(), Some("invalid_existing_config") ); assert_eq!(config_failure.failure_stage, "import_request_failed"); assert!( config_failure .message .contains("invalid existing config.toml"), "unexpected failure: {config_failure:?}" ); let commands_result = completed .item_type_results .iter() .find(|result| result.item_type == ExternalAgentConfigMigrationItemType::Commands) .expect("commands result"); assert!(commands_result.successes.is_empty()); assert!(commands_result.failures.is_empty()); let events = timeout(DEFAULT_TIMEOUT, async { loop { let contents = match std::fs::read_to_string(&analytics_capture_file) { Ok(contents) => contents, Err(err) if err.kind() == std::io::ErrorKind::NotFound => { tokio::time::sleep(Duration::from_millis(25)).await; continue; } Err(err) => return Err(err.into()), }; let mut captured_events = Vec::new(); for line in contents.lines() { let payload: serde_json::Value = serde_json::from_str(line)?; let Some(events) = payload["events"].as_array() else { continue; }; captured_events.extend(events.iter().cloned()); } if captured_events.iter().any(|event| { event["event_type"] == "codex_onboarding_external_agent_import_complete" && event["event_params"]["type"] == "COMMANDS" }) { return Ok::, anyhow::Error>(captured_events); } tokio::time::sleep(Duration::from_millis(25)).await; } }) .await??; let event = events .iter() .find(|event| { event["event_type"] == "codex_onboarding_external_agent_import_failure" && event["event_params"]["type"] == "CONFIG" }) .expect("config failure analytics event"); let event_params = &event["event_params"]; assert_eq!(event_params["import_id"], import_id); assert_eq!(event_params["source"], "test_import"); assert_eq!(event_params["type"], "CONFIG"); assert_eq!(event_params["failure_stage"], "import_request_failed"); assert_eq!(event_params["error_type"], "invalid_existing_config"); assert!(event_params.get("raw_errors").is_none()); assert!(event_params.get("message").is_none()); assert!(!events.iter().any(|event| { event["event_type"] == "codex_onboarding_external_agent_import_failure" && event["event_params"]["type"] == "COMMANDS" })); Ok(()) } #[tokio::test] async fn external_agent_config_import_completed_tracks_analytics_event() -> Result<()> { let analytics_server = start_analytics_events_server().await?; let codex_home = TempDir::new()?; write_analytics_config(codex_home.path(), &analytics_server.uri())?; write_chatgpt_auth( codex_home.path(), ChatGptAuthFixture::new("chatgpt-token") .account_id("account-123") .chatgpt_user_id("user-123") .chatgpt_account_id("account-123"), AuthCredentialsStoreMode::File, )?; let missing_session_path = external_agent_home(codex_home.path()).join("projects/repo/missing.jsonl"); let project_root = codex_home.path().join("repo"); let home_dir = codex_home.path().display().to_string(); let mut mcp = TestAppServer::new_with_env(codex_home.path(), &[("HOME", Some(home_dir.as_str()))]) .await?; timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??; let request_id = mcp .send_raw_request( "externalAgentConfig/import", Some(serde_json::json!({ "source": "test_import", "migrationItems": [{ "itemType": "SESSIONS", "description": "Migrate recent sessions", "cwd": null, "details": { "sessions": [{ "path": missing_session_path, "cwd": project_root, "title": "missing session" }] } }] })), ) .await?; let response: JSONRPCResponse = timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; let response: ExternalAgentConfigImportResponse = to_response(response)?; let import_id = assert_import_response(response); let notification = timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_notification_message("externalAgentConfig/import/completed"), ) .await??; let completed: ExternalAgentConfigImportCompletedNotification = serde_json::from_value(notification.params.expect("completed params"))?; assert_eq!(completed.import_id, import_id); assert_eq!(completed.item_type_results.len(), 1); assert_eq!(completed.item_type_results[0].successes.len(), 0); assert_eq!(completed.item_type_results[0].failures.len(), 1); let event = wait_for_analytics_event( &analytics_server, DEFAULT_TIMEOUT, "codex_onboarding_external_agent_import_complete", ) .await?; let event_params = &event["event_params"]; assert_eq!(event_params["import_id"], serde_json::json!(import_id)); assert_eq!(event_params["source"], "test_import"); assert_eq!(event_params["type"], "SESSIONS"); assert_eq!(event_params["success_count"], 0); assert_eq!(event_params["failed_count"], 1); assert!(event_params.get("raw_errors").is_none()); let event = wait_for_analytics_event( &analytics_server, DEFAULT_TIMEOUT, "codex_onboarding_external_agent_import_failure", ) .await?; let event_params = &event["event_params"]; assert_eq!(event_params["import_id"], serde_json::json!(import_id)); assert_eq!(event_params["source"], "test_import"); assert_eq!(event_params["type"], "SESSIONS"); assert_eq!(event_params["failure_stage"], "session_missing"); assert_eq!(event_params["error_type"], "session_missing"); assert!(event_params.get("raw_errors").is_none()); assert!(event_params.get("message").is_none()); Ok(()) } #[tokio::test] async fn external_agent_config_import_sends_completion_notification_for_local_plugins() -> Result<()> { let codex_home = TempDir::new()?; let marketplace_root = codex_home.path().join("marketplace"); let plugin_root = marketplace_root.join("plugins").join("sample"); std::fs::create_dir_all(marketplace_root.join(".agents/plugins"))?; std::fs::create_dir_all(plugin_root.join(".codex-plugin"))?; std::fs::write( marketplace_root.join(".agents/plugins/marketplace.json"), r#"{ "name": "debug", "plugins": [ { "name": "sample", "source": { "source": "local", "path": "./plugins/sample" } } ] }"#, )?; std::fs::write( plugin_root.join(".codex-plugin/plugin.json"), r#"{"name":"sample","version":"0.1.0"}"#, )?; let source_home = external_agent_home(codex_home.path()); std::fs::create_dir_all(&source_home)?; let settings = serde_json::json!({ "enabledPlugins": { "sample@debug": true }, "extraKnownMarketplaces": { "debug": { "source": "local", "path": marketplace_root, } } }); std::fs::write( source_home.join("settings.json"), serde_json::to_string_pretty(&settings)?, )?; let home_dir = codex_home.path().display().to_string(); let mut mcp = TestAppServer::new_with_env(codex_home.path(), &[("HOME", Some(home_dir.as_str()))]) .await?; timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??; let request_id = mcp .send_raw_request( "externalAgentConfig/import", Some(serde_json::json!({ "migrationItems": [{ "itemType": "PLUGINS", "description": "Import plugins", "cwd": null, "details": { "plugins": [{ "marketplaceName": "debug", "pluginNames": ["sample"] }] } }] })), ) .await?; let response: JSONRPCResponse = timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; let response: ExternalAgentConfigImportResponse = to_response(response)?; let import_id = assert_import_response(response); let notification = timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_notification_message("externalAgentConfig/import/completed"), ) .await??; assert_eq!(notification.method, "externalAgentConfig/import/completed"); let completed: ExternalAgentConfigImportCompletedNotification = serde_json::from_value(notification.params.expect("completed params"))?; assert_eq!(completed.import_id, import_id); let request_id = mcp .send_plugin_list_request(PluginListParams { cwds: None, marketplace_kinds: None, }) .await?; let response: JSONRPCResponse = timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; let response: PluginListResponse = to_response(response)?; let plugin = response .marketplaces .iter() .find(|marketplace| marketplace.name == "debug") .and_then(|marketplace| { marketplace .plugins .iter() .find(|plugin| plugin.name == "sample") }) .expect("expected imported plugin to be listed"); assert!(plugin.installed); assert!(plugin.enabled); Ok(()) } #[tokio::test] async fn external_agent_config_import_sends_completion_notification_after_pending_plugins_finish() -> Result<()> { let codex_home = TempDir::new()?; let source_home = external_agent_home(codex_home.path()); std::fs::create_dir_all(&source_home)?; // This test only needs a pending non-local plugin import. Use an invalid // source so the background completion path cannot make a real network clone. std::fs::write( source_home.join("settings.json"), r#"{ "enabledPlugins": { "formatter@acme-tools": true }, "extraKnownMarketplaces": { "acme-tools": { "source": "not a valid marketplace source" } } }"#, )?; let home_dir = codex_home.path().display().to_string(); let mut mcp = TestAppServer::new_with_env(codex_home.path(), &[("HOME", Some(home_dir.as_str()))]) .await?; timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??; let request_id = mcp .send_raw_request( "externalAgentConfig/import", Some(serde_json::json!({ "migrationItems": [{ "itemType": "PLUGINS", "description": "Import plugins", "cwd": null, "details": { "plugins": [{ "marketplaceName": "acme-tools", "pluginNames": ["formatter"] }] } }] })), ) .await?; let response: JSONRPCResponse = timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; let response: ExternalAgentConfigImportResponse = to_response(response)?; let import_id = assert_import_response(response); let notification = timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_notification_message("externalAgentConfig/import/completed"), ) .await??; assert_eq!(notification.method, "externalAgentConfig/import/completed"); let completed: ExternalAgentConfigImportCompletedNotification = serde_json::from_value(notification.params.expect("completed params"))?; assert_eq!(completed.import_id, import_id); Ok(()) } #[tokio::test] async fn external_agent_config_import_creates_session_rollouts() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("follow-up answer").await; let codex_home = TempDir::new()?; create_config_toml(codex_home.path(), &server.uri())?; let project_root = codex_home.path().join("repo"); let recent_timestamp = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true); let session_dir = external_agent_home(codex_home.path()).join("projects/repo"); let session_path = session_dir.join("session.jsonl"); std::fs::create_dir_all(&project_root)?; std::fs::create_dir_all(&session_dir)?; std::fs::write( &session_path, [ serde_json::json!({ "type": "user", "cwd": &project_root, "timestamp": &recent_timestamp, "message": { "content": "first request" }, }) .to_string(), serde_json::json!({ "type": "assistant", "cwd": &project_root, "timestamp": &recent_timestamp, "message": { "content": "first answer" }, }) .to_string(), serde_json::json!({ "type": "custom-title", "customTitle": "source session title", }) .to_string(), ] .join("\n"), )?; let home_dir = codex_home.path().display().to_string(); let mut mcp = TestAppServer::new_with_env(codex_home.path(), &[("HOME", Some(home_dir.as_str()))]) .await?; timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??; let request_id = mcp .send_raw_request( "externalAgentConfig/detect", Some(serde_json::json!({ "includeHome": true, })), ) .await?; let response: JSONRPCResponse = timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; let detected: ExternalAgentConfigDetectResponse = to_response(response)?; assert_eq!(detected.items.len(), 1); let request_id = mcp .send_raw_request( "externalAgentConfig/import", Some(serde_json::json!({ "migrationItems": detected.items })), ) .await?; let response: JSONRPCResponse = timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; let response: ExternalAgentConfigImportResponse = to_response(response)?; let import_id = assert_import_response(response); let notification = timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_notification_message("externalAgentConfig/import/completed"), ) .await??; assert_eq!(notification.method, "externalAgentConfig/import/completed"); let completed: ExternalAgentConfigImportCompletedNotification = serde_json::from_value(notification.params.expect("completed params"))?; assert_eq!(completed.import_id, import_id); assert_eq!(completed.item_type_results.len(), 1); let session_result = &completed.item_type_results[0]; assert_eq!( session_result.item_type, ExternalAgentConfigMigrationItemType::Sessions ); assert_eq!(session_result.failures, Vec::new()); assert_eq!(session_result.successes.len(), 1); let session_success = &session_result.successes[0]; assert_eq!( session_success.item_type, ExternalAgentConfigMigrationItemType::Sessions ); assert_eq!(session_success.cwd, None); let session_source = std::fs::canonicalize(&session_path)?.display().to_string(); assert_eq!( session_success.source.as_deref(), Some(session_source.as_str()) ); let imported_thread_id = session_success .target .as_deref() .expect("session success should include imported thread id") .to_string(); let request_id = mcp .send_thread_list_request(ThreadListParams { cursor: None, limit: None, sort_key: None, sort_direction: None, model_providers: None, source_kinds: None, archived: None, cwd: None, use_state_db_only: false, search_term: None, parent_thread_id: None, ancestor_thread_id: None, }) .await?; let response: JSONRPCResponse = timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; let response: ThreadListResponse = to_response(response)?; let thread = response .data .first() .expect("expected imported thread") .clone(); assert_eq!(imported_thread_id, thread.id.to_string()); assert_eq!(thread.preview, "first request"); assert_eq!(thread.name.as_deref(), Some("source session title")); let request_id = mcp .send_thread_read_request(ThreadReadParams { thread_id: thread.id.clone(), include_turns: true, }) .await?; let response: JSONRPCResponse = timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; let response: ThreadReadResponse = to_response(response)?; assert_eq!(response.thread.turns.len(), 1); let items = &response.thread.turns[0].items; assert_eq!(items.len(), 3); assert_eq!( items.last(), Some(&ThreadItem::AgentMessage { id: "item-3".into(), text: "".into(), phase: None, memory_citation: None, }) ); let request_id = mcp .send_thread_resume_request(ThreadResumeParams { thread_id: thread.id.clone(), ..Default::default() }) .await?; let response: JSONRPCResponse = timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; let _: ThreadResumeResponse = to_response(response)?; let request_id = mcp .send_turn_start_request(TurnStartParams { thread_id: thread.id.clone(), client_user_message_id: None, input: vec![UserInput::Text { text: "follow up".to_string(), text_elements: Vec::new(), }], ..Default::default() }) .await?; timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_notification_message("turn/completed"), ) .await??; let request_id = mcp .send_thread_read_request(ThreadReadParams { thread_id: thread.id, include_turns: true, }) .await?; let response: JSONRPCResponse = timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; let response: ThreadReadResponse = to_response(response)?; assert_eq!(response.thread.turns.len(), 2); match &response.thread.turns[1].items[1] { ThreadItem::AgentMessage { text, .. } => assert_eq!(text, "follow-up answer"), other => panic!("expected agent message item, got {other:?}"), } Ok(()) } #[tokio::test] async fn external_agent_config_import_does_not_initialize_required_mcp() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("unused").await; let codex_home = TempDir::new()?; create_config_toml(codex_home.path(), &server.uri())?; let mut config = std::fs::read_to_string(codex_home.path().join("config.toml"))?; config.push_str( r#" [mcp_servers.required_broken] command = "this-command-does-not-exist" required = true "#, ); std::fs::write(codex_home.path().join("config.toml"), config)?; let project_root = codex_home.path().join("repo"); let recent_timestamp = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true); let session_dir = external_agent_home(codex_home.path()).join("projects/repo"); let session_path = session_dir.join("session.jsonl"); std::fs::create_dir_all(&project_root)?; std::fs::create_dir_all(&session_dir)?; std::fs::write( &session_path, serde_json::json!({ "type": "user", "cwd": &project_root, "timestamp": &recent_timestamp, "message": { "content": "first request" }, }) .to_string(), )?; let home_dir = codex_home.path().display().to_string(); let mut mcp = TestAppServer::new_with_env(codex_home.path(), &[("HOME", Some(home_dir.as_str()))]) .await?; timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??; let request_id = mcp .send_raw_request( "externalAgentConfig/import", Some(serde_json::json!({ "migrationItems": [{ "itemType": "SESSIONS", "description": "Migrate recent sessions", "cwd": null, "details": { "sessions": [{ "path": session_path, "cwd": project_root, "title": "first request" }] } }] })), ) .await?; timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_notification_message("externalAgentConfig/import/completed"), ) .await??; let request_id = mcp .send_thread_list_request(ThreadListParams { cursor: None, limit: None, sort_key: None, sort_direction: None, model_providers: None, source_kinds: None, archived: None, cwd: None, use_state_db_only: false, search_term: None, parent_thread_id: None, ancestor_thread_id: None, }) .await?; let response: JSONRPCResponse = timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; let response: ThreadListResponse = to_response(response)?; assert_eq!(response.data.len(), 1); Ok(()) } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn external_agent_config_import_accepts_detected_session_payload_after_restart() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("unused").await; let codex_home = TempDir::new()?; create_config_toml(codex_home.path(), &server.uri())?; let project_root = codex_home.path().join("repo"); let recent_timestamp = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true); let session_dir = external_agent_home(codex_home.path()).join("projects/repo"); let session_path = session_dir.join("session.jsonl"); std::fs::create_dir_all(&project_root)?; std::fs::create_dir_all(&session_dir)?; std::fs::write( &session_path, serde_json::json!({ "type": "user", "cwd": &project_root, "timestamp": &recent_timestamp, "message": { "content": "first request" }, }) .to_string(), )?; let home_dir = codex_home.path().display().to_string(); let mut mcp = TestAppServer::new_with_env(codex_home.path(), &[("HOME", Some(home_dir.as_str()))]) .await?; timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??; let request_id = mcp .send_raw_request( "externalAgentConfig/import", Some(serde_json::json!({ "migrationItems": [{ "itemType": "SESSIONS", "description": "Migrate recent sessions", "cwd": null, "details": { "sessions": [{ "path": session_path, "cwd": project_root, "title": "first request" }] } }] })), ) .await?; let response: JSONRPCResponse = timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; let response: ExternalAgentConfigImportResponse = to_response(response)?; let import_id = assert_import_response(response); let notification = timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_notification_message("externalAgentConfig/import/completed"), ) .await??; assert_eq!(notification.method, "externalAgentConfig/import/completed"); let completed: ExternalAgentConfigImportCompletedNotification = serde_json::from_value(notification.params.expect("completed params"))?; assert_eq!(completed.import_id, import_id); let request_id = mcp .send_thread_list_request(ThreadListParams { cursor: None, limit: None, sort_key: None, sort_direction: None, model_providers: None, source_kinds: None, archived: None, cwd: None, use_state_db_only: false, search_term: None, parent_thread_id: None, ancestor_thread_id: None, }) .await?; let response: JSONRPCResponse = timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; let response: ThreadListResponse = to_response(response)?; assert_eq!(response.data.len(), 1); Ok(()) } #[tokio::test] async fn external_agent_config_import_skips_already_imported_session_versions() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("unused").await; let codex_home = TempDir::new()?; create_config_toml(codex_home.path(), &server.uri())?; let project_root = codex_home.path().join("repo"); let recent_timestamp = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true); let session_dir = external_agent_home(codex_home.path()).join("projects/repo"); let session_path = session_dir.join("session.jsonl"); std::fs::create_dir_all(&project_root)?; std::fs::create_dir_all(&session_dir)?; std::fs::write( &session_path, serde_json::json!({ "type": "user", "cwd": &project_root, "timestamp": &recent_timestamp, "message": { "content": "first request" }, }) .to_string(), )?; let home_dir = codex_home.path().display().to_string(); let mut mcp = TestAppServer::new_with_env(codex_home.path(), &[("HOME", Some(home_dir.as_str()))]) .await?; timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??; let request_id = mcp .send_raw_request( "externalAgentConfig/detect", Some(serde_json::json!({ "includeHome": true })), ) .await?; let response: JSONRPCResponse = timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; let detected: ExternalAgentConfigDetectResponse = to_response(response)?; for _ in 0..2 { let request_id = mcp .send_raw_request( "externalAgentConfig/import", Some(serde_json::json!({ "migrationItems": detected.items.clone() })), ) .await?; let response: JSONRPCResponse = timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; let response: ExternalAgentConfigImportResponse = to_response(response)?; let import_id = assert_import_response(response); let notification = timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_notification_message("externalAgentConfig/import/completed"), ) .await??; assert_eq!(notification.method, "externalAgentConfig/import/completed"); let completed: ExternalAgentConfigImportCompletedNotification = serde_json::from_value(notification.params.expect("completed params"))?; assert_eq!(completed.import_id, import_id); } let request_id = mcp .send_thread_list_request(ThreadListParams { cursor: None, limit: None, sort_key: None, sort_direction: None, model_providers: None, source_kinds: None, archived: None, cwd: None, use_state_db_only: false, search_term: None, parent_thread_id: None, ancestor_thread_id: None, }) .await?; let response: JSONRPCResponse = timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; let response: ThreadListResponse = to_response(response)?; assert_eq!(response.data.len(), 1); Ok(()) } #[cfg(unix)] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn external_agent_config_import_returns_before_background_session_import_finishes() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("unused").await; let codex_home = TempDir::new()?; create_config_toml(codex_home.path(), &server.uri())?; let project_root = codex_home.path().join("repo"); let recent_timestamp = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true); let session_dir = external_agent_home(codex_home.path()).join("projects/repo"); let session_path = session_dir.join("session.jsonl"); std::fs::create_dir_all(&project_root)?; std::fs::create_dir_all(&session_dir)?; let session_contents = serde_json::json!({ "type": "user", "cwd": &project_root, "timestamp": &recent_timestamp, "message": { "content": "first request" }, }) .to_string(); std::fs::write(&session_path, &session_contents)?; let home_dir = codex_home.path().display().to_string(); let mut mcp = TestAppServer::new_with_env(codex_home.path(), &[("HOME", Some(home_dir.as_str()))]) .await?; timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??; let request_id = mcp .send_raw_request( "externalAgentConfig/detect", Some(serde_json::json!({ "includeHome": true })), ) .await?; let response: JSONRPCResponse = timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; let detected: ExternalAgentConfigDetectResponse = to_response(response)?; assert_eq!(detected.items.len(), 1); let detected_items = detected.items; std::fs::remove_file(&session_path)?; let status = std::process::Command::new("mkfifo") .arg(&session_path) .status()?; assert!(status.success()); let request_id = mcp .send_raw_request( "externalAgentConfig/import", Some(serde_json::json!({ "migrationItems": detected_items.clone() })), ) .await?; let response: JSONRPCResponse = timeout( Duration::from_secs(5), mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; let response: ExternalAgentConfigImportResponse = to_response(response)?; let import_id = assert_import_response(response); assert!( timeout( Duration::from_millis(200), mcp.read_stream_until_notification_message("externalAgentConfig/import/completed") ) .await .is_err(), "session import completed before the blocked background import was unblocked" ); let duplicate_request_id = mcp .send_raw_request( "externalAgentConfig/import", Some(serde_json::json!({ "migrationItems": detected_items })), ) .await?; let response: JSONRPCResponse = timeout( Duration::from_secs(5), mcp.read_stream_until_response_message(RequestId::Integer(duplicate_request_id)), ) .await??; let response: ExternalAgentConfigImportResponse = to_response(response)?; let duplicate_import_id = assert_import_response(response); let mut completed_import_ids = Vec::new(); for _ in 0..2 { timeout(DEFAULT_TIMEOUT, async { let mut file = tokio::fs::OpenOptions::new() .write(true) .open(&session_path) .await?; file.write_all(session_contents.as_bytes()).await }) .await??; let notification = timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_notification_message("externalAgentConfig/import/completed"), ) .await??; assert_eq!(notification.method, "externalAgentConfig/import/completed"); let completed: ExternalAgentConfigImportCompletedNotification = serde_json::from_value(notification.params.expect("completed params"))?; completed_import_ids.push(completed.import_id); } completed_import_ids.sort(); let mut expected_import_ids = vec![import_id, duplicate_import_id]; expected_import_ids.sort(); assert_eq!(completed_import_ids, expected_import_ids); let request_id = mcp .send_thread_list_request(ThreadListParams { cursor: None, limit: None, sort_key: None, sort_direction: None, model_providers: None, source_kinds: None, archived: None, cwd: None, use_state_db_only: false, search_term: None, parent_thread_id: None, ancestor_thread_id: None, }) .await?; let response: JSONRPCResponse = timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; let response: ThreadListResponse = to_response(response)?; assert_eq!(response.data.len(), 1); Ok(()) } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn external_agent_config_import_compacts_huge_session_before_first_follow_up() -> Result<()> { let server = responses::start_mock_server().await; let response_log = responses::mount_sse_sequence( &server, vec![ responses::sse(vec![ responses::ev_assistant_message("m1", "LOCAL_SUMMARY"), responses::ev_completed_with_tokens("r1", /*total_tokens*/ 120), ]), responses::sse(vec![ responses::ev_assistant_message("m2", "follow-up answer"), responses::ev_completed_with_tokens("r2", /*total_tokens*/ 80), ]), ], ) .await; let codex_home = TempDir::new()?; write_mock_responses_config_toml( codex_home.path(), &server.uri(), &BTreeMap::default(), /*auto_compact_limit*/ 200, /*requires_openai_auth*/ None, "mock_provider", "Summarize the conversation.", )?; let project_root = codex_home.path().join("repo"); let recent_timestamp = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true); let session_dir = external_agent_home(codex_home.path()).join("projects/repo"); let session_path = session_dir.join("session.jsonl"); std::fs::create_dir_all(&project_root)?; std::fs::create_dir_all(&session_dir)?; let huge_user = "u".repeat(20_000); let huge_assistant = "a".repeat(20_000); std::fs::write( &session_path, [ serde_json::json!({ "type": "user", "cwd": &project_root, "timestamp": &recent_timestamp, "message": { "content": &huge_user }, }) .to_string(), serde_json::json!({ "type": "assistant", "cwd": &project_root, "timestamp": &recent_timestamp, "message": { "content": &huge_assistant }, }) .to_string(), ] .join("\n"), )?; let home_dir = codex_home.path().display().to_string(); let mut mcp = TestAppServer::new_with_env(codex_home.path(), &[("HOME", Some(home_dir.as_str()))]) .await?; timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??; let request_id = mcp .send_raw_request( "externalAgentConfig/detect", Some(serde_json::json!({ "includeHome": true, })), ) .await?; let response: JSONRPCResponse = timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; let detected: ExternalAgentConfigDetectResponse = to_response(response)?; assert_eq!(detected.items.len(), 1); let request_id = mcp .send_raw_request( "externalAgentConfig/import", Some(serde_json::json!({ "migrationItems": detected.items })), ) .await?; let response: JSONRPCResponse = timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; let response: ExternalAgentConfigImportResponse = to_response(response)?; let import_id = assert_import_response(response); let notification = timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_notification_message("externalAgentConfig/import/completed"), ) .await??; assert_eq!(notification.method, "externalAgentConfig/import/completed"); let completed: ExternalAgentConfigImportCompletedNotification = serde_json::from_value(notification.params.expect("completed params"))?; assert_eq!(completed.import_id, import_id); let request_id = mcp .send_thread_list_request(ThreadListParams { cursor: None, limit: None, sort_key: None, sort_direction: None, model_providers: None, source_kinds: None, archived: None, cwd: None, use_state_db_only: false, search_term: None, parent_thread_id: None, ancestor_thread_id: None, }) .await?; let response: JSONRPCResponse = timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; let response: ThreadListResponse = to_response(response)?; let thread = response .data .first() .expect("expected imported thread") .clone(); let request_id = mcp .send_thread_resume_request(ThreadResumeParams { thread_id: thread.id.clone(), ..Default::default() }) .await?; let response: JSONRPCResponse = timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; let _: ThreadResumeResponse = to_response(response)?; let request_id = mcp .send_turn_start_request(TurnStartParams { thread_id: thread.id.clone(), client_user_message_id: None, input: vec![UserInput::Text { text: "follow up".to_string(), text_elements: Vec::new(), }], ..Default::default() }) .await?; timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(request_id)), ) .await??; timeout( DEFAULT_TIMEOUT, mcp.read_stream_until_notification_message("turn/completed"), ) .await??; let requests = response_log.requests(); assert_eq!(requests.len(), 2); let first = requests[0].body_json().to_string(); let second = requests[1].body_json().to_string(); assert!(first.contains("Summarize the conversation.")); assert!(!first.contains("follow up")); assert!(second.contains("follow up")); assert!(second.contains("LOCAL_SUMMARY")); Ok(()) } fn create_config_toml(codex_home: &std::path::Path, server_uri: &str) -> std::io::Result<()> { std::fs::write( codex_home.join("config.toml"), format!( r#" model = "mock-model" approval_policy = "never" sandbox_mode = "read-only" model_provider = "mock_provider" [model_providers.mock_provider] name = "Mock provider for test" base_url = "{server_uri}/v1" wire_api = "responses" request_max_retries = 0 stream_max_retries = 0 "# ), ) } fn write_analytics_config(codex_home: &std::path::Path, base_url: &str) -> std::io::Result<()> { std::fs::write( codex_home.join("config.toml"), format!("chatgpt_base_url = \"{base_url}\"\n"), ) }