codex/codex-rs/app-server/tests/suite/v2/external_agent_config.rs

use std::time::Duration;

use anyhow::Result;
use app_test_support::TestAppServer;
use app_test_support::create_mock_responses_server_repeating_assistant;
use app_test_support::to_response;
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::JSONRPCError;
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 core_test_support::responses;
use pretty_assertions::assert_eq;
use std::collections::BTreeMap;
use tempfile::TempDir;
#[cfg(unix)]
use tokio::io::AsyncWriteExt;
use tokio::time::timeout;

const DEFAULT_TIMEOUT: Duration = Duration::from_secs(60);

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::<Vec<_>>();
    let expected_failures = completed
        .item_type_results
        .iter()
        .flat_map(|type_result| type_result.failures.iter())
        .collect::<Vec<_>>();
    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_returns_error_for_failed_sync_import() -> Result<()> {
    let codex_home = TempDir::new()?;
    std::fs::create_dir_all(codex_home.path().join(".claude"))?;
    std::fs::write(
        codex_home.path().join(".claude").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 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": "CONFIG",
                    "description": "Import config",
                    "cwd": null
                }]
            })),
        )
        .await?;

    let error: JSONRPCError = timeout(
        DEFAULT_TIMEOUT,
        mcp.read_stream_until_error_message(RequestId::Integer(request_id)),
    )
    .await??;
    assert_eq!(error.error.code, -32603);
    assert!(
        error.error.message.contains("invalid existing config.toml"),
        "unexpected error: {error:?}"
    );

    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"}"#,
    )?;
    std::fs::create_dir_all(codex_home.path().join(".claude"))?;
    let settings = serde_json::json!({
        "enabledPlugins": {
            "sample@debug": true
        },
        "extraKnownMarketplaces": {
            "debug": {
                "source": "local",
                "path": marketplace_root,
            }
        }
    });
    std::fs::write(
        codex_home.path().join(".claude").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()?;
    std::fs::create_dir_all(codex_home.path().join(".claude"))?;
    // 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(
        codex_home.path().join(".claude").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 = codex_home.path().join(".claude/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,
        })
        .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: "<EXTERNAL SESSION IMPORTED>".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 = codex_home.path().join(".claude/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,
        })
        .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 = codex_home.path().join(".claude/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,
        })
        .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 = codex_home.path().join(".claude/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,
        })
        .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 = codex_home.path().join(".claude/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,
        })
        .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 = codex_home.path().join(".claude/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,
        })
        .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
"#
        ),
    )
}