refactor(client): share stream opening across wire adapters
Route Anthropic Messages and OpenAI Responses SSE opens through the client/stream_entry.rs transport seam that Chat Completions already used, closing the post-0.9.1 follow-up in docs/POST_0_9_1_SEAMS.md. The seam gains one shared open function, open_sse_response(): it bounds the response-header wait (stream_open_timeout, now owned by the seam with its CODEWHALE_STREAM_OPEN_TIMEOUT_SECS override and 5..=300s clamp), selects the dual or HTTP/1.1-twin client per policy, and on a classified HTTP/2 header stall retries exactly once on the HTTP/1.1 twin. An H1-pinned request never retries, a provider error before headers never triggers the fallback, and nothing retries once response headers exist — stream-body errors stay in each adapter's decode loop. Adapter edges keep their wire-specific behavior: Chat Completions keeps its JSON retry path on the dual client and single-shot H1 sends; Anthropic keeps its rate-limit wait, Accept header, error-envelope parsing, and circuit-breaker marks (the non-streaming Messages path is untouched); Responses keeps its OAuth bearer default headers, beta opt-in, originator and account-id headers, and its pre-existing rate-limit/transient retry loop inside each open attempt. Both newly routed adapters also gain the shared idle-timeout diagnostics message (bytes received, stream age, ms since last chunk) in place of a bare "Stream idle timeout". New tests pin the seam contract: dual-policy first-attempt success, retry-exactly-once via H1 on a stall, no retry when H1-pinned, no retry for non-stall errors, dual-protocol timeout text, wire-header preservation through the seam for Anthropic and Responses (including bearer auth), and fail-fast error-envelope preservation. The stream-open-timeout clamp test moved with the function. Verified: cargo test -p codewhale-tui --bin codewhale-tui for client::stream_entry (8), client::anthropic (16), client::responses (15), client::chat (74) — all passing; crate clippy --all-targets clean.
This commit is contained in:
Hunter B
@@ -20,6 +20,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
|
||||
|
||||
### Changed
|
||||
|
||||
- Consolidate Anthropic Messages and OpenAI Responses stream opening
|
||||
through the shared `client/stream_entry.rs` transport seam already used
|
||||
by Chat Completions: one bounded response-header wait, shared
|
||||
dual/HTTP-1.1 client policy selection, at most one HTTP/1.1 fallback
|
||||
retry on a classified HTTP/2 header stall (never after a stream body
|
||||
has begun), and the shared idle-timeout diagnostics format. Both
|
||||
adapters gain the bounded open wait and `CODEWHALE_FORCE_HTTP1`
|
||||
H1 pinning; wire-specific headers, authentication, endpoints, and
|
||||
stream decoding stay at the adapter edge, and the Responses provider
|
||||
retry loop for rate limits / transient upstream errors is preserved.
|
||||
- Rename the internal delegated-worker role type from `SubAgentType` to
|
||||
`FleetRole` with canonical variants (`Worker`, `Scout`, `Planner`,
|
||||
`Reviewer`, `Builder`, `Verifier`, `Custom`) matching the public Fleet
|
||||
|
||||
@@ -180,7 +180,15 @@ impl DeepSeekClient {
|
||||
.send()
|
||||
.await
|
||||
.context("Anthropic Messages API request failed")?;
|
||||
self.check_anthropic_response(response).await
|
||||
}
|
||||
|
||||
/// Shared status/error-envelope handling for streaming and
|
||||
/// non-streaming Messages responses.
|
||||
async fn check_anthropic_response(
|
||||
&self,
|
||||
response: reqwest::Response,
|
||||
) -> Result<reqwest::Response> {
|
||||
let status = response.status();
|
||||
if !status.is_success() {
|
||||
let raw = bounded_error_text(response, ERROR_BODY_MAX_BYTES).await;
|
||||
@@ -193,13 +201,54 @@ impl DeepSeekClient {
|
||||
Ok(response)
|
||||
}
|
||||
|
||||
/// Open the streaming Messages request through the shared stream-entry
|
||||
/// transport policy: bounded header wait, dual-client selection, and at
|
||||
/// most one HTTP/1.1 fallback retry on a classified H2 header stall.
|
||||
/// Wire-specific request construction (headers, endpoint, body) stays
|
||||
/// here at the adapter edge.
|
||||
async fn open_anthropic_stream_response(&self, body: &Value) -> Result<reqwest::Response> {
|
||||
let url = anthropic_messages_url(&self.base_url);
|
||||
let open_req = super::stream_entry::StreamOpenRequest::new(
|
||||
super::stream_entry::stream_open_timeout(),
|
||||
self.stream_idle_timeout,
|
||||
);
|
||||
let opened = super::stream_entry::open_sse_response(&open_req, |policy| {
|
||||
let url = url.clone();
|
||||
async move {
|
||||
self.wait_for_rate_limit().await;
|
||||
let client = super::stream_entry::client_for_policy(
|
||||
&self.http_client,
|
||||
self.http1_fallback_client(),
|
||||
policy,
|
||||
);
|
||||
client
|
||||
.post(&url)
|
||||
.header("Accept", "text/event-stream")
|
||||
.json(body)
|
||||
.send()
|
||||
.await
|
||||
.context("Anthropic Messages API request failed")
|
||||
}
|
||||
})
|
||||
.await;
|
||||
let response = match opened {
|
||||
Ok(response) => response,
|
||||
Err(err) => {
|
||||
self.mark_request_failure(&format!("anthropic stream open: {err}"))
|
||||
.await;
|
||||
return Err(err);
|
||||
}
|
||||
};
|
||||
self.check_anthropic_response(response).await
|
||||
}
|
||||
|
||||
/// Handle a streaming Messages API request.
|
||||
pub(super) async fn handle_anthropic_stream(
|
||||
&self,
|
||||
request: MessageRequest,
|
||||
) -> Result<StreamEventBox> {
|
||||
let body = self.build_anthropic_body(&request, true);
|
||||
let response = self.send_anthropic_request(&body).await?;
|
||||
let response = self.open_anthropic_stream_response(&body).await?;
|
||||
|
||||
let stream_idle_timeout = self.stream_idle_timeout;
|
||||
let byte_stream = response.bytes_stream();
|
||||
@@ -212,6 +261,9 @@ impl DeepSeekClient {
|
||||
// corrupted to U+FFFD. Line boundaries ('\n') are ASCII and can
|
||||
// never fall inside a multi-byte sequence. (Mirrors chat.rs.)
|
||||
let mut buffer: Vec<u8> = Vec::new();
|
||||
let stream_start = std::time::Instant::now();
|
||||
let mut last_chunk_at = std::time::Instant::now();
|
||||
let mut bytes_received: usize = 0;
|
||||
tokio::pin!(byte_stream);
|
||||
|
||||
loop {
|
||||
@@ -223,11 +275,18 @@ impl DeepSeekClient {
|
||||
}
|
||||
Ok(None) => break,
|
||||
Err(_) => {
|
||||
yield Err(anyhow::anyhow!("Stream idle timeout"));
|
||||
yield Err(anyhow::anyhow!(super::stream_entry::idle_timeout_message(
|
||||
stream_idle_timeout,
|
||||
bytes_received,
|
||||
stream_start.elapsed(),
|
||||
last_chunk_at.elapsed(),
|
||||
)));
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
bytes_received += chunk.len();
|
||||
last_chunk_at = std::time::Instant::now();
|
||||
buffer.extend_from_slice(&chunk);
|
||||
|
||||
while let Some(line) = super::take_sse_line(&mut buffer) {
|
||||
@@ -1168,4 +1227,76 @@ mod tests {
|
||||
"https://api.minimaxi.com/anthropic/v1/messages"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn anthropic_stream_opens_through_shared_seam_preserving_headers() {
|
||||
use futures_util::StreamExt;
|
||||
use wiremock::matchers::{header, method, path};
|
||||
use wiremock::{Mock, MockServer, ResponseTemplate};
|
||||
|
||||
let server = MockServer::start().await;
|
||||
// The wire-specific Accept header must survive the shared stream-entry
|
||||
// open path; the mock only answers when it is present.
|
||||
Mock::given(method("POST"))
|
||||
.and(path("/v1/messages"))
|
||||
.and(header("Accept", "text/event-stream"))
|
||||
.respond_with(
|
||||
ResponseTemplate::new(200)
|
||||
.insert_header("Content-Type", "text/event-stream")
|
||||
.set_body_string("data: {\"type\":\"message_stop\"}\n\n"),
|
||||
)
|
||||
.expect(1)
|
||||
.mount(&server)
|
||||
.await;
|
||||
|
||||
let client = deepseek_test_client(&server.uri());
|
||||
let mut stream = client
|
||||
.handle_anthropic_stream(request_with("deepseek-v4", None, None, None))
|
||||
.await
|
||||
.expect("stream opens through the shared seam");
|
||||
|
||||
let mut saw_stop = false;
|
||||
tokio::time::timeout(std::time::Duration::from_secs(5), async {
|
||||
while let Some(event) = stream.next().await {
|
||||
if matches!(event.expect("stream event"), StreamEvent::MessageStop) {
|
||||
saw_stop = true;
|
||||
}
|
||||
}
|
||||
})
|
||||
.await
|
||||
.expect("stream finishes after message_stop");
|
||||
assert!(saw_stop, "message_stop should arrive through the seam");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn anthropic_stream_open_error_is_not_retried() {
|
||||
use wiremock::matchers::{method, path};
|
||||
use wiremock::{Mock, MockServer, ResponseTemplate};
|
||||
|
||||
let server = MockServer::start().await;
|
||||
// A definitive provider error before any stream body must fail fast:
|
||||
// exactly one request, no H1 fallback, envelope preserved.
|
||||
Mock::given(method("POST"))
|
||||
.and(path("/v1/messages"))
|
||||
.respond_with(ResponseTemplate::new(401).set_body_string(
|
||||
"{\"error\":{\"type\":\"authentication_error\",\"message\":\"bad key\"}}",
|
||||
))
|
||||
.expect(1)
|
||||
.mount(&server)
|
||||
.await;
|
||||
|
||||
let client = deepseek_test_client(&server.uri());
|
||||
let err = match client
|
||||
.handle_anthropic_stream(request_with("deepseek-v4", None, None, None))
|
||||
.await
|
||||
{
|
||||
Ok(_) => panic!("auth errors must fail fast"),
|
||||
Err(err) => err,
|
||||
};
|
||||
let text = err.to_string();
|
||||
assert!(
|
||||
text.contains("HTTP 401") && text.contains("authentication_error"),
|
||||
"error envelope should be preserved: {text}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
+27
-122
@@ -19,34 +19,10 @@ use crate::config::{
|
||||
wire_model_for_provider_route,
|
||||
};
|
||||
|
||||
/// Default timeout for the initial streaming response headers.
|
||||
///
|
||||
/// `doctor` uses a bounded non-streaming request, but normal TUI turns first
|
||||
/// wait for the SSE response to open. On some Windows/proxy paths that wait can
|
||||
/// hang before any stream chunk exists, leaving the UI stuck at "Working...".
|
||||
const DEFAULT_STREAM_OPEN_TIMEOUT: Duration = Duration::from_secs(45);
|
||||
|
||||
/// Reads `CODEWHALE_STREAM_OPEN_TIMEOUT_SECS` (legacy alias:
|
||||
/// `DEEPSEEK_STREAM_OPEN_TIMEOUT_SECS`) as a bounded override for the
|
||||
/// response-header wait. This is intentionally shorter than the per-chunk idle
|
||||
/// timeout because it only covers connection setup and upstream header return,
|
||||
/// not model thinking time after streaming has started.
|
||||
fn stream_open_timeout() -> Duration {
|
||||
stream_open_timeout_from_env(
|
||||
std::env::var("CODEWHALE_STREAM_OPEN_TIMEOUT_SECS")
|
||||
.or_else(|_| std::env::var("DEEPSEEK_STREAM_OPEN_TIMEOUT_SECS"))
|
||||
.ok()
|
||||
.as_deref(),
|
||||
)
|
||||
}
|
||||
|
||||
fn stream_open_timeout_from_env(value: Option<&str>) -> Duration {
|
||||
let secs = value
|
||||
.and_then(|v| v.parse::<u64>().ok())
|
||||
.unwrap_or(DEFAULT_STREAM_OPEN_TIMEOUT.as_secs())
|
||||
.clamp(5, 300);
|
||||
Duration::from_secs(secs)
|
||||
}
|
||||
// The bounded response-header wait (`stream_open_timeout`) and its env
|
||||
// override live in the shared stream-entry seam; every streaming adapter
|
||||
// (Chat Completions / Anthropic Messages / Responses) uses the same policy.
|
||||
use super::stream_entry::stream_open_timeout;
|
||||
|
||||
fn stream_idle_timeout_message(
|
||||
idle: Duration,
|
||||
@@ -497,85 +473,35 @@ impl DeepSeekClient {
|
||||
url: &str,
|
||||
body: &Value,
|
||||
) -> Result<(reqwest::Response, Duration)> {
|
||||
let open_timeout = stream_open_timeout();
|
||||
let open_req =
|
||||
super::stream_entry::StreamOpenRequest::new(open_timeout, self.stream_idle_timeout);
|
||||
let idle_timeout = open_req.idle_timeout;
|
||||
let open_client = super::stream_entry::client_for_policy(
|
||||
&self.http_client,
|
||||
self.http1_fallback_client(),
|
||||
open_req.policy,
|
||||
let open_req = super::stream_entry::StreamOpenRequest::new(
|
||||
stream_open_timeout(),
|
||||
self.stream_idle_timeout,
|
||||
);
|
||||
let response = match tokio_timeout(open_req.open_timeout, async {
|
||||
if matches!(
|
||||
open_req.policy,
|
||||
super::stream_entry::StreamHttpPolicy::Http1Only
|
||||
) {
|
||||
Ok(open_client
|
||||
.post(url)
|
||||
.header(reqwest::header::CONTENT_TYPE, "application/json")
|
||||
.json(body)
|
||||
.send()
|
||||
.await?)
|
||||
} else {
|
||||
self.send_json_with_retry(url, body).await
|
||||
}
|
||||
})
|
||||
.await
|
||||
{
|
||||
Ok(result) => result?,
|
||||
Err(_elapsed) => {
|
||||
// A header stall on the dual client is eligible for one
|
||||
// explicit retry through the prebuilt HTTP/1.1 twin.
|
||||
if super::stream_entry::should_retry_with_h1(open_req.policy, "http2 stream closed")
|
||||
{
|
||||
let h1_req = open_req.with_h1_only();
|
||||
let h1_client = super::stream_entry::client_for_policy(
|
||||
let idle_timeout = open_req.idle_timeout;
|
||||
let response = super::stream_entry::open_sse_response(&open_req, |policy| async move {
|
||||
match policy {
|
||||
// The prebuilt HTTP/1.1 twin carries the same default
|
||||
// headers/auth; send once, without the JSON retry loop
|
||||
// (matching the pre-seam H1-pin behavior).
|
||||
super::stream_entry::StreamHttpPolicy::Http1Only => {
|
||||
let client = super::stream_entry::client_for_policy(
|
||||
&self.http_client,
|
||||
self.http1_fallback_client(),
|
||||
h1_req.policy,
|
||||
);
|
||||
crate::logging::warn(
|
||||
"SSE stream headers timed out over HTTP/2; retrying once with HTTP/1.1",
|
||||
);
|
||||
match tokio_timeout(
|
||||
h1_req.open_timeout,
|
||||
h1_client
|
||||
.post(url)
|
||||
.header(reqwest::header::CONTENT_TYPE, "application/json")
|
||||
.json(body)
|
||||
.send(),
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(Ok(response)) => response,
|
||||
Ok(Err(err)) => {
|
||||
anyhow::bail!(
|
||||
"SSE stream request failed after HTTP/1.1 fallback: {err}. \
|
||||
`codewhale doctor` can still pass when non-streaming requests work; \
|
||||
on Windows or proxy networks, try `CODEWHALE_FORCE_HTTP1=1` and rerun `codewhale`."
|
||||
);
|
||||
}
|
||||
Err(_elapsed) => {
|
||||
anyhow::bail!(
|
||||
"SSE stream request did not receive response headers after {}s \
|
||||
(HTTP/2 and HTTP/1.1). `codewhale doctor` can still pass when \
|
||||
non-streaming requests work; try `CODEWHALE_FORCE_HTTP1=1` and \
|
||||
rerun `codewhale`.",
|
||||
open_timeout.as_secs()
|
||||
);
|
||||
}
|
||||
}
|
||||
} else {
|
||||
anyhow::bail!(
|
||||
"SSE stream request did not receive response headers after {}s. \
|
||||
`codewhale doctor` can still pass when non-streaming requests work; \
|
||||
on Windows or proxy networks, try `CODEWHALE_FORCE_HTTP1=1` and rerun `codewhale`.",
|
||||
open_timeout.as_secs()
|
||||
policy,
|
||||
);
|
||||
Ok(client
|
||||
.post(url)
|
||||
.header(reqwest::header::CONTENT_TYPE, "application/json")
|
||||
.json(body)
|
||||
.send()
|
||||
.await?)
|
||||
}
|
||||
super::stream_entry::StreamHttpPolicy::DualWithH1Fallback => {
|
||||
self.send_json_with_retry(url, body).await
|
||||
}
|
||||
}
|
||||
};
|
||||
})
|
||||
.await?;
|
||||
Ok((response, idle_timeout))
|
||||
}
|
||||
|
||||
@@ -3303,27 +3229,6 @@ mod stream_diagnostics_tests {
|
||||
use super::*;
|
||||
use reqwest::header::{HeaderMap, HeaderValue};
|
||||
|
||||
#[test]
|
||||
fn stream_open_timeout_defaults_and_clamps_env_values() {
|
||||
assert_eq!(stream_open_timeout_from_env(None), Duration::from_secs(45));
|
||||
assert_eq!(
|
||||
stream_open_timeout_from_env(Some("not-a-number")),
|
||||
Duration::from_secs(45)
|
||||
);
|
||||
assert_eq!(
|
||||
stream_open_timeout_from_env(Some("1")),
|
||||
Duration::from_secs(5)
|
||||
);
|
||||
assert_eq!(
|
||||
stream_open_timeout_from_env(Some("120")),
|
||||
Duration::from_secs(120)
|
||||
);
|
||||
assert_eq!(
|
||||
stream_open_timeout_from_env(Some("999")),
|
||||
Duration::from_secs(300)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stream_idle_timeout_reports_progress_and_timing() {
|
||||
let message = stream_idle_timeout_message(
|
||||
|
||||
@@ -86,29 +86,50 @@ impl DeepSeekClient {
|
||||
let url = format!("{}{}", self.base_url, CODEX_RESPONSES_PATH);
|
||||
|
||||
// The bearer Authorization header is already installed as a default
|
||||
// header on `http_client` (resolved from the Codex OAuth access token),
|
||||
// so it must not be set again here or it would be duplicated. The
|
||||
// ChatGPT backend additionally requires the account id and the
|
||||
// experimental Responses beta opt-in.
|
||||
// header on both the dual and the HTTP/1.1 twin client (resolved from
|
||||
// the Codex OAuth access token), so it must not be set again here or
|
||||
// it would be duplicated. The ChatGPT backend additionally requires
|
||||
// the account id and the experimental Responses beta opt-in.
|
||||
//
|
||||
// The open itself goes through the shared stream-entry transport
|
||||
// policy: bounded header wait, policy-selected client, and at most
|
||||
// one HTTP/1.1 fallback retry on a classified H2 header stall. The
|
||||
// pre-existing provider retry loop (rate limit / transient upstream)
|
||||
// stays inside each open attempt, before any stream body exists.
|
||||
let account_id = self.codex_account_id.clone();
|
||||
let request_body =
|
||||
serde_json::to_vec(&body).context("Failed to serialize Responses API request body")?;
|
||||
let response = self
|
||||
.send_with_retry(|| {
|
||||
let mut builder = self
|
||||
.http_client
|
||||
.post(&url)
|
||||
.header("Content-Type", "application/json")
|
||||
.header("Accept", "text/event-stream")
|
||||
.header("OpenAI-Beta", "responses=experimental")
|
||||
.header("originator", "codex_cli_rs");
|
||||
if let Some(account_id) = &account_id {
|
||||
builder = builder.header("chatgpt-account-id", account_id);
|
||||
}
|
||||
builder.body(request_body.clone())
|
||||
})
|
||||
.await
|
||||
.context("Responses API request failed")?;
|
||||
let open_req = super::stream_entry::StreamOpenRequest::new(
|
||||
super::stream_entry::stream_open_timeout(),
|
||||
self.stream_idle_timeout,
|
||||
);
|
||||
let response = super::stream_entry::open_sse_response(&open_req, |policy| {
|
||||
let url = url.clone();
|
||||
let account_id = account_id.clone();
|
||||
let request_body = request_body.clone();
|
||||
async move {
|
||||
let client = super::stream_entry::client_for_policy(
|
||||
&self.http_client,
|
||||
self.http1_fallback_client(),
|
||||
policy,
|
||||
);
|
||||
self.send_with_retry(|| {
|
||||
let mut builder = client
|
||||
.post(&url)
|
||||
.header("Content-Type", "application/json")
|
||||
.header("Accept", "text/event-stream")
|
||||
.header("OpenAI-Beta", "responses=experimental")
|
||||
.header("originator", "codex_cli_rs");
|
||||
if let Some(account_id) = &account_id {
|
||||
builder = builder.header("chatgpt-account-id", account_id);
|
||||
}
|
||||
builder.body(request_body.clone())
|
||||
})
|
||||
.await
|
||||
.context("Responses API request failed")
|
||||
}
|
||||
})
|
||||
.await?;
|
||||
|
||||
let status = response.status();
|
||||
if !status.is_success() {
|
||||
@@ -151,6 +172,9 @@ impl DeepSeekClient {
|
||||
let mut buffer: Vec<u8> = Vec::new();
|
||||
let mut done = false;
|
||||
let mut content_block_counter: u32 = 0;
|
||||
let stream_start = std::time::Instant::now();
|
||||
let mut last_chunk_at = std::time::Instant::now();
|
||||
let mut bytes_received: usize = 0;
|
||||
|
||||
tokio::pin!(byte_stream);
|
||||
|
||||
@@ -163,11 +187,18 @@ impl DeepSeekClient {
|
||||
}
|
||||
Ok(None) => break,
|
||||
Err(_) => {
|
||||
yield Err(anyhow::anyhow!("Stream idle timeout"));
|
||||
yield Err(anyhow::anyhow!(super::stream_entry::idle_timeout_message(
|
||||
stream_idle_timeout,
|
||||
bytes_received,
|
||||
stream_start.elapsed(),
|
||||
last_chunk_at.elapsed(),
|
||||
)));
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
bytes_received += chunk.len();
|
||||
last_chunk_at = std::time::Instant::now();
|
||||
buffer.extend_from_slice(&chunk);
|
||||
|
||||
// Process complete SSE lines.
|
||||
@@ -963,6 +994,52 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn responses_stream_open_preserves_wire_headers_through_shared_seam() {
|
||||
use wiremock::matchers::header;
|
||||
|
||||
let server = MockServer::start().await;
|
||||
// Every wire-specific header (SSE accept, Responses beta opt-in,
|
||||
// originator, bearer auth from the default headers) must survive the
|
||||
// shared stream-entry open path; the mock only answers when all are
|
||||
// present.
|
||||
Mock::given(method("POST"))
|
||||
.and(path(CODEX_RESPONSES_PATH))
|
||||
.and(header("Accept", "text/event-stream"))
|
||||
.and(header("OpenAI-Beta", "responses=experimental"))
|
||||
.and(header("originator", "codex_cli_rs"))
|
||||
.and(header("Authorization", "Bearer test-token"))
|
||||
.respond_with(
|
||||
ResponseTemplate::new(200)
|
||||
.insert_header("Content-Type", "text/event-stream")
|
||||
.set_body_string("data: [DONE]\n\n"),
|
||||
)
|
||||
.expect(1)
|
||||
.mount(&server)
|
||||
.await;
|
||||
|
||||
let client = {
|
||||
let _env_lock = crate::test_support::lock_test_env();
|
||||
let _codex_token =
|
||||
crate::test_support::EnvVarGuard::set("OPENAI_CODEX_ACCESS_TOKEN", "test-token");
|
||||
let _legacy_codex_token =
|
||||
crate::test_support::EnvVarGuard::remove("CODEX_ACCESS_TOKEN");
|
||||
DeepSeekClient::new(&test_codex_config(&server)).unwrap()
|
||||
};
|
||||
let mut stream = client
|
||||
.handle_responses_stream(minimal_responses_request())
|
||||
.await
|
||||
.expect("stream opens with preserved headers");
|
||||
|
||||
tokio::time::timeout(std::time::Duration::from_secs(5), async {
|
||||
while let Some(event) = stream.next().await {
|
||||
event.unwrap();
|
||||
}
|
||||
})
|
||||
.await
|
||||
.expect("stream should finish after [DONE]");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn responses_stream_inserts_boundary_between_reasoning_summary_parts() {
|
||||
let server = MockServer::start().await;
|
||||
|
||||
@@ -8,10 +8,37 @@
|
||||
//! Full piagent-style provider collapse is deferred — see
|
||||
//! `docs/notes/post-0.9.1-thin-tui-and-stream.md`.
|
||||
|
||||
use std::future::Future;
|
||||
use std::time::Duration;
|
||||
|
||||
use anyhow::Result;
|
||||
use reqwest::Client;
|
||||
|
||||
/// Default bounded wait for SSE response headers. Intentionally shorter than
|
||||
/// the per-chunk idle timeout: it covers connection setup and upstream header
|
||||
/// return only, never model thinking time after streaming has started.
|
||||
pub(crate) const DEFAULT_STREAM_OPEN_TIMEOUT: Duration = Duration::from_secs(45);
|
||||
|
||||
/// Env override (`CODEWHALE_STREAM_OPEN_TIMEOUT_SECS`, legacy
|
||||
/// `DEEPSEEK_STREAM_OPEN_TIMEOUT_SECS`) for the response-header wait,
|
||||
/// shared by every streaming adapter.
|
||||
pub(crate) fn stream_open_timeout() -> Duration {
|
||||
stream_open_timeout_from_env(
|
||||
std::env::var("CODEWHALE_STREAM_OPEN_TIMEOUT_SECS")
|
||||
.or_else(|_| std::env::var("DEEPSEEK_STREAM_OPEN_TIMEOUT_SECS"))
|
||||
.ok()
|
||||
.as_deref(),
|
||||
)
|
||||
}
|
||||
|
||||
pub(crate) fn stream_open_timeout_from_env(value: Option<&str>) -> Duration {
|
||||
let secs = value
|
||||
.and_then(|v| v.parse::<u64>().ok())
|
||||
.unwrap_or(DEFAULT_STREAM_OPEN_TIMEOUT.as_secs())
|
||||
.clamp(5, 300);
|
||||
Duration::from_secs(secs)
|
||||
}
|
||||
|
||||
/// How the shared stream open path should pin HTTP version.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum StreamHttpPolicy {
|
||||
@@ -79,6 +106,63 @@ pub fn should_retry_with_h1(policy: StreamHttpPolicy, err_text: &str) -> bool {
|
||||
|| lower.contains("frame size")
|
||||
}
|
||||
|
||||
/// Open an SSE response through the shared transport policy.
|
||||
///
|
||||
/// `attempt` builds and sends one wire-specific request on the client
|
||||
/// selected for the given policy (via [`client_for_policy`]); everything
|
||||
/// transport-shared lives here:
|
||||
///
|
||||
/// - the response-header wait is bounded by `open_req.open_timeout`;
|
||||
/// - a header stall on the dual client retries exactly once on the
|
||||
/// HTTP/1.1 twin ([`should_retry_with_h1`] classification);
|
||||
/// - a stall on an already H1-pinned request never retries;
|
||||
/// - once response headers have been received the seam never retries —
|
||||
/// body/stream errors belong to the adapter's decode loop.
|
||||
pub(crate) async fn open_sse_response<F, Fut>(
|
||||
open_req: &StreamOpenRequest,
|
||||
attempt: F,
|
||||
) -> Result<reqwest::Response>
|
||||
where
|
||||
F: Fn(StreamHttpPolicy) -> Fut,
|
||||
Fut: Future<Output = Result<reqwest::Response>>,
|
||||
{
|
||||
match tokio::time::timeout(open_req.open_timeout, attempt(open_req.policy)).await {
|
||||
Ok(result) => result,
|
||||
Err(_elapsed) => {
|
||||
// A header stall on the dual client is eligible for one explicit
|
||||
// retry through the prebuilt HTTP/1.1 twin.
|
||||
if should_retry_with_h1(open_req.policy, "http2 stream closed") {
|
||||
let h1_req = open_req.clone().with_h1_only();
|
||||
crate::logging::warn(
|
||||
"SSE stream headers timed out over HTTP/2; retrying once with HTTP/1.1",
|
||||
);
|
||||
match tokio::time::timeout(h1_req.open_timeout, attempt(h1_req.policy)).await {
|
||||
Ok(Ok(response)) => Ok(response),
|
||||
Ok(Err(err)) => Err(anyhow::anyhow!(
|
||||
"SSE stream request failed after HTTP/1.1 fallback: {err}. \
|
||||
`codewhale doctor` can still pass when non-streaming requests work; \
|
||||
on Windows or proxy networks, try `CODEWHALE_FORCE_HTTP1=1` and rerun `codewhale`."
|
||||
)),
|
||||
Err(_elapsed) => Err(anyhow::anyhow!(
|
||||
"SSE stream request did not receive response headers after {}s \
|
||||
(HTTP/2 and HTTP/1.1). `codewhale doctor` can still pass when \
|
||||
non-streaming requests work; try `CODEWHALE_FORCE_HTTP1=1` and \
|
||||
rerun `codewhale`.",
|
||||
open_req.open_timeout.as_secs()
|
||||
)),
|
||||
}
|
||||
} else {
|
||||
Err(anyhow::anyhow!(
|
||||
"SSE stream request did not receive response headers after {}s. \
|
||||
`codewhale doctor` can still pass when non-streaming requests work; \
|
||||
on Windows or proxy networks, try `CODEWHALE_FORCE_HTTP1=1` and rerun `codewhale`.",
|
||||
open_req.open_timeout.as_secs()
|
||||
))
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Format a stable idle-timeout message shared across adapters.
|
||||
#[must_use]
|
||||
pub fn idle_timeout_message(
|
||||
@@ -99,8 +183,166 @@ pub fn idle_timeout_message(
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
|
||||
use wiremock::matchers::method;
|
||||
use wiremock::{Mock, MockServer, ResponseTemplate};
|
||||
|
||||
use super::*;
|
||||
|
||||
fn open_req(policy: StreamHttpPolicy, open_timeout: Duration) -> StreamOpenRequest {
|
||||
StreamOpenRequest {
|
||||
policy,
|
||||
open_timeout,
|
||||
idle_timeout: Duration::from_secs(30),
|
||||
}
|
||||
}
|
||||
|
||||
async fn ok_server() -> MockServer {
|
||||
let server = MockServer::start().await;
|
||||
Mock::given(method("POST"))
|
||||
.respond_with(ResponseTemplate::new(200))
|
||||
.mount(&server)
|
||||
.await;
|
||||
server
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn open_returns_first_attempt_response_on_dual_policy() {
|
||||
let server = ok_server().await;
|
||||
let _ = rustls::crypto::ring::default_provider().install_default();
|
||||
let client = reqwest::Client::new();
|
||||
let attempts = Arc::new(AtomicUsize::new(0));
|
||||
let response = open_sse_response(
|
||||
&open_req(StreamHttpPolicy::DualWithH1Fallback, Duration::from_secs(5)),
|
||||
|policy| {
|
||||
assert_eq!(policy, StreamHttpPolicy::DualWithH1Fallback);
|
||||
let attempts = Arc::clone(&attempts);
|
||||
let client = client.clone();
|
||||
let url = server.uri();
|
||||
async move {
|
||||
attempts.fetch_add(1, Ordering::SeqCst);
|
||||
Ok(client.post(url).send().await?)
|
||||
}
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("first attempt succeeds");
|
||||
assert_eq!(response.status(), 200);
|
||||
assert_eq!(attempts.load(Ordering::SeqCst), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn header_stall_on_dual_policy_retries_exactly_once_on_h1() {
|
||||
let server = ok_server().await;
|
||||
let _ = rustls::crypto::ring::default_provider().install_default();
|
||||
let client = reqwest::Client::new();
|
||||
let attempts = Arc::new(AtomicUsize::new(0));
|
||||
let response = open_sse_response(
|
||||
&open_req(
|
||||
StreamHttpPolicy::DualWithH1Fallback,
|
||||
Duration::from_millis(150),
|
||||
),
|
||||
|policy| {
|
||||
let attempts = Arc::clone(&attempts);
|
||||
let client = client.clone();
|
||||
let url = server.uri();
|
||||
async move {
|
||||
let attempt = attempts.fetch_add(1, Ordering::SeqCst);
|
||||
if attempt == 0 {
|
||||
// First attempt stalls before response headers.
|
||||
assert_eq!(policy, StreamHttpPolicy::DualWithH1Fallback);
|
||||
std::future::pending::<()>().await;
|
||||
}
|
||||
assert_eq!(policy, StreamHttpPolicy::Http1Only);
|
||||
Ok(client.post(url).send().await?)
|
||||
}
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("H1 fallback retry succeeds");
|
||||
assert_eq!(response.status(), 200);
|
||||
assert_eq!(
|
||||
attempts.load(Ordering::SeqCst),
|
||||
2,
|
||||
"exactly one fallback retry"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn header_stall_when_h1_pinned_never_retries_and_reports_timeout_text() {
|
||||
let attempts = Arc::new(AtomicUsize::new(0));
|
||||
let err = open_sse_response(
|
||||
&open_req(StreamHttpPolicy::Http1Only, Duration::from_millis(100)),
|
||||
|_| {
|
||||
let attempts = Arc::clone(&attempts);
|
||||
async move {
|
||||
attempts.fetch_add(1, Ordering::SeqCst);
|
||||
std::future::pending::<()>().await;
|
||||
unreachable!("stalled attempt never resolves")
|
||||
}
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect_err("H1-pinned stall fails without retry");
|
||||
assert_eq!(attempts.load(Ordering::SeqCst), 1, "no retry when pinned");
|
||||
let text = err.to_string();
|
||||
assert!(text.contains("did not receive response headers"), "{text}");
|
||||
assert!(text.contains("CODEWHALE_FORCE_HTTP1=1"), "{text}");
|
||||
assert!(
|
||||
!text.contains("HTTP/2 and HTTP/1.1"),
|
||||
"single-protocol stall must not claim a dual-protocol attempt: {text}"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn attempt_error_before_headers_is_not_h1_retried() {
|
||||
let attempts = Arc::new(AtomicUsize::new(0));
|
||||
let err = open_sse_response(
|
||||
&open_req(StreamHttpPolicy::DualWithH1Fallback, Duration::from_secs(5)),
|
||||
|_| {
|
||||
let attempts = Arc::clone(&attempts);
|
||||
async move {
|
||||
attempts.fetch_add(1, Ordering::SeqCst);
|
||||
Err(anyhow::anyhow!("HTTP 401: invalid api key"))
|
||||
}
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect_err("provider error propagates");
|
||||
assert_eq!(
|
||||
attempts.load(Ordering::SeqCst),
|
||||
1,
|
||||
"non-stall errors are never H1-retried"
|
||||
);
|
||||
assert!(err.to_string().contains("HTTP 401"), "{err}");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn double_stall_reports_both_protocols_in_timeout_text() {
|
||||
let attempts = Arc::new(AtomicUsize::new(0));
|
||||
let err = open_sse_response(
|
||||
&open_req(
|
||||
StreamHttpPolicy::DualWithH1Fallback,
|
||||
Duration::from_millis(100),
|
||||
),
|
||||
|_| {
|
||||
let attempts = Arc::clone(&attempts);
|
||||
async move {
|
||||
attempts.fetch_add(1, Ordering::SeqCst);
|
||||
std::future::pending::<()>().await;
|
||||
unreachable!("stalled attempt never resolves")
|
||||
}
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect_err("double stall fails");
|
||||
assert_eq!(attempts.load(Ordering::SeqCst), 2, "one fallback, no more");
|
||||
let text = err.to_string();
|
||||
assert!(text.contains("HTTP/2 and HTTP/1.1"), "{text}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn h1_retry_only_on_dual_policy() {
|
||||
assert!(should_retry_with_h1(
|
||||
@@ -113,6 +355,27 @@ mod tests {
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stream_open_timeout_defaults_and_clamps_env_values() {
|
||||
assert_eq!(stream_open_timeout_from_env(None), Duration::from_secs(45));
|
||||
assert_eq!(
|
||||
stream_open_timeout_from_env(Some("not-a-number")),
|
||||
Duration::from_secs(45)
|
||||
);
|
||||
assert_eq!(
|
||||
stream_open_timeout_from_env(Some("1")),
|
||||
Duration::from_secs(5)
|
||||
);
|
||||
assert_eq!(
|
||||
stream_open_timeout_from_env(Some("120")),
|
||||
Duration::from_secs(120)
|
||||
);
|
||||
assert_eq!(
|
||||
stream_open_timeout_from_env(Some("999")),
|
||||
Duration::from_secs(300)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn idle_message_is_stable() {
|
||||
let msg = idle_timeout_message(
|
||||
|
||||
@@ -20,18 +20,33 @@ monoliths without necessity):
|
||||
Business logic must not land in `ui.rs` / `app.rs` / `widgets/mod.rs` unless it
|
||||
is pure view wiring.
|
||||
|
||||
## StreamFn consolidation (partial)
|
||||
## StreamFn consolidation (landed post-0.9.1)
|
||||
|
||||
`client/stream_entry.rs` is the shared open-path seam:
|
||||
`client/stream_entry.rs` is the shared open-path seam, and all three
|
||||
streaming adapters open through it:
|
||||
|
||||
- HTTP policy (`DualWithH1Fallback` / `Http1Only`)
|
||||
- H1 retry classification
|
||||
- idle-timeout message format
|
||||
- HTTP policy (`DualWithH1Fallback` / `Http1Only`, env pin via
|
||||
`CODEWHALE_FORCE_HTTP1`)
|
||||
- dual/H1-twin client selection (`client_for_policy`)
|
||||
- bounded response-header wait (`stream_open_timeout`, env override
|
||||
`CODEWHALE_STREAM_OPEN_TIMEOUT_SECS`)
|
||||
- one shared open function (`open_sse_response`): a classified H2 header
|
||||
stall on the dual client retries exactly once on the HTTP/1.1 twin;
|
||||
an H1-pinned request never retries; nothing retries once response
|
||||
headers (and therefore any stream body) exist
|
||||
- H1 retry classification (`should_retry_with_h1`)
|
||||
- idle-timeout message format (`idle_timeout_message`, with
|
||||
bytes/age/last-chunk diagnostics)
|
||||
|
||||
Wire-protocol adapters remain at the edge (`chat.rs`, `anthropic.rs`,
|
||||
`responses.rs`). Chat Completions now opens through `StreamOpenRequest`;
|
||||
**follow-up:** route the Anthropic Messages and Responses adapters through the
|
||||
same policy before collapsing further toward a piagent-style single StreamFn.
|
||||
Wire-protocol request construction and stream decoding remain at the
|
||||
adapter edge (`chat.rs`, `anthropic.rs`, `responses.rs`): each adapter
|
||||
builds its own endpoint URL, headers, auth, and body inside the attempt
|
||||
closure it hands to `open_sse_response`. The pre-existing Responses
|
||||
provider retry loop (rate limit / transient upstream, `send_with_retry`)
|
||||
stays inside each open attempt, before any stream body exists.
|
||||
|
||||
Remaining follow-up: collapsing further toward a piagent-style single
|
||||
StreamFn (shared decode loop) is still deferred.
|
||||
|
||||
## Thin TUI over core (north star)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user