File size: 27,059 Bytes

52a9af3

use std::process::Command;
use std::sync::Arc;

use codex_core::ModelClient;
use codex_core::Prompt;
use codex_core::ResponseEvent;
use codex_features::Feature;
use codex_login::CodexAuth;
use codex_login::auth::AgentIdentityAuthPolicy;
use codex_model_provider_info::ModelProviderInfo;
use codex_model_provider_info::WireApi;
use codex_otel::SessionTelemetry;
use codex_otel::TelemetryAuthMode;
use codex_protocol::ThreadId;
use codex_protocol::config_types::ReasoningSummary;
use codex_protocol::models::ContentItem;
use codex_protocol::models::ResponseItem;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use core_test_support::TestCodexResponsesRequestKind;
use core_test_support::load_default_config_for_test;
use core_test_support::responses;
use core_test_support::responses_metadata as test_responses_metadata;
use core_test_support::test_codex::test_codex;
use futures::StreamExt;
use pretty_assertions::assert_eq;
use tempfile::TempDir;
use wiremock::matchers::header;

fn normalize_git_remote_url(url: &str) -> String {
    let normalized = url.trim().trim_end_matches('/');
    normalized
        .strip_suffix(".git")
        .unwrap_or(normalized)
        .to_string()
}

const TEST_INSTALLATION_ID: &str = "11111111-1111-4111-8111-111111111111";
fn test_turn_responses_metadata(
    _client: &ModelClient,
    thread_id: ThreadId,
    session_source: &SessionSource,
) -> codex_core::CodexResponsesMetadata {
    let thread_id = thread_id.to_string();
    test_responses_metadata(
        TEST_INSTALLATION_ID,
        &thread_id,
        &thread_id,
        /*turn_id*/ None,
        format!("{thread_id}:0"),
        session_source,
        /*parent_thread_id*/ None,
        TestCodexResponsesRequestKind::Turn,
    )
}

#[tokio::test]
async fn responses_stream_includes_subagent_header_on_review() {
    core_test_support::skip_if_no_network!();

    let server = responses::start_mock_server().await;
    let response_body = responses::sse(vec![
        responses::ev_response_created("resp-1"),
        responses::ev_completed("resp-1"),
    ]);

    let request_recorder = responses::mount_sse_once_match(
        &server,
        header("x-openai-subagent", "review"),
        response_body,
    )
    .await;

    let provider = ModelProviderInfo {
        name: "mock".into(),
        base_url: Some(format!("{}/v1", server.uri())),
        env_key: None,
        env_key_instructions: None,
        experimental_bearer_token: None,
        auth: None,
        aws: None,
        wire_api: WireApi::Responses,
        query_params: None,
        http_headers: None,
        env_http_headers: None,
        request_max_retries: Some(0),
        stream_max_retries: Some(0),
        stream_idle_timeout_ms: Some(5_000),
        websocket_connect_timeout_ms: None,
        requires_openai_auth: false,
        supports_websockets: false,
        supports_standalone_web_search: false,
    };

    let codex_home = TempDir::new().expect("failed to create TempDir");
    let mut config = load_default_config_for_test(&codex_home).await;
    config.model_provider_id = provider.name.clone();
    config.model_provider = provider.clone();
    let effort = config.model_reasoning_effort.clone();
    let summary = config.model_reasoning_summary;
    let model = codex_core::test_support::get_model_offline(config.model.as_deref());
    config.model = Some(model.clone());
    let config = Arc::new(config);

    let thread_id = ThreadId::new();
    let auth_mode = TelemetryAuthMode::Chatgpt;
    let session_source = SessionSource::SubAgent(SubAgentSource::Review);
    let model_info =
        codex_core::test_support::construct_model_info_offline(model.as_str(), &config);
    let expected_window_id = format!("{thread_id}:0");
    let session_telemetry = SessionTelemetry::new(
        thread_id,
        model.as_str(),
        model_info.slug.as_str(),
        /*account_id*/ None,
        Some("test@test.com".to_string()),
        Some(auth_mode),
        "test_originator".to_string(),
        /*log_user_prompts*/ false,
        "test".to_string(),
        session_source.clone(),
    );

    let client = ModelClient::new(
        /*auth_manager*/ None,
        AgentIdentityAuthPolicy::JwtOnly,
        thread_id,
        provider.clone(),
        session_source.clone(),
        "test_originator".to_string(),
        config.model_verbosity,
        config.features.enabled(Feature::ContentItemKinds),
        /*enable_request_compression*/ false,
        /*include_timing_metrics*/ false,
        /*beta_features_header*/ None,
        /*concurrent_reasoning_summaries_enabled*/ false,
        /*attestation_provider*/ None,
        config.http_client_factory(),
        config.workspace_routing_context(),
    );
    let responses_metadata = test_turn_responses_metadata(&client, thread_id, &session_source);
    let mut client_session = client.new_session();

    let mut prompt = Prompt::default();
    prompt.input = vec![ResponseItem::Message {
        id: None,
        role: "user".into(),
        content: vec![ContentItem::InputText {
            text: "hello".into(),
        }],
        phase: None,
        internal_chat_message_metadata_passthrough: None,
    }];

    let mut stream = client_session
        .stream(
            &prompt,
            &model_info,
            &session_telemetry,
            effort,
            summary.unwrap_or(model_info.default_reasoning_summary),
            /*service_tier*/ None,
            &responses_metadata,
            &codex_rollout_trace::InferenceTraceContext::disabled(),
        )
        .await
        .expect("stream failed");
    while let Some(event) = stream.next().await {
        if matches!(event, Ok(ResponseEvent::Completed { .. })) {
            break;
        }
    }

    let request = request_recorder.single_request();
    assert_eq!(
        request.header("x-openai-subagent").as_deref(),
        Some("review")
    );
    assert_eq!(
        request.header("x-codex-window-id").as_deref(),
        Some(expected_window_id.as_str())
    );
    assert_eq!(request.header("x-codex-parent-thread-id"), None);
    assert_eq!(
        request.body_json()["client_metadata"]["x-codex-installation-id"].as_str(),
        Some(TEST_INSTALLATION_ID)
    );
    assert_eq!(
        request.body_json()["client_metadata"]["x-codex-window-id"].as_str(),
        Some(expected_window_id.as_str())
    );
    assert_eq!(request.header("x-codex-sandbox"), None);
}

