mirrors/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-05-04 05:11:37 +03:00
Code Issues Packages Projects Releases Wiki Activity

Simplification 2

Browse Source
This commit is contained in:
jif-oai
2025-11-12 15:53:18 +00:00
parent f5918d7e1b
commit 166ca2fce7
9 changed files with 395 additions and 241 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
codex-rs/api-client/src/responses.rs
View File

@@ -5,7 +5,6 @@ use codex_otel::otel_event_manager::OtelEventManager;
use codex_protocol::ConversationId;
use futures::TryStreamExt;
use serde_json::Value;
use tokio::sync::mpsc;
use tracing::debug;
use tracing::trace;
@@ -79,10 +78,7 @@ impl ResponsesApiClient {
),
];
owned_headers.extend(self.config.extra_headers.iter().cloned());
let extra_headers: Vec<(&str, String)> = owned_headers
.iter()
.map(|(k, v)| (k.as_str(), v.clone()))
.collect();
let extra_headers = crate::client::http::header_pairs(&owned_headers);
let mut req_builder = crate::client::http::build_request(
&self.config.http_client,
&self.config.provider,
@@ -112,32 +108,7 @@ impl ResponsesApiClient {
request_id: None,
})?;
let (tx_event, rx_event) = mpsc::channel::<Result<crate::stream::WireEvent>>(1600);
if let Some(snapshot) = crate::client::rate_limits::parse_rate_limit_snapshot(res.headers())
&& tx_event
.send(Ok(crate::stream::WireEvent::RateLimits(
crate::stream::WireRateLimitSnapshot {
primary: snapshot
.primary
.map(|w| crate::stream::WireRateLimitWindow {
used_percent: Some(w.used_percent),
window_minutes: w.window_minutes,
resets_at: w.resets_at,
}),
secondary: snapshot
.secondary
.map(|w| crate::stream::WireRateLimitWindow {
used_percent: Some(w.used_percent),
window_minutes: w.window_minutes,
resets_at: w.resets_at,
}),
},
)))
.await
.is_err()
{
debug!("receiver dropped rate limit snapshot event");
}
let snapshot = crate::client::rate_limits::parse_rate_limit_snapshot(res.headers());
let stream = res
.bytes_stream()
@@ -145,16 +116,22 @@ impl ResponsesApiClient {
source: err,
request_id: None,
});
let idle_timeout = self.config.provider.stream_idle_timeout();
let otel = self.config.otel_event_manager.clone();
tokio::spawn(crate::client::sse::process_sse_wire(
stream,
tx_event,
idle_timeout,
otel,
crate::decode_wire::responses::WireResponsesSseDecoder,
));
Ok(crate::stream::EventStream::from_receiver(rx_event))
let (tx_event, rx_event) = crate::client::sse::spawn_wire_stream(
stream,
&self.config.provider,
self.config.otel_event_manager.clone(),
crate::decode_wire::responses::WireResponsesSseDecoder,
);
if let Some(snapshot) = snapshot
&& tx_event
.send(Ok(crate::stream::WireEvent::RateLimits(snapshot.into())))
.await
.is_err()
{
debug!("receiver dropped rate limit snapshot event");
}
Ok(rx_event)
}
}