diff --git a/crates/agentic-server/src/handler/websocket/responses.rs b/crates/agentic-server/src/handler/websocket/responses.rs index ea22d73..25ca651 100644 --- a/crates/agentic-server/src/handler/websocket/responses.rs +++ b/crates/agentic-server/src/handler/websocket/responses.rs @@ -12,8 +12,10 @@ use serde_json::Value; use tokio_util::sync::CancellationToken; use tracing::{debug, warn}; -use agentic_core::executor::{BoxStream, ExecuteRequest, ExecutorError}; +use agentic_core::ResponseUsage; +use agentic_core::executor::{BoxStream, ExecuteRequest, ExecutorError, RequestContext, rehydrate_conversation}; use agentic_core::types::request_response::RequestPayload; +use agentic_core::utils::common::utcnow_str; use super::super::common::{MAX_BODY_SIZE, extract_bearer}; use super::error::WsError; @@ -113,6 +115,7 @@ async fn handle_ws_text( return Err(WsError::UnexpectedType); } + let generate = value.get("generate").and_then(Value::as_bool); let mut payload = serde_json::from_value::(value).map_err(ExecutorError::from)?; let requested_stream = payload.stream; let requested_store = payload.store; @@ -125,10 +128,16 @@ async fn handle_ws_text( forced_store = payload.store, has_previous_response_id = payload.previous_response_id.is_some(), has_conversation_id = payload.conversation_id.is_some(), + ?generate, tools = payload.tools.as_ref().map_or(0, Vec::len), "accepted websocket response.create" ); + if generate == Some(false) { + debug!("handling non-generating websocket request locally"); + return complete_without_inference(sender, state, payload).await; + } + let auth = extract_bearer(headers, state.openai_api_key.as_deref()); let result = ExecuteRequest::new(payload, Arc::clone(&state.exec_ctx)) .with_auth(auth) @@ -143,6 +152,57 @@ async fn handle_ws_text( stream_ws_response(sender, receiver, stream, shutdown_token, queue).await } +async fn complete_without_inference( + sender: &mut WsSender, + state: &AppState, + payload: RequestPayload, +) -> Result<(), WsError> { + let ctx = rehydrate_conversation(payload, &state.exec_ctx).await?; + let created_at = utcnow_str(); + let created_event = empty_response_event(&ctx, created_at, "response.created", "in_progress", 0, None); + let completed_event = empty_response_event( + &ctx, + created_at, + "response.completed", + "completed", + 1, + Some(ResponseUsage::default()), + ); + + state.exec_ctx.resp_handler.execute_turn(ctx, Vec::new()).await?; + + send_ws_json(sender, created_event).await?; + send_ws_json(sender, completed_event).await +} + +fn empty_response_event( + ctx: &RequestContext, + created_at: i64, + event_type: &str, + status: &str, + sequence_number: u32, + usage: Option, +) -> Value { + serde_json::json!({ + "type": event_type, + "sequence_number": sequence_number, + "response": { + "id": &ctx.response_id, + "object": "response", + "created_at": created_at, + "model": &ctx.enriched_request.model, + "status": status, + "output": [], + "usage": usage, + "incomplete_details": null, + "error": null, + "previous_response_id": &ctx.original_request.previous_response_id, + "conversation_id": &ctx.conversation_id, + "instructions": &ctx.enriched_request.instructions, + }, + }) +} + /// Stream a response from the executor to the client. /// /// Requests arriving from the client while the stream is active are pushed diff --git a/crates/agentic-server/tests/responses_websocket_test.rs b/crates/agentic-server/tests/responses_websocket_test.rs index a7c5310..f44a143 100644 --- a/crates/agentic-server/tests/responses_websocket_test.rs +++ b/crates/agentic-server/tests/responses_websocket_test.rs @@ -479,6 +479,87 @@ async fn test_websocket_first_turn_forwards_incremental_events_and_final_payload assert!(requests[0].get("type").is_none()); } +#[tokio::test] +async fn test_websocket_generate_false_is_local_and_reusable() { + let mock = MockResponsesServer::start(vec![sse_response("resp_upstream_1", "msg_upstream_1", "HELLO")]).await; + let fixture = storage_backed_state(&mock.url).await; + let (gateway_url, _gateway) = spawn_gateway(fixture.state.clone()).await; + let mut ws = connect_responses_ws(&gateway_url).await; + + send_json( + &mut ws, + json!({ + "type": "response.create", + "model": "test-model", + "input": [], + "generate": false, + "store": false, + "stream": true + }), + ) + .await; + + let warmup = recv_until_completed(&mut ws).await; + assert_eq!(warmup.len(), 2); + assert_eq!(warmup[0]["type"], "response.created"); + assert_eq!(warmup[1]["type"], "response.completed"); + assert_eq!(warmup[0]["response"]["id"], warmup[1]["response"]["id"]); + assert_eq!(warmup[1]["response"]["output"], json!([])); + assert_eq!(warmup[1]["response"]["usage"]["total_tokens"], 0); + assert!(mock.request_bodies().await.is_empty()); + + let warmup_id = warmup[1]["response"]["id"].as_str().unwrap(); + send_json( + &mut ws, + json!({ + "type": "response.create", + "model": "test-model", + "previous_response_id": warmup_id, + "input": [{"type": "message", "role": "user", "content": "hello"}], + "store": false, + "stream": true + }), + ) + .await; + + let response = recv_until_completed(&mut ws).await; + assert_eq!(response.last().unwrap()["response"]["previous_response_id"], warmup_id); + let requests = mock.request_bodies().await; + assert_eq!(requests.len(), 1); + assert_eq!(requests[0]["input"].as_array().unwrap().len(), 1); + assert_eq!(requests[0]["input"][0]["role"], "user"); + assert_eq!(requests[0]["input"][0]["content"], "hello"); +} + +#[tokio::test] +async fn test_websocket_empty_input_without_generate_reaches_upstream() { + let mock = MockResponsesServer::start(vec![sse_response("resp_upstream_1", "msg_upstream_1", "HELLO")]).await; + let fixture = storage_backed_state(&mock.url).await; + let (gateway_url, _gateway) = spawn_gateway(fixture.state.clone()).await; + let mut ws = connect_responses_ws(&gateway_url).await; + + send_json( + &mut ws, + json!({ + "type": "response.create", + "model": "test-model", + "input": [], + "store": false, + "stream": true + }), + ) + .await; + + let response = recv_until_completed(&mut ws).await; + assert_eq!( + response.last().unwrap()["response"]["output"][0]["content"][0]["text"], + "HELLO" + ); + let requests = mock.request_bodies().await; + assert_eq!(requests.len(), 1); + assert_eq!(requests[0]["input"], json!([])); +} + #[tokio::test] async fn test_websocket_restores_namespace_tool_call_events() { let mock = MockResponsesServer::start(vec![sse_function_call_response(