From 5d930154a9bcb84d85fe6d09796bc9a9ade440e3 Mon Sep 17 00:00:00 2001 From: Deepak Jain Date: Tue, 22 Sep 2026 12:09:08 +0000 Subject: [PATCH 1/2] fix(client): record conversation id with canonical cross-format history Signed-off-by: Deepak Jain --- crates/libsy-llm-client/src/run.rs | 113 +++++++++++++++++++++++++++-- 1 file changed, 106 insertions(+), 7 deletions(-) diff --git a/crates/libsy-llm-client/src/run.rs b/crates/libsy-llm-client/src/run.rs index 01e1c36fc..3339b8801 100644 --- a/crates/libsy-llm-client/src/run.rs +++ b/crates/libsy-llm-client/src/run.rs @@ -719,8 +719,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, + input, + conversation.as_deref(), + ) + .map_err(|error| LibsyError::client_call(model.clone(), error))?; } LlmResponse::Agg(agg) } @@ -757,6 +763,7 @@ impl ClientRouter { &model, store, input, + conversation.as_deref(), )?; } return Ok(None); @@ -842,6 +849,7 @@ impl ClientRouter { model: &ModelId, store: bool, input: &CanonicalInput, + conversation: Option<&str>, ) -> std::result::Result<(), LlmClientError> { let response_id = response.id.as_deref().filter(|_| store); if response_id.is_none() { @@ -856,11 +864,12 @@ 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"); @@ -1283,6 +1292,96 @@ mod tests { Ok(()) } + #[tokio::test] + async fn cross_format_conversation_lookup_finds_canonical_history() -> Result<()> { + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(|_request: &wiremock::Request| { + ResponseTemplate::new(200).set_body_json(json!({ + "id": "chatcmpl_seed", "object": "chat.completion", "model": "weak", + "choices": [{ + "index": 0, + "message": {"role": "assistant", "content": "seed answer"}, + "finish_reason": "stop" + }], + "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2} + })) + }) + .mount(&server) + .await; + + let client: Arc = Arc::new( + TranslatingLlmClient::new(&[ModelConfig::new( + "weak", + Backend::OpenAiChat(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 mut seed = Request { + llm_request: switchyard_translation::decode_request( + WireFormat::OpenAiResponses, + &json!({"model": "route", "input": "SEED", "store": true}), + ) + .map_err(|error| LibsyError::external("decoding seed request", error))?, + raw_request: None, + metadata: None, + }; + seed.llm_request + .extensions + .fields + .insert("conversation".to_string(), json!("conv_1")); + + let (_, response) = run( + Arc::new(switchyard_libsy::Passthrough), + clients.clone(), + seed, + models, + None, + ) + .await?; + response + .llm_response + .into_agg() + .await + .map_err(|error| LibsyError::client_call("weak", error))?; + + let mut probe = Request { + llm_request: switchyard_translation::decode_request( + WireFormat::OpenAiResponses, + &json!({"model": "route", "input": "RECALL"}), + ) + .map_err(|error| LibsyError::external("decoding continuation request", error))?, + raw_request: None, + metadata: None, + }; + probe + .llm_request + .extensions + .fields + .insert("conversation".to_string(), json!("conv_1")); + + let owner = clients + .stored_state_owner(&probe) + .expect("conversation id finds the recorded canonical history"); + assert_eq!(owner.trigger, "conversation"); + let history = owner.history.expect("materialized canonical history"); + assert_eq!(history.segment.len(), 2); + Ok(()) + } + #[tokio::test] async fn responses_stored_tool_continuation_materializes_for_anthropic() -> Result<()> { let server = MockServer::start().await; From 676f55c0721892756349539e35be0e94df6f0c5f Mon Sep 17 00:00:00 2001 From: Deepak Jain Date: Tue, 22 Sep 2026 13:12:32 +0000 Subject: [PATCH 2/2] fix(client): refresh materialized conversation history on repeat store Signed-off-by: Deepak Jain --- crates/libsy-llm-client/src/run.rs | 104 ++++++++++++++++++++++++++++- 1 file changed, 103 insertions(+), 1 deletion(-) diff --git a/crates/libsy-llm-client/src/run.rs b/crates/libsy-llm-client/src/run.rs index 3339b8801..269773d50 100644 --- a/crates/libsy-llm-client/src/run.rs +++ b/crates/libsy-llm-client/src/run.rs @@ -518,7 +518,14 @@ impl StateOwners { } } if let Some(id) = conversation_id { - self.by_id.entry(id.to_owned()).or_insert(state); + // Materialized canonical history always describes the conversation's + // latest state, so a repeat store replaces the entry; provider-owned + // state keeps first-writer routing pins. + if materialized { + self.by_id.insert(id.to_owned(), state); + } else { + self.by_id.entry(id.to_owned()).or_insert(state); + } } Ok(()) } @@ -1382,6 +1389,101 @@ mod tests { Ok(()) } + #[tokio::test] + async fn conversation_lookup_returns_latest_materialized_history() -> Result<()> { + let server = MockServer::start().await; + Mock::given(method("POST")) + .respond_with(|_request: &wiremock::Request| { + ResponseTemplate::new(200).set_body_json(json!({ + "id": "chatcmpl_seed", "object": "chat.completion", "model": "weak", + "choices": [{ + "index": 0, + "message": {"role": "assistant", "content": "seed answer"}, + "finish_reason": "stop" + }], + "usage": {"prompt_tokens": 1, "completion_tokens": 1, "total_tokens": 2} + })) + }) + .mount(&server) + .await; + + let client: Arc = Arc::new( + TranslatingLlmClient::new(&[ModelConfig::new( + "weak", + Backend::OpenAiChat(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"]); + + for input in ["FIRST", "SECOND"] { + let mut request = Request { + llm_request: switchyard_translation::decode_request( + WireFormat::OpenAiResponses, + &json!({"model": "route", "input": input, "store": true}), + ) + .map_err(|error| LibsyError::external("decoding request", error))?, + raw_request: None, + metadata: None, + }; + request + .llm_request + .extensions + .fields + .insert("conversation".to_string(), json!("conv_1")); + + let (_, response) = run( + Arc::new(switchyard_libsy::Passthrough), + clients.clone(), + request, + models.clone(), + None, + ) + .await?; + response + .llm_response + .into_agg() + .await + .map_err(|error| LibsyError::client_call("weak", error))?; + } + + let mut probe = Request { + llm_request: switchyard_translation::decode_request( + WireFormat::OpenAiResponses, + &json!({"model": "route", "input": "RECALL"}), + ) + .map_err(|error| LibsyError::external("decoding continuation request", error))?, + raw_request: None, + metadata: None, + }; + probe + .llm_request + .extensions + .fields + .insert("conversation".to_string(), json!("conv_1")); + + let owner = clients + .stored_state_owner(&probe) + .expect("conversation id finds the recorded canonical history"); + assert_eq!(owner.trigger, "conversation"); + let history = owner.history.expect("materialized canonical history"); + // The second response stored the full four-message history; a stale + // first-writer entry would only hold the first two messages. + assert_eq!(history.segment.len(), 4); + Ok(()) + } + #[tokio::test] async fn responses_stored_tool_continuation_materializes_for_anthropic() -> Result<()> { let server = MockServer::start().await;