Emit API-key turn cost telemetry from app-server (#39006)

## What changed - Track app-server turn lifecycle events for OpenAI API-key sessions when OTLP telemetry export is enabled. - Query the turn-cost endpoint after completed or aborted turns and emit `codex.turn_cost` with the estimated USD cost, interruption state, model, speed, and reasoning effort. - Preserve provider organization and project headers, wait for all observed responses to be priced, and retry pending or transiently failed queries with bounded tracking. - Re-probe availability after authentication changes and shut the worker down with other app-server background tasks. ## Testing - Cover late API-key login, transient probe failures, telemetry after thread removal, and turns with multiple responses. - Verify turn-cost requests use API-key authentication and provider scope headers and deserialize priced response data. GitOrigin-RevId: a303162a54798b554fcb2e9e2b372e1cf9f49f39
2026-11-05 20:10:33 +00:00 · 2026-08-17 13:20:58 +00:00
parent 02e9bfaac7
commit 04caa22c82
11 changed files with 934 additions and 0 deletions
@@ -121,6 +121,7 @@ mod skills_watcher;
 mod thread_state;
 mod thread_status;
 mod transport;
+mod turn_cost_worker;

 pub use crate::code_mode_host::AppServerCodeModeHostArgs;
 pub use crate::code_mode_host::CodeModeHostTransport;
@@ -52,6 +52,7 @@ use crate::thread_state::ConnectionCapabilities;
 use crate::thread_state::ThreadStateManager;
 use crate::transport::AppServerTransport;
 use crate::transport::RemoteControlHandle;
+use crate::turn_cost_worker::TurnCostWorker;
 use codex_analytics::AnalyticsEventsClient;
 use codex_analytics::AppServerRpcTransport;
 use codex_app_server_protocol::ClientNotification;
@@ -131,6 +132,7 @@ fn reject_removed_permission_profile(request: &JSONRPCRequest) -> Result<(), JSO
 pub(crate) struct MessageProcessor {
    outgoing: Arc<OutgoingMessageSender>,
    models_refresh_worker: ModelsRefreshWorker,
+    turn_cost_worker: Option<TurnCostWorker>,
    skills_watcher: Arc<SkillsWatcher>,
    account_processor: AccountRequestProcessor,
    apps_processor: AppsRequestProcessor,
@@ -358,6 +360,8 @@ impl MessageProcessor {
        let models_manager = thread_manager.get_models_manager();
        let models_refresh_worker =
            crate::models_refresh_worker::spawn(&models_manager, config.http_client_factory());
+        let turn_cost_worker =
+            TurnCostWorker::spawn(Arc::clone(&config), Arc::clone(&auth_manager));
        thread_manager
            .plugins_manager()
            .set_analytics_events_client(analytics_events_client.clone());
@@ -491,6 +495,7 @@ impl MessageProcessor {
            state_db.clone(),
            log_db,
            Arc::clone(&skills_watcher),
+            turn_cost_worker.as_ref().map(TurnCostWorker::handle),
            config_warnings,
        );
        let turn_processor = TurnRequestProcessor::new(
@@ -506,6 +511,7 @@ impl MessageProcessor {
            thread_watch_manager,
            thread_list_state_permit,
            Arc::clone(&skills_watcher),
+            turn_cost_worker.as_ref().map(TurnCostWorker::handle),
        );
        if matches!(plugin_startup_tasks, crate::PluginStartupTasks::Start) {
            // Keep plugin startup warmups aligned at app-server startup.
@@ -546,6 +552,7 @@ impl MessageProcessor {
        Self {
            outgoing,
            models_refresh_worker,
+            turn_cost_worker,
            skills_watcher,
            account_processor,
            apps_processor,
@@ -755,6 +762,9 @@ impl MessageProcessor {

    pub(crate) async fn drain_background_tasks(&self) {
        self.models_refresh_worker.shutdown();
+        if let Some(worker) = &self.turn_cost_worker {
+            worker.shutdown();
+        }
        self.thread_processor.drain_background_tasks().await;
    }

@@ -17,6 +17,7 @@ pub(super) struct ListenerTaskContext {
    pub(super) fallback_model_provider: String,
    pub(super) codex_home: PathBuf,
    pub(super) skills_watcher: Arc<SkillsWatcher>,
+    pub(super) turn_cost_worker: Option<crate::turn_cost_worker::TurnCostWorkerHandle>,
 }

 struct UnloadingState {
@@ -274,6 +275,7 @@ pub(super) async fn ensure_listener_task_running(
        thread_list_state_permit,
        fallback_model_provider,
        codex_home,
+        turn_cost_worker,
        ..
    } = listener_task_context;
    let outgoing_for_task = Arc::clone(&outgoing);
@@ -311,6 +313,12 @@ pub(super) async fn ensure_listener_task_running(
                        }
                    };

+                    if let Some(worker) = &turn_cost_worker {
+                        worker.observe_event(conversation_id, &event, || {
+                            conversation.session_telemetry()
+                        });
+                    }
+
                    // Track the event before emitting any typed translations
                    // so thread-local state such as raw event opt-in stays
                    // synchronized with the conversation.
@@ -481,6 +481,7 @@ pub(crate) struct ThreadRequestProcessor {
    pub(super) log_db: Option<LogDbLayer>,
    pub(super) background_tasks: TaskTracker,
    pub(super) skills_watcher: Arc<SkillsWatcher>,
+    pub(super) turn_cost_worker: Option<crate::turn_cost_worker::TurnCostWorkerHandle>,
    pub(super) initial_config_warnings: Arc<Vec<ConfigWarningNotification>>,
 }

@@ -513,6 +514,7 @@ impl ThreadRequestProcessor {
        state_db: Option<StateDbHandle>,
        log_db: Option<LogDbLayer>,
        skills_watcher: Arc<SkillsWatcher>,
+        turn_cost_worker: Option<crate::turn_cost_worker::TurnCostWorkerHandle>,
        initial_config_warnings: Vec<ConfigWarningNotification>,
    ) -> Self {
        Self {
@@ -532,6 +534,7 @@ impl ThreadRequestProcessor {
            log_db,
            background_tasks: TaskTracker::new(),
            skills_watcher,
+            turn_cost_worker,
            initial_config_warnings: Arc::new(initial_config_warnings),
        }
    }
@@ -1058,6 +1061,7 @@ impl ThreadRequestProcessor {
            fallback_model_provider: self.config.model_provider_id.clone(),
            codex_home: self.config.codex_home.to_path_buf(),
            skills_watcher: Arc::clone(&self.skills_watcher),
+            turn_cost_worker: self.turn_cost_worker.clone(),
        }
    }

@@ -1188,6 +1192,7 @@ impl ThreadRequestProcessor {
            fallback_model_provider: self.config.model_provider_id.clone(),
            codex_home: self.config.codex_home.to_path_buf(),
            skills_watcher: Arc::clone(&self.skills_watcher),
+            turn_cost_worker: self.turn_cost_worker.clone(),
        };
        let request_trace = request_context.request_trace();
        let config_manager = self.config_manager.clone();
@@ -100,6 +100,7 @@ pub(crate) struct TurnRequestProcessor {
    thread_watch_manager: ThreadWatchManager,
    thread_list_state_permit: Arc<Semaphore>,
    skills_watcher: Arc<SkillsWatcher>,
+    turn_cost_worker: Option<crate::turn_cost_worker::TurnCostWorkerHandle>,
 }

 fn map_additional_context(
@@ -155,6 +156,7 @@ impl TurnRequestProcessor {
        thread_watch_manager: ThreadWatchManager,
        thread_list_state_permit: Arc<Semaphore>,
        skills_watcher: Arc<SkillsWatcher>,
+        turn_cost_worker: Option<crate::turn_cost_worker::TurnCostWorkerHandle>,
    ) -> Self {
        let agent_runner = AgentRunner::new(Arc::downgrade(&thread_manager));
        Self {
@@ -171,6 +173,7 @@ impl TurnRequestProcessor {
            thread_watch_manager,
            thread_list_state_permit,
            skills_watcher,
+            turn_cost_worker,
        }
    }

@@ -1516,6 +1519,7 @@ impl TurnRequestProcessor {
            fallback_model_provider: self.config.model_provider_id.clone(),
            codex_home: self.config.codex_home.to_path_buf(),
            skills_watcher: Arc::clone(&self.skills_watcher),
+            turn_cost_worker: self.turn_cost_worker.clone(),
        }
    }

@@ -0,0 +1,454 @@
+use codex_backend_client::ApiKeyTurnCost;
+use codex_backend_client::ApiKeyTurnCostStatus;
+use codex_backend_client::Client as BackendClient;
+use codex_config::types::OtelExporterKind;
+use codex_core::config::Config;
+use codex_login::AuthManager;
+use codex_login::CodexAuth;
+use codex_otel::SessionTelemetry;
+use codex_protocol::ThreadId;
+use codex_protocol::auth::AuthMode;
+use codex_protocol::protocol::Event;
+use codex_protocol::protocol::EventMsg;
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::Duration;
+use std::time::Instant;
+use tokio::sync::mpsc;
+use tokio::task::JoinHandle;
+use tokio_util::sync::CancellationToken;
+use tracing::warn;
+
+const POLL_INTERVAL: Duration = Duration::from_secs(150);
+const REQUEST_TIMEOUT: Duration = Duration::from_secs(15);
+const OBSERVATION_CHANNEL_CAPACITY: usize = 16_384;
+const MAX_TRACKED_TURNS: usize = 4_096;
+const MAX_QUERY_TURNS: usize = 100;
+const MAX_STALLED_POLL_ATTEMPTS: u8 = 5;
+
+pub(crate) struct TurnCostWorker {
+    handle: TurnCostWorkerHandle,
+    shutdown: CancellationToken,
+    _task: JoinHandle<()>,
+}
+
+#[derive(Clone)]
+pub(crate) struct TurnCostWorkerHandle {
+    sender: mpsc::Sender<TurnCostObservation>,
+    auth_manager: Arc<AuthManager>,
+}
+
+enum TurnCostObservationKind {
+    Started {
+        session_telemetry: Box<SessionTelemetry>,
+    },
+    ResponseCompleted,
+    Finished {
+        interrupted: bool,
+    },
+}
+
+struct TurnCostObservation {
+    thread_id: ThreadId,
+    turn_id: String,
+    kind: TurnCostObservationKind,
+}
+
+#[derive(Clone, Copy, PartialEq, Eq)]
+enum TurnCostStatus {
+    Running,
+    Completed,
+    Interrupted,
+}
+
+struct TurnCostEntry {
+    thread_id: ThreadId,
+    session_telemetry: SessionTelemetry,
+    expected_response_count: u64,
+    status: TurnCostStatus,
+    next_poll_at: Instant,
+    attempt_count: u8,
+}
+
+struct WorkerRuntime {
+    config: Arc<Config>,
+    auth_manager: Arc<AuthManager>,
+    turns: HashMap<String, TurnCostEntry>,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+enum BackendAvailability {
+    AwaitingAuthChange,
+    RetryProbe,
+    Ready,
+    Disabled,
+}
+
+impl TurnCostWorker {
+    pub(crate) fn spawn(config: Arc<Config>, auth_manager: Arc<AuthManager>) -> Option<Self> {
+        if !matches!(
+            config.otel.exporter,
+            OtelExporterKind::OtlpHttp { .. } | OtelExporterKind::OtlpGrpc { .. }
+        ) || !config.model_provider.is_openai()
+        {
+            return None;
+        }
+        let (sender, receiver) = mpsc::channel(OBSERVATION_CHANNEL_CAPACITY);
+        let shutdown = CancellationToken::new();
+        let runtime = WorkerRuntime {
+            config: Arc::clone(&config),
+            auth_manager: Arc::clone(&auth_manager),
+            turns: HashMap::new(),
+        };
+        let worker_shutdown = shutdown.clone();
+        let task = tokio::spawn(async move {
+            runtime.run(receiver, worker_shutdown).await;
+        });
+        Some(Self {
+            handle: TurnCostWorkerHandle {
+                sender,
+                auth_manager,
+            },
+            shutdown,
+            _task: task,
+        })
+    }
+
+    pub(crate) fn handle(&self) -> TurnCostWorkerHandle {
+        self.handle.clone()
+    }
+
+    pub(crate) fn shutdown(&self) {
+        self.shutdown.cancel();
+    }
+}
+
+impl Drop for TurnCostWorker {
+    fn drop(&mut self) {
+        self.shutdown();
+    }
+}
+
+impl TurnCostWorkerHandle {
+    pub(crate) fn observe_event(
+        &self,
+        thread_id: ThreadId,
+        event: &Event,
+        session_telemetry: impl FnOnce() -> SessionTelemetry,
+    ) {
+        let Some(auth) = self.auth_manager.auth_cached() else {
+            return;
+        };
+        if !auth.is_api_key_auth() {
+            return;
+        }
+        let kind = match &event.msg {
+            EventMsg::TurnStarted(_) => TurnCostObservationKind::Started {
+                session_telemetry: Box::new(session_telemetry()),
+            },
+            EventMsg::RawResponseCompleted(_) => TurnCostObservationKind::ResponseCompleted,
+            EventMsg::TurnComplete(_) => TurnCostObservationKind::Finished { interrupted: false },
+            EventMsg::TurnAborted(_) => TurnCostObservationKind::Finished { interrupted: true },
+            _ => return,
+        };
+        let _ = self.sender.try_send(TurnCostObservation {
+            thread_id,
+            turn_id: event.id.clone(),
+            kind,
+        });
+    }
+}
+
+impl WorkerRuntime {
+    async fn run(self, receiver: mpsc::Receiver<TurnCostObservation>, shutdown: CancellationToken) {
+        let auth_changes = self.auth_manager.auth_change_receiver();
+        let backend_availability = self.probe_backend().await;
+        self.run_with_backend_availability(receiver, shutdown, auth_changes, backend_availability)
+            .await;
+    }
+
+    async fn run_with_backend_availability(
+        mut self,
+        mut receiver: mpsc::Receiver<TurnCostObservation>,
+        shutdown: CancellationToken,
+        mut auth_changes: tokio::sync::watch::Receiver<u64>,
+        mut backend_availability: BackendAvailability,
+    ) {
+        let mut ticker = tokio::time::interval(POLL_INTERVAL);
+        ticker.tick().await;
+        loop {
+            tokio::select! {
+                biased;
+                _ = shutdown.cancelled() => break,
+                changed = auth_changes.changed() => {
+                    if changed.is_err() {
+                        break;
+                    }
+                    self.turns.clear();
+                    while receiver.try_recv().is_ok() {}
+                    backend_availability = self.probe_backend().await;
+                }
+                observation = receiver.recv() => {
+                    let Some(observation) = observation else {
+                        break;
+                    };
+                    if !matches!(
+                        backend_availability,
+                        BackendAvailability::Ready | BackendAvailability::RetryProbe
+                    ) {
+                        continue;
+                    }
+                    self.record_observation(observation);
+                }
+                _ = ticker.tick() => {
+                    match backend_availability {
+                        BackendAvailability::Ready => self.poll_due().await,
+                        BackendAvailability::RetryProbe => {
+                            backend_availability = self.probe_backend().await;
+                        }
+                        BackendAvailability::AwaitingAuthChange
+                        | BackendAvailability::Disabled => {}
+                    }
+                }
+            }
+        }
+    }
+
+    async fn probe_backend(&self) -> BackendAvailability {
+        let Some(auth) = self.auth_manager.auth().await else {
+            return BackendAvailability::AwaitingAuthChange;
+        };
+        if !auth.is_api_key_auth() {
+            return BackendAvailability::AwaitingAuthChange;
+        }
+        let provider = match self
+            .config
+            .model_provider
+            .to_api_provider(Some(AuthMode::ApiKey))
+        {
+            Ok(provider) => provider,
+            Err(error) => {
+                warn!("failed to resolve OpenAI API-key provider headers: {error}");
+                return BackendAvailability::RetryProbe;
+            }
+        };
+        let client = BackendClient::from_auth(
+            self.config.chatgpt_base_url.clone(),
+            &auth,
+            self.config.http_client_factory(),
+        );
+        let probe_turn_ids = [uuid::Uuid::new_v4().to_string()];
+        match tokio::time::timeout(
+            REQUEST_TIMEOUT,
+            client.query_api_key_turn_costs(&probe_turn_ids, &provider.headers),
+        )
+        .await
+        {
+            Ok(Ok(_)) => BackendAvailability::Ready,
+            Ok(Err(error)) => match error.status().map(|status| status.as_u16()) {
+                Some(401 | 403) => {
+                    tracing::debug!(
+                        "turn cost worker waiting for auth change after backend availability check: {error}"
+                    );
+                    BackendAvailability::AwaitingAuthChange
+                }
+                Some(429) => BackendAvailability::RetryProbe,
+                Some(400..=499) => {
+                    tracing::debug!(
+                        "turn cost worker disabled by backend availability check: {error}"
+                    );
+                    BackendAvailability::Disabled
+                }
+                _ => {
+                    tracing::debug!(
+                        "turn cost worker backend availability check failed temporarily: {error}"
+                    );
+                    BackendAvailability::RetryProbe
+                }
+            },
+            Err(_) => {
+                tracing::debug!(
+                    "turn cost worker backend availability check timed out; will retry"
+                );
+                BackendAvailability::RetryProbe
+            }
+        }
+    }
+
+    fn record_observation(&mut self, observation: TurnCostObservation) {
+        match observation.kind {
+            TurnCostObservationKind::Started { session_telemetry } => {
+                if self.turns.len() < MAX_TRACKED_TURNS {
+                    self.turns
+                        .entry(observation.turn_id)
+                        .or_insert(TurnCostEntry {
+                            thread_id: observation.thread_id,
+                            session_telemetry: *session_telemetry,
+                            expected_response_count: 0,
+                            status: TurnCostStatus::Running,
+                            next_poll_at: Instant::now(),
+                            attempt_count: 0,
+                        });
+                }
+            }
+            TurnCostObservationKind::ResponseCompleted => {
+                if let Some(entry) = self.turns.get_mut(&observation.turn_id)
+                    && entry.status == TurnCostStatus::Running
+                {
+                    entry.expected_response_count = entry.expected_response_count.saturating_add(1);
+                }
+            }
+            TurnCostObservationKind::Finished { interrupted } => {
+                let Some(entry) = self.turns.get_mut(&observation.turn_id) else {
+                    return;
+                };
+                if entry.status != TurnCostStatus::Running {
+                    return;
+                }
+                entry.status = if interrupted {
+                    TurnCostStatus::Interrupted
+                } else {
+                    TurnCostStatus::Completed
+                };
+                entry.next_poll_at = Instant::now();
+            }
+        }
+    }
+
+    async fn poll_due(&mut self) {
+        let Some(auth) = self.auth_manager.auth().await else {
+            return;
+        };
+        if !auth.is_api_key_auth() {
+            return;
+        }
+        let now = Instant::now();
+        let due_turn_ids: Vec<String> = self
+            .turns
+            .iter()
+            .filter(|(_, entry)| {
+                entry.status != TurnCostStatus::Running && entry.next_poll_at <= now
+            })
+            .take(MAX_QUERY_TURNS)
+            .map(|(turn_id, _)| turn_id.clone())
+            .collect();
+        if !due_turn_ids.is_empty() {
+            self.poll_api_key_entries(&due_turn_ids, &auth).await;
+        }
+    }
+
+    async fn poll_api_key_entries(&mut self, turn_ids: &[String], auth: &CodexAuth) {
+        let provider = match self
+            .config
+            .model_provider
+            .to_api_provider(Some(AuthMode::ApiKey))
+        {
+            Ok(provider) => provider,
+            Err(error) => {
+                warn!("failed to resolve OpenAI API-key provider headers: {error}");
+                self.retry_entries(turn_ids);
+                return;
+            }
+        };
+        let client = BackendClient::from_auth(
+            self.config.chatgpt_base_url.clone(),
+            auth,
+            self.config.http_client_factory(),
+        );
+        let costs = match tokio::time::timeout(
+            REQUEST_TIMEOUT,
+            client.query_api_key_turn_costs(turn_ids, &provider.headers),
+        )
+        .await
+        {
+            Ok(Ok(costs)) => costs,
+            Ok(Err(error)) => {
+                warn!("failed to query OpenAI API-key turn costs: {error}");
+                self.retry_entries(turn_ids);
+                return;
+            }
+            Err(_) => {
+                warn!("timed out querying OpenAI API-key turn costs");
+                self.retry_entries(turn_ids);
+                return;
+            }
+        };
+        let costs_by_turn: HashMap<String, ApiKeyTurnCost> = costs
+            .into_iter()
+            .map(|cost| (cost.turn_id.clone(), cost))
+            .collect();
+        for turn_id in turn_ids {
+            let Some(cost) = costs_by_turn.get(turn_id) else {
+                self.retry_entry(turn_id);
+                continue;
+            };
+            self.process_api_key_cost(turn_id, cost);
+        }
+    }
+
+    fn process_api_key_cost(&mut self, turn_id: &str, cost: &ApiKeyTurnCost) {
+        if cost.status != ApiKeyTurnCostStatus::Priced {
+            self.retry_entry(turn_id);
+            return;
+        }
+        let response_count = cost
+            .responses
+            .as_ref()
+            .map(|responses| responses.len() as u64)
+            .or(cost.event_count);
+        let (Some(total_usd), Some(response_count)) = (cost.total_usd.as_deref(), response_count)
+        else {
+            self.retry_entry(turn_id);
+            return;
+        };
+        let Some(entry) = self.turns.get(turn_id) else {
+            return;
+        };
+        if response_count < entry.expected_response_count {
+            self.retry_entry(turn_id);
+            return;
+        }
+        let mut session_telemetry = entry.session_telemetry.clone();
+        if let Some(model) = cost.model.as_deref() {
+            session_telemetry = session_telemetry.with_model(model, model);
+        }
+        let Some(entry) = self.turns.remove(turn_id) else {
+            return;
+        };
+        session_telemetry.record_turn_cost(
+            turn_id,
+            total_usd,
+            entry.status == TurnCostStatus::Interrupted,
+            cost.speed.as_deref(),
+            cost.reasoning_effort.as_deref(),
+        );
+    }
+
+    fn retry_entries(&mut self, turn_ids: &[String]) {
+        for turn_id in turn_ids {
+            self.retry_entry(turn_id);
+        }
+    }
+
+    fn retry_entry(&mut self, turn_id: &str) {
+        let Some(entry) = self.turns.get_mut(turn_id) else {
+            return;
+        };
+        entry.attempt_count = entry.attempt_count.saturating_add(1);
+        if entry.attempt_count >= MAX_STALLED_POLL_ATTEMPTS {
+            warn!(
+                thread_id = %entry.thread_id,
+                turn_id,
+                attempts = MAX_STALLED_POLL_ATTEMPTS,
+                "dropping turn cost event after repeated unsuccessful polls"
+            );
+            self.turns.remove(turn_id);
+            return;
+        }
+        entry.next_poll_at = Instant::now() + POLL_INTERVAL;
+    }
+}
+
+#[cfg(test)]
+#[path = "turn_cost_worker_tests.rs"]
+mod tests;
@@ -0,0 +1,241 @@
+use super::*;
+use codex_backend_client::ApiKeyResponseCost;
+use codex_core::config::ConfigBuilder;
+use codex_login::AuthCredentialsStoreMode;
+use codex_login::AuthKeyringBackendKind;
+use codex_login::login_with_api_key;
+use codex_otel::TelemetryAuthMode;
+use codex_protocol::protocol::SessionSource;
+use pretty_assertions::assert_eq;
+use tempfile::TempDir;
+use tokio::time::timeout;
+use wiremock::Mock;
+use wiremock::MockServer;
+use wiremock::ResponseTemplate;
+use wiremock::matchers::method;
+use wiremock::matchers::path;
+
+const TURN_COST_PATH: &str = "/v1/analytics/codex/turn-costs";
+
+#[tokio::test]
+async fn worker_waits_for_late_api_key_login() {
+    let server = MockServer::start().await;
+    Mock::given(method("POST"))
+        .and(path(TURN_COST_PATH))
+        .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
+            "turns": []
+        })))
+        .expect(1)
+        .mount(&server)
+        .await;
+    let auth_home = TempDir::new().expect("temporary auth home");
+    let auth_manager = Arc::new(
+        AuthManager::new(
+            auth_home.path().to_path_buf(),
+            /*enable_codex_api_key_env*/ false,
+            AuthCredentialsStoreMode::File,
+            /*forced_chatgpt_workspace_id*/ None,
+            /*chatgpt_base_url*/ None,
+            AuthKeyringBackendKind::default(),
+            codex_login::test_support::transport_default_auth_route_config(),
+        )
+        .await,
+    );
+    let runtime = test_runtime(&server, Arc::clone(&auth_manager)).await;
+    let (_sender, receiver) = mpsc::channel(OBSERVATION_CHANNEL_CAPACITY);
+    let shutdown = CancellationToken::new();
+    let mut task = tokio::spawn(runtime.run(receiver, shutdown.clone()));
+
+    assert!(
+        timeout(Duration::from_millis(/*millis*/ 50), &mut task)
+            .await
+            .is_err(),
+        "worker exited while waiting for login"
+    );
+    login_with_api_key(
+        auth_home.path(),
+        "sk-test",
+        AuthCredentialsStoreMode::File,
+        AuthKeyringBackendKind::default(),
+    )
+    .expect("write API-key auth");
+    assert!(auth_manager.reload().await);
+    wait_for_request_count(&server, /*expected*/ 1).await;
+
+    shutdown.cancel();
+    task.await.expect("worker task");
+    server.verify().await;
+}
+
+#[tokio::test]
+async fn transient_probe_failure_keeps_worker_alive() {
+    let server = MockServer::start().await;
+    Mock::given(method("POST"))
+        .and(path(TURN_COST_PATH))
+        .respond_with(ResponseTemplate::new(500).set_body_string("temporary failure"))
+        .expect(1)
+        .mount(&server)
+        .await;
+    let auth_manager = AuthManager::from_auth_for_testing(CodexAuth::from_api_key("sk-test"));
+    let runtime = test_runtime(&server, Arc::clone(&auth_manager)).await;
+    let backend_availability = runtime.probe_backend().await;
+    assert_eq!(backend_availability, BackendAvailability::RetryProbe);
+    let (_sender, receiver) = mpsc::channel(OBSERVATION_CHANNEL_CAPACITY);
+    let shutdown = CancellationToken::new();
+    let auth_changes = auth_manager.auth_change_receiver();
+    let mut task = tokio::spawn(runtime.run_with_backend_availability(
+        receiver,
+        shutdown.clone(),
+        auth_changes,
+        backend_availability,
+    ));
+
+    assert!(
+        timeout(Duration::from_millis(/*millis*/ 50), &mut task)
+            .await
+            .is_err(),
+        "worker exited after a transient probe failure"
+    );
+
+    shutdown.cancel();
+    task.await.expect("worker task");
+    server.verify().await;
+}
+
+#[tokio::test]
+async fn priced_cost_uses_telemetry_captured_before_thread_removal() {
+    let server = MockServer::start().await;
+    let auth_manager = AuthManager::from_auth_for_testing(CodexAuth::from_api_key("sk-test"));
+    let mut runtime = test_runtime(&server, auth_manager).await;
+    let thread_id = ThreadId::new();
+    let turn_id = "turn-1";
+
+    runtime.record_observation(TurnCostObservation {
+        thread_id,
+        turn_id: turn_id.to_string(),
+        kind: TurnCostObservationKind::Started {
+            session_telemetry: Box::new(test_session_telemetry(thread_id)),
+        },
+    });
+    runtime.record_observation(TurnCostObservation {
+        thread_id,
+        turn_id: turn_id.to_string(),
+        kind: TurnCostObservationKind::ResponseCompleted,
+    });
+    runtime.record_observation(TurnCostObservation {
+        thread_id,
+        turn_id: turn_id.to_string(),
+        kind: TurnCostObservationKind::Finished { interrupted: false },
+    });
+
+    runtime.process_api_key_cost(
+        turn_id,
+        &ApiKeyTurnCost {
+            turn_id: turn_id.to_string(),
+            status: ApiKeyTurnCostStatus::Priced,
+            total_usd: Some("1.25".to_string()),
+            event_count: Some(1),
+            responses: None,
+            model: Some("gpt-5.6".to_string()),
+            speed: Some("fast".to_string()),
+            reasoning_effort: Some("high".to_string()),
+        },
+    );
+
+    assert_eq!(runtime.turns.len(), 0);
+}
+
+#[tokio::test]
+async fn priced_cost_waits_for_every_response_when_response_costs_are_available() {
+    let server = MockServer::start().await;
+    let auth_manager = AuthManager::from_auth_for_testing(CodexAuth::from_api_key("sk-test"));
+    let mut runtime = test_runtime(&server, auth_manager).await;
+    let thread_id = ThreadId::new();
+    let turn_id = "turn-1";
+
+    runtime.turns.insert(
+        turn_id.to_string(),
+        TurnCostEntry {
+            thread_id,
+            session_telemetry: test_session_telemetry(thread_id),
+            expected_response_count: 2,
+            status: TurnCostStatus::Completed,
+            next_poll_at: Instant::now(),
+            attempt_count: 0,
+        },
+    );
+
+    let mut cost = ApiKeyTurnCost {
+        turn_id: turn_id.to_string(),
+        status: ApiKeyTurnCostStatus::Priced,
+        total_usd: Some("1.25".to_string()),
+        event_count: Some(2),
+        responses: Some(vec![ApiKeyResponseCost {
+            response_id: "resp-one".to_string(),
+            total_usd: "1.25".to_string(),
+        }]),
+        model: Some("gpt-5.6".to_string()),
+        speed: Some("fast".to_string()),
+        reasoning_effort: Some("high".to_string()),
+    };
+    runtime.process_api_key_cost(turn_id, &cost);
+
+    let entry = runtime.turns.get(turn_id).expect("turn remains tracked");
+    assert_eq!(entry.attempt_count, 1);
+
+    cost.event_count = None;
+    cost.responses
+        .as_mut()
+        .expect("response costs")
+        .push(ApiKeyResponseCost {
+            response_id: "resp-two".to_string(),
+            total_usd: "0.50".to_string(),
+        });
+    runtime.process_api_key_cost(turn_id, &cost);
+
+    assert_eq!(runtime.turns.len(), 0);
+}
+
+async fn test_runtime(server: &MockServer, auth_manager: Arc<AuthManager>) -> WorkerRuntime {
+    let codex_home = TempDir::new().expect("temporary Codex home");
+    let mut config = ConfigBuilder::default()
+        .codex_home(codex_home.path().to_path_buf())
+        .build()
+        .await
+        .expect("test config");
+    config.chatgpt_base_url = server.uri();
+    WorkerRuntime {
+        config: Arc::new(config),
+        auth_manager,
+        turns: HashMap::new(),
+    }
+}
+
+fn test_session_telemetry(thread_id: ThreadId) -> SessionTelemetry {
+    SessionTelemetry::new(
+        thread_id,
+        "gpt-5.6",
+        "gpt-5.6",
+        /*account_id*/ None,
+        /*account_email*/ None,
+        Some(TelemetryAuthMode::ApiKey),
+        "test".to_string(),
+        /*log_user_prompts*/ false,
+        "test".to_string(),
+        SessionSource::Cli,
+    )
+}
+
+async fn wait_for_request_count(server: &MockServer, expected: usize) {
+    timeout(Duration::from_secs(/*secs*/ 15), async {
+        loop {
+            let requests = server.received_requests().await.unwrap_or_default();
+            if requests.len() >= expected {
+                break;
+            }
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .expect("timed out waiting for turn-cost request");
+}
@@ -36,6 +36,7 @@ use std::fmt;

 mod rate_limit_resets;
 mod thread_usage;
+pub(crate) mod turn_usage;

 pub use thread_usage::ThreadUsage;
 pub use thread_usage::ThreadUsageBreakdownGroup;
@@ -0,0 +1,188 @@
+use super::Client;
+use super::RequestError;
+use http::HeaderMap;
+use http::Method;
+use http::header::CONTENT_TYPE;
+use http::header::HeaderValue;
+use serde::Deserialize;
+use serde::Serialize;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum ApiKeyTurnCostStatus {
+    Pending,
+    Priced,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
+pub struct ApiKeyResponseCost {
+    pub response_id: String,
+    pub total_usd: String,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
+pub struct ApiKeyTurnCost {
+    pub turn_id: String,
+    pub status: ApiKeyTurnCostStatus,
+    pub total_usd: Option<String>,
+    pub event_count: Option<u64>,
+    pub responses: Option<Vec<ApiKeyResponseCost>>,
+    pub model: Option<String>,
+    pub speed: Option<String>,
+    pub reasoning_effort: Option<String>,
+}
+
+#[derive(Serialize)]
+struct ApiKeyTurnCostsRequest<'a> {
+    turn_ids: &'a [String],
+}
+
+#[derive(Deserialize)]
+struct ApiKeyTurnCostsResponse {
+    turns: Vec<ApiKeyTurnCost>,
+}
+
+impl Client {
+    pub async fn query_api_key_turn_costs(
+        &self,
+        turn_ids: &[String],
+        provider_headers: &HeaderMap,
+    ) -> Result<Vec<ApiKeyTurnCost>, RequestError> {
+        let mut url =
+            url::Url::parse(&self.base_url).map_err(|error| RequestError::Other(error.into()))?;
+        let analytics_host = match url.host_str() {
+            Some("chatgpt.com" | "chat.openai.com") => Some("api.chatgpt.com"),
+            Some("chatgpt-staging.com") => Some("api.chatgpt-staging.com"),
+            _ => None,
+        };
+        if let Some(host) = analytics_host {
+            url.set_host(Some(host))
+                .map_err(|error| RequestError::Other(error.into()))?;
+        }
+        url.set_path("/v1/analytics/codex/turn-costs");
+        url.set_query(None);
+        url.set_fragment(None);
+        let url = url.to_string();
+        let mut headers = self.headers();
+        for header_name in ["openai-organization", "openai-project"] {
+            if let Some(value) = provider_headers.get(header_name) {
+                headers.insert(header_name, value.clone());
+            }
+        }
+        let request = self
+            .request(Method::POST, &url)
+            .headers(headers)
+            .header(CONTENT_TYPE, HeaderValue::from_static("application/json"))
+            .json(&ApiKeyTurnCostsRequest { turn_ids });
+        let (body, content_type) = self.exec_request_detailed(request, "POST", &url).await?;
+        let response: ApiKeyTurnCostsResponse = self
+            .decode_json(&url, &content_type, &body)
+            .map_err(RequestError::Other)?;
+        Ok(response.turns)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::ApiKeyResponseCost;
+    use super::ApiKeyTurnCost;
+    use super::ApiKeyTurnCostStatus;
+    use super::Client;
+    use codex_http_client::HttpClientFactory;
+    use codex_http_client::OutboundProxyPolicy;
+    use codex_login::CodexAuth;
+    use http::HeaderMap;
+    use http::HeaderValue;
+    use pretty_assertions::assert_eq;
+    use wiremock::Mock;
+    use wiremock::MockServer;
+    use wiremock::ResponseTemplate;
+    use wiremock::matchers::body_json;
+    use wiremock::matchers::header;
+    use wiremock::matchers::method;
+    use wiremock::matchers::path;
+
+    #[tokio::test]
+    async fn api_key_turn_cost_queries_use_api_key_auth_and_provider_scope() {
+        let server = MockServer::start().await;
+        Mock::given(method("POST"))
+            .and(path("/v1/analytics/codex/turn-costs"))
+            .and(header("authorization", "Bearer sk-test"))
+            .and(header("openai-organization", "org-test"))
+            .and(header("openai-project", "project-test"))
+            .and(body_json(serde_json::json!({
+                "turn_ids": ["turn-priced", "turn-response"]
+            })))
+            .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
+                "turns": [
+                    {
+                        "turn_id": "turn-priced",
+                        "status": "priced",
+                        "total_usd": "1.2500000001",
+                        "event_count": 2,
+                        "model": "gpt-5.6",
+                        "speed": "fast",
+                        "reasoning_effort": "high"
+                    },
+                    {
+                        "turn_id": "turn-response",
+                        "status": "priced",
+                        "total_usd": "0.5000000000",
+                        "responses": [{
+                            "response_id": "resp-one",
+                            "total_usd": "0.5000000000"
+                        }]
+                    }
+                ]
+            })))
+            .expect(1)
+            .mount(&server)
+            .await;
+
+        let auth = CodexAuth::from_api_key("sk-test");
+        let client = Client::from_auth(
+            format!("{}/backend-api", server.uri()),
+            &auth,
+            HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault),
+        );
+        let mut provider_headers = HeaderMap::new();
+        provider_headers.insert("openai-organization", HeaderValue::from_static("org-test"));
+        provider_headers.insert("openai-project", HeaderValue::from_static("project-test"));
+        let costs = client
+            .query_api_key_turn_costs(
+                &["turn-priced".to_string(), "turn-response".to_string()],
+                &provider_headers,
+            )
+            .await
+            .expect("query API key turn costs");
+
+        assert_eq!(
+            costs,
+            vec![
+                ApiKeyTurnCost {
+                    turn_id: "turn-priced".to_string(),
+                    status: ApiKeyTurnCostStatus::Priced,
+                    total_usd: Some("1.2500000001".to_string()),
+                    event_count: Some(2),
+                    responses: None,
+                    model: Some("gpt-5.6".to_string()),
+                    speed: Some("fast".to_string()),
+                    reasoning_effort: Some("high".to_string()),
+                },
+                ApiKeyTurnCost {
+                    turn_id: "turn-response".to_string(),
+                    status: ApiKeyTurnCostStatus::Priced,
+                    total_usd: Some("0.5000000000".to_string()),
+                    event_count: None,
+                    responses: Some(vec![ApiKeyResponseCost {
+                        response_id: "resp-one".to_string(),
+                        total_usd: "0.5000000000".to_string(),
+                    }]),
+                    model: None,
+                    speed: None,
+                    reasoning_effort: None,
+                },
+            ]
+        );
+    }
+}
@@ -6,6 +6,9 @@ pub use client::Client;
 pub use client::RequestError;
 pub use client::ThreadUsage;
 pub use client::ThreadUsageBreakdownGroup;
+pub use client::turn_usage::ApiKeyResponseCost;
+pub use client::turn_usage::ApiKeyTurnCost;
+pub use client::turn_usage::ApiKeyTurnCostStatus;
 pub use types::AccountEntry;
 pub use types::AccountsCheckResponse;
 pub use types::CodeTaskDetailsResponse;
@@ -260,6 +260,25 @@ impl SessionTelemetry {
        );
    }

+    pub fn record_turn_cost(
+        &self,
+        turn_id: &str,
+        estimated_usd: &str,
+        interrupted: bool,
+        speed: Option<&str>,
+        reasoning_effort: Option<&str>,
+    ) {
+        log_event!(
+            self,
+            event.name = "codex.turn_cost",
+            turn.id = turn_id,
+            usage.estimated_usd = estimated_usd,
+            turn.interrupted = interrupted,
+            speed = speed,
+            reasoning_effort = reasoning_effort,
+        );
+    }
+
    /// Records the moment a plugin or connector install elicitation is dispatched.
    pub fn record_plugin_install_elicitation_sent(
        &self,