LLM Client and Responses API

源码版本rust-v0.145.0

codex routes all model calls through the Responses API (/v1/responses). This layer is implemented in core/src/client.rs; its job is to take prompt + model config + auth, build the request, talk over WebSocket or HTTP, parse the SSE stream into ResponseEvent, and feed token usage, reasoning summary, and tool-call deltas to the upstream loop. The same ModelClient also handles remote compaction requests (compact_conversation_history) and auth prewarm, plus a standalone responses-api-proxy debug proxy.

Responsibilities

1. Session-level client: ModelClient::new takes 13 parameters including auth manager, provider info, thread id, HTTP client factory; it folds these cross-turn stable states into Arc<ModelClientState>. codex-rs/core/src/client.rs:413-460
2. Turn-level session: ModelClientSession is created by new_session, holding a WebsocketSession cache and turn_state: Arc<OnceLock<String>> to reuse the same turn's sticky-routing token across multiple requests. codex-rs/core/src/client.rs:480-486
3. Streaming entry: ModelClientSession::stream dispatches by WireApi; Responses API goes WebSocket (if available) → falls back to HTTP on failure, all paths return ResponseStream. codex-rs/core/src/client.rs:1792-1843
4. Remote compaction: the same ModelClient hosts compact_conversation_history, which lets the server fold history via the /responses/compact endpoint and returns a new list of ResponseItem. codex-rs/core/src/client.rs:538-547

Design motivation

The Responses API differs from traditional Chat Completions: it natively supports reasoning summary, server-side compaction, and streaming tool-call deltas. codex chose to route all model calls through /v1/responses (WireApi::Responses is currently the only variant) in exchange for a unified ResponseEvent stream-handling logic. WebSocket is an optional transport that, compared to HTTP SSE, lets codex send multiple incremental requests on the same connection (reusing context during multi-step reasoning within a turn), and uses the x-codex-turn-state sticky-routing token to guarantee that the same turn routes to the same backend instance; but WS is complex, so force_http_fallback permanently disables WS and clears the cache, and on error the whole session falls back to HTTP.

The two-layer split between ModelClient and ModelClientSession: the former is session-scoped (one copy for the entire Codex session), the latter is turn-scoped (one new_session() per turn) and must not be reused across turns — otherwise the turn_state token would pollute routing across turns. current_client_setup puts auth resolution under a single lock to ensure prewarm (background pre-connect) and the actual turn see a consistent auth / provider state. compact_conversation_history reusing the same client is a subtle point: it hits /responses/compact rather than /responses, but transport headers, auth headers, and telemetry all use the same builder, so compaction requests look indistinguishable from normal turns in monitoring.

Key files

codex-rs/core/src/client.rs:253-272 — the ModelClient and ModelClientSession struct definitions; the latter's doc requires calling new_session once per turn.codex-rs/core/src/client.rs:509-528 — force_http_fallback, permanently disabling WS + clearing the cache + emitting telemetry.codex-rs/core/src/client.rs:944-962 — current_client_setup, centralizing auth + provider resolution.codex-rs/codex-api/src/common.rs:74-119 — the ResponseEvent enum, all SSE event types: Created / OutputItemDone / Completed / ReasoningSummaryDelta, etc.codex-rs/responses-api-proxy/src/lib.rs:73-108 — run_main, a tiny_http server for debugging that forwards to upstream_url and optionally dumps traffic.

ModelClient::new has so many parameters it needs #[allow(clippy::too_many_arguments)], but each is a session-scoped necessity. create_model_provider(provider_info, auth_manager) turns provider info into a concrete provider (OpenAI / Bedrock / Ollama / LMStudio / Anthropic-style external); all subsequent auth resolution goes through this provider abstraction.

// core/src/client.rs:413-460 — ModelClient::new 装配 session-scoped 状态
pub fn new(
    auth_manager: Option<Arc<AuthManager>>,
    agent_identity_policy: AgentIdentityAuthPolicy,
    thread_id: ThreadId,
    provider_info: ModelProviderInfo,
    session_source: SessionSource,
    originator: String,
    model_verbosity: Option<VerbosityConfig>,
    enable_request_compression: bool,
    include_timing_metrics: bool,
    beta_features_header: Option<String>,
    item_ids_enabled: bool,
    concurrent_reasoning_summaries_enabled: bool,
    attestation_provider: Option<Arc<dyn AttestationProvider>>,
    http_client_factory: HttpClientFactory,
) -> Self {
    let model_provider = create_model_provider(provider_info, auth_manager);
    // ...
    Self { state: Arc::new(ModelClientState { /* ... */ }), /* ... */ }
}