#[tokio::test]
async fn responses_stream_includes_subagent_header_on_other() {
    core_test_support::skip_if_no_network!();

    let server = responses::start_mock_server().await;
    let response_body = responses::sse(vec![
        responses::ev_response_created("resp-1"),
        responses::ev_completed("resp-1"),
    ]);

    let request_recorder = responses::mount_sse_once_match(
        &server,
        header("x-openai-subagent", "my-task"),
        response_body,
    )
    .await;

    let provider = ModelProviderInfo {
        name: "mock".into(),
        base_url: Some(format!("{}/v1", server.uri())),
        env_key: None,
        env_key_instructions: None,
        experimental_bearer_token: None,
        auth: None,
        aws: None,
        wire_api: WireApi::Responses,
        query_params: None,
        http_headers: None,
        env_http_headers: None,
        request_max_retries: Some(0),
        stream_max_retries: Some(0),
        stream_idle_timeout_ms: Some(5_000),
        websocket_connect_timeout_ms: None,
        requires_openai_auth: false,
        supports_websockets: false,
        supports_standalone_web_search: false,
    };

    let codex_home = TempDir::new().expect("failed to create TempDir");
    let mut config = load_default_config_for_test(&codex_home).await;
    config.model_provider_id = provider.name.clone();
    config.model_provider = provider.clone();
    let effort = config.model_reasoning_effort.clone();
    let summary = config.model_reasoning_summary;
    let model = codex_core::test_support::get_model_offline(config.model.as_deref());
    config.model = Some(model.clone());
    let config = Arc::new(config);

    let thread_id = ThreadId::new();
    let auth_mode = TelemetryAuthMode::Chatgpt;
    let session_source = SessionSource::SubAgent(SubAgentSource::Other("my-task".to_string()));
    let model_info =
        codex_core::test_support::construct_model_info_offline(model.as_str(), &config);

    let session_telemetry = SessionTelemetry::new(
        thread_id,
        model.as_str(),
        model_info.slug.as_str(),
        /*account_id*/ None,
        Some("test@test.com".to_string()),
        Some(auth_mode),
        "test_originator".to_string(),
        /*log_user_prompts*/ false,
        "test".to_string(),
        session_source.clone(),
    );

    let client = ModelClient::new(
        /*auth_manager*/ None,
        AgentIdentityAuthPolicy::JwtOnly,
        thread_id,
        provider.clone(),
        session_source.clone(),
        "test_originator".to_string(),
        config.model_verbosity,
        config.features.enabled(Feature::ContentItemKinds),
        /*enable_request_compression*/ false,
        /*include_timing_metrics*/ false,
        /*beta_features_header*/ None,
        /*concurrent_reasoning_summaries_enabled*/ false,
        /*attestation_provider*/ None,
        config.http_client_factory(),
        config.workspace_routing_context(),
    );
    let responses_metadata = test_turn_responses_metadata(&client, thread_id, &session_source);
    let mut client_session = client.new_session();

    let mut prompt = Prompt::default();
    prompt.input = vec![ResponseItem::Message {
        id: None,
        role: "user".into(),
        content: vec![ContentItem::InputText {
            text: "hello".into(),
        }],
        phase: None,
        internal_chat_message_metadata_passthrough: None,
    }];

    let mut stream = client_session
        .stream(
            &prompt,
            &model_info,
            &session_telemetry,
            effort,
            summary.unwrap_or(model_info.default_reasoning_summary),
            /*service_tier*/ None,
            &responses_metadata,
            &codex_rollout_trace::InferenceTraceContext::disabled(),
        )
        .await
        .expect("stream failed");
    while let Some(event) = stream.next().await {
        if matches!(event, Ok(ResponseEvent::Completed { .. })) {
            break;
        }
    }

    let request = request_recorder.single_request();
    assert_eq!(
        request.header("x-openai-subagent").as_deref(),
        Some("my-task")
    );
}

