diff --git a/Cargo.lock b/Cargo.lock index 90e3112..632f188 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -423,7 +423,7 @@ dependencies = [ [[package]] name = "agentkit-tool-compose" -version = "0.10.11" +version = "0.10.12" dependencies = [ "agentkit-core", "agentkit-tools-core", @@ -3103,9 +3103,9 @@ dependencies = [ [[package]] name = "runlet" -version = "0.6.0" +version = "0.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "057e0864428c5a79a68683942d3750d05e9ffae541aa0185fde9ae1c87eca100" +checksum = "b710066183389b1bfa03d4bfd28c9558b582f9f584ca251556a639461e3efa58" dependencies = [ "hex", "regex", diff --git a/crates/agentkit-provider-openai/README.md b/crates/agentkit-provider-openai/README.md index d7f1cc6..38c40e1 100644 --- a/crates/agentkit-provider-openai/README.md +++ b/crates/agentkit-provider-openai/README.md @@ -124,12 +124,18 @@ WebSocket retries are intentionally more conservative than HTTP retries: errors never trigger automatic replay or authentication refresh. The narrowly scoped missing-continuation recovery is described below. An accepted response is never automatically replayed. -- An interrupted send, socket EOF, receive failure, or timeout after sending is - **not replayed**: the server may already have accepted the request. -- Visible WebSocket output is never superseded/replayed, even when the consumer - opts into HTTP response-attempt supersession. No WebSocket idempotency guarantee - is assumed. Wrapped error statuses and allowlisted retry headers are retained - in normal retry observations; raw provider error messages are not exposed. +- A send failure, socket close, receive failure, stall, or timeout after sending + reconnects and resends the full request (no `previous_response_id`) within the + retry budget, when the request policy sends `store: false`. The server keeps no + response state for such a request, so a lost or unfinished delivery cannot + create a duplicate stored response or corrupt a continuation. With any other + `store` setting these failures are not replayed. +- After visible output, that resend happens only when the consumer opts into + response-attempt supersession, exactly as for HTTP: the adapter emits + `ModelTurnEvent::ResponseAttemptSuperseded` before the replacement attempt. + Otherwise visible WebSocket output is never replayed. Wrapped error statuses and + allowlisted retry headers are retained in normal retry observations; raw + provider error messages are not exposed. - Dropping a pending `next_event` future during authentication refresh or a WebSocket opening operation makes a retained turn fail safely on its next poll. It does not resume with stale credentials or replay an uncertain send. @@ -141,6 +147,15 @@ have a 30-second ceiling; configured attempt, idle, logical retry-budget, and cancellation bounds also apply. As with HTTP, configure `with_resilience` to set stream idle and whole-turn deadlines. +Both transports also bound each attempt by response progress +(`with_progress_timeouts`, default 120s to the first progress event and 300s +between progress events). Only `response.*` events other than `response.queued`, +`response.in_progress` and `response.metadata` count as progress. Keepalives, +`codex.*` rate-limit/metadata events, SSE comments and ping/pong keep the idle +timer alive but cannot keep a stalled response open. A stall is a retryable +failure. The 300s gap matches the Codex CLI's stream idle timeout, leaving room +for reasoning that runs silently between events. + The optional [live continuation suite](../../docs/live-responses-websocket.md) verifies incremental wire requests and server acceptance using environment credentials. It is ignored by default and makes billable requests when explicitly enabled. @@ -177,8 +192,8 @@ resilience retry budget**. Recovery requires the documented 400 message formats, and malformed error events fail closed: they cannot safely correlate a rejection on a reused socket. Set `with_resilience` with `max_retries >= 1` to allow this recovery. The full retry has no previous ID, so this special recovery cannot -repeat. Accepted responses, visible output, and ambiguous sends/disconnects are -never replayed. Recovery uses the existing retry observations and deadline budget. +repeat. This recovery never replays accepted responses, visible output, or ambiguous +sends/disconnects. Recovery uses the existing retry observations and deadline budget. WebSocket upgrades use a dedicated reqwest HTTP/1 client with redirects and implicit HTTP retries disabled, using the existing reqwest TLS stack. A custom @@ -305,12 +320,18 @@ default. Context content is sent unchanged: the public profile keeps system and Context items as system messages, while the private profile downgrades both to developer messages, defaults `parallel_tool_calls` to `true`, -omits unsupported `max_output_tokens`, sends `originator`/`session-id`, and -replays validated `x-codex-turn-state` only within one logical turn and its -retries. HTTP turn state is accepted only from a successful SSE response; the -equivalent `response.metadata` headers can update retry context, and all header, -metadata, and retry values must agree. It does not perform credential discovery -or model catalog lookups. +omits unsupported `max_output_tokens`, and sends `originator`, +`x-codex-routing-hint: model=`, `thread-id` (the agentkit session ID) and +`session-id`. `session-id` is the request's prompt cache key when one is set, +otherwise the session ID, so sessions sharing a cache key (for example forked +subagents in other processes) route to the same backend and its prompt cache. +It also replays validated `x-codex-turn-state` for sticky routing. Turn state spans one user turn: every request from the latest +user message through its tool rounds and retries. A new user message or a changed +authentication binding clears it. HTTP turn state is accepted only from a +successful SSE response or upgrade; `response.metadata` and +`codex.response.metadata` headers can also update it, and a changed value replaces the old one. Over WebSocket it is also sent as +`client_metadata` on each `response.create`. It does not perform credential +discovery or model catalog lookups. Continuation metadata is versioned and bound to the authentication binding, session, provider item ID, and item kind. It records the originating model as diff --git a/crates/agentkit-provider-openai/src/responses.rs b/crates/agentkit-provider-openai/src/responses.rs index 046f5d1..e2c8720 100644 --- a/crates/agentkit-provider-openai/src/responses.rs +++ b/crates/agentkit-provider-openai/src/responses.rs @@ -46,7 +46,10 @@ const CONTINUATION_METADATA: &str = "openai.responses.continuation.v1"; const CONTINUATION_SCHEMA_VERSION: u64 = 3; const GENERATED_IMAGE_METADATA: &str = "openai.responses.generated_image.v1"; const X_CODEX_TURN_STATE: &str = "x-codex-turn-state"; +const X_CODEX_ROUTING_HINT: &str = "x-codex-routing-hint"; const MAX_CACHE_KEY_BYTES: usize = 256; +const DEFAULT_FIRST_PROGRESS_TIMEOUT: Duration = Duration::from_secs(120); +const DEFAULT_PROGRESS_GAP_TIMEOUT: Duration = Duration::from_secs(300); /// Selects the public Responses API or ChatGPT's private Codex-shaped profile. #[derive(Clone, Copy, Debug, PartialEq, Eq)] @@ -146,6 +149,8 @@ pub struct OpenAIResponsesConfig { originator: Option, transport: OpenAIResponsesTransport, websocket_no_proxy: bool, + first_progress_timeout: Option, + progress_gap_timeout: Option, } impl fmt::Debug for OpenAIResponsesConfig { @@ -157,6 +162,8 @@ impl fmt::Debug for OpenAIResponsesConfig { .field("profile", &self.profile) .field("transport", &self.transport) .field("websocket_no_proxy", &self.websocket_no_proxy) + .field("first_progress_timeout", &self.first_progress_timeout) + .field("progress_gap_timeout", &self.progress_gap_timeout) .field("header_names", &self.headers.keys().collect::>()) .field("request_policy", &self.request_policy) .field("reasoning_effort", &self.reasoning_effort) @@ -200,6 +207,8 @@ impl OpenAIResponsesConfig { originator: None, transport: OpenAIResponsesTransport::Http, websocket_no_proxy: false, + first_progress_timeout: Some(DEFAULT_FIRST_PROGRESS_TIMEOUT), + progress_gap_timeout: Some(DEFAULT_PROGRESS_GAP_TIMEOUT), } } @@ -227,6 +236,8 @@ impl OpenAIResponsesConfig { originator: None, transport: OpenAIResponsesTransport::Http, websocket_no_proxy: false, + first_progress_timeout: Some(DEFAULT_FIRST_PROGRESS_TIMEOUT), + progress_gap_timeout: Some(DEFAULT_PROGRESS_GAP_TIMEOUT), } } @@ -248,6 +259,29 @@ impl OpenAIResponsesConfig { self } + /// Bounds an attempt by response progress rather than by raw traffic. + /// + /// `first` limits the wait from opening an attempt to its first progress event + /// (normally `response.created`); `gap` limits the wait between later progress + /// events. Progress is any `response.*` event except `response.queued`, + /// `response.in_progress` and `response.metadata`: keepalives, rate-limit, + /// metadata and timing events can arrive indefinitely without the response + /// advancing, so they do not reset these timers. A stalled attempt is a + /// retryable failure. `None` disables a bound. + /// + /// Defaults are 120s and 300s. Reasoning can run silently between progress + /// events, so the gap matches the Codex CLI's 300s stream idle timeout: no stream + /// it would keep open is cut short by this bound. + pub fn with_progress_timeouts( + mut self, + first: Option, + gap: Option, + ) -> Self { + self.first_progress_timeout = first; + self.progress_gap_timeout = gap; + self + } + pub fn with_endpoint(mut self, endpoint: impl Into) -> Self { self.endpoint = endpoint.into(); self @@ -391,6 +425,8 @@ impl ModelAdapter for OpenAIResponsesAdapter { session: config, retry_observer: None, websocket: Arc::new(Mutex::new(websocket::Session::default())), + turn_state: Arc::new(Mutex::new(None)), + turn_scope: None, }) } @@ -406,6 +442,8 @@ pub struct OpenAIResponsesSession { session: SessionConfig, retry_observer: Option>, websocket: Arc>, + turn_state: Arc>>, + turn_scope: Option<(Option, agentkit_core::TurnId)>, } #[async_trait] @@ -486,6 +524,18 @@ impl ModelSession for OpenAIResponsesSession { .await; let (body, idempotency_key, auth, deadline, supersession_enabled) = prepared.map_err(|error| tracker.finish(error))?; + // x-codex-turn-state is scoped to one logical turn (every request from + // its input through its tool rounds) and one account. The turn ID, not + // a transcript position, identifies it: compaction can move or reuse + // positions. + let scope = (auth.binding().map(str::to_owned), request.turn_id.clone()); + if self.turn_scope.as_ref() != Some(&scope) { + self.turn_scope = Some(scope); + *self + .turn_state + .lock() + .map_err(|_| tracker.finish(local_error(ProviderFailureReason::Protocol)))? = None; + } OpenAIResponsesTurn::open( ResponsesRequestContext { client: self.client.clone(), @@ -493,7 +543,12 @@ impl ModelSession for OpenAIResponsesSession { body, idempotency_key, session_id: request.session_id.to_string(), - turn_state: Arc::new(Mutex::new(None)), + prompt_cache_key: request + .cache + .as_ref() + .filter(|cache| !matches!(cache.mode, PromptCacheMode::Disabled)) + .and_then(|cache| cache.key.clone()), + turn_state: self.turn_state.clone(), auth, deadline, retries: 0, @@ -666,6 +721,10 @@ impl OpenAIResponsesTurn { } let pending = if let Some(timeout) = expired_attempt_timeout { + self.attempt + .as_mut() + .expect("live attempt is open") + .connection_lost = true; Err(attempt_timeout_failure(timeout)) } else { self.attempt @@ -710,11 +769,21 @@ impl OpenAIResponsesTurn { .budget .saturating_sub(deadline.started_at.elapsed()) }); - let timeout = [idle_timeout, remaining, attempt_remaining] + let stall_remaining = self + .attempt + .as_mut() + .expect("live attempt is open") + .stall_remaining(&self.context.config); + let timeout = [idle_timeout, remaining, attempt_remaining, stall_remaining] .into_iter() .flatten() .min(); - let chunk = { + let chunk = if stall_remaining.is_some_and(|remaining| remaining.is_zero()) { + Ok(Err(HttpError::Timeout { + operation: "response progress", + timeout: Duration::ZERO, + })) + } else { let attempt = self.attempt.as_mut().expect("live attempt is open"); cancellable(attempt.body.next_chunk(timeout), cancellation).await }; @@ -733,16 +802,30 @@ impl OpenAIResponsesTurn { .expect("checked logical deadline") .budget, }))) - } else if let Some(timeout) = self - .attempt - .as_ref() - .and_then(|attempt| attempt.deadline.as_ref()) - .filter(|deadline| deadline.started_at.elapsed() >= deadline.budget) - .map(|deadline| deadline.budget) - { - Err(attempt_timeout_failure(timeout)) } else { - Err(transport_failure(error)) + let attempt = self.attempt.as_mut().expect("live attempt is open"); + attempt.connection_lost = true; + if let Some(timeout) = attempt + .deadline + .as_ref() + .filter(|deadline| deadline.started_at.elapsed() >= deadline.budget) + .map(|deadline| deadline.budget) + { + Err(attempt_timeout_failure(timeout)) + } else if attempt + .stall_remaining(&self.context.config) + .is_some_and(|remaining| remaining.is_zero()) + { + Err(AttemptFailure { + error: Box::new(local_error( + ProviderFailureReason::IdleTimeout, + )), + retryable: true, + headers: None, + }) + } else { + Err(transport_failure(error)) + } } } Ok(Ok(Some(chunk))) => { @@ -780,6 +863,7 @@ impl OpenAIResponsesTurn { Ok(Ok(None)) => { let attempt = self.attempt.as_mut().expect("live attempt is open"); attempt.eof = true; + attempt.connection_lost = true; attempt.truncated.finish().map_err(transport_failure) } } @@ -790,9 +874,9 @@ impl OpenAIResponsesTurn { .attempt .as_ref() .is_some_and(|attempt| matches!(attempt.body, LiveBody::WebSocket(_))); - // A sent WebSocket request has no idempotency guarantee. Only explicit - // provider rejections before visible output on a fresh socket - // can be retried; reused sockets can deliver uncorrelated stale errors. + // A sent WebSocket request has no idempotency guarantee. Explicit + // provider rejections before visible output can be retried only on a + // fresh socket; reused sockets can deliver uncorrelated stale errors. let missing_previous = self.attempt.as_ref().is_some_and(|attempt| { matches!(&attempt.body, LiveBody::WebSocket(connection) if connection.missing_previous(&attempt.decoder)) @@ -804,13 +888,29 @@ impl OpenAIResponsesTurn { if missing_previous { failure.retryable = true; } + // A stalled, timed-out or lost socket is resent in full on a fresh + // socket before visible output, or after it when the session accepts + // superseded attempts. With store:false the server keeps no response + // state, so an unfinished delivery cannot become a duplicate stored + // response or poison a continuation. + let resendable = websocket + && (!self.attempt_output_emitted || self.supersession_enabled) + && self.context.config.request_policy.store == Some(false) + && self + .attempt + .as_ref() + .is_some_and(|attempt| attempt.connection_lost); + if resendable { + failure.retryable = true; + } if websocket + && !resendable && (self.attempt_output_emitted || !self.attempt.as_ref().is_some_and(|attempt| { missing_previous || attempt.decoder.request_rejected && matches!(&attempt.body, LiveBody::WebSocket(connection) - if connection.can_retry_rejection()) + if connection.can_retry_rejection()) })) { return Err(*failure.error); @@ -1766,6 +1866,7 @@ struct ResponsesRequestContext { body: agentkit_http::Bytes, idempotency_key: String, session_id: String, + prompt_cache_key: Option, turn_state: Arc>>, auth: AuthenticationAttempt, deadline: Option, @@ -1803,9 +1904,43 @@ struct LiveAttempt { deadline: Option, eof: bool, closed: bool, + progress: u64, + progress_at: Instant, + connection_lost: bool, } impl LiveAttempt { + fn new( + body: LiveBody, + truncated: TruncatedStreamDetector, + decoder: ResponsesSseDecoder, + ) -> Self { + Self { + body, + truncated, + decoder, + deadline: None, + eof: false, + closed: false, + progress: 0, + progress_at: Instant::now(), + connection_lost: false, + } + } + + fn stall_remaining(&mut self, config: &OpenAIResponsesConfig) -> Option { + if self.decoder.state.progress != self.progress { + self.progress = self.decoder.state.progress; + self.progress_at = Instant::now(); + } + let budget = if self.progress == 0 { + config.first_progress_timeout + } else { + config.progress_gap_timeout + }?; + Some(budget.saturating_sub(self.progress_at.elapsed())) + } + fn pop_event(&mut self) -> Option { if !self.closed && matches!(self.decoder.peek_event(), Some(ModelTurnEvent::Finished(_))) { return None; @@ -1880,9 +2015,16 @@ async fn open_live_attempt( let attempt_deadline = attempt_timeout.map(LogicalDeadline::new); let logical_deadline = context.deadline.clone(); context.websocket_sent = false; + // The first-progress bound covers opening the attempt too (connecting, + // sending, and waiting for response headers), not only the stream after it. + let opened_at = Instant::now(); + let open_timeout = [attempt_timeout, context.config.first_progress_timeout] + .into_iter() + .flatten() + .min(); let mut result = attempt_with_timeout( send_live_attempt(context, cancellation), - attempt_timeout, + open_timeout, logical_deadline.as_ref(), cancellation, ) @@ -1890,8 +2032,18 @@ async fn open_live_attempt( if context.websocket_sent && let Err(failure) = &mut result { - // A timeout may have interrupted the send future after bytes left. - failure.retryable = false; + // A timeout may have interrupted the send future after bytes left. With + // store:false a lost or partial delivery holds no server state, so a + // transport-level send failure is resent in full on a fresh socket. + failure.retryable = context.config.request_policy.store == Some(false) + && failure.error.provider_failure().is_some_and(|failure| { + matches!( + failure.reason, + ProviderFailureReason::Transport + | ProviderFailureReason::AttemptTimeout + | ProviderFailureReason::IdleTimeout + ) + }); } if let Err(failure) = &result { context.tracker.note_failure(&failure.error); @@ -1899,6 +2051,7 @@ async fn open_live_attempt( match result { Ok(mut attempt) => { attempt.deadline = attempt_deadline; + attempt.progress_at = opened_at; return Ok(attempt); } Err(failure) if is_unauthorized(&failure.error) && !context.refreshed => { @@ -1984,27 +2137,36 @@ async fn send_live_attempt( .map_err(|_| nonretryable(local_error(ProviderFailureReason::InvalidRequest)))?, ); } - let sent_turn_state = if context.config.profile == OpenAIResponsesProfile::ChatGptPrivate { + if context.config.profile == OpenAIResponsesProfile::ChatGptPrivate { headers.remove(X_CODEX_TURN_STATE); headers .entry("originator") .or_insert(HeaderValue::from_static("agentkit")); headers.entry("session-id").or_insert( + HeaderValue::from_str( + context + .prompt_cache_key + .as_deref() + .unwrap_or(&context.session_id), + ) + .map_err(|_| nonretryable(local_error(ProviderFailureReason::InvalidRequest)))?, + ); + headers.entry("thread-id").or_insert( HeaderValue::from_str(&context.session_id) .map_err(|_| nonretryable(local_error(ProviderFailureReason::InvalidRequest)))?, ); - let state = context + if let Ok(hint) = HeaderValue::from_str(&format!("model={}", context.config.model)) { + headers.entry(X_CODEX_ROUTING_HINT).or_insert(hint); + } + if let Some(value) = context .turn_state .lock() .map_err(|_| protocol_failure("turn-state lock poisoned"))? - .clone(); - if let Some(value) = &state { - headers.insert(X_CODEX_TURN_STATE, value.clone()); + .clone() + { + headers.insert(X_CODEX_TURN_STATE, value); } - state - } else { - None - }; + } headers.extend(context.auth.headers().clone()); if context .websocket @@ -2075,22 +2237,16 @@ async fn send_live_attempt( if context.config.profile == OpenAIResponsesProfile::ChatGptPrivate && let Some(captured) = validated_turn_state_header(response.headers())? { - if sent_turn_state - .as_ref() - .is_some_and(|expected| expected != captured) - { - return Err(protocol_failure("provider changed x-codex-turn-state")); - } *context .turn_state .lock() .map_err(|_| protocol_failure("turn-state lock poisoned"))? = Some(captured); } let truncated = TruncatedStreamDetector::from_headers(response.headers()); - Ok(LiveAttempt { - body: LiveBody::Http(response.bytes_stream()), + Ok(LiveAttempt::new( + LiveBody::Http(response.bytes_stream()), truncated, - decoder: ResponsesSseDecoder::with_policy( + ResponsesSseDecoder::with_policy( &context.config.model, &context.session_id, context.config.profile, @@ -2099,10 +2255,7 @@ async fn send_live_attempt( context.turn_state.clone(), context.config.limits, ), - deadline: None, - eof: false, - closed: false, - }) + )) } fn retryable_response_status(status: StatusCode, profile: OpenAIResponsesProfile) -> bool { @@ -2479,6 +2632,7 @@ struct ResponsesState { provider_finish_reason: Option, tool_call: bool, next_media: usize, + progress: u64, } impl ResponsesState { @@ -2525,6 +2679,7 @@ impl ResponsesState { provider_finish_reason: None, tool_call: false, next_media: 0, + progress: 0, } } @@ -2534,6 +2689,14 @@ impl ResponsesState { "Responses event followed a terminal event", )); } + if kind.starts_with("response.") + && !matches!( + kind, + "response.queued" | "response.in_progress" | "response.metadata" + ) + { + self.progress = self.progress.saturating_add(1); + } if let Some(sequence) = value.get("sequence_number").and_then(Value::as_u64) { let expected = self .sequence @@ -2603,25 +2766,27 @@ impl ResponsesState { }) } "keepalive" => Ok(()), + "codex.response.metadata" => { + if let Some(captured) = responses_turn_state(value)? { + *self + .turn_state + .lock() + .map_err(|_| protocol_failure("turn-state lock poisoned"))? = + Some(captured); + } + Ok(()) + } "response.metadata" => { self.require_created()?; if let Some(response) = value.get("response") { self.observe_response(Some(response), "response.metadata")?; } if let Some(captured) = responses_turn_state(value)? { - let mut turn_state = self + *self .turn_state .lock() - .map_err(|_| protocol_failure("turn-state lock poisoned"))?; - if turn_state - .as_ref() - .is_some_and(|expected| expected != captured) - { - return Err(protocol_failure( - "response metadata changed x-codex-turn-state", - )); - } - *turn_state = Some(captured); + .map_err(|_| protocol_failure("turn-state lock poisoned"))? = + Some(captured); } Ok(()) } @@ -3797,6 +3962,7 @@ mod tests { use agentkit_core::{Item, SessionId, TurnId}; use agentkit_http::{HeaderValue, HttpClient, HttpRequest, HttpResponse, StatusCode, header}; + use agentkit_loop::PromptCacheRequest; use async_trait::async_trait; use futures_util::stream; @@ -3853,6 +4019,52 @@ mod tests { } } + /// The first request's headers arrive after `delay` (never when `None`) and + /// its stream carries only keepalives; later requests succeed. + struct DelayedHeadersClient { + delay: Option, + sent: Mutex>, + } + + #[async_trait] + impl HttpClient for DelayedHeadersClient { + async fn execute(&self, request: HttpRequest) -> Result { + let first = { + let mut sent = self.sent.lock().unwrap(); + sent.push(Instant::now()); + sent.len() == 1 + }; + if !first { + let body = stream::once(async { + Ok(agentkit_http::Bytes::from_static(SUCCESS.as_bytes())) + }); + return Ok(HttpResponse::new( + StatusCode::OK, + sse_headers(), + request.url, + Box::pin(body), + )); + } + match self.delay { + Some(delay) => sleep(delay).await, + None => std::future::pending::<()>().await, + } + let body = stream::unfold((), |_| async { + sleep(Duration::from_millis(1)).await; + Some(( + Ok(agentkit_http::Bytes::from_static(b": keepalive\n\n")), + (), + )) + }); + Ok(HttpResponse::new( + StatusCode::OK, + sse_headers(), + request.url, + Box::pin(body), + )) + } + } + struct RefreshingAuth { calls: Arc, } @@ -4459,26 +4671,28 @@ data: {"type":"response.completed","sequence_number":18,"response":{"id":"resp-1 } #[tokio::test] - async fn private_profile_scopes_turn_state_to_accepted_logical_turn() { - let mut headers = sse_headers(); - headers.insert(X_CODEX_TURN_STATE, HeaderValue::from_static("state-1")); + async fn private_turn_state_spans_user_turn_adopts_changes_and_resets_on_new_input() { + let state = |value: &'static str| { + let mut headers = sse_headers(); + headers.insert(X_CODEX_TURN_STATE, HeaderValue::from_static(value)); + headers + }; + let ok = |headers| WireResponse { + status: StatusCode::OK, + headers, + body: SUCCESS, + }; let client = Arc::new(ScriptedClient { responses: Mutex::new(VecDeque::from([ WireResponse { status: StatusCode::SERVICE_UNAVAILABLE, - headers: headers.clone(), + headers: state("state-1"), body: "", }, - WireResponse { - status: StatusCode::OK, - headers: headers.clone(), - body: SUCCESS, - }, - WireResponse { - status: StatusCode::OK, - headers, - body: SUCCESS, - }, + ok(state("state-1")), + ok(state("state-2")), + ok(sse_headers()), + ok(sse_headers()), ])), requests: Mutex::new(Vec::new()), }); @@ -4497,16 +4711,26 @@ data: {"type":"response.completed","sequence_number":18,"response":{"id":"resp-1 .start_session(SessionConfig::new("session")) .await .unwrap(); - session.begin_turn(request(), None).await.unwrap(); - let mut second = request(); - second.turn_id = TurnId::new("turn-2"); - session.begin_turn(second, None).await.unwrap(); + let mut request = request(); + session.begin_turn(request.clone(), None).await.unwrap(); + for _ in 0..2 { + request + .transcript + .push(Item::text(ItemKind::Assistant, "tool round")); + session.begin_turn(request.clone(), None).await.unwrap(); + } + request.transcript.push(Item::text(ItemKind::User, "next")); + request.turn_id = TurnId::new("next-turn"); + session.begin_turn(request, None).await.unwrap(); let requests = client.requests.lock().unwrap(); assert_eq!(requests[0].headers["originator"], "agentkit"); assert_eq!(requests[0].headers["session-id"], "session"); + assert_eq!(requests[0].headers[X_CODEX_ROUTING_HINT], "model=gpt-test"); assert!(!requests[0].headers.contains_key(X_CODEX_TURN_STATE)); assert!(!requests[1].headers.contains_key(X_CODEX_TURN_STATE)); - assert!(!requests[2].headers.contains_key(X_CODEX_TURN_STATE)); + assert_eq!(requests[2].headers[X_CODEX_TURN_STATE], "state-1"); + assert_eq!(requests[3].headers[X_CODEX_TURN_STATE], "state-2"); + assert!(!requests[4].headers.contains_key(X_CODEX_TURN_STATE)); assert_eq!(requests[0].body, requests[1].body); assert_eq!( requests[0].headers["idempotency-key"], @@ -4514,6 +4738,187 @@ data: {"type":"response.completed","sequence_number":18,"response":{"id":"resp-1 ); } + #[tokio::test] + async fn private_turn_state_follows_turn_identity_not_transcript_position() { + let state = |value: &'static str| { + let mut headers = sse_headers(); + headers.insert(X_CODEX_TURN_STATE, HeaderValue::from_static(value)); + headers + }; + let ok = |headers| WireResponse { + status: StatusCode::OK, + headers, + body: SUCCESS, + }; + let client = Arc::new(ScriptedClient { + responses: Mutex::new(VecDeque::from([ + ok(state("state-1")), + ok(sse_headers()), + ok(state("state-2")), + ok(sse_headers()), + ])), + requests: Mutex::new(Vec::new()), + }); + let adapter = OpenAIResponsesAdapter::with_client( + OpenAIResponsesConfig::chatgpt_private("gpt-test", Authentication::bearer("secret")), + Http::from_arc(client.clone()), + ); + let mut session = adapter + .start_session(SessionConfig::new("session")) + .await + .unwrap(); + let mut first = request(); + first.transcript = vec![ + Item::text(ItemKind::User, "old"), + Item::text(ItemKind::Assistant, "old answer"), + Item::text(ItemKind::User, "current"), + ]; + session.begin_turn(first.clone(), None).await.unwrap(); + // Compaction drops a prefix during a tool round: the user message moves. + let mut compacted = first.clone(); + compacted.transcript = vec![ + Item::text(ItemKind::User, "current"), + Item::text(ItemKind::Assistant, "tool round"), + ]; + session.begin_turn(compacted, None).await.unwrap(); + // A rolling window puts the next turn's user message at the old index. + let mut next = first.clone(); + next.turn_id = TurnId::new("next-turn"); + next.transcript = vec![ + Item::text(ItemKind::User, "current"), + Item::text(ItemKind::Assistant, "answer"), + Item::text(ItemKind::User, "next"), + ]; + session.begin_turn(next.clone(), None).await.unwrap(); + next.transcript + .push(Item::text(ItemKind::Assistant, "tool round")); + session.begin_turn(next, None).await.unwrap(); + let requests = client.requests.lock().unwrap(); + assert!(!requests[0].headers.contains_key(X_CODEX_TURN_STATE)); + assert_eq!(requests[1].headers[X_CODEX_TURN_STATE], "state-1"); + assert!(!requests[2].headers.contains_key(X_CODEX_TURN_STATE)); + assert_eq!(requests[3].headers[X_CODEX_TURN_STATE], "state-2"); + } + + #[tokio::test] + async fn private_session_id_header_follows_prompt_cache_key() { + let ok = || WireResponse { + status: StatusCode::OK, + headers: sse_headers(), + body: SUCCESS, + }; + let client = Arc::new(ScriptedClient { + responses: Mutex::new(VecDeque::from([ok(), ok(), ok()])), + requests: Mutex::new(Vec::new()), + }); + let adapter = OpenAIResponsesAdapter::with_client( + OpenAIResponsesConfig::chatgpt_private("gpt-test", Authentication::bearer("secret")), + Http::from_arc(client.clone()), + ); + let mut session = adapter + .start_session(SessionConfig::new("session")) + .await + .unwrap(); + for cache in [ + None, + Some(PromptCacheRequest::automatic().with_key("shared-key")), + Some(PromptCacheRequest::disabled().with_key("shared-key")), + ] { + let mut request = request(); + request.cache = cache; + session.begin_turn(request, None).await.unwrap(); + } + let requests = client.requests.lock().unwrap(); + for (request, session_id) in requests.iter().zip(["session", "shared-key", "session"]) { + assert_eq!(request.headers["session-id"], session_id); + assert_eq!(request.headers["thread-id"], "session"); + } + } + + #[test] + fn codex_response_metadata_before_created_captures_turn_state() { + let turn_state = Arc::new(Mutex::new(Some(HeaderValue::from_static("old")))); + let mut decoder = ResponsesSseDecoder::with_policy( + "gpt-test", + "session", + OpenAIResponsesProfile::ChatGptPrivate, + true, + None, + turn_state.clone(), + OpenAIResponsesLimits::default(), + ); + decoder + .push_json( + br#"{"type":"codex.response.metadata","headers":{"x-codex-turn-state":"new"}}"#, + ) + .unwrap(); + decoder + .push_json(br#"{"type":"response.created","sequence_number":1,"response":{"id":"resp-1","model":"gpt-test"}}"#) + .unwrap(); + assert_eq!( + *turn_state.lock().unwrap(), + Some(HeaderValue::from_static("new")) + ); + } + + #[tokio::test] + async fn private_turn_state_resets_on_authentication_binding_change() { + let mut headers = sse_headers(); + headers.insert(X_CODEX_TURN_STATE, HeaderValue::from_static("state-1")); + let response = WireResponse { + status: StatusCode::OK, + headers, + body: SUCCESS, + }; + let client = Arc::new(ScriptedClient { + responses: Mutex::new(VecDeque::from([response.clone(), response])), + requests: Mutex::new(Vec::new()), + }); + let config = OpenAIResponsesConfig::chatgpt_private( + "gpt-test", + Authentication::new(BindingChangingAuth { + calls: Arc::new(AtomicUsize::new(0)), + }), + ); + let adapter = OpenAIResponsesAdapter::with_client(config, Http::from_arc(client.clone())); + let mut session = adapter + .start_session(SessionConfig::new("session")) + .await + .unwrap(); + let mut request = request(); + session.begin_turn(request.clone(), None).await.unwrap(); + request + .transcript + .push(Item::text(ItemKind::Assistant, "tool round")); + session.begin_turn(request, None).await.unwrap(); + let requests = client.requests.lock().unwrap(); + assert!(!requests[1].headers.contains_key(X_CODEX_TURN_STATE)); + } + + #[tokio::test] + async fn public_profile_sends_no_routing_hint() { + let client = Arc::new(ScriptedClient { + responses: Mutex::new(VecDeque::from([WireResponse { + status: StatusCode::OK, + headers: sse_headers(), + body: SUCCESS, + }])), + requests: Mutex::new(Vec::new()), + }); + let adapter = OpenAIResponsesAdapter::with_client( + OpenAIResponsesConfig::public("gpt-test", Authentication::bearer("secret")), + Http::from_arc(client.clone()), + ); + let mut session = adapter + .start_session(SessionConfig::new("session")) + .await + .unwrap(); + session.begin_turn(request(), None).await.unwrap(); + let requests = client.requests.lock().unwrap(); + assert!(!requests[0].headers.contains_key(X_CODEX_ROUTING_HINT)); + assert!(!requests[0].headers.contains_key(X_CODEX_TURN_STATE)); + } + #[tokio::test] async fn response_metadata_turn_state_updates_same_turn_retry_context() { const FAILED: &str = concat!( @@ -5299,6 +5704,90 @@ data: {"type":"response.completed","sequence_number":18,"response":{"id":"resp-1 assert_eq!(client.requests.load(Ordering::SeqCst), 1); } + #[tokio::test] + async fn keepalive_only_sse_stream_stalls_and_retries() { + let client = Arc::new(ActiveStreamClient { + requests: AtomicUsize::new(0), + }); + let config = OpenAIResponsesConfig::new("secret", "gpt-test") + .with_progress_timeouts(Some(Duration::from_millis(20)), None) + .with_resilience(ResilienceConfig { + max_retries: 1, + retry_budget: Duration::from_secs(1), + attempt_timeout: None, + stream_idle_timeout: Some(Duration::from_millis(100)), + initial_backoff: Duration::ZERO, + max_backoff: Duration::ZERO, + }); + let adapter = OpenAIResponsesAdapter::with_client(config, Http::from_arc(client.clone())); + let mut session = adapter + .start_session(SessionConfig::new("session")) + .await + .unwrap(); + let mut turn = session.begin_turn(request(), None).await.unwrap(); + let error = turn.next_event(None).await.unwrap_err(); + let failure = error.provider_failure().unwrap(); + assert_eq!(failure.reason, ProviderFailureReason::RetryExhausted); + assert_eq!( + failure.last_attempt_reason, + Some(ProviderFailureReason::IdleTimeout) + ); + assert_eq!(client.requests.load(Ordering::SeqCst), 2); + } + + #[tokio::test] + async fn first_progress_bound_covers_waiting_for_response_headers() { + const FIRST: Duration = Duration::from_millis(600); + const DELAY: Duration = Duration::from_millis(450); + for delay in [None, Some(DELAY)] { + let client = Arc::new(DelayedHeadersClient { + delay, + sent: Mutex::new(Vec::new()), + }); + let config = OpenAIResponsesConfig::new("secret", "gpt-test") + .with_progress_timeouts(Some(FIRST), None) + .with_resilience(ResilienceConfig { + max_retries: 1, + retry_budget: Duration::from_secs(5), + attempt_timeout: None, + stream_idle_timeout: None, + initial_backoff: Duration::ZERO, + max_backoff: Duration::ZERO, + }); + let adapter = + OpenAIResponsesAdapter::with_client(config, Http::from_arc(client.clone())); + let mut session = adapter + .start_session(SessionConfig::new("session")) + .await + .unwrap(); + let mut turn = session.begin_turn(request(), None).await.unwrap(); + let events = tokio::time::timeout(Duration::from_secs(3), async { + let mut events = Vec::new(); + while let Some(event) = turn.next_event(None).await.unwrap() { + events.push(event); + } + events + }) + .await + .expect("an attempt without headers or progress must not hang"); + assert!( + events + .iter() + .any(|event| matches!(event, ModelTurnEvent::Finished(_))) + ); + let sent = client.sent.lock().unwrap(); + assert_eq!(sent.len(), 2, "{delay:?}"); + let first_attempt = sent[1] - sent[0]; + assert!(first_attempt >= FIRST, "{delay:?}: {first_attempt:?}"); + // Waiting for headers counts against the same bound; without that the + // delayed attempt would last at least FIRST + DELAY. + assert!( + first_attempt < FIRST + DELAY, + "{delay:?}: {first_attempt:?}" + ); + } + } + #[tokio::test] async fn aggregate_wire_limit_spans_retries() { let client = Arc::new(ScriptedClient { diff --git a/crates/agentkit-provider-openai/src/responses/websocket.rs b/crates/agentkit-provider-openai/src/responses/websocket.rs index b06ca27..eb7df6f 100644 --- a/crates/agentkit-provider-openai/src/responses/websocket.rs +++ b/crates/agentkit-provider-openai/src/responses/websocket.rs @@ -314,8 +314,22 @@ impl Connection { pub(super) async fn send( context: &mut ResponsesRequestContext, - mut headers: HeaderMap, + headers: HeaderMap, ) -> Result, AttemptFailure> { + // Encode the complete credential-bound transcript first. The connection-local + // checkpoint is only an optimization, never an alternative source of history. + let mut value: Value = serde_json::from_slice(&context.body) + .map_err(|_| protocol_failure("invalid encoded Responses request"))?; + let fields = value + .as_object_mut() + .ok_or_else(|| protocol_failure("invalid Responses request object"))?; + fields.remove("stream"); + fields.remove("background"); + fields.insert("type".into(), json!("response.create")); + let full = Zeroizing::new( + serde_json::to_string(&value) + .map_err(|_| protocol_failure("could not serialize WebSocket request"))?, + ); let previous = context .websocket .as_mut() @@ -335,116 +349,11 @@ pub(super) async fn send( context.tracker.accounting.attempts = context.tracker.accounting.attempts.saturating_add(1); connection } else { - headers.remove("idempotency-key"); // WebSocket response.create has no idempotency contract. - headers.remove("accept"); - headers.remove("content-type"); - headers.insert( - "openai-beta", - HeaderValue::from_static("responses_websockets=2026-02-06"), - ); - let key = generate_key(); - headers.insert("connection", HeaderValue::from_static("Upgrade")); - headers.insert("upgrade", HeaderValue::from_static("websocket")); - headers.insert("sec-websocket-version", HeaderValue::from_static("13")); - headers.insert( - "sec-websocket-key", - HeaderValue::from_str(&key).map_err(|_| protocol_failure("invalid WebSocket nonce"))?, - ); - let client = reqwest::Client::builder() - .http1_only() - .redirect(reqwest::redirect::Policy::none()) - .retry(reqwest::retry::never()) - .connect_timeout(HANDSHAKE_TIMEOUT) - .timeout(HANDSHAKE_TIMEOUT); - let client = if context.config.websocket_no_proxy { - client.no_proxy() - } else { - client - }; - let client = client - .build() - .map_err(|_| protocol_failure("could not build WebSocket upgrade client"))?; - context.tracker.accounting.attempts = context.tracker.accounting.attempts.saturating_add(1); - // HTTP/1 upgrade on https uses reqwest's existing rustls trust/proxy stack; - // it is the same wire operation as connecting to the corresponding wss URL. - let response = client - .get(&context.config.endpoint) - .headers(headers) - .send() - .await - .map_err(|error| transport_failure(HttpError::request(error)))?; - let status = response.status(); - if status == StatusCode::UPGRADE_REQUIRED - && context.config.transport == OpenAIResponsesTransport::Auto - { - context - .websocket - .as_mut() - .expect("WebSocket lease") - .fallback()?; + let Some(connection) = connect(context, headers).await? else { return Ok(None); - } - if status != StatusCode::SWITCHING_PROTOCOLS { - return Err(AttemptFailure { - error: Box::new(provider_error( - ProviderFailureReason::HttpStatus, - ProviderClassification { - http_status: Some(status.as_u16()), - ..ProviderClassification::default() - }, - )), - retryable: retryable_response_status(status, context.config.profile), - headers: retry_headers(response.headers()), - }); - } - validate_handshake(response.headers(), &key)?; - let turn_state = if context.config.profile == OpenAIResponsesProfile::ChatGptPrivate { - validated_turn_state_header(response.headers())? - } else { - None }; - if let Some(captured) = turn_state { - let mut state = context - .turn_state - .lock() - .map_err(|_| protocol_failure("turn-state lock poisoned"))?; - if state.as_ref().is_some_and(|expected| expected != captured) { - return Err(protocol_failure("provider changed x-codex-turn-state")); - } - *state = Some(captured); - } - let socket = response - .upgrade() - .await - .map_err(|_| protocol_failure("WebSocket upgrade failed"))?; - let limit = context.config.limits.max_attempt_bytes; - let config = WebSocketConfig::default() - .max_message_size(Some(limit)) - .max_frame_size(Some(limit)) - .write_buffer_size(0); - Box::new(Connection { - stream: WebSocketStream::from_raw_socket(socket, Role::Client, Some(config)).await, - auth_headers: context.auth.headers().clone(), - binding: context.auth.binding().map(str::to_owned), - completed_ids: BTreeSet::new(), - checkpoint: None, - previous_response_id: None, - }) + connection }; - // Encode the complete credential-bound transcript first. The connection-local - // checkpoint is only an optimization, never an alternative source of history. - let mut value: Value = serde_json::from_slice(&context.body) - .map_err(|_| protocol_failure("invalid encoded Responses request"))?; - let fields = value - .as_object_mut() - .ok_or_else(|| protocol_failure("invalid Responses request object"))?; - fields.remove("stream"); - fields.remove("background"); - fields.insert("type".into(), json!("response.create")); - let full = Zeroizing::new( - serde_json::to_string(&value) - .map_err(|_| protocol_failure("could not serialize WebSocket request"))?, - ); connection.previous_response_id = None; if let Some(checkpoint) = connection.checkpoint.take() && let Some(suffix) = checkpoint.suffix(&value) @@ -462,16 +371,9 @@ pub(super) async fn send( bytes: 0, completed: false, }); - let serialized = serde_json::to_string(&value); + let request = serialize_request(context, &mut value); zeroize_encrypted_content(&mut value); - let request = Zeroizing::new( - serialized.map_err(|_| protocol_failure("could not serialize WebSocket request"))?, - ); - if request.len() > context.config.limits.max_request_bytes { - return Err(protocol_failure( - "Responses WebSocket request exceeds byte limit", - )); - } + let request = request?; // Once send is polled, delivery is ambiguous. Never retry send/timeout errors. context.websocket_sent = true; run_bounded_http( @@ -488,10 +390,10 @@ pub(super) async fn send( ) .await .map_err(|e| nonretryable(http_loop_error(e)))?; - Ok(Some(LiveAttempt { - body: LiveBody::WebSocket(connection), - truncated: TruncatedStreamDetector::from_headers(&HeaderMap::new()), - decoder: ResponsesSseDecoder::with_policy( + Ok(Some(LiveAttempt::new( + LiveBody::WebSocket(connection), + TruncatedStreamDetector::from_headers(&HeaderMap::new()), + ResponsesSseDecoder::with_policy( &context.config.model, &context.session_id, context.config.profile, @@ -500,10 +402,128 @@ pub(super) async fn send( context.turn_state.clone(), context.config.limits, ), - deadline: None, - eof: false, - closed: false, - })) + ))) +} + +/// Adds per-message fields that are excluded from checkpoint comparison. +fn serialize_request( + context: &ResponsesRequestContext, + value: &mut Value, +) -> Result, AttemptFailure> { + if context.config.profile == OpenAIResponsesProfile::ChatGptPrivate + && let Some(state) = context + .turn_state + .lock() + .map_err(|_| protocol_failure("turn-state lock poisoned"))? + .as_ref() + .and_then(|state| state.to_str().ok()) + { + value["client_metadata"] = json!({ X_CODEX_TURN_STATE: state }); + } + let request = Zeroizing::new( + serde_json::to_string(value) + .map_err(|_| protocol_failure("could not serialize WebSocket request"))?, + ); + if request.len() > context.config.limits.max_request_bytes { + return Err(protocol_failure( + "Responses WebSocket request exceeds byte limit", + )); + } + Ok(request) +} + +async fn connect( + context: &mut ResponsesRequestContext, + mut headers: HeaderMap, +) -> Result>, AttemptFailure> { + headers.remove("idempotency-key"); // WebSocket response.create has no idempotency contract. + headers.remove("accept"); + headers.remove("content-type"); + headers.insert( + "openai-beta", + HeaderValue::from_static("responses_websockets=2026-02-06"), + ); + let key = generate_key(); + headers.insert("connection", HeaderValue::from_static("Upgrade")); + headers.insert("upgrade", HeaderValue::from_static("websocket")); + headers.insert("sec-websocket-version", HeaderValue::from_static("13")); + headers.insert( + "sec-websocket-key", + HeaderValue::from_str(&key).map_err(|_| protocol_failure("invalid WebSocket nonce"))?, + ); + let client = reqwest::Client::builder() + .http1_only() + .redirect(reqwest::redirect::Policy::none()) + .retry(reqwest::retry::never()) + .connect_timeout(HANDSHAKE_TIMEOUT) + .timeout(HANDSHAKE_TIMEOUT); + let client = if context.config.websocket_no_proxy { + client.no_proxy() + } else { + client + }; + let client = client + .build() + .map_err(|_| protocol_failure("could not build WebSocket upgrade client"))?; + context.tracker.accounting.attempts = context.tracker.accounting.attempts.saturating_add(1); + // HTTP/1 upgrade on https uses reqwest's existing rustls trust/proxy stack; + // it is the same wire operation as connecting to the corresponding wss URL. + let response = client + .get(&context.config.endpoint) + .headers(headers) + .send() + .await + .map_err(|error| transport_failure(HttpError::request(error)))?; + let status = response.status(); + if status == StatusCode::UPGRADE_REQUIRED + && context.config.transport == OpenAIResponsesTransport::Auto + { + context + .websocket + .as_mut() + .expect("WebSocket lease") + .fallback()?; + return Ok(None); + } + if status != StatusCode::SWITCHING_PROTOCOLS { + return Err(AttemptFailure { + error: Box::new(provider_error( + ProviderFailureReason::HttpStatus, + ProviderClassification { + http_status: Some(status.as_u16()), + ..ProviderClassification::default() + }, + )), + retryable: retryable_response_status(status, context.config.profile), + headers: retry_headers(response.headers()), + }); + } + validate_handshake(response.headers(), &key)?; + if context.config.profile == OpenAIResponsesProfile::ChatGptPrivate + && let Some(captured) = validated_turn_state_header(response.headers())? + { + *context + .turn_state + .lock() + .map_err(|_| protocol_failure("turn-state lock poisoned"))? = Some(captured); + } + let socket = response + .upgrade() + .await + .map_err(|_| protocol_failure("WebSocket upgrade failed"))?; + let limit = context.config.limits.max_attempt_bytes; + let config = WebSocketConfig::default() + .max_message_size(Some(limit)) + .max_frame_size(Some(limit)) + .write_buffer_size(0); + Ok(Some(Box::new(Connection { + stream: WebSocketStream::from_raw_socket(socket, Role::Client, Some(config)).await, + auth_headers: context.auth.headers().clone(), + binding: context.auth.binding().map(str::to_owned), + completed_ids: BTreeSet::new(), + checkpoint: None, + previous_response_id: None, + }))) } fn validate_handshake(headers: &HeaderMap, key: &str) -> Result<(), AttemptFailure> { diff --git a/crates/agentkit-provider-openai/src/responses/websocket/tests.rs b/crates/agentkit-provider-openai/src/responses/websocket/tests.rs index 50c09df..c8e8268 100644 --- a/crates/agentkit-provider-openai/src/responses/websocket/tests.rs +++ b/crates/agentkit-provider-openai/src/responses/websocket/tests.rs @@ -1,6 +1,7 @@ //! Loopback-only transport tests; no live inference. use super::*; use agentkit_core::{SessionId, TurnId}; +use agentkit_loop::PromptCacheRequest; use std::io::{Read, Write}; use std::net::{TcpListener, TcpStream}; use std::thread; @@ -1783,3 +1784,266 @@ fn websocket_proxy_policy_child_process() { blocked.join().unwrap(); direct.join().unwrap(); } + +fn private_config(endpoint: &str) -> OpenAIResponsesConfig { + let base = config(endpoint, OpenAIResponsesTransport::WebSocket); + OpenAIResponsesConfig::chatgpt_private("gpt-test", Authentication::bearer("loopback-test")) + .with_endpoint(endpoint) + .with_transport(OpenAIResponsesTransport::WebSocket) + .with_resilience(base.resilience.unwrap()) +} + +#[tokio::test] +#[allow( + clippy::result_large_err, + reason = "tungstenite fixes the handshake callback error type" +)] +async fn private_handshake_routes_by_cache_key_and_sends_turn_state_per_message() { + for key in [None, Some("shared-key")] { + let (endpoint, peer) = server(move |listener| { + let mut ws = tungstenite::accept_hdr( + accept(&listener), + |request: &tungstenite::handshake::server::Request, + mut response: tungstenite::handshake::server::Response| { + let headers = request.headers(); + assert_eq!(headers[X_CODEX_ROUTING_HINT], "model=gpt-test"); + assert_eq!(headers["session-id"], key.unwrap_or("ws-session")); + assert_eq!(headers["thread-id"], "ws-session"); + assert!(!headers.contains_key(X_CODEX_TURN_STATE)); + response + .headers_mut() + .insert(X_CODEX_TURN_STATE, "ws-state".parse().unwrap()); + Ok(response) + }, + ) + .unwrap(); + let first = receive(&mut ws); + assert_eq!(first["client_metadata"][X_CODEX_TURN_STATE], "ws-state"); + success(&mut ws); + let next = receive_continuation(&mut ws); + assert_eq!(next["previous_response_id"], "resp-1"); + assert_eq!(next["client_metadata"][X_CODEX_TURN_STATE], "ws-state"); + success_id(&mut ws, "resp-2"); + }); + let adapter = OpenAIResponsesAdapter::new(private_config(&endpoint)).unwrap(); + let mut session = adapter + .start_session(SessionConfig::new("ws-session")) + .await + .unwrap(); + let mut request = request(); + request.session_id = SessionId::new("ws-session"); + request.cache = key.map(|key| PromptCacheRequest::automatic().with_key(key)); + let mut turn = session.begin_turn(request.clone(), None).await.unwrap(); + let events = drain(&mut turn).await.unwrap(); + finished(&events); + request.transcript.extend( + events + .into_iter() + .find_map(|event| match event { + ModelTurnEvent::Finished(result) => Some(result.output_items), + _ => None, + }) + .unwrap(), + ); + append_tool_result(&mut request); + let mut turn = session.begin_turn(request, None).await.unwrap(); + finished(&drain(&mut turn).await.unwrap()); + peer.join().unwrap(); + } +} + +const STALL: Duration = Duration::from_millis(300); + +fn stall_config(endpoint: &str) -> OpenAIResponsesConfig { + let mut cfg = private_config(endpoint).with_progress_timeouts(Some(STALL), Some(STALL)); + cfg.resilience = Some(ResilienceConfig { + max_retries: 1, + retry_budget: WAIT, + attempt_timeout: Some(Duration::from_secs(4)), + stream_idle_timeout: Some(Duration::from_secs(4)), + initial_backoff: Duration::ZERO, + max_backoff: Duration::ZERO, + }); + cfg +} +fn created(ws: &mut Socket) { + ws.send(Message::Text( + r#"{"type":"response.created","sequence_number":1,"response":{"id":"stalled","model":"gpt-test"}}"#.into(), + )) + .unwrap(); +} +// Keeps a stalled socket busy with non-progress traffic until the client drops it. +fn keepalive_until_closed(ws: &mut Socket) { + let start = Instant::now(); + while start.elapsed() < WAIT { + let sent = ws + .send(Message::Text(r#"{"type":"keepalive"}"#.into())) + .and_then(|()| { + ws.send(Message::Text( + r#"{"type":"codex.rate_limits","primary":{"used_percent":1}}"#.into(), + )) + }); + if sent.is_err() { + return; + } + thread::sleep(Duration::from_millis(20)); + } + panic!("client kept a stalled socket open"); +} +fn no_connection_within(listener: &TcpListener, wait: Duration) { + thread::sleep(wait); + assert!( + listener + .accept() + .is_err_and(|e| e.kind() == std::io::ErrorKind::WouldBlock), + "unexpected retry connection" + ); +} + +#[tokio::test] +async fn non_progress_traffic_stalls_attempt_and_retries_full_request_on_fresh_socket() { + for accepted in [false, true] { + let (endpoint, peer) = server(move |listener| { + let mut first = socket(&listener); + let original = receive(&mut first); + if accepted { + created(&mut first); + } + keepalive_until_closed(&mut first); + let mut second = socket(&listener); + assert_eq!(receive(&mut second), original); + success(&mut second); + }); + let adapter = OpenAIResponsesAdapter::new(stall_config(&endpoint)).unwrap(); + let mut session = adapter + .start_session(SessionConfig::new("session")) + .await + .unwrap(); + let mut turn = session.begin_turn(request(), None).await.unwrap(); + finished(&drain(&mut turn).await.unwrap()); + peer.join().unwrap(); + } +} + +#[tokio::test] +async fn silent_socket_after_send_is_retried_on_fresh_socket() { + let (endpoint, peer) = server(|listener| { + let mut first = socket(&listener); + receive(&mut first); + let mut second = socket(&listener); + receive(&mut second); + success(&mut second); + drop(first); + }); + let adapter = OpenAIResponsesAdapter::new(stall_config(&endpoint)).unwrap(); + let mut session = adapter + .start_session(SessionConfig::new("session")) + .await + .unwrap(); + let start = Instant::now(); + let mut turn = session.begin_turn(request(), None).await.unwrap(); + finished(&drain(&mut turn).await.unwrap()); + assert!(start.elapsed() >= STALL); + peer.join().unwrap(); +} + +#[tokio::test] +async fn stall_after_visible_output_fails_without_retry() { + let (endpoint, peer) = server(|listener| { + let mut ws = socket(&listener); + receive(&mut ws); + for line in SUCCESS + .replace("resp-1", "stalled") + .lines() + .filter_map(|line| line.strip_prefix("data: ")) + .take(4) + { + ws.send(Message::Text(line.into())).unwrap(); + } + keepalive_until_closed(&mut ws); + no_connection_within(&listener, Duration::from_millis(500)); + }); + let adapter = OpenAIResponsesAdapter::new(stall_config(&endpoint)).unwrap(); + let mut session = adapter + .start_session(SessionConfig::new("session")) + .await + .unwrap(); + let mut turn = session.begin_turn(request(), None).await.unwrap(); + let error = drain(&mut turn).await.unwrap_err(); + assert_eq!( + error.provider_failure().unwrap().reason, + ProviderFailureReason::IdleTimeout + ); + peer.join().unwrap(); +} + +#[tokio::test] +async fn lost_socket_after_visible_output_supersedes_and_resends_when_opted_in() { + for stalled in [true, false] { + let (endpoint, peer) = server(move |listener| { + let mut first = socket(&listener); + let original = receive(&mut first); + for line in SUCCESS + .replace("resp-1", "lost") + .lines() + .filter_map(|line| line.strip_prefix("data: ")) + .take(4) + { + first.send(Message::Text(line.into())).unwrap(); + } + if stalled { + keepalive_until_closed(&mut first); + } else { + drop(first); + } + let mut second = socket(&listener); + assert_eq!(receive(&mut second), original); + success(&mut second); + }); + let adapter = OpenAIResponsesAdapter::new(stall_config(&endpoint)).unwrap(); + let mut session = adapter + .start_session(SessionConfig::new("session").with_response_attempt_supersession()) + .await + .unwrap(); + let mut turn = session.begin_turn(request(), None).await.unwrap(); + let events = drain(&mut turn).await.unwrap(); + let superseded = events + .iter() + .position(|event| matches!(event, ModelTurnEvent::ResponseAttemptSuperseded)) + .expect("visible attempt was not superseded"); + assert!( + events[superseded..] + .iter() + .any(|event| matches!(event, ModelTurnEvent::Finished(_))) + ); + peer.join().unwrap(); + } +} + +#[tokio::test] +async fn slow_progressing_stream_outlives_progress_gap() { + let (endpoint, peer) = server(|listener| { + let mut ws = socket(&listener); + receive(&mut ws); + for line in SUCCESS + .lines() + .filter_map(|line| line.strip_prefix("data: ")) + { + thread::sleep(STALL / 2); + ws.send(Message::Text(r#"{"type":"keepalive"}"#.into())) + .unwrap(); + ws.send(Message::Text(line.into())).unwrap(); + } + no_connection_within(&listener, Duration::from_millis(100)); + }); + let adapter = OpenAIResponsesAdapter::new(stall_config(&endpoint)).unwrap(); + let mut session = adapter + .start_session(SessionConfig::new("session")) + .await + .unwrap(); + let mut turn = session.begin_turn(request(), None).await.unwrap(); + let start = Instant::now(); + finished(&drain(&mut turn).await.unwrap()); + assert!(start.elapsed() > STALL * 2); + peer.join().unwrap(); +} diff --git a/crates/agentkit-tool-compose/Cargo.toml b/crates/agentkit-tool-compose/Cargo.toml index 4dd9366..36b0371 100644 --- a/crates/agentkit-tool-compose/Cargo.toml +++ b/crates/agentkit-tool-compose/Cargo.toml @@ -4,7 +4,7 @@ homepage.workspace = true name = "agentkit-tool-compose" readme = "README.md" repository.workspace = true -version = "0.10.11" +version = "0.10.12" edition.workspace = true license.workspace = true rust-version.workspace = true @@ -14,7 +14,7 @@ agentkit-core = { version = "0.10.5", path = "../agentkit-core" } agentkit-tools-core = { version = "0.10.5", path = "../agentkit-tools-core" } async-trait.workspace = true mlua = { workspace = true, optional = true } -runlet = { version = "0.6.0", optional = true } +runlet = { version = "0.6.1", optional = true } serde = { workspace = true, features = ["derive"] } serde_json.workspace = true serde_toon2 = { version = "0.2.0", optional = true } diff --git a/crates/agentkit-tool-compose/src/lib.rs b/crates/agentkit-tool-compose/src/lib.rs index 95bf80c..f8d21e3 100644 --- a/crates/agentkit-tool-compose/src/lib.rs +++ b/crates/agentkit-tool-compose/src/lib.rs @@ -701,12 +701,12 @@ impl ComposeTool { json!({ "type": "object", "properties": { + "input": { + "description": "Optional JSON value exposed to the script as global input. Its strings are used verbatim. When text would need escaping to fit in a script string literal, such as file contents or code containing quotes or backslashes, pass it here and reference it instead of escaping it." + }, "script": { "type": "string", "description": backend.script_description() - }, - "input": { - "description": "Optional JSON value exposed to the script as global input." } }, "required": ["script"], diff --git a/crates/agentkit-tool-compose/src/runlet_backend.rs b/crates/agentkit-tool-compose/src/runlet_backend.rs index 4438895..914f0ef 100644 --- a/crates/agentkit-tool-compose/src/runlet_backend.rs +++ b/crates/agentkit-tool-compose/src/runlet_backend.rs @@ -175,8 +175,9 @@ const RULES_PRIMER: &str = "Run a Runlet program that composes available tools. `result = after earlier { return publish_item({ id: 4 }) }`. Calls lexically created \ inside the `after` block wait for every prerequisite to succeed. Source order alone \ does not sequence calls.\n\ - - Objects `{ key: value }`, lists `[a, b]`, strings, integers, floats, booleans, \ - null. String concat with `+`; `a + b` on two objects merges them shallowly (right \ + - Objects `{ key: value }`, lists `[a, b]`, strings (`\"…\"` or `'…'`, newlines \ + allowed; escapes \\n \\t \\\" \\' \\\\ \\uXXXX, any other backslash stays as written), \ + integers, floats, booleans, null. String concat with `+`; `a + b` on two objects merges them shallowly (right \ side wins). `{ [expr]: value }` computes a property key (scalars stringify: \ `{ [u.id]: u }` keys by \"42\"). Comparisons, `and`/`or`/`not`, and `x in xs` \ (list membership, substring, object key).\n\ @@ -237,7 +238,9 @@ const RULES_PRIMER: &str = "Run a Runlet program that composes available tools. time.parse(\"2026-07-12T09:30:00Z\") -> epoch ms, time.format(ms) -> RFC 3339; \ time math is plain integer ms (86400000 per day). \ Substring check is the `in` operator (`\"x\" in s`) — there is no text.contains.\n\ - - The global `input` holds the JSON value passed alongside the program.\n\n\ + - The global `input` holds the JSON value passed alongside the program; its strings \ + are used verbatim, never parsed as Runlet. Text that would need escaping as a literal \ + goes there.\n\n\ Example — fetch and filter concurrently, aggregate with fold:\n\ listing = list_items({ page: 1 })\n\ open_items = for item in listing.items {\n\ @@ -273,9 +276,12 @@ remaining = for page in list.range(2, first.total_pages + 1) { return result.items } listing = fold acc = first.items for page in remaining { return acc + page } -config = json.parse(input.settings) # `input` is the JSON value submitted with the program +saved = write_file({ path: input.path, content: input.source }) + # `input` is the JSON value submitted with the program; + # its strings are used verbatim, never parsed as Runlet. + # Text that would need escaping as a literal goes there. -# `config` and the listing work share no data, so they run IN PARALLEL — this +# `saved` and the listing work share no data, so they run IN PARALLEL — this # includes effectful calls. There is no await; ordinary result references create # dependencies. Source order alone never sequences independent calls. @@ -302,8 +308,8 @@ shaped = for record in listing { # concurrent loop; the hos # lexically created in this block wait until `shaped` succeeds; this is a real # ordering edge, unlike source order. _ = after shaped { # explicit discard still evaluates even pure work - return log_event({ kind: "shaping_complete" }) -} + return log_event({ kind: 'shaping_complete' }) # strings: "…" or '…', newlines allowed; escapes \n \t \" \' +} # \\ \uXXXX, any other \x stays as written total = fold acc = 0 for row in shaped { # fold is THE way to aggregate: sequential reduce, return acc + row.amount # the body's return becomes the next accumulator; @@ -314,7 +320,7 @@ by_id = fold acc = {} for row in shaped { } # object + object merges shallowly, right side wins. # grouping idiom: acc + { [row.kind]: (acc[row.kind] if row.kind in acc else []) + [row] } -label = if total >= config.threshold { # block-bodied if is an expression; every branch +label = if total >= input.threshold { # block-bodied if is an expression; every branch audit = log_event({ kind: "high", total }) # is a block ending in `return`; only the selected return "high" # branch runs (including its writes). } else if total > 0 {