stream is the core entry within a turn. First check whether wire_api is Responses (currently the only option), then responses_websocket_enabled() decides WS vs HTTP. If the WS path returns WebsocketStreamOutcome::FallbackToHttp, it calls try_switch_fallback_transport to permanently switch to HTTP, then falls back to stream_responses_api. This way a WS failure does not stall the turn; it just loses the reuse benefit.

// core/src/client.rs:1792-1843 — ModelClientSession::stream 分派
pub async fn stream(
    &mut self,
    prompt: &Prompt,
    model_info: &ModelInfo,
    session_telemetry: &SessionTelemetry,
    effort: Option<ReasoningEffortConfig>,
    summary: ReasoningSummaryConfig,
    service_tier: Option<String>,
    responses_metadata: &CodexResponsesMetadata,
    inference_trace: &InferenceTraceContext,
) -> Result<ResponseStream> {
    let wire_api = self.client.state.provider.info().wire_api;
    match wire_api {
        WireApi::Responses => {
            if self.client.responses_websocket_enabled() {
                match self.stream_responses_websocket(/* ... */).await? {
                    WebsocketStreamOutcome::Stream(stream) => return Ok(stream),
                    WebsocketStreamOutcome::FallbackToHttp => {
                        self.try_switch_fallback_transport(session_telemetry, model_info);
                    }
                }
            }
            self.stream_responses_api(/* ... */).await
        }
    }
}

current_client_setup is the shared entry for prewarm and turn. It takes auth() and api_provider() from the provider, then resolves the agent identity scope by ProviderAuthScope — agent_identity_policy decides whether ChatGPT auth can auto-upgrade to agent identity, session_source decides the scope range. It returns CurrentClientSetup, and the subsequent build_api_transport only reads this struct.

// core/src/client.rs:944-962 — 集中 auth + provider 解析
async fn current_client_setup(&self) -> Result<CurrentClientSetup> {
    let auth = self.state.provider.auth().await;
    let api_provider = self.state.provider.api_provider().await?;
    let resolved_auth = self
        .state
        .provider
        .api_auth_for_scope(ProviderAuthScope {
            agent_identity_policy: self.agent_identity_policy,
            session_source: self.state.session_source.clone(),
            agent_identity_session_fallback: self.state.agent_identity_session_fallback.clone(),
        })
        .await?;
    Ok(CurrentClientSetup {
        auth,
        api_provider,
        api_auth: resolved_auth.auth,
        agent_identity_telemetry: resolved_auth.agent_identity_telemetry,
    })
}

Data flow

Boundaries and failures

• WS fallback is permanent: force_http_fallback uses AtomicBool::swap to permanently disable WS and store_cached_websocket_session(WebsocketSession::default()) to clear the established connection. Once a fallback is triggered inside a turn, all remaining turns of the session go HTTP and never retry WS. codex-rs/core/src/client.rs:509-528
• Turn session is not reused: the ModelClientSession docs explicitly require one new_session per turn; reusing across turns would leak the previous turn's x-codex-turn-state sticky-routing token into the next, triggering server-side routing chaos. codex-rs/core/src/client.rs:260-272
• Empty prompt skips compact: compact_conversation_history returns an empty Vec directly when prompt.input.is_empty(), wasting no network call. codex-rs/core/src/client.rs:548-550
• Auth prewarm matches the turn: prewarm_auth also goes through current_client_setup, ensuring the background pre-connect and the real turn see the exact same auth / provider state, so prewarm does not get evicted by a new token. codex-rs/core/src/client.rs:979-981

Summary

ModelClient routes all model calls through the Responses API, uses new_session to carve out a turn-scoped handle, and stream picks between WebSocket and HTTP, permanently switching to HTTP on failure. compact_conversation_history reuses the same transport and auth headers via /responses/compact. responses-api-proxy provides a standalone debug proxy. This layer does not participate in loop scheduling; it just streams prompts out and ResponseEvents back. What actually drives the loop is the Agent main loop. Auth state comes from the auth.json persisted by ChatGPT login and auth.