#[tokio::test]
async fn responses_respects_model_info_overrides_from_config() {
    core_test_support::skip_if_no_network!();

    let server = responses::start_mock_server().await;
    let response_body = responses::sse(vec![
        responses::ev_response_created("resp-1"),
        responses::ev_completed("resp-1"),
    ]);

    let request_recorder = responses::mount_sse_once(&server, response_body).await;

    let provider = ModelProviderInfo {
        name: "mock".into(),
        base_url: Some(format!("{}/v1", server.uri())),
        env_key: None,
        env_key_instructions: None,
        experimental_bearer_token: None,
        auth: None,
        aws: None,
        wire_api: WireApi::Responses,
        query_params: None,
        http_headers: None,
        env_http_headers: None,
        request_max_retries: Some(0),
        stream_max_retries: Some(0),
        stream_idle_timeout_ms: Some(5_000),
        websocket_connect_timeout_ms: None,
        requires_openai_auth: false,
        supports_websockets: false,
        supports_standalone_web_search: false,
    };

    let codex_home = TempDir::new().expect("failed to create TempDir");
    let mut config = load_default_config_for_test(&codex_home).await;
    config.model = Some("gpt-3.5-turbo".to_string());
    config.model_provider_id = provider.name.clone();
    config.model_provider = provider.clone();
    config.model_reasoning_summary = Some(ReasoningSummary::Detailed);
    let effort = config.model_reasoning_effort.clone();
    let summary = config.model_reasoning_summary;
    let model = config.model.clone().expect("model configured");
    let config = Arc::new(config);

    let thread_id = ThreadId::new();
    let auth_mode =
        codex_core::test_support::auth_manager_from_auth(CodexAuth::from_api_key("Test API Key"))
            .auth_mode()
            .map(TelemetryAuthMode::from);
    let session_source =
        SessionSource::SubAgent(SubAgentSource::Other("override-check".to_string()));
    let model_info =
        codex_core::test_support::construct_model_info_offline(model.as_str(), &config);
    let session_telemetry = SessionTelemetry::new(
        thread_id,
        model.as_str(),
        model_info.slug.as_str(),
        /*account_id*/ None,
        Some("test@test.com".to_string()),
        auth_mode,
        "test_originator".to_string(),
        /*log_user_prompts*/ false,
        "test".to_string(),
        session_source.clone(),
    );

    let client = ModelClient::new(
        /*auth_manager*/ None,
        AgentIdentityAuthPolicy::JwtOnly,
        thread_id,
        provider.clone(),
        session_source.clone(),
        "test_originator".to_string(),
        config.model_verbosity,
        config.features.enabled(Feature::ContentItemKinds),
        /*enable_request_compression*/ false,
        /*include_timing_metrics*/ false,
        /*beta_features_header*/ None,
        /*concurrent_reasoning_summaries_enabled*/ false,
        /*attestation_provider*/ None,
        config.http_client_factory(),
        config.workspace_routing_context(),
    );
    let responses_metadata = test_turn_responses_metadata(&client, thread_id, &session_source);
    let mut client_session = client.new_session();

    let mut prompt = Prompt::default();
    prompt.input = vec![ResponseItem::Message {
        id: None,
        role: "user".into(),
        content: vec![ContentItem::InputText {
            text: "hello".into(),
        }],
        phase: None,
        internal_chat_message_metadata_passthrough: None,
    }];

    let mut stream = client_session
        .stream(
            &prompt,
            &model_info,
            &session_telemetry,
            effort,
            summary.unwrap_or(model_info.default_reasoning_summary),
            /*service_tier*/ None,
            &responses_metadata,
            &codex_rollout_trace::InferenceTraceContext::disabled(),
        )
        .await
        .expect("stream failed");
    while let Some(event) = stream.next().await {
        if matches!(event, Ok(ResponseEvent::Completed { .. })) {
            break;
        }
    }

    let request = request_recorder.single_request();
    let body = request.body_json();
    let reasoning = body
        .get("reasoning")
        .and_then(|value| value.as_object())
        .cloned();

    assert!(
        reasoning.is_some(),
        "reasoning should be present when config enables summaries"
    );

    assert_eq!(
        reasoning
            .as_ref()
            .and_then(|value| value.get("summary"))
            .and_then(|value| value.as_str()),
        Some("detailed")
    );
}

