From 3588a873a2440d1c33cefe0f4066aa9e072ffd93 Mon Sep 17 00:00:00 2001 From: daniel Date: Wed, 23 Sep 2026 14:11:54 -0700 Subject: [PATCH] feat(openai): send incremental Responses WebSocket input --- Cargo.lock | 2 +- crates/agentkit-provider-openai/Cargo.toml | 2 +- crates/agentkit-provider-openai/README.md | 54 +- .../agentkit-provider-openai/src/responses.rs | 67 +- .../src/responses/websocket.rs | 191 +++- .../src/responses/websocket/tests.rs | 879 +++++++++++++++++- .../tests/live_responses_websocket.rs | 502 ++++++++++ docs/live-responses-websocket.md | 110 +++ 8 files changed, 1780 insertions(+), 27 deletions(-) create mode 100644 crates/agentkit-provider-openai/tests/live_responses_websocket.rs create mode 100644 docs/live-responses-websocket.md diff --git a/Cargo.lock b/Cargo.lock index 852ed00..9648c80 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -353,7 +353,7 @@ dependencies = [ [[package]] name = "agentkit-provider-openai" -version = "0.10.10" +version = "0.10.11" dependencies = [ "agentkit-adapter-completions", "agentkit-core", diff --git a/crates/agentkit-provider-openai/Cargo.toml b/crates/agentkit-provider-openai/Cargo.toml index 5da3482..ce25adb 100644 --- a/crates/agentkit-provider-openai/Cargo.toml +++ b/crates/agentkit-provider-openai/Cargo.toml @@ -7,7 +7,7 @@ edition.workspace = true license.workspace = true repository.workspace = true rust-version.workspace = true -version = "0.10.10" +version = "0.10.11" [dependencies] tokio-tungstenite = { version = "=0.29.0", default-features = false, features = ["handshake"] } diff --git a/crates/agentkit-provider-openai/README.md b/crates/agentkit-provider-openai/README.md index a34bb99..c585af7 100644 --- a/crates/agentkit-provider-openai/README.md +++ b/crates/agentkit-provider-openai/README.md @@ -120,9 +120,10 @@ WebSocket retries are intentionally more conservative than HTTP retries: - Handshake status failures use the existing bounded retry policy and observer. - A wrapped HTTP error before `response.created` (and before visible output) may reconnect and retry only on a fresh socket. On reused sockets, errors lack - reliable request correlation and may belong to a previous turn, so they never - trigger automatic replay or authentication refresh. An accepted response is - never automatically replayed. + reliable request correlation and may belong to a previous turn, so generic + 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 @@ -140,15 +141,44 @@ 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. -### Current limitations - -The **full credential-bound transcript is always authoritative and always sent**. -This release deliberately does not send `previous_response_id` or maintain an -incremental response-ID cache: reconstructing an exact prefix from normalized -text/tool/reasoning/image output without losing provider fields requires a -separate lossless compatibility proof. Reuse therefore saves connection setup, -not request transcript bytes. Compaction, changed inputs, and reconnection do not -risk a stale server-side prefix. +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. + +### Incremental continuation and limitations + +The **full credential-bound transcript remains authoritative**. After a successful +`response.completed`, the live connection retains a bounded, memory-only checkpoint: +the response ID, full encoded request, raw completed output items, and their normal +transcript replay encoding. When all non-input request fields match and the next +input begins with the previous input plus that replay output, the next +`response.create` sends `previous_response_id` and only the new input suffix. +This includes tool results and later user messages, including rounds containing +encrypted reasoning or generated images. No continuation state is persisted. + +Compatibility uses the same credential-bound encoder as an ordinary full request. +In particular, function arguments are serialized from parsed JSON, message output +metadata such as status and annotations is not replayed, and reasoning replays its +ID and encrypted content with an empty summary. These are the adapter's transcript +semantics, not byte equality against the server's raw output. Changes to retained +text, call IDs/arguments, protected reasoning or image data fail the prefix proof. +Unreadable/unrepresentable or oversized checkpoints disable the optimization. +Changes to model, tools, instructions, reasoning settings, or compacted/edited +history send the full input without an ID. Reconnection, authentication changes, +cancellation and failures discard the connection checkpoint. Checkpoint request +and output buffers are bounded by the configured request/item limits and zeroized +on drop; the socket also bounds completed response-ID correlation history. + +An explicit `previous_response_not_found` rejection before acceptance or visible +output may reconnect and retry once with the full request, **within the configured +resilience retry budget**. Recovery requires the documented 400 +`invalid_request_error` message naming the exact predecessor sent by this request +(`Previous response with id '' not found.`). Code-only errors, older IDs, changed +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. WebSocket upgrades use a dedicated reqwest HTTP/1 client with redirects and implicit HTTP retries disabled, using the existing reqwest TLS stack. A custom diff --git a/crates/agentkit-provider-openai/src/responses.rs b/crates/agentkit-provider-openai/src/responses.rs index 3a32f68..84d8657 100644 --- a/crates/agentkit-provider-openai/src/responses.rs +++ b/crates/agentkit-provider-openai/src/responses.rs @@ -739,10 +739,17 @@ impl OpenAIResponsesTurn { } else { let attempt = self.attempt.as_mut().expect("live attempt is open"); attempt.truncated.observe(&chunk); - match &attempt.body { + match &mut attempt.body { LiveBody::Http(_) => attempt.decoder.push(&chunk), - LiveBody::WebSocket(_) => { + LiveBody::WebSocket(connection) => { let result = attempt.decoder.push_json(&chunk); + if result.is_ok() { + connection.observe( + &chunk, + &attempt.decoder.state, + &self.context, + ); + } if result.is_ok() && attempt.decoder.state.terminal { attempt.eof = true; } @@ -761,7 +768,7 @@ impl OpenAIResponsesTurn { } } }; - if let Err(failure) = result { + if let Err(mut failure) = result { self.context.tracker.note_failure(&failure.error); let websocket = self .attempt @@ -770,11 +777,23 @@ impl OpenAIResponsesTurn { // 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. + let missing_previous = self.attempt.as_ref().is_some_and(|attempt| { + matches!(&attempt.body, LiveBody::WebSocket(connection) + if connection.missing_previous(&attempt.decoder)) + }); + // A precise missing-continuation rejection is recoverable by dropping + // this socket and using the authoritative full request. The new socket + // has no checkpoint, so this exception cannot repeat. It shares the + // existing retry budget, observer and cancellation-safe reopen path. + if missing_previous { + failure.retryable = true; + } if websocket && (self.attempt_output_emitted || !self.attempt.as_ref().is_some_and(|attempt| { - attempt.decoder.request_rejected - && matches!(&attempt.body, LiveBody::WebSocket(connection) + missing_previous + || attempt.decoder.request_rejected + && matches!(&attempt.body, LiveBody::WebSocket(connection) if connection.can_retry_rejection()) })) { @@ -2138,6 +2157,7 @@ struct ResponsesSseDecoder { buffer_start: usize, received: usize, request_rejected: bool, + previous_response_missing: Option, max_attempt_bytes: usize, state: ResponsesState, } @@ -2170,6 +2190,7 @@ impl ResponsesSseDecoder { buffer_start: 0, received: 0, request_rejected: false, + previous_response_missing: None, max_attempt_bytes: limits.max_attempt_bytes, state: ResponsesState::new( model, @@ -2328,10 +2349,46 @@ impl ResponsesSseDecoder { zeroize_encrypted_content(&mut value); return Err(protocol_failure("Responses SSE event name/type mismatch")); } + self.previous_response_missing = None; let mut result = self.state.consume(&kind, &value); if kind == "error" && let Err(failure) = &mut result + && failure + .error + .provider_failure() + .is_some_and(|failure| failure.reason == ProviderFailureReason::ResponseFailed) { + // Only a validated rejection may authorize recovery, never a sequence + // or lifecycle protocol failure containing an error-shaped payload. + let error = value.get("error").unwrap_or(&value); + if !self.state.created + && value + .get("status") + .or_else(|| value.get("status_code")) + .and_then(Value::as_u64) + == Some(400) + && error.get("type").and_then(Value::as_str) == Some("invalid_request_error") + && error + .get("param") + .is_none_or(|param| param == "previous_response_id") + && error + .get("code") + .or_else(|| value.get("code")) + .and_then(Value::as_str) + == Some("previous_response_not_found") + { + // https://developers.openai.com/api/docs/guides/websocket-mode#errors-to-handle + // The code alone is uncorrelated on a reused socket. Extract only + // the bounded ID from the provider's precise rejection message; + // never retain or expose arbitrary provider message text. + self.previous_response_missing = error + .get("message") + .and_then(Value::as_str) + .and_then(|message| message.strip_prefix("Previous response with id '")) + .and_then(|message| message.strip_suffix("' not found.")) + .filter(|id| !id.is_empty() && id.len() <= 512) + .map(str::to_owned); + } if let Some(status) = value .get("status") .or_else(|| value.get("status_code")) diff --git a/crates/agentkit-provider-openai/src/responses/websocket.rs b/crates/agentkit-provider-openai/src/responses/websocket.rs index 6dadf12..c29ffe8 100644 --- a/crates/agentkit-provider-openai/src/responses/websocket.rs +++ b/crates/agentkit-provider-openai/src/responses/websocket.rs @@ -67,11 +67,17 @@ impl Lease { } pub(super) fn complete(&mut self, mut connection: Box, response_id: Option<&str>) { - // Keep bounded replay correlation, not a server-side continuation cache. + // Only successful response.completed checkpoints may survive the lease. if let Some(id) = response_id + && id.len() <= 1024 && connection.completed_ids.len() < 1024 { connection.completed_ids.insert(id.to_owned()); + if !connection.checkpoint.as_ref().is_some_and(|c| c.completed) { + connection.checkpoint = None; + } else if let Some(checkpoint) = &mut connection.checkpoint { + checkpoint.response_id = id.to_owned(); + } self.connection = Some(connection); self.reusable = true; } @@ -101,9 +107,163 @@ pub(super) struct Connection { auth_headers: HeaderMap, binding: Option, completed_ids: BTreeSet, + checkpoint: Option, + previous_response_id: Option, +} + +// All fields are private to the leased connection: no new lock or persistent +// state. Serialized buffers zeroize encrypted reasoning on every exit path. +struct Checkpoint { + request: Zeroizing, + response_id: String, + raw_output: BTreeMap>, + replay_output: BTreeMap>, + bytes: usize, + completed: bool, +} + +impl Checkpoint { + fn suffix(&self, current: &Value) -> Option> { + if !self.completed || self.response_id.is_empty() { + return None; + } + let mut previous: Value = serde_json::from_str(&self.request).ok()?; + let result = (|| { + let old = previous.as_object()?; + let new = current.as_object()?; + if old.len() != new.len() + || old + .iter() + .any(|(key, value)| key != "input" && new.get(key) != Some(value)) + { + return None; + } + let old_input = old.get("input")?.as_array()?; + let input = new.get("input")?.as_array()?; + if !input.starts_with(old_input) { + return None; + } + let mut offset = old_input.len(); + for encoded in self.replay_output.values() { + let mut replay: Value = serde_json::from_str(encoded).ok()?; + let matches = replay.as_array().is_some_and(|items| { + let end = offset + items.len(); + let matches = input.get(offset..end) == Some(items.as_slice()); + offset = end; + matches + }); + zeroize_encrypted_content(&mut replay); + if !matches { + return None; + } + } + Some(input[offset..].to_vec()) + })(); + zeroize_encrypted_content(&mut previous); + result + } } impl Connection { + pub(super) fn missing_previous(&self, decoder: &ResponsesSseDecoder) -> bool { + // Successful-completion-only, single-flight reuse and unique response IDs + // ensure this predecessor was never used by an earlier request. A delayed + // rejection naming an older predecessor cannot authorize a replay here. + self.previous_response_id.as_ref().is_some_and(|id| { + decoder.previous_response_missing.as_ref() == Some(id) && !decoder.state.created + }) + } + + pub(super) fn observe( + &mut self, + chunk: &[u8], + state: &ResponsesState, + context: &ResponsesRequestContext, + ) { + let Some(checkpoint) = &mut self.checkpoint else { + return; + }; + let Ok(mut event) = serde_json::from_slice::(chunk) else { + self.checkpoint = None; + return; + }; + let valid = (|| { + match event.get("type").and_then(Value::as_str) { + Some("response.completed") => { + let response = event.get("response")?; + if response + .get("status") + .is_some_and(|status| status != "completed") + { + return None; + } + // Streamed output_item.done items are the indexed source of + // truth used by Finished and the checkpoint. Live WS completion + // envelopes can omit output or send output:[] despite those + // items. An empty terminal array must not erase them. Validate + // any nonempty terminal copy against the observed raw items. + if let Some(output) = response.get("output") { + let output = output.as_array()?; + if !output.is_empty() && output.len() != checkpoint.raw_output.len() { + return None; + } + for (item, raw) in output.iter().zip(checkpoint.raw_output.values()) { + let mut observed: Value = serde_json::from_str(raw).ok()?; + let same = *item == observed; + zeroize_encrypted_content(&mut observed); + if !same { + return None; + } + } + } + checkpoint.completed = true; + } + Some("response.output_item.done") => { + let index = event.get("output_index")?.as_u64()?; + // The same representation the next real transcript produces: + // status/annotations and readable reasoning summaries are not + // request continuation fields in this adapter. + let item = state.output.get(&index)?; + let mut replay = Value::Array( + encode_item( + &context.config, + &context.session_id, + context.auth.binding(), + item, + ) + .ok()?, + ); + let encoded = serde_json::to_string(&replay).ok().map(Zeroizing::new); + let empty = replay.as_array().is_none_or(Vec::is_empty); + zeroize_encrypted_content(&mut replay); + if empty { + return None; + } + let encoded = encoded?; + let raw = Zeroizing::new(serde_json::to_string(event.get("item")?).ok()?); + checkpoint.bytes = checkpoint + .bytes + .checked_add(raw.len())? + .checked_add(encoded.len())?; + if checkpoint.raw_output.len() >= context.config.limits.max_items + || checkpoint.bytes > context.config.limits.max_request_bytes + { + return None; + } + checkpoint.raw_output.insert(index, raw); + checkpoint.replay_output.insert(index, encoded); + } + _ => {} + } + Some(()) + })() + .is_some(); + zeroize_encrypted_content(&mut event); + if !valid { + self.checkpoint = None; + } + } + pub(super) fn can_retry_rejection(&self) -> bool { // Wrapped errors have no reliable request correlation. On a reused // socket they may belong to a previous turn, even after a pending read. @@ -261,20 +421,41 @@ pub(super) async fn send( auth_headers: context.auth.headers().clone(), binding: context.auth.binding().map(str::to_owned), completed_ids: BTreeSet::new(), + checkpoint: None, + previous_response_id: None, }) }; - // Always send the authoritative, credential-bound full transcript. Incremental - // previous_response_id is intentionally not used without a lossless prefix proof. + // 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"))?; - // Public Responses WebSocket mode excludes HTTP transport controls. - // https://developers.openai.com/api/docs/guides/websocket-mode 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) + { + connection.previous_response_id = Some(checkpoint.response_id.clone()); + zeroize_encrypted_content(&mut value["input"]); + value["input"] = Value::Array(suffix); + value["previous_response_id"] = json!(checkpoint.response_id); + } + connection.checkpoint = Some(Checkpoint { + request: full, + response_id: String::new(), + raw_output: BTreeMap::new(), + replay_output: BTreeMap::new(), + bytes: 0, + completed: false, + }); let serialized = serde_json::to_string(&value); zeroize_encrypted_content(&mut value); let request = Zeroizing::new( diff --git a/crates/agentkit-provider-openai/src/responses/websocket/tests.rs b/crates/agentkit-provider-openai/src/responses/websocket/tests.rs index 73a9349..8b6f4c2 100644 --- a/crates/agentkit-provider-openai/src/responses/websocket/tests.rs +++ b/crates/agentkit-provider-openai/src/responses/websocket/tests.rs @@ -6,7 +6,9 @@ use std::net::{TcpListener, TcpStream}; use std::thread; use tokio_tungstenite::tungstenite::{self, WebSocket}; -// Same text/tool/reasoning/usage fixture as the HTTP decoder tests. +// Same text/tool/reasoning/usage events as the HTTP decoder tests. Live WS +// completion envelopes can contain output:[] even after output_item.done; the +// streamed items, not this empty terminal field, supply the real transcript. const SUCCESS: &str = r#"event: response.created data: {"type":"response.created","sequence_number":1,"response":{"id":"resp-1","model":"gpt-test"}} @@ -59,7 +61,7 @@ event: response.output_item.done data: {"type":"response.output_item.done","sequence_number":17,"output_index":2,"item":{"id":"call-item","type":"function_call","call_id":"call-1","name":"lookup","arguments":"{\"q\":1}"}} event: response.completed -data: {"type":"response.completed","sequence_number":18,"response":{"id":"resp-1","model":"gpt-test","usage":{"input_tokens":3,"output_tokens":5,"output_tokens_details":{"reasoning_tokens":2}}}} +data: {"type":"response.completed","sequence_number":18,"response":{"id":"resp-1","model":"gpt-test","output":[],"usage":{"input_tokens":3,"output_tokens":5,"output_tokens_details":{"reasoning_tokens":2}}}} "#; @@ -97,10 +99,14 @@ fn socket(listener: &TcpListener) -> Socket { tungstenite::accept(accept(listener)).unwrap() } fn receive(ws: &mut Socket) -> Value { + let value = receive_continuation(ws); + assert!(value.get("previous_response_id").is_none()); + value +} +fn receive_continuation(ws: &mut Socket) -> Value { let message = ws.read().unwrap(); let value: Value = serde_json::from_str(message.to_text().unwrap()).unwrap(); assert_eq!(value["type"], "response.create"); - assert!(value.get("previous_response_id").is_none()); assert!( value.get("stream").is_none(), "HTTP stream field on WS wire" @@ -824,3 +830,870 @@ async fn dropped_refresh_future_repolls_safely_without_replaying_stale_credentia peer.join().unwrap(); } } + +// Like the loop, append Finished's canonical items rather than synthesizing +// continuation metadata from raw wire output or incremental text deltas. +async fn append_completed(turn: &mut OpenAIResponsesTurn, request: &mut TurnRequest) { + let events = drain(turn).await.unwrap(); + finished(&events); + let result = events + .into_iter() + .find_map(|event| match event { + ModelTurnEvent::Finished(result) => Some(result), + _ => None, + }) + .unwrap(); + assert!( + result + .output_items + .iter() + .flat_map(|item| &item.parts) + .any(|part| matches!(part, Part::Reasoning(_))) + ); + assert!( + result + .output_items + .iter() + .flat_map(|item| &item.parts) + .any(|part| matches!(part, Part::ToolCall(_))) + ); + request.transcript.extend(result.output_items); +} +fn append_tool_result(request: &mut TurnRequest) { + request.transcript.push(Item::new( + ItemKind::Tool, + vec![Part::ToolResult(agentkit_core::ToolResultPart::success( + "call-1", + ToolOutput::text("found"), + ))], + )); +} +fn missing_previous(ws: &mut Socket) { + ws.send(Message::Text(r#"{"type":"error","status":400,"error":{"type":"invalid_request_error","code":"previous_response_not_found","message":"Previous response with id 'resp-1' not found."}}"#.into())).unwrap(); +} + +#[tokio::test] +async fn continuation_reduces_real_tool_reasoning_rounds_and_new_user_turn() { + let (endpoint, peer) = server(|listener| { + let mut ws = socket(&listener); + receive(&mut ws); + normalized_success(&mut ws); + for (previous, next) in [("resp-1", "resp-2"), ("resp-2", "resp-3")] { + let wire = receive_continuation(&mut ws); + assert_eq!(wire["previous_response_id"], previous); + assert_eq!( + wire["input"], + json!([{ + "type": "function_call_output", "call_id": "call-1", "output": "found" + }]) + ); + success_id(&mut ws, next); + } + let wire = receive_continuation(&mut ws); + assert_eq!(wire["previous_response_id"], "resp-3"); + assert_eq!(wire["input"].as_array().unwrap().len(), 2); + assert_eq!(wire["input"][0]["type"], "function_call_output"); + assert_eq!(wire["input"][1]["role"], "user"); + assert_eq!(wire["input"][1]["content"][0]["text"], "next question"); + success_id(&mut ws, "resp-4"); + }); + let adapter = + OpenAIResponsesAdapter::new(config(&endpoint, OpenAIResponsesTransport::WebSocket)) + .unwrap(); + let mut session = adapter + .start_session(SessionConfig::new("session")) + .await + .unwrap(); + let mut request = request(); + for _ in 0..3 { + let mut turn = session.begin_turn(request.clone(), None).await.unwrap(); + append_completed(&mut turn, &mut request).await; + append_tool_result(&mut request); + } + request.turn_id = TurnId::new("next-user-turn"); + request + .transcript + .push(Item::text(ItemKind::User, "next question")); + let mut turn = session.begin_turn(request, None).await.unwrap(); + finished(&drain(&mut turn).await.unwrap()); + peer.join().unwrap(); +} + +#[tokio::test] +async fn continuation_falls_back_to_full_input_for_mutated_history_or_settings() { + for mutation in ["user", "tools", "system", "compaction", "call", "reasoning"] { + let (endpoint, peer) = server(move |listener| { + let mut ws = socket(&listener); + let initial = receive(&mut ws); + decorated_success(&mut ws); + let full = receive(&mut ws); + assert!( + full["input"].as_array().unwrap().len() + > initial["input"].as_array().unwrap().len() + ); + assert!( + full["input"] + .as_array() + .unwrap() + .iter() + .any(|item| item["type"] == "reasoning") + ); + assert!( + full["input"] + .as_array() + .unwrap() + .iter() + .any(|item| item["type"] == "function_call") + ); + if mutation == "tools" { + assert_ne!(full["tools"], initial["tools"]); + } else if matches!(mutation, "system" | "compaction") { + assert_ne!(full["input"][0], initial["input"][0]); + } + assert_canonical_output( + &full, + if mutation == "reasoning" { + "changed-opaque" + } else { + "opaque" + }, + if mutation == "call" { + json!({"q":2}) + } else { + json!({"q":1}) + }, + ); + success_id(&mut ws, "resp-2"); + }); + let adapter = + OpenAIResponsesAdapter::new(config(&endpoint, OpenAIResponsesTransport::WebSocket)) + .unwrap(); + let mut session = adapter + .start_session(SessionConfig::new("session")) + .await + .unwrap(); + let mut request = request(); + request + .transcript + .insert(0, Item::text(ItemKind::System, "original instructions")); + let mut turn = session.begin_turn(request.clone(), None).await.unwrap(); + append_completed(&mut turn, &mut request).await; + append_tool_result(&mut request); + if mutation == "tools" { + request.available_tools.push( + serde_json::from_value(json!({ + "name": "lookup", "description": "changed tools", + "input_schema": {"type": "object"}, "annotations": {"read_only_hint": false, "destructive_hint": false, + "idempotent_hint": false, "needs_approval_hint": false, + "supports_streaming_hint": false}, "metadata": {} + })) + .unwrap(), + ); + } else { + match mutation { + "user" => request.transcript[1] = Item::text(ItemKind::User, "edited history"), + "system" => { + request.transcript[0] = Item::text(ItemKind::System, "new instructions") + } + "compaction" => { + request.transcript.drain(..2); + request + .transcript + .insert(0, Item::text(ItemKind::User, "compacted history")); + } + "call" | "reasoning" => { + for part in request + .transcript + .iter_mut() + .flat_map(|item| &mut item.parts) + { + match part { + Part::ToolCall(call) if mutation == "call" => { + call.input = json!({"q":2}) + } + Part::Reasoning(reasoning) if mutation == "reasoning" => { + reasoning.metadata.get_mut(CONTINUATION_METADATA).unwrap()["encrypted_content"] = + json!("changed-opaque"); + } + _ => {} + } + } + } + _ => unreachable!(), + } + } + let mut next = session.begin_turn(request, None).await.unwrap(); + finished(&drain(&mut next).await.unwrap()); + peer.join().unwrap(); + } +} + +#[tokio::test] +async fn continuation_missing_previous_recovers_once_on_fresh_socket() { + for reject_full_retry in [false, true] { + let (done_tx, done_rx) = std::sync::mpsc::channel(); + let (endpoint, peer) = + server(move |listener| { + let mut ws = socket(&listener); + receive(&mut ws); + success(&mut ws); + let suffix = receive_continuation(&mut ws); + assert_eq!(suffix["previous_response_id"], "resp-1"); + assert_eq!(suffix["input"].as_array().unwrap().len(), 1); + missing_previous(&mut ws); + assert!(ws.read().is_err(), "rejected socket must be discarded"); + let mut fresh = socket(&listener); + let full = receive(&mut fresh); + let input = full["input"].as_array().unwrap(); + assert_eq!(input.len(), 5); // user, text, reasoning, call, result + assert_eq!(input.last(), suffix["input"].as_array().unwrap().last()); + assert_eq!(input[0]["role"], "user"); + assert!(input.iter().any( + |item| item["type"] == "reasoning" && item["encrypted_content"] == "opaque" + )); + if reject_full_retry { + missing_previous(&mut fresh); + } else { + success_id(&mut fresh, "resp-recovered"); + } + done_rx.recv_timeout(WAIT).unwrap(); + assert_eq!( + listener.accept().unwrap_err().kind(), + std::io::ErrorKind::WouldBlock + ); + }); + let mut cfg = config(&endpoint, OpenAIResponsesTransport::WebSocket); + cfg.resilience.as_mut().unwrap().max_retries = 3; + let adapter = OpenAIResponsesAdapter::new(cfg).unwrap(); + let mut session = adapter + .start_session(SessionConfig::new("session")) + .await + .unwrap(); + let mut request = request(); + let mut turn = session.begin_turn(request.clone(), None).await.unwrap(); + append_completed(&mut turn, &mut request).await; + append_tool_result(&mut request); + let mut next = session.begin_turn(request, None).await.unwrap(); + let result = drain(&mut next).await; + if reject_full_retry { + assert!(result.is_err()); + assert!(next.next_event(None).await.unwrap().is_none()); + } else { + finished(&result.unwrap()); + } + done_tx.send(()).unwrap(); + peer.join().unwrap(); + } +} + +#[tokio::test] +async fn continuation_missing_previous_after_created_or_output_never_replays() { + for case in ["created", "output", "code_only", "stale_id", "out_of_order"] { + let (done_tx, done_rx) = std::sync::mpsc::channel(); + let (endpoint, peer) = server(move |listener| { + let mut ws = socket(&listener); + receive(&mut ws); + success(&mut ws); + assert_eq!( + receive_continuation(&mut ws)["previous_response_id"], + "resp-1" + ); + for line in SUCCESS + .replace("resp-1", "accepted") + .lines() + .filter_map(|line| line.strip_prefix("data: ")) + .take(match case { + "output" => 4, + "created" => 1, + _ => 0, + }) + { + ws.send(Message::Text(line.into())).unwrap(); + } + if matches!(case, "created" | "output") { + missing_previous(&mut ws); + } else { + let mut error = json!({"type":"error", "status":400, "error":{ + "type":"invalid_request_error", "code":"previous_response_not_found" + }}); + if case == "stale_id" { + error["error"]["message"] = + json!("Previous response with id 'resp-stale' not found."); + } else if case == "out_of_order" { + ws.send(Message::Text( + json!({"type":"keepalive", "sequence_number":1}) + .to_string() + .into(), + )) + .unwrap(); + error["sequence_number"] = json!(1); + error["error"]["message"] = + json!("Previous response with id 'resp-1' not found."); + } + ws.send(Message::Text(error.to_string().into())).unwrap(); + } + assert!(ws.read().is_err(), "failed socket must be discarded"); + done_rx.recv_timeout(WAIT).unwrap(); + assert_eq!( + listener.accept().unwrap_err().kind(), + std::io::ErrorKind::WouldBlock + ); + }); + let mut cfg = config(&endpoint, OpenAIResponsesTransport::WebSocket); + cfg.resilience.as_mut().unwrap().max_retries = 3; + let adapter = OpenAIResponsesAdapter::new(cfg).unwrap(); + let mut session = adapter + .start_session(SessionConfig::new("session")) + .await + .unwrap(); + let mut request = request(); + let mut turn = session.begin_turn(request.clone(), None).await.unwrap(); + append_completed(&mut turn, &mut request).await; + append_tool_result(&mut request); + let mut next = session.begin_turn(request, None).await.unwrap(); + assert!(drain(&mut next).await.is_err()); + assert!(next.next_event(None).await.unwrap().is_none()); + done_tx.send(()).unwrap(); + peer.join().unwrap(); + } +} + +#[tokio::test] +async fn continuation_raw_output_is_bounded_and_overflow_uses_full_request() { + for overflow in [false, true] { + let (endpoint, peer) = server(move |listener| { + let mut ws = socket(&listener); + receive(&mut ws); + for line in SUCCESS + .lines() + .filter_map(|line| line.strip_prefix("data: ")) + { + let mut event: Value = serde_json::from_str(line).unwrap(); + if event["type"] == "response.output_item.done" && event["output_index"] == 0 { + // Real output may contain provider-only fields absent from the + // canonical replay representation. They still consume the bound. + event["item"]["provider_padding"] = + json!("x".repeat(if overflow { 8192 } else { 16 })); + } + ws.send(Message::Text(event.to_string().into())).unwrap(); + } + let wire = if overflow { + receive(&mut ws) + } else { + receive_continuation(&mut ws) + }; + if overflow { + assert_eq!(wire["input"].as_array().unwrap().len(), 5); + } else { + assert_eq!(wire["previous_response_id"], "resp-1"); + assert_eq!(wire["input"].as_array().unwrap().len(), 1); + } + success_id(&mut ws, "resp-2"); + }); + let mut cfg = config(&endpoint, OpenAIResponsesTransport::WebSocket); + cfg.limits.max_request_bytes = 4096; + cfg.limits.max_text_bytes = 4096; + let adapter = OpenAIResponsesAdapter::new(cfg).unwrap(); + let mut session = adapter + .start_session(SessionConfig::new("session")) + .await + .unwrap(); + let mut request = request(); + let mut turn = session.begin_turn(request.clone(), None).await.unwrap(); + append_completed(&mut turn, &mut request).await; + { + let state = session.websocket.lock().unwrap(); + let connection = state.idle.as_ref().unwrap(); + if overflow { + assert!(connection.checkpoint.is_none()); + } else { + let checkpoint = connection.checkpoint.as_ref().unwrap(); + assert_eq!(checkpoint.raw_output.len(), 3); + assert!(checkpoint.bytes <= 4096); + let raw: Value = serde_json::from_str(&checkpoint.raw_output[&0]).unwrap(); + assert_eq!(raw["provider_padding"], "x".repeat(16)); + } + } + append_tool_result(&mut request); + let mut next = session.begin_turn(request, None).await.unwrap(); + finished(&drain(&mut next).await.unwrap()); + peer.join().unwrap(); + } +} + +fn decorated_success(ws: &mut Socket) { + let fixture = SUCCESS.replace(r#"{\"q\":1}"#, r#"{ \"q\" : 1 }"#); + for line in fixture + .lines() + .filter_map(|line| line.strip_prefix("data: ")) + { + let mut event: Value = serde_json::from_str(line).unwrap(); + if event["type"] == "response.output_item.done" { + event["item"]["status"] = json!("completed"); + event["item"]["annotations"] = json!([{"type":"provider_only"}]); + } + ws.send(Message::Text(event.to_string().into())).unwrap(); + } +} +fn assert_canonical_output(wire: &Value, encrypted: &str, arguments: Value) { + let input = wire["input"].as_array().unwrap(); + assert_eq!( + input + .iter() + .find(|item| item["type"] == "reasoning") + .unwrap(), + &json!({ + "id":"reason-1", "type":"reasoning", "summary":[], "encrypted_content":encrypted + }) + ); + assert_eq!( + input + .iter() + .find(|item| item["type"] == "function_call") + .unwrap(), + &json!({ + "id":"call-item", "type":"function_call", "call_id":"call-1", "name":"lookup", + "arguments":arguments.to_string() + }) + ); +} + +#[tokio::test] +async fn continuation_generated_image_reduces_into_real_transcript() { + let (endpoint, peer) = server(|listener| { + let mut ws = socket(&listener); + receive(&mut ws); + let item = json!({"id":"image-1", "type":"image_generation_call", "status":"completed", + "revised_prompt":"safer prompt", "result":"AQID"}); + for event in [ + json!({"type":"response.created", "sequence_number":1, "response":{"id":"resp-image", "model":"gpt-test"}}), + json!({"type":"response.output_item.added", "sequence_number":2, "output_index":0, "item":{"id":"image-1", "type":"image_generation_call"}}), + json!({"type":"response.output_item.done", "sequence_number":3, "output_index":0, "item":item}), + json!({"type":"response.completed", "sequence_number":4, "response":{"id":"resp-image", "model":"gpt-test", "status":"completed", "output":[item]}}), + ] { + ws.send(Message::Text(event.to_string().into())).unwrap(); + } + let wire = receive_continuation(&mut ws); + assert_eq!(wire["previous_response_id"], "resp-image"); + assert_eq!( + wire["input"], + json!([{"type":"message", "role":"user", "content":[{"type":"input_text", "text":"describe image"}]}]) + ); + success_id(&mut ws, "resp-after-image"); + }); + let adapter = + OpenAIResponsesAdapter::new(config(&endpoint, OpenAIResponsesTransport::WebSocket)) + .unwrap(); + let mut session = adapter + .start_session(SessionConfig::new("session")) + .await + .unwrap(); + let mut request = request(); + let mut turn = session.begin_turn(request.clone(), None).await.unwrap(); + let events = drain(&mut turn).await.unwrap(); + let result = events + .into_iter() + .find_map(|event| match event { + ModelTurnEvent::Finished(result) => Some(result), + _ => None, + }) + .unwrap(); + let Part::Media(media) = &result.output_items[0].parts[0] else { + panic!("missing generated image") + }; + assert_eq!(media.data, DataRef::InlineBytes(vec![1, 2, 3])); + request.transcript.extend(result.output_items); + request + .transcript + .push(Item::text(ItemKind::User, "describe image")); + let mut next = session.begin_turn(request, None).await.unwrap(); + finished(&drain(&mut next).await.unwrap()); + peer.join().unwrap(); +} + +#[test] +fn continuation_checkpoint_requires_identical_model_and_reasoning_settings() { + let previous = json!({"type":"response.create", "model":"gpt-test", "reasoning":{"effort":"low", "summary":"auto"}, "input":[{"role":"user","content":"hello"}]}); + let output = json!({"type":"reasoning","id":"r", "summary":[], "encrypted_content":"opaque"}); + let checkpoint = Checkpoint { + request: Zeroizing::new(previous.to_string()), + response_id: "resp-1".into(), + raw_output: BTreeMap::new(), + replay_output: BTreeMap::from([(0, Zeroizing::new(json!([output]).to_string()))]), + bytes: 0, + completed: true, + }; + let mut current = previous.clone(); + current["input"] + .as_array_mut() + .unwrap() + .extend([output, json!({"role":"user","content":"next"})]); + assert_eq!( + checkpoint.suffix(¤t), + Some(vec![json!({"role":"user","content":"next"})]) + ); + for (key, value) in [ + ("model", json!("other-model")), + ("reasoning", json!({"effort":"high","summary":"auto"})), + ] { + let mut changed = current.clone(); + changed[key] = value; + assert!(checkpoint.suffix(&changed).is_none(), "changed {key}"); + } +} + +#[tokio::test] +async fn continuation_terminal_output_preserves_streamed_items_or_discards_conflicts() { + for variant in [ + "malformed", + "missing_item", + "mismatch", + "status", + "matching", + "empty", + ] { + let (endpoint, peer) = server(move |listener| { + let mut ws = socket(&listener); + receive(&mut ws); + let mut output = vec![]; + for line in SUCCESS + .lines() + .filter_map(|line| line.strip_prefix("data: ")) + { + let mut event: Value = serde_json::from_str(line).unwrap(); + if event["type"] == "response.output_item.done" { + output.push(event["item"].clone()); + } + if event["type"] == "response.completed" { + event["response"]["status"] = json!("completed"); + event["response"]["output"] = json!(output); + match variant { + "malformed" => event["response"]["output"] = json!({"unexpected":"object"}), + "missing_item" => { + event["response"]["output"].as_array_mut().unwrap().pop(); + } + "mismatch" => { + event["response"]["output"][1]["encrypted_content"] = json!("different") + } + "status" => event["response"]["status"] = json!("in_progress"), + "matching" => {} + "empty" => event["response"]["output"] = json!([]), + _ => unreachable!(), + } + } + ws.send(Message::Text(event.to_string().into())).unwrap(); + } + if matches!(variant, "matching" | "empty") { + let wire = receive_continuation(&mut ws); + assert_eq!(wire["previous_response_id"], "resp-1"); + assert_eq!(wire["input"].as_array().unwrap().len(), 1); + } else { + let full = receive(&mut ws); + assert_eq!(full["input"].as_array().unwrap().len(), 5); + assert_canonical_output(&full, "opaque", json!({"q":1})); + } + success_id(&mut ws, "resp-2"); + }); + let adapter = + OpenAIResponsesAdapter::new(config(&endpoint, OpenAIResponsesTransport::WebSocket)) + .unwrap(); + let mut session = adapter + .start_session(SessionConfig::new("session")) + .await + .unwrap(); + let mut request = request(); + let mut turn = session.begin_turn(request.clone(), None).await.unwrap(); + append_completed(&mut turn, &mut request).await; + append_tool_result(&mut request); + let mut next = session.begin_turn(request, None).await.unwrap(); + finished(&drain(&mut next).await.unwrap()); + peer.join().unwrap(); + } +} + +// Change every arguments-bearing frame, not just output_item.done: the real +// decoder observes identical values with noncanonical whitespace and key order. +fn normalized_success(ws: &mut Socket) { + let fixture = SUCCESS.replace(r#"{\"q\":1}"#, r#"{ \"z\" : 2, \"q\" : 1 }"#); + assert_ne!(fixture, SUCCESS); + for line in fixture + .lines() + .filter_map(|line| line.strip_prefix("data: ")) + { + let mut event: Value = serde_json::from_str(line).unwrap(); + if event["type"] == "response.output_item.done" { + event["item"]["status"] = json!("completed"); + if event["item"]["type"] == "message" { + event["item"]["content"][0]["annotations"] = json!([]); + } + } + ws.send(Message::Text(event.to_string().into())).unwrap(); + } +} + +struct RotatingInitialAuth { + calls: std::sync::atomic::AtomicUsize, + change_binding: bool, +} +#[async_trait] +impl AuthenticationProvider for RotatingInitialAuth { + async fn authenticate( + &self, + previous: Option<&AuthenticationAttempt>, + ) -> Result { + assert!( + previous.is_none(), + "this test must not refresh a rejected attempt" + ); + let generation = self.calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + let mut headers = HeaderMap::new(); + headers.insert( + "authorization", + HeaderValue::from_str(&format!("Bearer loopback-{generation}")).unwrap(), + ); + Ok( + AuthenticationAttempt::stateless(headers).with_binding(if self.change_binding { + format!("loopback-identity-{generation}") + } else { + "loopback-identity".to_owned() + }), + ) + } +} + +#[tokio::test] +async fn continuation_reconnect_or_rotated_headers_sends_full_reduced_history() { + for rotate_headers in [false, true] { + let (close_tx, close_rx) = tokio::sync::oneshot::channel(); + let (endpoint, peer) = server(move |listener| { + let mut first = socket(&listener); + receive(&mut first); + success(&mut first); + if !rotate_headers { + first.close(None).unwrap(); + close_tx.send(()).unwrap(); + assert!(matches!(first.read(), Ok(Message::Close(_)))); + drop(first); + } + let mut fresh = socket(&listener); + let full = receive(&mut fresh); + assert_eq!(full["input"].as_array().unwrap().len(), 5); + assert_canonical_output(&full, "opaque", json!({"q":1})); + assert_eq!(full["input"][0]["role"], "user"); + assert_eq!( + full["input"][4], + json!({"type":"function_call_output", "call_id":"call-1", "output":"found"}) + ); + success_id(&mut fresh, "resp-fresh"); + }); + let mut cfg = config(&endpoint, OpenAIResponsesTransport::WebSocket); + if rotate_headers { + cfg = cfg.with_authentication_provider(RotatingInitialAuth { + calls: std::sync::atomic::AtomicUsize::new(0), + change_binding: false, + }); + } + let adapter = OpenAIResponsesAdapter::new(cfg).unwrap(); + let mut session = adapter + .start_session(SessionConfig::new("session")) + .await + .unwrap(); + let mut request = request(); + let mut turn = session.begin_turn(request.clone(), None).await.unwrap(); + append_completed(&mut turn, &mut request).await; + append_tool_result(&mut request); + if !rotate_headers { + tokio::time::timeout(WAIT, close_rx).await.unwrap().unwrap(); + // Synchronize receipt of the peer's Close, not a sleep or a race + // against send's nonblocking boundary check. Leave the completed + // checkpoint on the closed connection for normal checkout to reject. + let mut connection = session.websocket.lock().unwrap().idle.take().unwrap(); + let frame = tokio::time::timeout(WAIT, connection.stream.next()) + .await + .unwrap(); + assert!(matches!(frame, Some(Ok(Message::Close(_))))); + connection.stream.flush().await.unwrap(); + assert!( + tokio::time::timeout(WAIT, connection.stream.next()) + .await + .unwrap() + .is_none() + ); + assert!(connection.checkpoint.as_ref().unwrap().completed); + session.websocket.lock().unwrap().idle = Some(connection); + } + let mut next = session.begin_turn(request, None).await.unwrap(); + finished(&drain(&mut next).await.unwrap()); + peer.join().unwrap(); + } +} + +#[tokio::test] +async fn continuation_changed_binding_reconnects_with_full_text_history() { + let (endpoint, peer) = server(|listener| { + let mut first = socket(&listener); + receive(&mut first); + // Text-only output has no authentication-bound protected metadata. + for line in SUCCESS + .lines() + .filter_map(|line| line.strip_prefix("data: ")) + .take(7) + { + first.send(Message::Text(line.into())).unwrap(); + } + first + .send(Message::Text( + json!({"type":"response.completed", "sequence_number":8, + "response":{"id":"resp-1", "model":"gpt-test", "status":"completed"}}) + .to_string() + .into(), + )) + .unwrap(); + let mut fresh = socket(&listener); + let full = receive(&mut fresh); + assert_eq!( + full["input"], + json!([ + {"type":"message", "role":"user", "content":[{"type":"input_text", "text":"hello"}]}, + {"type":"message", "role":"assistant", "content":[{"type":"output_text", "text":"hello"}]}, + {"type":"message", "role":"user", "content":[{"type":"input_text", "text":"next"}]} + ]) + ); + success_id(&mut fresh, "resp-2"); + }); + let cfg = config(&endpoint, OpenAIResponsesTransport::WebSocket).with_authentication_provider( + RotatingInitialAuth { + calls: std::sync::atomic::AtomicUsize::new(0), + change_binding: true, + }, + ); + let adapter = OpenAIResponsesAdapter::new(cfg).unwrap(); + let mut session = adapter + .start_session(SessionConfig::new("session")) + .await + .unwrap(); + let mut request = request(); + let mut turn = session.begin_turn(request.clone(), None).await.unwrap(); + let events = drain(&mut turn).await.unwrap(); + finished(&events); + for event in events { + if let ModelTurnEvent::Finished(result) = event { + request.transcript.extend(result.output_items); + } + } + request.transcript.push(Item::text(ItemKind::User, "next")); + let mut next = session.begin_turn(request, None).await.unwrap(); + finished(&drain(&mut next).await.unwrap()); + peer.join().unwrap(); +} + +#[tokio::test] +async fn interrupted_missing_previous_backoff_releases_session_without_replay() { + use std::future::Future; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::task::Poll; + + for cancel in [false, true] { + let (resume_tx, resume_rx) = std::sync::mpsc::channel(); + let (checked_tx, checked_rx) = tokio::sync::oneshot::channel(); + let (endpoint, peer) = server(move |listener| { + let mut ws = socket(&listener); + receive(&mut ws); + success(&mut ws); + let incremental = receive_continuation(&mut ws); + assert_eq!(incremental["previous_response_id"], "resp-1"); + assert_eq!(incremental["input"].as_array().unwrap().len(), 1); + missing_previous(&mut ws); + assert!(ws.read().is_err(), "rejected socket must be discarded"); + // The client signals only after interrupting the suspended backoff. + // No recovery connection may have been opened in the meantime. + resume_rx.recv_timeout(WAIT).unwrap(); + assert_eq!( + listener.accept().unwrap_err().kind(), + std::io::ErrorKind::WouldBlock + ); + checked_tx.send(()).unwrap(); + let mut fresh = socket(&listener); + let full = receive(&mut fresh); + let input = full["input"].as_array().unwrap(); + assert_eq!(input.len(), 6); + assert_eq!(input[0]["content"][0]["text"], "hello"); + assert_eq!(input[2]["type"], "reasoning"); + assert_eq!(input[2]["encrypted_content"], "opaque"); + assert_eq!(input[4], incremental["input"][0]); + assert_eq!(input[5]["content"][0]["text"], "after interrupted recovery"); + success_id(&mut fresh, "fresh-after-interruption"); + }); + let mut cfg = config(&endpoint, OpenAIResponsesTransport::WebSocket); + let resilience = cfg.resilience.as_mut().unwrap(); + resilience.max_retries = 1; + // Never wait for this duration: an observed Scheduled event followed by + // Poll::Pending proves suspension, without sleep-based timing assertions. + resilience.initial_backoff = Duration::from_secs(3600); + resilience.max_backoff = Duration::from_secs(3600); + resilience.retry_budget = Duration::from_secs(7200); + let adapter = OpenAIResponsesAdapter::new(cfg).unwrap(); + let mut session = adapter + .start_session(SessionConfig::new("session")) + .await + .unwrap(); + let mut history = request(); + let mut first = session.begin_turn(history.clone(), None).await.unwrap(); + append_completed(&mut first, &mut history).await; + append_tool_result(&mut history); + + let scheduled = Arc::new(AtomicBool::new(false)); + let observed = scheduled.clone(); + session.set_retry_observer(Some(Arc::new(move |event: ProviderRetryEvent| { + if let ProviderRetryEvent::Scheduled(progress) = event { + assert_eq!(progress.accounting.attempts, 1); + assert_eq!(progress.accounting.completed_backoff, Duration::ZERO); + assert!(!progress.next_delay.is_zero()); + assert!(!observed.swap(true, Ordering::SeqCst)); + } + }))); + let controller = agentkit_core::CancellationController::new(); + let cancellation = controller.handle().checkpoint(); + let mut interrupted = session.begin_turn(history.clone(), None).await.unwrap(); + let mut pending = Box::pin(interrupted.next_event(Some(cancellation.clone()))); + tokio::time::timeout( + WAIT, + futures_util::future::poll_fn(|cx| { + assert!(pending.as_mut().poll(cx).is_pending()); + if scheduled.load(Ordering::SeqCst) { + Poll::Ready(()) + } else { + Poll::Pending + } + }), + ) + .await + .unwrap(); + if cancel { + controller.interrupt(); + assert!(matches!(pending.await, Err(LoopError::Cancelled))); + // Cancellation finalizes the lease even while the owned turn lives. + assert!(interrupted.next_event(None).await.unwrap().is_none()); + } else { + drop(pending); + drop(interrupted); + } + resume_tx.send(()).unwrap(); + // Do not open the legitimate next socket until the server has completed + // its no-replay assertion; channel ordering, not elapsed time, controls it. + tokio::time::timeout(WAIT, checked_rx) + .await + .unwrap() + .unwrap(); + history + .transcript + .push(Item::text(ItemKind::User, "after interrupted recovery")); + let mut fresh = session.begin_turn(history, None).await.unwrap(); + let events = drain(&mut fresh).await.unwrap(); + finished(&events); + assert!(events.iter().any(|event| matches!(event, + ModelTurnEvent::Finished(result) + if result.response_id.as_deref() == Some("fresh-after-interruption")))); + peer.join().unwrap(); + } +} diff --git a/crates/agentkit-provider-openai/tests/live_responses_websocket.rs b/crates/agentkit-provider-openai/tests/live_responses_websocket.rs new file mode 100644 index 0000000..afaf434 --- /dev/null +++ b/crates/agentkit-provider-openai/tests/live_responses_websocket.rs @@ -0,0 +1,502 @@ +//! Opt-in, billable network tests. See docs/live-responses-websocket.md. +//! Never reads credential files or logs headers, payloads, IDs, or provider errors. +use agentkit_core::{ + Item, ItemKind, MetadataMap, Part, SessionId, ToolOutput, ToolResultPart, TurnId, +}; +use agentkit_loop::{ + ModelAdapter, ModelSession, ModelTurn, ModelTurnEvent, ModelTurnResult, SessionConfig, + TurnRequest, +}; +use agentkit_provider_openai::{ + OpenAIResponsesAdapter, OpenAIResponsesConfig, OpenAIResponsesTransport, +}; +use futures_util::{SinkExt, StreamExt}; +use reqwest::header::{HeaderMap, HeaderValue}; +use serde_json::{Value, json}; +use std::{ + env, + sync::{Arc, Mutex}, + time::Duration, +}; +use tokio::{net::TcpListener, task::JoinHandle}; +use tokio_tungstenite::{ + WebSocketStream, + tungstenite::{ + Message, handshake::client::generate_key, handshake::derive_accept_key, protocol::Role, + }, +}; + +// Never format external errors: they can contain credentials or echoed payloads. +fn checked(result: Result, label: &'static str) -> T { + result.unwrap_or_else(|_| panic!("{label} (details suppressed)")) +} +struct Credentials { + headers: HeaderMap, + private: bool, + model: String, +} +impl Credentials { + fn load() -> Option { + let token = env::var("AGENTKIT_LIVE_CHATGPT_BEARER") + .ok() + .filter(|s| !s.trim().is_empty()); + let account = env::var("AGENTKIT_LIVE_CHATGPT_ACCOUNT_ID") + .ok() + .filter(|s| !s.trim().is_empty()); + let key = env::var("OPENAI_API_KEY") + .ok() + .filter(|s| !s.trim().is_empty()); + let (secret, account, private) = match (token, account, key) { + (Some(token), Some(account), _) => (token, Some(account), true), + (None, None, Some(key)) => (key, None, false), + (None, None, None) => { + assert!( + env::var("AGENTKIT_LIVE_REQUIRED").as_deref() != Ok("1"), + "live credentials required but absent" + ); + eprintln!( + "SKIP: set AGENTKIT_LIVE_CHATGPT_BEARER + AGENTKIT_LIVE_CHATGPT_ACCOUNT_ID, or OPENAI_API_KEY; AGENTKIT_LIVE_REQUIRED=1 makes absence fail" + ); + return None; + } + _ => panic!("incomplete ChatGPT credential environment"), + }; + let mut headers = HeaderMap::new(); + let mut bearer = checked( + HeaderValue::from_str(&format!("Bearer {secret}")), + "invalid bearer header", + ); + bearer.set_sensitive(true); + headers.insert("authorization", bearer); + if let Some(account) = account { + let mut value = checked(HeaderValue::from_str(&account), "invalid account header"); + value.set_sensitive(true); + headers.insert("chatgpt-account-id", value); + } + let model = env::var("AGENTKIT_LIVE_MODEL") + .unwrap_or_else(|_| if private { "gpt-5.4" } else { "gpt-4.1-mini" }.into()); + Some(Self { + headers, + private, + model, + }) + } +} +#[derive(Clone, Debug)] +struct Shape { + connection: usize, + previous: bool, + previous_matches_completed: bool, + inputs: usize, + user_messages: usize, + tool_outputs: usize, + completed_replay_inputs: Option, +} +struct Proxy { + endpoint: String, + shapes: Arc>>, + task: JoinHandle<()>, +} +impl Drop for Proxy { + fn drop(&mut self) { + self.task.abort(); + if std::thread::panicking() { + // Diagnostics only: a proof assertion can poison this read-only capture + // lock. Recover its safe counts without recovering production state. + let shapes = self + .shapes + .lock() + .unwrap_or_else(|poison| poison.into_inner()); + eprintln!("sanitized request shapes: {shapes:?}"); + } + } +} +impl Proxy { + async fn start(credentials: Credentials) -> Self { + let listener = checked(TcpListener::bind("127.0.0.1:0").await, "bind proxy"); + let endpoint = format!( + "http://{}/responses", + checked(listener.local_addr(), "proxy address") + ); + let shapes = Arc::new(Mutex::new(Vec::new())); + let captured = shapes.clone(); + let budget = if env::var("AGENTKIT_LIVE_RECONNECT").as_deref() == Ok("1") { + 4 + } else { + 3 + }; + let task = tokio::spawn(async move { + // No detached children: dropping Proxy cancels all networking. + for connection in 0..2 { + let (socket, _) = checked(listener.accept().await, "accept local websocket"); + let mut forwarded = HeaderMap::new(); + // Tungstenite fixes the callback error type; this closure never rejects. + #[allow(clippy::result_large_err)] + let mut local = checked( + tokio_tungstenite::accept_hdr_async( + socket, + |request: &tokio_tungstenite::tungstenite::handshake::server::Request, + response| { + for name in [ + "originator", + "user-agent", + "session_id", + "x-codex-turn-state", + ] { + if let Some(value) = request.headers().get(name) { + forwarded.insert(name, value.clone()); + } + } + Ok(response) + }, + ) + .await, + "local websocket handshake", + ); + let key = generate_key(); + let mut headers = credentials.headers.clone(); + headers.extend(forwarded); + 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", + checked(HeaderValue::from_str(&key), "websocket key"), + ); + headers.insert( + "openai-beta", + HeaderValue::from_static("responses_websockets=2026-02-06"), + ); + let client = checked( + reqwest::Client::builder() + .http1_only() + .no_proxy() + .redirect(reqwest::redirect::Policy::none()) + .retry(reqwest::retry::never()) + .connect_timeout(Duration::from_secs(20)) + .timeout(Duration::from_secs(30)) + .build(), + "upstream client", + ); + let endpoint = if credentials.private { + "https://chatgpt.com/backend-api/codex/responses" + } else { + "https://api.openai.com/v1/responses" + }; + let response = checked( + client.get(endpoint).headers(headers).send().await, + "upstream handshake", + ); + assert!( + response.status() == reqwest::StatusCode::SWITCHING_PROTOCOLS, + "upstream did not accept websocket (status/body suppressed; check credentials and access)" + ); + assert!( + response + .headers() + .get("sec-websocket-accept") + .and_then(|v| v.to_str().ok()) + == Some(derive_accept_key(key.as_bytes()).as_str()), + "invalid upstream websocket accept" + ); + let upgraded = checked(response.upgrade().await, "upstream upgrade"); + let mut upstream = + WebSocketStream::from_raw_socket(upgraded, Role::Client, None).await; + let mut completed_id: Option = None; + let mut streamed_replay_inputs = 0; + loop { + tokio::select! { + message = local.next() => { + let Some(Ok(message)) = message else { break }; + if message.is_close() { break; } + if let Message::Text(text) = &message { + let value: Value = checked(serde_json::from_str(text), "outbound JSON"); + assert!(value["type"] == "response.create", "unexpected outbound message type"); + let input = value["input"].as_array().unwrap_or_else(|| panic!("input array missing")); + let previous = value["previous_response_id"].as_str(); + assert!(checked(captured.lock(), "capture lock").len() < budget, "live request budget exceeded; refusing to forward"); + checked(captured.lock(), "capture lock").push(Shape { + connection, previous: previous.is_some(), completed_replay_inputs: None, + previous_matches_completed: previous.is_some() && previous == completed_id.as_deref(), + inputs: input.len(), + user_messages: input.iter().filter(|v| v["role"] == "user").count(), + tool_outputs: input.iter().filter(|v| v["type"] == "function_call_output").count(), + }); + } + checked(upstream.send(message).await, "forward request"); + } + message = upstream.next() => { + let Some(Ok(message)) = message else { break }; + if let Message::Text(text) = &message { + let value: Value = checked(serde_json::from_str(text), "inbound JSON"); + if value["type"] == "error" || value["type"] == "response.failed" { + let error = value.get("error").or_else(|| value.pointer("/response/error")); + let message = error.and_then(|e| e["message"].as_str()).unwrap_or("").to_ascii_lowercase(); + eprintln!("sanitized upstream error: instructions={} model={} unsupported={} invalid_request={} auth={}", + message.contains("instructions"), message.contains("model"), message.contains("unsupported"), + error.is_some_and(|e| e["type"] == "invalid_request_error"), + message.contains("auth") || message.contains("token")); + } + if value["type"] == "response.output_item.done" { + streamed_replay_inputs += completed_replay_inputs(&json!({"response":{"output":[value["item"].clone()]}})); + } + if value["type"] == "response.completed" { + completed_id = value.pointer("/response/id").and_then(Value::as_str).map(str::to_owned); + // Some real endpoints send an empty terminal output array: + // item.done remains the authoritative streamed output. + let count = if streamed_replay_inputs > 0 { streamed_replay_inputs } else { completed_replay_inputs(&value) }; + streamed_replay_inputs = 0; + let mut shapes = checked(captured.lock(), "capture lock"); + let shape = shapes.last_mut().unwrap_or_else(|| panic!("completion without request")); + assert!(shape.completed_replay_inputs.is_none(), "duplicate completion"); + shape.completed_replay_inputs = Some(count); + } + } + if local.send(message).await.is_err() { break; } + } + } + } + } + }); + Self { + endpoint, + shapes, + task, + } + } +} + +// Count replayable wire outputs, independently of AgentKit's credential-bound +// metadata encoding. Restrict this tiny fixture to text/tool/reasoning outputs. +// Summary-only reasoning has no replayable encrypted state and is omitted. +fn completed_replay_inputs(event: &Value) -> usize { + event + .pointer("/response/output") + .and_then(Value::as_array) + .unwrap_or_else(|| panic!("completed response missing output array")) + .iter() + .map(|item| match item["type"].as_str() { + Some("message") => usize::from( + item["content"] + .as_array() + .is_some_and(|parts| !parts.is_empty()), + ), + Some("function_call") => 1, + Some("reasoning") => usize::from( + item["encrypted_content"] + .as_str() + .is_some_and(|text| !text.is_empty()), + ), + _ => panic!("unexpected output type for live fixture"), + }) + .sum() +} + +fn request(transcript: &[Item], turn: &str) -> TurnRequest { + TurnRequest { + session_id: SessionId::new("live-ws"), + turn_id: TurnId::new(turn), + transcript: transcript.to_vec(), + available_tools: vec![checked( + serde_json::from_value( + json!({"name":"read_probe", "description":"Read the probe value. Call exactly once when asked to read it.", "metadata":{},"annotations":{"read_only_hint":true,"destructive_hint":false,"idempotent_hint":true,"needs_approval_hint":false,"supports_streaming_hint":false},"input_schema":{"type":"object","properties":{},"required":[],"additionalProperties":false}}), + ), + "probe tool specification", + )], + cache: None, + metadata: MetadataMap::new(), + } +} +async fn run_turn( + session: &mut impl ModelSession, + transcript: &mut Vec, + turn: &str, +) -> ModelTurnResult { + let future = async { + let mut stream = checked( + session.begin_turn(request(transcript, turn), None).await, + "begin model turn", + ); + loop { + match checked(stream.next_event(None).await, "read model event") { + Some(ModelTurnEvent::Finished(result)) => { + // Same authoritative reduction as LoopDriver: preserve all output items and + // opaque continuation metadata, not a hand-built assistant/tool-call replay. + transcript.extend(result.output_items.clone()); + return result; + } + Some(_) => {} + None => panic!("model stream ended before Finished"), + } + } + }; + checked( + tokio::time::timeout(Duration::from_secs(90), future).await, + "model turn timed out", + ) +} +fn answer_contains(result: &ModelTurnResult, expected: &str) -> bool { + result + .output_items + .iter() + .flat_map(|item| &item.parts) + .any(|part| matches!(part, Part::Text(text) if text.text.contains(expected))) +} +#[tokio::test] +#[ignore = "billable live WebSocket test; explicit credentials required; see docs/live-responses-websocket.md"] +async fn incremental_tool_and_user_roundtrip() { + let Some(credentials) = Credentials::load() else { + return; + }; + let config = if credentials.private { + OpenAIResponsesConfig::chatgpt_private(credentials.model.clone(), "local-proxy-placeholder") + .with_originator("codex_cli_rs") + .with_user_agent("agentkit-live-websocket-test") + } else { + OpenAIResponsesConfig::public(credentials.model.clone(), "local-proxy-placeholder") + .with_max_output_tokens(512) + }; + let proxy = Proxy::start(credentials).await; + let config = config + .with_endpoint(&proxy.endpoint) + .with_transport(OpenAIResponsesTransport::WebSocket) + .with_parallel_tool_calls(false); + let adapter = checked(OpenAIResponsesAdapter::new(config), "create adapter"); + let mut session = checked( + adapter.start_session(SessionConfig::new("live-ws")).await, + "start session", + ); + let mut transcript = vec![ + Item::text( + ItemKind::System, + "Follow these instructions exactly. When asked to read the probe, call read_probe exactly once with no arguments. After the tool result, answer only with its value. For later user messages reply only with the literal text requested; do not call tools again.", + ), + Item::text(ItemKind::User, "Read the probe now."), + ]; + let first = run_turn(&mut session, &mut transcript, "tool").await; + let calls: Vec<_> = first + .output_items + .iter() + .flat_map(|i| &i.parts) + .filter_map(|p| { + if let Part::ToolCall(call) = p { + Some(call) + } else { + None + } + }) + .collect(); + assert!( + calls.len() == 1 && calls[0].name.as_str() == "read_probe", + "expected exactly the probe tool call" + ); + transcript.push(Item::new( + ItemKind::Tool, + vec![Part::ToolResult(ToolResultPart::success( + calls[0].id.clone(), + ToolOutput::text("PROBE_731"), + ))], + )); + let second = run_turn(&mut session, &mut transcript, "tool-output").await; + assert!( + answer_contains(&second, "PROBE_731"), + "tool output was not reflected in answer" + ); + transcript.push(Item::text(ItemKind::User, "Reply only with USER_927.")); + let third = run_turn(&mut session, &mut transcript, "next-user").await; + assert!( + answer_contains(&third, "USER_927"), + "subsequent user turn not answered" + ); + { + let shapes = checked(proxy.shapes.lock(), "capture lock"); + assert_eq!(shapes.len(), 3, "unexpected retries or extra requests"); + assert!( + !shapes[0].previous && shapes[0].user_messages == 1, + "first request must be full input" + ); + for shape in &shapes[1..] { + assert!( + shape.connection == 0 && shape.previous && shape.previous_matches_completed, + "continuation must reference the completed response on the same connection" + ); + assert_eq!( + shape.inputs, 1, + "continuation must send only one new input, never replay" + ); + } + assert_eq!(shapes[1].tool_outputs, 1); + assert_eq!(shapes[1].user_messages, 0); + assert_eq!(shapes[2].tool_outputs, 0); + assert_eq!(shapes[2].user_messages, 1); + eprintln!( + "PASS: three accepted requests; continuation inputs: tool_output=1, user=1; prior IDs matched completed responses" + ); + } + // Optional fourth billable request. A new public session forces a new socket; + // replay real transcript items without synthesizing provider metadata. + if env::var("AGENTKIT_LIVE_RECONNECT").as_deref() == Ok("1") { + drop(session); + transcript.push(Item::text(ItemKind::User, "Reply only with RECONNECT_419.")); + // Public encode_request deliberately cannot validate credential-bound output + // metadata. Use independent, sanitized wire accounting instead; never strip + // metadata or manufacture a continuation to make the public encoder work. + let expected_inputs = { + let shapes = checked(proxy.shapes.lock(), "capture lock"); + shapes[0].inputs + + shapes + .iter() + .map(|shape| { + shape + .completed_replay_inputs + .unwrap_or_else(|| panic!("missing completed output accounting")) + }) + .sum::() + + 3 // local tool output, second user turn, reconnect user turn + }; + let mut session = checked( + adapter.start_session(SessionConfig::new("live-ws")).await, + "restart session", + ); + let fourth = run_turn(&mut session, &mut transcript, "reconnect").await; + assert!( + answer_contains(&fourth, "RECONNECT_419"), + "reconnect response missing" + ); + let shapes = checked(proxy.shapes.lock(), "capture lock"); + assert_eq!(shapes.len(), 4); + assert!( + shapes[3].connection == 1 && !shapes[3].previous, + "new socket must not use previous_response_id" + ); + assert_eq!( + shapes[3].inputs, expected_inputs, + "new socket must replay full input" + ); + assert!( + shapes[3].inputs > 1 && shapes[3].tool_outputs == 1 && shapes[3].user_messages == 3 + ); + eprintln!("PASS: new session/socket accepted full transcript without previous_response_id"); + } +} + +#[test] +fn fixture_encodes_without_credentials_or_network() { + let config = OpenAIResponsesConfig::chatgpt_private("unused", "unused"); + let value = checked( + config.encode_request(&request(&[Item::text(ItemKind::User, "probe")], "fixture")), + "fixture encoding", + ); + assert_eq!(value["tools"][0]["name"], "read_probe"); +} + +#[test] +fn completed_wire_accounting_omits_summary_only_reasoning() { + let event = json!({"response":{"output":[ + {"type":"function_call"}, + {"type":"message","content":[{"type":"output_text","text":"fixture"}]}, + {"type":"reasoning","summary":[]}, + {"type":"reasoning","encrypted_content":"fixture-only"} + ]}}); + assert_eq!(completed_replay_inputs(&event), 3); +} diff --git a/docs/live-responses-websocket.md b/docs/live-responses-websocket.md new file mode 100644 index 0000000..713b15f --- /dev/null +++ b/docs/live-responses-websocket.md @@ -0,0 +1,110 @@ +# Live Responses WebSocket continuation test + +`crates/agentkit-provider-openai/tests/live_responses_websocket.rs` uses only public +AgentKit APIs. The network test is ignored by default. It makes three billable +model requests, or four with the optional fresh-session replay check. Run it only +with explicit authorization and a small inference budget. + +## Environment interface + +Choose one credential source: + +- ChatGPT subscription: `AGENTKIT_LIVE_CHATGPT_BEARER` **and** + `AGENTKIT_LIVE_CHATGPT_ACCOUNT_ID` (both required). +- Public API: `OPENAI_API_KEY`. + +The complete ChatGPT pair takes precedence. A partial pair is an error, even if +an API key exists. Empty values count as absent. The test never reads `.env`, Kit +credential files, keychains, or refresh tokens, and never refreshes credentials. + +Other settings: + +| Variable | Meaning | +| --- | --- | +| `AGENTKIT_LIVE_REQUIRED=1` | Fail if credentials are absent. Otherwise print `SKIP` and return before opening any socket. Use this in live CI to prevent a false-green missing-credentials run. | +| `AGENTKIT_LIVE_MODEL` | Model override. Defaults to `gpt-5.4` for ChatGPT and `gpt-4.1-mini` for the public API. Model/account must support Responses WebSocket and tools. | +| `AGENTKIT_LIVE_RECONNECT=1` | Add a fourth request using a fresh session/socket and the full real transcript. | + +With credentials already securely injected into the environment: + +```sh +AGENTKIT_LIVE_REQUIRED=1 RUST_LOG=off RUST_BACKTRACE=0 \ + cargo test -p agentkit-provider-openai --test live_responses_websocket \ + -- --ignored --nocapture +``` + +Prefer compiling **before** injecting credentials, then invoking the compiled +integration-test binary directly. This keeps credentials out of cargo, build +scripts, and compiler subprocesses. Never put secrets in command-line arguments, +print environment contents, enable shell tracing, or enable HTTP/wire logging. +The test itself does not install a tracing subscriber. Disable core dumps when +running with real credentials. Use a trusted local machine: credentials and +payloads necessarily exist in process memory. + +## What is proved + +A loopback-only forwarding WebSocket proxy upgrades to the fixed HTTPS OpenAI +endpoint with real authentication. Authentication headers are added only to the +upstream request; the provider sees a nonsecret placeholder on loopback. Redirects, +environment proxies, and upstream HTTP retries are disabled. The provider uses +`OpenAIResponsesTransport::WebSocket`, **not** `Auto`: no SSE fallback. + +The test passes the full application transcript to every ordinary +`ModelSession::begin_turn` call, consumes actual `ModelTurnEvent`s, and appends +`Finished.output_items` unchanged (including opaque provider metadata), as the +agent loop does. Only the local tool result and subsequent user messages are +constructed by the application. It does not invent continuation IDs or metadata. + +The proxy forwards unmodified frames. It retains only request counts, connection +indices, input counts/types, and booleans indicating whether `previous_response_id` +matched the latest completed response. Response IDs are held transiently in memory +for equality checks, never logged or included in assertions. No transcript, header, +response body, token, account ID, or raw upstream error is printed or persisted. + +Assertions require: + +1. A full first request produces exactly the requested local tool call. +2. Tool output is accepted and appears in the answer. The outgoing request has the + correct `previous_response_id` and exactly one input: the new tool output. +3. A subsequent user turn is accepted and answered. Its request has the correct + `previous_response_id` and exactly one input: the new user message. +4. With the optional replay setting, a **fresh public session/socket** accepts the + complete transcript without `previous_response_id`. This is session restart + coverage, not a forced same-session network-failure/retry test. Its exact full + input count is calculated from the first wire request plus replayable outputs + observed in the three real completed response streams, plus the new tool/user inputs. + `response.output_item.done` supplies output counts when the terminal completion + carries an empty output array. Summary-only reasoning is excluded; encrypted reasoning is counted. This does + not call the public request encoder on credential-bound metadata, strip that + metadata, or synthesize a continuation. + +The proxy refuses to forward beyond the three/four-request budget. Strict +request-count assertions also reject hidden retry/replay success. Each model +turn has a 90-second deadline; upstream connection/handshake limits are 20/30 +seconds. Prompts request short literal answers. Public API responses are capped at +512 output tokens; the private profile does not support that field. Drop cancels +the forwarding task. Authentication failure is a failure, never a skip; error +details are deliberately suppressed. + +## Safely using an existing Kit file credential (external runner only) + +The suite must remain environment-only. A separately authorized, **uncommitted** +runner may read the user's selected Kit file store and pass only the access token +and account ID into the already-compiled test process environment. + +Kit's `src/provider/openai_auth.rs` uses the `openai-subscription` namespace and +`subscription` identity. `src/credentials.rs` names the record +`BLAKE3(b"openai-subscription\0subscription").hex() + ".json"`, where `\0` +denotes one NUL byte. Resolve the store directory from the nonsecret +`credential_store` / `credential_dir` configuration fields; do not dump the config. +The record contains `access_token`, `account_id`, and Unix-seconds `expires_at`, +as well as other fields the runner must not expose or pass to the test. + +Before launching, check expiry with a five-minute margin. If expired, stop and ask +the owner to refresh through Kit; do not independently rotate the refresh token. +Do not rewrite the credential record. Catch parse/read errors without printing +record contents or exception representations. Construct a minimal child +environment; omit inherited debug/logging/proxy/preload settings and all unrelated +credentials. Disable core dumps and impose an overall subprocess timeout. Do not +start the runner until the production implementation is ready and live inference +is explicitly approved.