From 670e6f5bd1db0c82ac00c1d5afe594b543b18fe1 Mon Sep 17 00:00:00 2001 From: Atharva-Kanherkar <142440039+Atharva-Kanherkar@users.noreply.github.com> Date: Sun, 20 Sep 2026 16:31:38 +0530 Subject: [PATCH 1/6] fix(llm-client): store canonical history under the conversation id remember_canonical_response recorded cross-format materialized history under the response id only, so a Responses continuation by conversation missed its state and reached the backend without prior turns. Pass the request conversation id through and overwrite the conversation entry on materialized records so multi-turn chains keep the latest history. Fixes NVIDIA-NeMo/Switchyard#802 Signed-off-by: Atharva-Kanherkar <142440039+Atharva-Kanherkar@users.noreply.github.com> --- crates/libsy-llm-client/src/run.rs | 28 ++++++++++++++++++++-------- 1 file changed, 20 insertions(+), 8 deletions(-) diff --git a/crates/libsy-llm-client/src/run.rs b/crates/libsy-llm-client/src/run.rs index 01e1c36fc..faf38b240 100644 --- a/crates/libsy-llm-client/src/run.rs +++ b/crates/libsy-llm-client/src/run.rs @@ -518,7 +518,11 @@ impl StateOwners { } } if let Some(id) = conversation_id { - self.by_id.entry(id.to_owned()).or_insert(state); + if materialized { + self.by_id.insert(id.to_owned(), state); + } else { + self.by_id.entry(id.to_owned()).or_insert(state); + } } Ok(()) } @@ -719,8 +723,14 @@ impl ClientRouter { .map_err(|error| LibsyError::client_call(model.clone(), error))?; } } else if let Some(input) = &canonical_input { - self.remember_canonical_response(&agg, &model, store, input) - .map_err(|error| LibsyError::client_call(model.clone(), error))?; + self.remember_canonical_response( + &agg, + &model, + store, + conversation.as_deref(), + input, + ) + .map_err(|error| LibsyError::client_call(model.clone(), error))?; } LlmResponse::Agg(agg) } @@ -756,6 +766,7 @@ impl ClientRouter { &accumulator.finish(), &model, store, + conversation.as_deref(), input, )?; } @@ -841,6 +852,7 @@ impl ClientRouter { response: &AggLlmResponse, model: &ModelId, store: bool, + conversation: Option<&str>, input: &CanonicalInput, ) -> std::result::Result<(), LlmClientError> { let response_id = response.id.as_deref().filter(|_| store); @@ -856,11 +868,11 @@ impl ClientRouter { input.parent.clone(), Arc::from(segment), )); - let result = - self.inner - .state_owners - .lock() - .remember(response_id, None, model, Some(history)); + let result = self + .inner + .state_owners + .lock() + .remember(response_id, conversation, model, Some(history)); if let Err(error) = result { if matches!(error, LlmClientError::ResponseStateLimitExceeded { .. }) { tracing::warn!(%error, "cross-format Responses state capacity reached; history was not retained"); From c43b2c7e70d0fe7ebc233d8758bdeede6458b439 Mon Sep 17 00:00:00 2001 From: Atharva-Kanherkar <142440039+Atharva-Kanherkar@users.noreply.github.com> Date: Sun, 20 Sep 2026 16:33:44 +0530 Subject: [PATCH 2/6] test(llm-client): cover conversation continuations on both cross-format paths Adds a StateOwners level check that materialized conversation entries keep the latest history, a buffered Responses to Anthropic conversation continuation test asserting the upstream receives the full chain, and a streamed conversation continuation test asserting materialization after completion. Signed-off-by: Atharva-Kanherkar <142440039+Atharva-Kanherkar@users.noreply.github.com> --- crates/libsy-llm-client/src/run.rs | 238 +++++++++++++++++++++++++++++ 1 file changed, 238 insertions(+) diff --git a/crates/libsy-llm-client/src/run.rs b/crates/libsy-llm-client/src/run.rs index faf38b240..1ece7e49b 100644 --- a/crates/libsy-llm-client/src/run.rs +++ b/crates/libsy-llm-client/src/run.rs @@ -1502,6 +1502,134 @@ mod tests { Ok(()) } + #[tokio::test] + async fn responses_conversation_continuation_materializes_for_anthropic() -> Result<()> { + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(|request: &wiremock::Request| { + let body: Value = serde_json::from_slice(&request.body).expect("request JSON"); + let text = body.to_string(); + let id = if text.contains("thanks question") { + "msg_third" + } else if text.contains("recall question") { + "msg_follow" + } else { + "msg_seed" + }; + ResponseTemplate::new(200).set_body_json(json!({ + "id": id, "type": "message", "role": "assistant", "model": "weak", + "content": [{"type": "text", "text": "ok"}], + "stop_reason": "end_turn", + "usage": {"input_tokens": 1, "output_tokens": 1} + })) + }) + .mount(&server) + .await; + + let client: Arc = Arc::new( + TranslatingLlmClient::new(&[ModelConfig::new( + "weak", + Backend::Anthropic(HttpBackendConfig { + base_url: server.uri(), + api_key: None, + forward_auth: false, + extra_headers: BTreeMap::new(), + extra_body: BTreeMap::new(), + reasoning_effort: None, + max_retries: 0, + timeout: None, + }), + None, + )]) + .map_err(|error| LibsyError::external("building test client", error))?, + ); + let clients = ClientRouter::new(HashMap::from([(ModelId::from("weak"), client)])); + let models = to_category_map(&["weak"]); + + let request = |input: &str| { + let llm_request = switchyard_translation::decode_request( + WireFormat::OpenAiResponses, + &json!({ + "model": "route", + "input": input, + "conversation": "conv_081", + "store": true + }), + ) + .map_err(|error| LibsyError::external("decoding request", error)); + Ok::(Request { + llm_request: llm_request?, + raw_request: None, + metadata: None, + }) + }; + + let (_, response) = run( + Arc::new(switchyard_libsy::Passthrough), + clients.clone(), + request("seed question")?, + models.clone(), + None, + ) + .await?; + assert_eq!( + response + .llm_response + .as_agg() + .and_then(|agg| agg.id.as_deref()), + Some("msg_seed") + ); + + let (_, response) = run( + Arc::new(switchyard_libsy::Passthrough), + clients.clone(), + request("recall question")?, + models.clone(), + None, + ) + .await?; + assert_eq!( + completion_text( + response + .llm_response + .as_agg() + .expect("buffered follow-up response") + ), + "ok" + ); + + let (_, response) = run( + Arc::new(switchyard_libsy::Passthrough), + clients, + request("thanks question")?, + models, + None, + ) + .await?; + assert_eq!( + completion_text( + response + .llm_response + .as_agg() + .expect("buffered chained response") + ), + "ok" + ); + + let requests = server.received_requests().await.expect("request recording"); + assert_eq!(requests.len(), 3); + let seed = String::from_utf8_lossy(&requests[0].body); + assert!(seed.contains("seed question")); + assert!(!seed.contains("recall question")); + let follow = String::from_utf8_lossy(&requests[1].body); + assert!(follow.contains("seed question")); + assert!(follow.contains("recall question")); + let third = String::from_utf8_lossy(&requests[2].body); + assert!(third.contains("recall question")); + assert!(third.contains("thanks question")); + Ok(()) + } + #[tokio::test] async fn streamed_cross_format_state_is_recorded_only_after_completion() -> Result<()> { let client = Arc::new(CandidateClient { @@ -1579,6 +1707,76 @@ mod tests { Ok(()) } + #[tokio::test] + async fn streamed_cross_format_conversation_state_is_recorded_after_completion() -> Result<()> { + let client = Arc::new(CandidateClient { + calls: Mutex::new(Vec::new()), + requests: Mutex::new(Vec::new()), + first: FirstOutcome::StreamSuccess, + }); + let clients = ClientRouter::new(HashMap::from([( + ModelId::from("weak"), + client as Arc, + )])); + let mut seed = request(); + seed.llm_request.messages = vec![Message::text(Role::User, "seed question")]; + seed.llm_request + .extensions + .fields + .insert("conversation".to_string(), json!("conv_stream")); + let mut response = stream_response(vec![ + LlmResponseChunk::MessageStart { + id: Some("msg_stream".to_string()), + model: Some("weak".to_string()), + }, + LlmResponseChunk::TextDelta { + index: 0, + text: "streamed".to_string(), + }, + LlmResponseChunk::MessageStop { + reason: Some("stop".to_string()), + }, + ]); + response.set_served_model(&ModelId::from("weak")); + let response = clients.remember_state_owner(&seed, response)?; + + let mut follow = request(); + follow + .llm_request + .extensions + .fields + .insert("conversation".to_string(), json!("conv_stream")); + follow.llm_request.messages = vec![Message::text(Role::User, "recall question")]; + assert!(clients.stored_state_owner(&follow).is_none()); + response + .llm_response + .into_agg() + .await + .map_err(|error| LibsyError::client_call("weak", error))?; + + let outcome = continue_on( + clients + .stored_state_owner(&follow) + .expect("completed conversation stream state"), + "passthrough", + follow, + ); + assert_eq!(outcome.request.llm_request.messages.len(), 3); + assert!(matches!( + outcome.request.llm_request.messages[0].content.as_slice(), + [ContentBlock::Text { text }] if text == "seed question" + )); + assert!(matches!( + outcome.request.llm_request.messages[1].content.as_slice(), + [ContentBlock::Text { text }] if text == "streamed" + )); + assert!(matches!( + outcome.request.llm_request.messages[2].content.as_slice(), + [ContentBlock::Text { text }] if text == "recall question" + )); + Ok(()) + } + #[test] fn cross_format_response_with_store_false_is_not_retained() -> Result<()> { let client = Arc::new(CandidateClient { @@ -1721,6 +1919,46 @@ mod tests { Ok(()) } + #[test] + fn materialized_conversation_state_keeps_latest_history() -> std::result::Result<(), LlmClientError> { + let mut owners = StateOwners::default(); + let model = ModelId::from("model/a"); + let first = Arc::new(MessageHistory::new( + None, + Arc::from(vec![Message::text(Role::User, "seed question")]), + )); + let second = Arc::new(MessageHistory::new( + None, + Arc::from(vec![Message::text(Role::User, "recall question")]), + )); + owners.remember(Some("resp_1"), Some("conv_1"), &model, Some(first))?; + owners.remember(Some("resp_2"), Some("conv_1"), &model, Some(second))?; + let latest = owners + .owner("conv_1") + .and_then(|state| state.history.clone()) + .expect("materialized conversation state"); + assert!(matches!( + latest.segment.as_ref(), + [Message { content, .. }] if matches!( + content.as_slice(), + [ContentBlock::Text { text }] if text == "recall question" + ) + )); + owners.remember(Some("resp_3"), Some("conv_1"), &model, None)?; + let kept = owners + .owner("conv_1") + .and_then(|state| state.history.clone()) + .expect("conversation state survived a provider-owned record"); + assert!(matches!( + kept.segment.as_ref(), + [Message { content, .. }] if matches!( + content.as_slice(), + [ContentBlock::Text { text }] if text == "recall question" + ) + )); + Ok(()) + } + #[test] fn concurrent_state_owners_do_not_exceed_capacity() { let model = ModelId::from("model/a"); From e30606ae4aaae0c5ae39381d48a85515de6119ff Mon Sep 17 00:00:00 2001 From: Atharva-Kanherkar <142440039+Atharva-Kanherkar@users.noreply.github.com> Date: Sun, 20 Sep 2026 16:36:53 +0530 Subject: [PATCH 3/6] style: apply rustfmt Signed-off-by: Atharva-Kanherkar <142440039+Atharva-Kanherkar@users.noreply.github.com> --- crates/libsy-llm-client/src/run.rs | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/crates/libsy-llm-client/src/run.rs b/crates/libsy-llm-client/src/run.rs index 1ece7e49b..9955c5458 100644 --- a/crates/libsy-llm-client/src/run.rs +++ b/crates/libsy-llm-client/src/run.rs @@ -868,11 +868,12 @@ impl ClientRouter { input.parent.clone(), Arc::from(segment), )); - let result = self - .inner - .state_owners - .lock() - .remember(response_id, conversation, model, Some(history)); + let result = self.inner.state_owners.lock().remember( + response_id, + conversation, + model, + Some(history), + ); if let Err(error) = result { if matches!(error, LlmClientError::ResponseStateLimitExceeded { .. }) { tracing::warn!(%error, "cross-format Responses state capacity reached; history was not retained"); @@ -1920,7 +1921,8 @@ mod tests { } #[test] - fn materialized_conversation_state_keeps_latest_history() -> std::result::Result<(), LlmClientError> { + fn materialized_conversation_state_keeps_latest_history() + -> std::result::Result<(), LlmClientError> { let mut owners = StateOwners::default(); let model = ModelId::from("model/a"); let first = Arc::new(MessageHistory::new( From 5cbea1824869465c6a8e77b5af6481b2bb5096c7 Mon Sep 17 00:00:00 2001 From: Atharva-Kanherkar <142440039+Atharva-Kanherkar@users.noreply.github.com> Date: Sun, 20 Sep 2026 16:47:07 +0530 Subject: [PATCH 4/6] test(llm-client): assert the seed turn survives and document invariants Signed-off-by: Atharva-Kanherkar <142440039+Atharva-Kanherkar@users.noreply.github.com> --- crates/libsy-llm-client/src/run.rs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/crates/libsy-llm-client/src/run.rs b/crates/libsy-llm-client/src/run.rs index 9955c5458..759fb7637 100644 --- a/crates/libsy-llm-client/src/run.rs +++ b/crates/libsy-llm-client/src/run.rs @@ -1503,6 +1503,7 @@ mod tests { Ok(()) } + // Every prior turn must reach the backend on each conversation continuation. #[tokio::test] async fn responses_conversation_continuation_materializes_for_anthropic() -> Result<()> { let server = MockServer::start().await; @@ -1626,6 +1627,7 @@ mod tests { assert!(follow.contains("seed question")); assert!(follow.contains("recall question")); let third = String::from_utf8_lossy(&requests[2].body); + assert!(third.contains("seed question")); assert!(third.contains("recall question")); assert!(third.contains("thanks question")); Ok(()) @@ -1708,6 +1710,7 @@ mod tests { Ok(()) } + // Conversation history is recorded only once the stream completes. #[tokio::test] async fn streamed_cross_format_conversation_state_is_recorded_after_completion() -> Result<()> { let client = Arc::new(CandidateClient { @@ -1920,6 +1923,7 @@ mod tests { Ok(()) } + // Materialized conversation records keep the latest history. #[test] fn materialized_conversation_state_keeps_latest_history() -> std::result::Result<(), LlmClientError> { From d620444e5e6e4d40702e0523f3cb763d26353503 Mon Sep 17 00:00:00 2001 From: Atharva-Kanherkar <142440039+Atharva-Kanherkar@users.noreply.github.com> Date: Sun, 20 Sep 2026 17:52:56 +0530 Subject: [PATCH 5/6] fix(llm-client): preserve conversation state and tool continuations Signed-off-by: Atharva-Kanherkar <142440039+Atharva-Kanherkar@users.noreply.github.com> --- crates/libsy-llm-client/src/run.rs | 84 ++++++++++++++++--- .../src/codecs/responses/buffered.rs | 13 ++- 2 files changed, 82 insertions(+), 15 deletions(-) diff --git a/crates/libsy-llm-client/src/run.rs b/crates/libsy-llm-client/src/run.rs index 759fb7637..25b23f447 100644 --- a/crates/libsy-llm-client/src/run.rs +++ b/crates/libsy-llm-client/src/run.rs @@ -677,12 +677,11 @@ impl ClientRouter { } fn canonical_input(&self, request: &Request) -> CanonicalInput { - let parent = request - .llm_request - .extensions - .fields + let fields = &request.llm_request.extensions.fields; + let parent = fields .get("previous_response_id") .and_then(Value::as_str) + .or_else(|| conversation_id(fields)) .and_then(|id| self.inner.state_owners.lock().owner(id)?.history.clone()); let parent_len = parent.as_ref().map_or(0, |history| history.len); let (parent, messages) = match request.llm_request.messages.get(parent_len..) { @@ -856,7 +855,8 @@ impl ClientRouter { input: &CanonicalInput, ) -> std::result::Result<(), LlmClientError> { let response_id = response.id.as_deref().filter(|_| store); - if response_id.is_none() { + // `store: false` disables response-ID lookup, not conversation history. + if response_id.is_none() && conversation.is_none() { return Ok(()); } let mut segment = input.messages.to_vec(); @@ -1298,6 +1298,22 @@ mod tests { #[tokio::test] async fn responses_stored_tool_continuation_materializes_for_anthropic() -> Result<()> { + for (field, conversation, store) in [ + ("previous_response_id", Value::Null, true), + ("conversation", json!("conv_tools"), true), + ("conversation", json!({"id": "conv_tools"}), true), + ("conversation", json!("conv_tools"), false), + ] { + check_stored_tool_continuation(field, conversation, store).await?; + } + Ok(()) + } + + async fn check_stored_tool_continuation( + field: &str, + conversation: Value, + store: bool, + ) -> Result<()> { let server = MockServer::start().await; Mock::given(method("POST")) .respond_with(|request: &wiremock::Request| { @@ -1382,11 +1398,12 @@ mod tests { &json!({ "model": "route", "input": "Call get_weather for Paris", + "conversation": conversation, "tools": [{ "type": "function", "name": "get_weather", "parameters": {"type": "object"} }], - "store": true + "store": store }), ) .map_err(|error| LibsyError::external("decoding seed request", error))?, @@ -1414,7 +1431,7 @@ mod tests { WireFormat::OpenAiResponses, &json!({ "model": "route", - "previous_response_id": "msg_seed", + (field): if field == "conversation" { conversation.clone() } else { json!("msg_seed") }, "input": [{ "type": "function_call_output", "call_id": "toolu_weather", @@ -1424,7 +1441,7 @@ mod tests { "type": "function", "name": "get_weather", "parameters": {"type": "object"} }], - "store": true + "store": store }), ) .map_err(|error| LibsyError::external("decoding continuation request", error))?, @@ -1454,9 +1471,9 @@ mod tests { WireFormat::OpenAiResponses, &json!({ "model": "route", - "previous_response_id": "msg_follow", + (field): if field == "conversation" { conversation.clone() } else { json!("msg_follow") }, "input": "thanks", - "store": true + "store": store }), ) .map_err(|error| LibsyError::external("decoding chained request", error))?, @@ -1470,6 +1487,24 @@ mod tests { assert_eq!(history.len, 4); assert_eq!(history.segment.len(), 2); assert!(history.parent.is_some()); + assert_eq!( + clients + .inner + .state_owners + .lock() + .owner("msg_seed") + .is_some(), + store + ); + assert_eq!( + clients + .inner + .state_owners + .lock() + .owner("msg_follow") + .is_some(), + store + ); let (_, response) = run( Arc::new(switchyard_libsy::Passthrough), clients, @@ -1713,6 +1748,20 @@ mod tests { // Conversation history is recorded only once the stream completes. #[tokio::test] async fn streamed_cross_format_conversation_state_is_recorded_after_completion() -> Result<()> { + for (store, response_id) in [ + (true, Some("msg_stream")), + (false, Some("msg_stream")), + (true, None), + ] { + check_streamed_conversation_state(store, response_id).await?; + } + Ok(()) + } + + async fn check_streamed_conversation_state( + store: bool, + response_id: Option<&str>, + ) -> Result<()> { let client = Arc::new(CandidateClient { calls: Mutex::new(Vec::new()), requests: Mutex::new(Vec::new()), @@ -1724,13 +1773,17 @@ mod tests { )])); let mut seed = request(); seed.llm_request.messages = vec![Message::text(Role::User, "seed question")]; + seed.llm_request + .extensions + .fields + .insert("store".to_string(), json!(store)); seed.llm_request .extensions .fields .insert("conversation".to_string(), json!("conv_stream")); let mut response = stream_response(vec![ LlmResponseChunk::MessageStart { - id: Some("msg_stream".to_string()), + id: response_id.map(str::to_owned), model: Some("weak".to_string()), }, LlmResponseChunk::TextDelta { @@ -1766,6 +1819,15 @@ mod tests { follow, ); assert_eq!(outcome.request.llm_request.messages.len(), 3); + assert_eq!( + clients + .inner + .state_owners + .lock() + .owner("msg_stream") + .is_some(), + store && response_id.is_some() + ); assert!(matches!( outcome.request.llm_request.messages[0].content.as_slice(), [ContentBlock::Text { text }] if text == "seed question" diff --git a/crates/switchyard-translation/src/codecs/responses/buffered.rs b/crates/switchyard-translation/src/codecs/responses/buffered.rs index 998505913..e31763d65 100644 --- a/crates/switchyard-translation/src/codecs/responses/buffered.rs +++ b/crates/switchyard-translation/src/codecs/responses/buffered.rs @@ -91,12 +91,17 @@ impl FormatCodec for OpenAiResponsesCodec { }], }); } - // With `previous_response_id`, the provider holds the earlier turns, so a tool output - // may answer a call that is not in this body. + // Stored continuations may answer a tool call that is not in this body. let stored_state = body .get("previous_response_id") .and_then(Value::as_str) - .is_some_and(|id| !id.is_empty()); + .is_some_and(|id| !id.is_empty()) + || body.get("conversation").is_some_and(|conversation| { + conversation + .as_str() + .or_else(|| conversation.get("id").and_then(Value::as_str)) + .is_some_and(|id| !id.is_empty()) + }); let mut custom_call_outputs = Vec::new(); let (messages, instructions) = decode_responses_input( body.get("input").unwrap_or(&Value::String(String::new())), @@ -623,7 +628,7 @@ fn decode_responses_input( }], }; // An output whose call is not in this body answers a call the provider - // holds behind `previous_response_id`; it stays a tool result so routing + // holds behind a continuation ID; it stays a tool result so routing // sees a tool continuation, not a new user turn. Without stored state // the request is malformed, and the output becomes readable user text. let answers_pending_call = pending_tool_calls From 671ffe9e8549d6e44594dc6fd778855d54bbb3f3 Mon Sep 17 00:00:00 2001 From: Atharva-Kanherkar <142440039+Atharva-Kanherkar@users.noreply.github.com> Date: Wed, 23 Sep 2026 01:14:41 +0530 Subject: [PATCH 6/6] test(llm-client): fold conversation coverage into existing tests Drop the standalone buffered conversation test and assert the seed turn in the stored tool continuation test. Merge the streamed conversation test into the existing streamed state test so one setup checks both the response id and conversation lookups. Signed-off-by: Atharva-Kanherkar <142440039+Atharva-Kanherkar@users.noreply.github.com> --- crates/libsy-llm-client/src/run.rs | 321 ++++++----------------------- 1 file changed, 67 insertions(+), 254 deletions(-) diff --git a/crates/libsy-llm-client/src/run.rs b/crates/libsy-llm-client/src/run.rs index 25b23f447..4128dd03c 100644 --- a/crates/libsy-llm-client/src/run.rs +++ b/crates/libsy-llm-client/src/run.rs @@ -1531,6 +1531,10 @@ mod tests { assert_eq!(follow["messages"][2]["content"][0]["type"], "tool_result"); let third: Value = serde_json::from_slice(&requests[2].body) .map_err(|error| LibsyError::external("decoding captured request", error))?; + assert_eq!( + third["messages"][0]["content"], + "Call get_weather for Paris" + ); assert_eq!(third["messages"][1]["content"][0]["type"], "tool_use"); assert_eq!(third["messages"][2]["content"][0]["type"], "tool_result"); assert_eq!(third["messages"][3]["content"], "sunny"); @@ -1538,230 +1542,20 @@ mod tests { Ok(()) } - // Every prior turn must reach the backend on each conversation continuation. - #[tokio::test] - async fn responses_conversation_continuation_materializes_for_anthropic() -> Result<()> { - let server = MockServer::start().await; - Mock::given(method("POST")) - .respond_with(|request: &wiremock::Request| { - let body: Value = serde_json::from_slice(&request.body).expect("request JSON"); - let text = body.to_string(); - let id = if text.contains("thanks question") { - "msg_third" - } else if text.contains("recall question") { - "msg_follow" - } else { - "msg_seed" - }; - ResponseTemplate::new(200).set_body_json(json!({ - "id": id, "type": "message", "role": "assistant", "model": "weak", - "content": [{"type": "text", "text": "ok"}], - "stop_reason": "end_turn", - "usage": {"input_tokens": 1, "output_tokens": 1} - })) - }) - .mount(&server) - .await; - - let client: Arc = Arc::new( - TranslatingLlmClient::new(&[ModelConfig::new( - "weak", - Backend::Anthropic(HttpBackendConfig { - base_url: server.uri(), - api_key: None, - forward_auth: false, - extra_headers: BTreeMap::new(), - extra_body: BTreeMap::new(), - reasoning_effort: None, - max_retries: 0, - timeout: None, - }), - None, - )]) - .map_err(|error| LibsyError::external("building test client", error))?, - ); - let clients = ClientRouter::new(HashMap::from([(ModelId::from("weak"), client)])); - let models = to_category_map(&["weak"]); - - let request = |input: &str| { - let llm_request = switchyard_translation::decode_request( - WireFormat::OpenAiResponses, - &json!({ - "model": "route", - "input": input, - "conversation": "conv_081", - "store": true - }), - ) - .map_err(|error| LibsyError::external("decoding request", error)); - Ok::(Request { - llm_request: llm_request?, - raw_request: None, - metadata: None, - }) - }; - - let (_, response) = run( - Arc::new(switchyard_libsy::Passthrough), - clients.clone(), - request("seed question")?, - models.clone(), - None, - ) - .await?; - assert_eq!( - response - .llm_response - .as_agg() - .and_then(|agg| agg.id.as_deref()), - Some("msg_seed") - ); - - let (_, response) = run( - Arc::new(switchyard_libsy::Passthrough), - clients.clone(), - request("recall question")?, - models.clone(), - None, - ) - .await?; - assert_eq!( - completion_text( - response - .llm_response - .as_agg() - .expect("buffered follow-up response") - ), - "ok" - ); - - let (_, response) = run( - Arc::new(switchyard_libsy::Passthrough), - clients, - request("thanks question")?, - models, - None, - ) - .await?; - assert_eq!( - completion_text( - response - .llm_response - .as_agg() - .expect("buffered chained response") - ), - "ok" - ); - - let requests = server.received_requests().await.expect("request recording"); - assert_eq!(requests.len(), 3); - let seed = String::from_utf8_lossy(&requests[0].body); - assert!(seed.contains("seed question")); - assert!(!seed.contains("recall question")); - let follow = String::from_utf8_lossy(&requests[1].body); - assert!(follow.contains("seed question")); - assert!(follow.contains("recall question")); - let third = String::from_utf8_lossy(&requests[2].body); - assert!(third.contains("seed question")); - assert!(third.contains("recall question")); - assert!(third.contains("thanks question")); - Ok(()) - } - + // Stream state is recorded under the response and conversation ids only after completion. #[tokio::test] async fn streamed_cross_format_state_is_recorded_only_after_completion() -> Result<()> { - let client = Arc::new(CandidateClient { - calls: Mutex::new(Vec::new()), - requests: Mutex::new(Vec::new()), - first: FirstOutcome::StreamSuccess, - }); - let clients = ClientRouter::new(HashMap::from([( - ModelId::from("weak"), - client as Arc, - )])); - let mut seed = request(); - seed.llm_request.messages = vec![Message::text(Role::User, "inspect")]; - seed.llm_request - .preservation - .requests - .insert(WireFormat::OpenAiResponses.into(), json!({})); - let mut response = stream_response(vec![ - LlmResponseChunk::MessageStart { - id: Some("msg_stream".to_string()), - model: Some("weak".to_string()), - }, - LlmResponseChunk::ToolCallDelta { - index: 0, - id: Some("toolu_stream".to_string()), - name: Some("lookup".to_string()), - arguments_delta: Some("{}".to_string()), - }, - LlmResponseChunk::MessageStop { - reason: Some("tool_use".to_string()), - }, - ]); - response.set_served_model(&ModelId::from("weak")); - let response = clients.remember_state_owner(&seed, response)?; - - let mut follow = request(); - follow - .llm_request - .extensions - .fields - .insert("previous_response_id".to_string(), json!("msg_stream")); - follow.llm_request.messages = vec![Message { - role: Role::Tool, - content: vec![ContentBlock::ToolResult(ToolResult { - tool_call_id: "toolu_stream".to_string(), - content: vec![ContentBlock::Text { - text: "ready".to_string(), - }], - is_error: None, - })], - }]; - assert!(clients.stored_state_owner(&follow).is_none()); - response - .llm_response - .into_agg() - .await - .map_err(|error| LibsyError::client_call("weak", error))?; - - let outcome = continue_on( - clients - .stored_state_owner(&follow) - .expect("completed stream state"), - "passthrough", - follow, - ); - assert_eq!(outcome.request.llm_request.messages.len(), 3); - assert!(matches!( - outcome.request.llm_request.messages[1].content.as_slice(), - [ContentBlock::ToolCall(call)] if call.id == "toolu_stream" - )); - assert!(matches!( - outcome.request.llm_request.messages[2].content.as_slice(), - [ContentBlock::ToolResult(result)] if result.tool_call_id == "toolu_stream" - )); - Ok(()) - } - - // Conversation history is recorded only once the stream completes. - #[tokio::test] - async fn streamed_cross_format_conversation_state_is_recorded_after_completion() -> Result<()> { for (store, response_id) in [ (true, Some("msg_stream")), (false, Some("msg_stream")), (true, None), ] { - check_streamed_conversation_state(store, response_id).await?; + check_streamed_state(store, response_id).await?; } Ok(()) } - async fn check_streamed_conversation_state( - store: bool, - response_id: Option<&str>, - ) -> Result<()> { + async fn check_streamed_state(store: bool, response_id: Option<&str>) -> Result<()> { let client = Arc::new(CandidateClient { calls: Mutex::new(Vec::new()), requests: Mutex::new(Vec::new()), @@ -1772,7 +1566,11 @@ mod tests { client as Arc, )])); let mut seed = request(); - seed.llm_request.messages = vec![Message::text(Role::User, "seed question")]; + seed.llm_request.messages = vec![Message::text(Role::User, "inspect")]; + seed.llm_request + .preservation + .requests + .insert(WireFormat::OpenAiResponses.into(), json!({})); seed.llm_request .extensions .fields @@ -1786,60 +1584,75 @@ mod tests { id: response_id.map(str::to_owned), model: Some("weak".to_string()), }, - LlmResponseChunk::TextDelta { + LlmResponseChunk::ToolCallDelta { index: 0, - text: "streamed".to_string(), + id: Some("toolu_stream".to_string()), + name: Some("lookup".to_string()), + arguments_delta: Some("{}".to_string()), }, LlmResponseChunk::MessageStop { - reason: Some("stop".to_string()), + reason: Some("tool_use".to_string()), }, ]); response.set_served_model(&ModelId::from("weak")); let response = clients.remember_state_owner(&seed, response)?; - let mut follow = request(); - follow - .llm_request - .extensions - .fields - .insert("conversation".to_string(), json!("conv_stream")); - follow.llm_request.messages = vec![Message::text(Role::User, "recall question")]; - assert!(clients.stored_state_owner(&follow).is_none()); + let follow = |field: &str, value: Value| { + let mut follow = request(); + follow + .llm_request + .extensions + .fields + .insert(field.to_string(), value); + follow.llm_request.messages = vec![Message { + role: Role::Tool, + content: vec![ContentBlock::ToolResult(ToolResult { + tool_call_id: "toolu_stream".to_string(), + content: vec![ContentBlock::Text { + text: "ready".to_string(), + }], + is_error: None, + })], + }]; + follow + }; + let follows = [ + ( + follow("previous_response_id", json!("msg_stream")), + store && response_id.is_some(), + ), + (follow("conversation", json!("conv_stream")), true), + ]; + for (follow, _) in &follows { + assert!(clients.stored_state_owner(follow).is_none()); + } response .llm_response .into_agg() .await .map_err(|error| LibsyError::client_call("weak", error))?; - let outcome = continue_on( - clients - .stored_state_owner(&follow) - .expect("completed conversation stream state"), - "passthrough", - follow, - ); - assert_eq!(outcome.request.llm_request.messages.len(), 3); - assert_eq!( - clients - .inner - .state_owners - .lock() - .owner("msg_stream") - .is_some(), - store && response_id.is_some() - ); - assert!(matches!( - outcome.request.llm_request.messages[0].content.as_slice(), - [ContentBlock::Text { text }] if text == "seed question" - )); - assert!(matches!( - outcome.request.llm_request.messages[1].content.as_slice(), - [ContentBlock::Text { text }] if text == "streamed" - )); - assert!(matches!( - outcome.request.llm_request.messages[2].content.as_slice(), - [ContentBlock::Text { text }] if text == "recall question" - )); + for (follow, retained) in follows { + let Some(owner) = clients.stored_state_owner(&follow) else { + assert!(!retained); + continue; + }; + assert!(retained); + let outcome = continue_on(owner, "passthrough", follow); + assert_eq!(outcome.request.llm_request.messages.len(), 3); + assert!(matches!( + outcome.request.llm_request.messages[0].content.as_slice(), + [ContentBlock::Text { text }] if text == "inspect" + )); + assert!(matches!( + outcome.request.llm_request.messages[1].content.as_slice(), + [ContentBlock::ToolCall(call)] if call.id == "toolu_stream" + )); + assert!(matches!( + outcome.request.llm_request.messages[2].content.as_slice(), + [ContentBlock::ToolResult(result)] if result.tool_call_id == "toolu_stream" + )); + } Ok(()) }