#[tokio::test]
async fn responses_stream_includes_turn_metadata_header_for_git_workspace_e2e() {
    core_test_support::skip_if_no_network!();

    let server = responses::start_mock_server().await;
    let response_body = responses::sse(vec![
        responses::ev_response_created("resp-1"),
        responses::ev_completed("resp-1"),
    ]);

    let test = test_codex()
        .with_config(|config| config.analytics_enabled = Some(true))
        .build(&server)
        .await
        .expect("build test codex");
    let cwd = test.cwd_path();

    let first_request = responses::mount_sse_once(&server, response_body.clone()).await;
    test.submit_turn("hello")
        .await
        .expect("submit first turn prompt");
    let initial_request = first_request.single_request();
    let initial_header = initial_request
        .header("x-codex-turn-metadata")
        .expect("x-codex-turn-metadata header should be present");
    let initial_parsed: serde_json::Value =
        serde_json::from_str(&initial_header).expect("x-codex-turn-metadata should be valid JSON");
    let initial_turn_id = initial_parsed
        .get("turn_id")
        .and_then(serde_json::Value::as_str)
        .expect("turn_id should be present")
        .to_string();
    assert!(
        !initial_turn_id.is_empty(),
        "turn_id should not be empty in x-codex-turn-metadata"
    );
    let initial_turn_started_at_unix_ms = initial_parsed
        .get("turn_started_at_unix_ms")
        .and_then(serde_json::Value::as_i64)
        .expect("turn_started_at_unix_ms should be present");
    assert!(
        initial_turn_started_at_unix_ms > 0,
        "turn_started_at_unix_ms should be positive"
    );
    assert_eq!(
        initial_parsed
            .get("sandbox")
            .and_then(serde_json::Value::as_str),
        Some("none")
    );
    assert_eq!(
        initial_parsed
            .get("sandbox_mode")
            .and_then(serde_json::Value::as_str),
        Some("danger-full-access")
    );
    let body_metadata: serde_json::Value = serde_json::from_str(
        initial_request.body_json()["client_metadata"]["x-codex-turn-metadata"]
            .as_str()
            .expect("request body should include x-codex-turn-metadata"),
    )
    .expect("body x-codex-turn-metadata should be valid JSON");
    assert_eq!(
        (
            initial_parsed["analytics_enabled"].as_bool(),
            body_metadata["analytics_enabled"].as_bool(),
        ),
        (Some(true), Some(true)),
    );
    assert_eq!(
        body_metadata
            .get("sandbox_mode")
            .and_then(serde_json::Value::as_str),
        Some("danger-full-access")
    );
    assert_eq!(
        initial_parsed
            .get("thread_source")
            .and_then(serde_json::Value::as_str),
        None
    );

    let git_config_global = cwd.join("empty-git-config");
    std::fs::write(&git_config_global, "").expect("write empty git config");
    let run_git = |args: &[&str]| {
        let output = Command::new("git")
            .env("GIT_CONFIG_GLOBAL", &git_config_global)
            .env("GIT_CONFIG_NOSYSTEM", "1")
            .args(args)
            .current_dir(cwd)
            .output()
            .expect("git command should run");
        assert!(
            output.status.success(),
            "git {:?} failed: stdout={} stderr={}",
            args,
            String::from_utf8_lossy(&output.stdout),
            String::from_utf8_lossy(&output.stderr)
        );
        output
    };

    run_git(&["init"]);
    run_git(&["config", "user.name", "Test User"]);
    run_git(&["config", "user.email", "test@example.com"]);
    std::fs::write(cwd.join("README.md"), "hello").expect("write README");
    run_git(&["add", "."]);
    run_git(&["commit", "-m", "initial commit"]);
    run_git(&[
        "remote",
        "add",
        "origin",
        "https://github.com/openai/codex.git",
    ]);

    let expected_head = String::from_utf8(run_git(&["rev-parse", "HEAD"]).stdout)
        .expect("git rev-parse output should be valid UTF-8")
        .trim()
        .to_string();
    let expected_origin = String::from_utf8(run_git(&["remote", "get-url", "origin"]).stdout)
        .expect("git remote get-url output should be valid UTF-8")
        .trim()
        .to_string();

    let first_response = responses::sse(vec![
        responses::ev_response_created("resp-2"),
        responses::ev_reasoning_item("rsn-1", &["thinking"], &[]),
        responses::ev_exec_command_call("call-1", "echo turn-metadata"),
        responses::ev_completed("resp-2"),
    ]);
    let follow_up_response = responses::sse(vec![
        responses::ev_response_created("resp-3"),
        responses::ev_assistant_message("msg-1", "done"),
        responses::ev_completed("resp-3"),
    ]);
    let request_log = responses::mount_response_sequence(
        &server,
        vec![
            responses::sse_response(first_response),
            responses::sse_response(follow_up_response),
        ],
    )
    .await;

    test.submit_turn("hello")
        .await
        .expect("submit post-git turn prompt");

    let requests = request_log.requests();
    assert_eq!(requests.len(), 2, "expected two requests in one turn");

    let first_parsed: serde_json::Value = serde_json::from_str(
        &requests[0]
            .header("x-codex-turn-metadata")
            .expect("first request should include turn metadata"),
    )
    .expect("first metadata should be valid json");
    let second_parsed: serde_json::Value = serde_json::from_str(
        &requests[1]
            .header("x-codex-turn-metadata")
            .expect("second request should include turn metadata"),
    )
    .expect("second metadata should be valid json");

    let first_turn_id = first_parsed
        .get("turn_id")
        .and_then(serde_json::Value::as_str)
        .expect("first turn_id should be present");
    let second_turn_id = second_parsed
        .get("turn_id")
        .and_then(serde_json::Value::as_str)
        .expect("second turn_id should be present");
    let first_turn_started_at_unix_ms = first_parsed
        .get("turn_started_at_unix_ms")
        .and_then(serde_json::Value::as_i64)
        .expect("first turn_started_at_unix_ms should be present");
    let second_turn_started_at_unix_ms = second_parsed
        .get("turn_started_at_unix_ms")
        .and_then(serde_json::Value::as_i64)
        .expect("second turn_started_at_unix_ms should be present");
    assert!(
        first_turn_started_at_unix_ms > 0,
        "first turn_started_at_unix_ms should be positive"
    );
    assert_eq!(
        first_turn_started_at_unix_ms, second_turn_started_at_unix_ms,
        "requests in the same turn should share turn_started_at_unix_ms"
    );
    assert_eq!(
        first_parsed
            .get("thread_source")
            .and_then(serde_json::Value::as_str),
        None
    );
    assert_eq!(
        second_parsed
            .get("thread_source")
            .and_then(serde_json::Value::as_str),
        None
    );
    assert_eq!(
        first_turn_id, second_turn_id,
        "requests should share turn_id"
    );
    assert_ne!(
        second_turn_id,
        initial_turn_id.as_str(),
        "post-git turn should have a new turn_id"
    );

    assert_eq!(
        second_parsed
            .get("sandbox")
            .and_then(serde_json::Value::as_str),
        Some("none")
    );

    let workspace = second_parsed
        .get("workspaces")
        .and_then(serde_json::Value::as_object)
        .and_then(|workspaces| workspaces.values().next())
        .cloned()
        .expect("second request should include git workspace metadata");
    assert_eq!(
        workspace
            .get("latest_git_commit_hash")
            .and_then(serde_json::Value::as_str),
        Some(expected_head.as_str())
    );
    if let Some(actual_origin) = workspace
        .get("associated_remote_urls")
        .and_then(serde_json::Value::as_object)
        .and_then(|remotes| remotes.get("origin"))
        .and_then(serde_json::Value::as_str)
    {
        assert_eq!(
            normalize_git_remote_url(actual_origin),
            normalize_git_remote_url(&expected_origin)
        );
    }
    assert_eq!(
        workspace
            .get("has_changes")
            .and_then(serde_json::Value::as_bool),
        Some(false)
    );
}

/// Credential-bearing Git remotes must be redacted from both metadata transports.
#[tokio::test]
async fn responses_stream_redacts_git_remote_credentials_from_turn_metadata() {
    core_test_support::skip_if_no_network!();

    let server = responses::start_mock_server().await;
    let test = test_codex()
        .with_model("test-gpt-5.1-codex")
        .build(&server)
        .await
        .expect("build test codex");
    let cwd = test.cwd_path();
    let remote_token = "test-git-secret-token";
    let credential_bearing_origin =
        "https://alice:test-git-secret-token@example.invalid/openai/codex.git";

    for args in [
        ["init"].as_slice(),
        ["remote", "add", "origin", credential_bearing_origin].as_slice(),
    ] {
        let output = Command::new("git")
            .args(args)
            .current_dir(cwd)
            .output()
            .expect("git command should run");
        assert!(
            output.status.success(),
            "git {args:?} failed: {}",
            String::from_utf8_lossy(&output.stderr)
        );
    }

    let first_response = responses::sse(vec![
        responses::ev_response_created("resp-1"),
        responses::ev_function_call(
            "wait-for-git",
            "test_sync_tool",
            r#"{"wait_for_git_enrichment":true}"#,
        ),
        responses::ev_completed("resp-1"),
    ]);
    let follow_up_response = responses::sse(vec![
        responses::ev_response_created("resp-2"),
        responses::ev_assistant_message("msg-1", "done"),
        responses::ev_completed("resp-2"),
    ]);
    let request_log = responses::mount_response_sequence(
        &server,
        vec![
            responses::sse_response(first_response),
            responses::sse_response(follow_up_response),
        ],
    )
    .await;

    test.submit_turn("inspect the workspace")
        .await
        .expect("submit turn prompt");

    let requests = request_log.requests();
    assert_eq!(requests.len(), 2, "expected a follow-up model request");
    let header_metadata = requests[1]
        .header("x-codex-turn-metadata")
        .expect("request header should include turn metadata");
    let request_body = requests[1].body_json();
    let body_metadata = request_body["client_metadata"]["x-codex-turn-metadata"]
        .as_str()
        .expect("request body should include turn metadata");

    for (location, metadata) in [
        ("request header", header_metadata.as_str()),
        ("request body", body_metadata),
    ] {
        assert!(
            !metadata.contains(remote_token),
            "{location} should not expose git remote credentials"
        );
        let metadata: serde_json::Value =
            serde_json::from_str(metadata).expect("turn metadata should be valid JSON");
        let remotes = metadata
            .get("workspaces")
            .and_then(serde_json::Value::as_object)
            .and_then(|workspaces| workspaces.values().next())
            .and_then(|workspace| workspace.get("associated_remote_urls"))
            .expect("turn metadata should include git remote URLs");
        assert_eq!(
            remotes,
            &serde_json::json!({ "origin": "https://example.invalid/openai/codex.git" }),
            "{location} should contain the sanitized remote"
        );
    }
}