From 00e9f14ca96e0ff7776ae3a153cb4b5badf359f8 Mon Sep 17 00:00:00 2001 From: nachiketb Date: Fri, 2 Oct 2026 14:33:33 -0700 Subject: [PATCH 1/8] feat(client): serve System One decision calls Signed-off-by: nachiketb --- Cargo.lock | 1 + crates/libsy-llm-client/Cargo.toml | 1 + crates/libsy-llm-client/src/client.rs | 4 +- crates/libsy-llm-client/src/lib.rs | 7 +- crates/libsy-llm-client/src/observation.rs | 9 +- crates/libsy-llm-client/src/run.rs | 276 ++++++++++++++++-- crates/libsy-llm-client/src/system_one.rs | 242 +++++++++++++++ crates/protocol/src/client.rs | 18 +- .../src/runtime.rs | 21 +- crates/switchyard-server/src/lib.rs | 2 +- 10 files changed, 543 insertions(+), 38 deletions(-) create mode 100644 crates/libsy-llm-client/src/system_one.rs diff --git a/Cargo.lock b/Cargo.lock index aeeeaf37b..48b15ca63 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2358,6 +2358,7 @@ dependencies = [ "opentelemetry_sdk", "parking_lot", "reqwest", + "serde", "serde_json", "switchyard-libsy", "switchyard-protocol", diff --git a/crates/libsy-llm-client/Cargo.toml b/crates/libsy-llm-client/Cargo.toml index cb8ea4f93..ded0ecab8 100644 --- a/crates/libsy-llm-client/Cargo.toml +++ b/crates/libsy-llm-client/Cargo.toml @@ -29,6 +29,7 @@ parking_lot.workspace = true http.workspace = true httpdate.workspace = true serde_json.workspace = true +serde.workspace = true tokio.workspace = true tracing.workspace = true tracing-opentelemetry.workspace = true diff --git a/crates/libsy-llm-client/src/client.rs b/crates/libsy-llm-client/src/client.rs index 6ba089d9c..b66a38b42 100644 --- a/crates/libsy-llm-client/src/client.rs +++ b/crates/libsy-llm-client/src/client.rs @@ -1,7 +1,7 @@ // SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 -//! [`TranslatingLlmClient`] — the crate's single public entry point: encode a neutral +//! [`TranslatingLlmClient`]: encode a neutral //! request, call the configured backend over HTTP, decode the neutral response. use std::collections::{BTreeMap, BTreeSet, HashMap}; @@ -896,7 +896,7 @@ fn record_gen_ai_request(url: &str, model: &str, streaming: bool) { } } -fn convert_reqwest_error(error: reqwest::Error) -> LlmClientError { +pub(crate) fn convert_reqwest_error(error: reqwest::Error) -> LlmClientError { // Reqwest labels truncated or otherwise unreadable response bodies as decode // errors, so distinguish them from serde JSON failures at the call site. let error = error.without_url(); diff --git a/crates/libsy-llm-client/src/lib.rs b/crates/libsy-llm-client/src/lib.rs index d5f579c7d..db0cf9c0c 100644 --- a/crates/libsy-llm-client/src/lib.rs +++ b/crates/libsy-llm-client/src/lib.rs @@ -11,6 +11,9 @@ //! back to a [`switchyard_protocol::Response`] — supporting both buffered and //! streamed responses. //! +//! [`SystemOneClient`] serves typed decision requests. Register decision targets +//! with [`ClientRouter::with_decision_clients`] to serve them alongside LLM calls. +//! //! [`run()`] pairs the client with a libsy algorithm: it drives //! [`switchyard_libsy::Algorithm::run_stream`], serves routing-time calls, and makes the terminal //! answer call from the routing outcome when needed. A host that just wants the answer does not @@ -24,14 +27,16 @@ mod observability; mod observation; pub mod raw; pub mod run; +mod system_one; pub use backend::{Backend, DEFAULT_MAX_RETRIES, HttpBackendConfig}; pub use client::{AuxiliaryOperation, ModelConfig, TranslatingLlmClient}; pub use error::{LlmClientError, Result}; -pub use observation::{LlmCallObservation, RunObservation, RunObserver}; +pub use observation::{LlmCallObservation, ModelCallObservation, RunObservation, RunObserver}; pub use raw::RawResponse; pub use run::{ClientRouter, decide, run}; pub use switchyard_translation::RawEventStream; +pub use system_one::SystemOneClient; /// Registers process-wide compatibility gauges with the global meter provider. pub fn initialize_metrics() { diff --git a/crates/libsy-llm-client/src/observation.rs b/crates/libsy-llm-client/src/observation.rs index 308ad4b28..0fc97a5e5 100644 --- a/crates/libsy-llm-client/src/observation.rs +++ b/crates/libsy-llm-client/src/observation.rs @@ -11,7 +11,7 @@ use switchyard_protocol::{ModelId, Usage}; /// One completed model call observed while serving an algorithm run. #[derive(Clone, Debug)] -pub struct LlmCallObservation { +pub struct ModelCallObservation { /// Model selected for the completed call. pub selected_model: ModelId, /// Whether the call completed successfully. @@ -22,13 +22,18 @@ pub struct LlmCallObservation { pub usage: Option, } -/// Events emitted inline while [`crate::run`] serves a routing request. +/// An LLM call's observation, sharing the fields used by decision calls. +pub type LlmCallObservation = ModelCallObservation; + +/// Events emitted inline while [`crate::run()`] serves a routing request. #[derive(Clone, Debug)] pub enum RunObservation { /// Metadata attached to the completed routing outcome. Outcome(OutcomeMetadata), /// A completed model call requested by the algorithm for routing work. LlmCall(LlmCallObservation), + /// A completed decision call requested by the algorithm for routing work. + DecisionCall(ModelCallObservation), /// A completed terminal model call made from the routing outcome. AnswerCall(LlmCallObservation), /// Routing time recorded by the `switchyard.routing_overhead_ms` metric. diff --git a/crates/libsy-llm-client/src/run.rs b/crates/libsy-llm-client/src/run.rs index a3dfc339a..f694fa05c 100644 --- a/crates/libsy-llm-client/src/run.rs +++ b/crates/libsy-llm-client/src/run.rs @@ -32,12 +32,12 @@ use switchyard_libsy::{ }; use switchyard_protocol::{ AggLlmResponse, LlmClientError, LlmResponse, LlmResponseChunk, LlmResponseStream, Message, - ModelId, Request, Response, ResponseAccumulator, RoutedLlmClient, RoutingFallbackReason, - WireFormat, + ModelId, Request, Response, ResponseAccumulator, RoutedDecisionClient, RoutedLlmClient, + RoutingFallbackReason, WireFormat, }; use switchyard_translation::prepare_request_for_target; -use crate::observation::{LlmCallObservation, RunObservation, RunObserver}; +use crate::observation::{LlmCallObservation, ModelCallObservation, RunObservation, RunObserver}; use crate::{metrics, observability}; /// Run one request to completion, serving every offloaded model call with `client`. @@ -152,18 +152,38 @@ pub async fn decide( Ok(outcome) } -async fn unsupported_decision(call: CallDecision) -> Result<()> { - let model = call.model.clone(); - call.respond(Err(LibsyError::client_call( - model, - LlmClientError::General("decision calls are not supported by this client".to_string()), - ))) +#[tracing::instrument( + name = "libsy.decision_client_call", + skip_all, + fields(algorithm = %call.algorithm, selected_model = %call.model, outcome = tracing::field::Empty), +)] +async fn serve_decision( + clients: &ClientRouter, + call: CallDecision, + observations: &Option>>>, +) -> Result<()> { + let client = clients.route_decision(&call.model); + let started = Instant::now(); + let result = async { client?.call(call.request.clone()).await }.await; + tracing::Span::current().record("outcome", if result.is_ok() { "ok" } else { "error" }); + if let Some(observations) = observations { + observations + .lock() + .push(RunObservation::DecisionCall(ModelCallObservation { + selected_model: call.model.clone(), + is_success: result.is_ok(), + duration: started.elapsed(), + usage: result.as_ref().ok().map(|response| response.usage.clone()), + })); + } + let result = result.map_err(|error| LibsyError::client_call(call.model.clone(), error)); + call.respond(result) } /// Emits completed routing calls after the outcome reveals whether one response became the answer. fn emit_routing_observations( observer: &Option, - observations: &Option>>>, + observations: &Option>>>, answered_model: Option<&ModelId>, ) { let (Some(observer), Some(observations)) = (observer, observations) else { @@ -171,31 +191,36 @@ fn emit_routing_observations( }; let mut answer_observed = false; for observation in observations.lock().drain(..) { - if !answer_observed && answered_model == Some(&observation.selected_model) { - answer_observed = true; - observer(RunObservation::AnswerCall(observation)); - } else { - observer(RunObservation::LlmCall(observation)); - } + observer(match observation { + RunObservation::LlmCall(call) + if !answer_observed && answered_model == Some(&call.selected_model) => + { + answer_observed = true; + RunObservation::AnswerCall(call) + } + observation => observation, + }); } } /// Serve one offloaded call and fulfill its promise. /// -/// LLM failures stop the run unless the call enables recovery. Unsupported decisions -/// return an error to the algorithm. +/// LLM failures stop the run unless the call enables recovery. Decision results, +/// including client failures, go back to the algorithm for its routing policy. async fn serve( clients: ClientRouter, call: Call, - observations: Option>>>, + observations: Option>>>, ) -> Result<()> { let call = match call { Call::Model(call) => *call, - Call::Decision(call) => return unsupported_decision(*call).await, + Call::Decision(call) => return serve_decision(&clients, *call, &observations).await, }; let observe = |observation| { if let Some(observations) = &observations { - observations.lock().push(observation); + observations + .lock() + .push(RunObservation::LlmCall(observation)); } }; let target = call.models.first().ok_or(LibsyError::NoTargets)?; @@ -557,13 +582,13 @@ fn conversation_id(fields: &serde_json::Map) -> Option<&str> { /// /// An algorithm routes among named targets; which provider each target lives on is the /// host's concern, and two targets in one run may sit on different providers. A router owns -/// that mapping. It is *not* itself a client: it hands back a [`RoutedLlmClient`] and the -/// caller makes the call. +/// that mapping and resolves the client for each kind of call. /// /// Cloning is cheap — the mapping is shared, so one router can serve every request. #[derive(Clone)] pub struct ClientRouter { inner: Arc, + decision_clients: Arc>>, } struct ClientRouting { @@ -613,7 +638,7 @@ impl ClientRouter { /// Build a router with an explicit list of models that can answer requests. /// - /// `by_model` contains every callable model, including classifiers. Only the model IDs in + /// `by_model` contains callable LLMs, including LLM classifiers. Only the model IDs in /// `completion_targets` determine whether native Responses state needs a routing pin. /// Cross-format state is recorded lazily when a Responses request uses a Chat or Anthropic /// target. `target_prompts` and `routing_answer_target` have the same meaning as in @@ -631,6 +656,7 @@ impl ClientRouter { .next() .is_some_and(|first| clients.any(|client| !Arc::ptr_eq(first, client))); Self { + decision_clients: Arc::default(), inner: Arc::new(ClientRouting { routing: Routing::ByModel(by_model), target_prompts, @@ -648,6 +674,7 @@ impl ClientRouter { /// only duplicate that. pub fn single(client: Arc) -> Self { Self { + decision_clients: Arc::default(), inner: Arc::new(ClientRouting { routing: Routing::Single(client), target_prompts: HashMap::new(), @@ -678,6 +705,27 @@ impl ClientRouter { } } + /// Register clients for decision targets independently of completion targets. + pub fn with_decision_clients( + mut self, + clients: HashMap>, + ) -> Self { + self.decision_clients = Arc::new(clients); + self + } + + /// Resolve a decision target without falling back to an LLM client. + pub fn route_decision( + &self, + model: &ModelId, + ) -> std::result::Result<&Arc, LlmClientError> { + self.decision_clients + .get(model) + .ok_or_else(|| LlmClientError::Configuration { + message: format!("no decision client is configured for model {model:?}"), + }) + } + /// Return the recorded model for the requested response or conversation ID. fn stored_state_owner(&self, request: &Request) -> Option { let fields = &request.llm_request.extensions.fields; @@ -1072,6 +1120,185 @@ mod tests { } } + #[tokio::test] + async fn system_one_serves_decisions_and_returns_failures_to_the_classifier() + -> std::result::Result<(), Box> { + use std::time::Duration; + use switchyard_libsy::{ + CapabilityJudgeConfig, DecisionJudgeConfig, LlmClassifierConfig, LlmTaskClassifier, + TaskClassifierConfig, + }; + use switchyard_protocol::{ + BooleanEstimate, DecisionRequest, DecisionValue, Probability, ScoreValue, + }; + use wiremock::matchers::{body_partial_json, header, path}; + + let server = MockServer::start().await; + let client = Arc::new(crate::SystemOneClient::new( + format!("{}/v1/systemone", server.uri()), + "test-key".into(), + Duration::from_secs(2), + )?); + let mixed: DecisionRequest = serde_json::from_value(json!({ + "model": "judge", "context": {"task": "A simple task"}, + "questions": { + "boolean": {"instructions": "Is it simple?", "kind": {"type": "boolean", "data": { + "true_description": ["Simple"], "false_description": null + }}}, + "route": {"instructions": {"task": "Choose"}, "kind": {"type": "choice", "data": { + "options": [{"id": "advantage", "description": "A wins"}, {"id": "no_advantage"}] + }}}, + "score": {"instructions": "Rate difficulty", "kind": {"type": "score", "data": { + "levels": ["Easy", {"description": "Hard"}] + }}} + } + }))?; + let mut body = json!({ + "model": "jev-1.13.0", "usage": {"input_tokens": 42, "output_tokens": 3}, + "answers": { + "boolean": {"type": "noul", "noul": 0.9}, + "route": {"type": "choice", "choice": "no_advantage", "confidence": 0.7, + "probabilities": {"advantage": 0.2, "no_advantage": 0.8}}, + "score": {"type": "score", "score": 0.6, "confidence": 0.3, + "legend": {"0": "Easy", "1": {"description": "Hard"}}, + "probabilities": {"1": 0.6, "0": 0.4}} + } + }); + Mock::given(method("POST")) + .and(path("/v1/systemone")) + .and(header("authorization", "Bearer test-key")) + .and(body_partial_json(json!({ + "model": "judge", "state": mixed.context, + "questions": { + "boolean": {"type": "noul", "criteria": {"true": ["Simple"]}}, + "route": {"type": "choice", "instructions": {"task": "Choose"}, + "criteria": {"advantage": "A wins", "no_advantage": null}}, + "score": {"type": "score", "criteria": ["Easy", {"description": "Hard"}]} + } + }))) + .respond_with(ResponseTemplate::new(200).set_body_json(&body)) + .expect(1) + .mount(&server) + .await; + let response = client.call(mixed).await?; + assert_eq!(response.model.as_deref(), Some("jev-1.13.0")); + assert_eq!(response.id, None); + assert_eq!(response.usage.input_tokens, Some(42)); + assert_eq!(response.usage.total_tokens, None); + assert_eq!( + response.answers["boolean"].value, + DecisionValue::Boolean(BooleanEstimate::ProbabilityTrue(Probability(0.9))) + ); + assert_eq!( + response.answers["score"].value, + DecisionValue::Score { + value: ScoreValue(0.6), + probabilities: Some(vec![Probability(0.4), Probability(0.6)]), + } + ); + assert_eq!( + response.answers["route"].provider_confidence.map(|c| c.0), + Some(0.7) + ); + server.verify().await; + + body["answers"].as_object_mut().unwrap().remove("boolean"); + body["answers"].as_object_mut().unwrap().remove("score"); + let models = Arc::new(RuntimeModels::new(HashMap::from([ + (Category::Judge, vec!["judge".into()]), + (Category::Capable, vec!["capable".into()]), + (Category::Efficient, vec!["efficient".into()]), + (Category::Any, vec!["efficient".into(), "capable".into()]), + ]))); + let algorithm = || -> Result> { + Ok(Arc::new(LlmTaskClassifier::new( + LlmClassifierConfig::Capability { + config: TaskClassifierConfig { + judge: CapabilityJudgeConfig::Decision(DecisionJudgeConfig { + cutoff: 0.4, + instructions: None, + candidates: BTreeMap::from([ + ("a".into(), "capable".into()), + ("b".into(), "efficient".into()), + ]), + evidence: json!({}), + }), + ..TaskClassifierConfig::default() + }, + }, + )?)) + }; + for (template, expected, is_success) in [ + ( + ResponseTemplate::new(200).set_body_json(&body), + "efficient", + true, + ), + ( + ResponseTemplate::new(200).set_body_string("invalid JSON"), + "capable", + false, + ), + ( + ResponseTemplate::new(503).set_body_string("unavailable"), + "capable", + false, + ), + ] { + server.reset().await; + Mock::given(method("POST")) + .and(path("/v1/systemone")) + .respond_with(template) + .expect(2) + .mount(&server) + .await; + let llm = Arc::new(CandidateClient { + calls: Mutex::default(), + requests: Mutex::default(), + first: FirstOutcome::Unauthorized, + }); + let clients = + ClientRouter::single(llm.clone()).with_decision_clients(HashMap::from([( + ModelId::from("judge"), + client.clone() as Arc, + )])); + let outcome = decide(algorithm()?, clients.clone(), request(), models.clone()).await?; + assert_eq!(outcome.selected_model_id()?, expected); + assert!(llm.calls.lock().is_empty()); + let events = Arc::new(Mutex::new(Vec::new())); + let captured = events.clone(); + let (selected, _) = run( + algorithm()?, + clients, + request(), + models.clone(), + Some(Arc::new(move |event| captured.lock().push(event))), + ) + .await?; + assert_eq!(selected, expected); + assert_eq!(&*llm.calls.lock(), &[ModelId::from(expected)]); + { + let events = events.lock(); + let RunObservation::DecisionCall(call) = &events[0] else { + panic!("missing decision observation") + }; + assert_eq!(call.selected_model, "judge"); + assert_eq!(call.is_success, is_success); + assert_eq!( + call.usage.as_ref().and_then(|u| u.input_tokens), + is_success.then_some(42) + ); + assert!( + events + .iter() + .any(|event| matches!(event, RunObservation::AnswerCall(_))) + ); + } + server.verify().await; + } + Ok(()) + } + fn instruction_text(request: &Request) -> Vec<&str> { request .llm_request @@ -1165,6 +1392,7 @@ mod tests { duration: std::time::Duration::ZERO, usage: None, }) + .map(RunObservation::LlmCall) .into(), ))); let emitted = Arc::new(Mutex::new(Vec::new())); diff --git a/crates/libsy-llm-client/src/system_one.rs b/crates/libsy-llm-client/src/system_one.rs new file mode 100644 index 000000000..caf4881e5 --- /dev/null +++ b/crates/libsy-llm-client/src/system_one.rs @@ -0,0 +1,242 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Buffered System One calls using the provider-neutral decision IR. + +use std::collections::BTreeMap; +use std::time::Duration; + +use async_trait::async_trait; +use serde::Deserialize; +use serde_json::{Value, json}; +use switchyard_protocol::{ + BooleanEstimate, DecisionAnswer, DecisionKind, DecisionRequest, DecisionResponse, + DecisionValue, ModelId, Probability, ProviderConfidence, RoutedDecisionClient, ScoreValue, + Usage, +}; + +use crate::client::convert_reqwest_error; +use crate::{LlmClientError, Result}; + +/// Serves decision requests through a System One endpoint, such as TypeSafe's Jev API. +pub struct SystemOneClient { + client: reqwest::Client, + endpoint: String, + api_key: String, +} + +impl SystemOneClient { + /// `endpoint` is the full URL, including `/v1/systemone`. Each call makes one + /// attempt; `timeout` covers sending the request and reading the response body. + pub fn new(endpoint: impl Into, api_key: String, timeout: Duration) -> Result { + let client = reqwest::Client::builder() + .timeout(timeout) + .redirect(reqwest::redirect::Policy::none()) + .build() + .map_err(convert_reqwest_error)?; + Ok(Self { + client, + endpoint: endpoint.into(), + api_key, + }) + } +} + +#[async_trait] +impl RoutedDecisionClient for SystemOneClient { + async fn call(&self, request: DecisionRequest) -> Result { + let response = self + .client + .post(&self.endpoint) + .bearer_auth(&self.api_key) + .json(&encode(&request)?) + .send() + .await + .map_err(convert_reqwest_error)?; + let status = response.status(); + let body = response.bytes().await.map_err(convert_reqwest_error)?; + if !status.is_success() { + return Err(LlmClientError::UpstreamHttp { + status, + body: String::from_utf8_lossy(&body).replace(&self.api_key, "[REDACTED]"), + }); + } + let response: WireResponse = + serde_json::from_slice(&body).map_err(|source| LlmClientError::InvalidResponse { + source: Box::new(source), + })?; + let answers = response + .answers + .into_iter() + .map(|(id, answer)| { + let value = match answer.value { + WireValue::Noul { noul } => { + DecisionValue::Boolean(BooleanEstimate::ProbabilityTrue(noul)) + } + WireValue::Choice { + choice, + probabilities, + } => DecisionValue::Choice { + selected: choice, + probabilities, + }, + WireValue::Score { + score, + probabilities, + } => { + let probabilities = probabilities + .map(|mut probabilities| { + let Some(DecisionKind::Score { levels }) = + request.questions.get(&id).map(|q| &q.kind) + else { + return Err(LlmClientError::ResponseTranslation(format!( + "score answer {id:?} has no matching score question" + ))); + }; + let invalid_rubric = || { + LlmClientError::ResponseTranslation(format!( + "score probabilities for {id:?} do not match its rubric" + )) + }; + if probabilities.len() != levels.len() { + return Err(invalid_rubric()); + } + // JSON keys are strings; order probabilities by the request's rubric. + (0..levels.len()) + .map(|index| { + probabilities + .remove(&index.to_string()) + .ok_or_else(invalid_rubric) + }) + .collect() + }) + .transpose()?; + DecisionValue::Score { + value: score, + probabilities, + } + } + }; + Ok(( + id, + DecisionAnswer { + value, + provider_confidence: answer.confidence, + }, + )) + }) + .collect::>()?; + Ok(DecisionResponse { + id: response.id, + model: response.model, + answers, + usage: response.usage, + }) + } +} + +fn encode(request: &DecisionRequest) -> Result { + let model = request + .model + .as_ref() + .ok_or_else(|| LlmClientError::InvalidRequest { + message: "System One requires a selected model".into(), + })?; + content(&request.context, false, "context")?; + let mut questions = serde_json::Map::new(); + for (id, question) in &request.questions { + let path = format!("questions.{id}"); + content( + &question.instructions, + true, + &format!("{path}.instructions"), + )?; + let (kind, criteria) = match &question.kind { + DecisionKind::Boolean { + true_description, + false_description, + } => { + let mut criteria = serde_json::Map::new(); + for (key, description) in [("true", true_description), ("false", false_description)] + { + if let Some(description) = description { + content(description, true, &format!("{path}.{key}_description"))?; + criteria.insert(key.into(), description.clone()); + } + } + ("noul", Value::Object(criteria)) + } + DecisionKind::Choice { options } => { + let mut criteria = serde_json::Map::new(); + for option in options { + let description = option.description.as_ref().unwrap_or(&Value::Null); + content(description, true, &format!("{path}.options.{}", option.id))?; + if criteria + .insert(option.id.clone(), description.clone()) + .is_some() + { + return Err(LlmClientError::InvalidRequest { + message: format!("duplicate option {:?} in {path}", option.id), + }); + } + } + ("choice", Value::Object(criteria)) + } + DecisionKind::Score { levels } => { + for (index, level) in levels.iter().enumerate() { + content(level, false, &format!("{path}.levels.{index}"))?; + } + ("score", json!(levels)) + } + }; + questions.insert( + id.clone(), + json!({ + "type": kind, "instructions": question.instructions, "criteria": criteria, + }), + ); + } + Ok(json!({"model": model, "state": request.context, "questions": questions})) +} + +fn content(value: &Value, nullable: bool, path: &str) -> Result<()> { + if value.is_string() || value.is_object() || value.is_array() || (nullable && value.is_null()) { + return Ok(()); + } + Err(LlmClientError::RequestEncoding(format!( + "System One cannot represent {path} as {}", + value + ))) +} + +#[derive(Deserialize)] +struct WireResponse { + id: Option, + model: Option, + answers: BTreeMap, + #[serde(default)] + usage: Usage, +} + +#[derive(Deserialize)] +struct WireAnswer { + #[serde(flatten)] + value: WireValue, + confidence: Option, +} + +#[derive(Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +enum WireValue { + Noul { + noul: Probability, + }, + Choice { + choice: String, + probabilities: Option>, + }, + Score { + score: ScoreValue, + probabilities: Option>, + }, +} diff --git a/crates/protocol/src/client.rs b/crates/protocol/src/client.rs index 8d312e72d..fa37f01c1 100644 --- a/crates/protocol/src/client.rs +++ b/crates/protocol/src/client.rs @@ -1,17 +1,15 @@ // SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 -//! The routed-call server trait and its shared error types. +//! Routed-call client contracts and shared error types. //! -//! [`RoutedLlmClient`] is the one piece of I/O the protocol does not own: a host -//! implements it to actually perform a model call. It lives here — rather than in -//! libsy's orchestration crate — so a client crate that depends only on the protocol -//! can serve routed calls without pulling in the orchestrator. +//! Hosts implement these traits to perform model calls. Keeping the contracts here +//! lets clients depend on the protocol without pulling in libsy's orchestration. use async_trait::async_trait; use thiserror::Error; -use crate::{ModelId, Request, Response}; +use crate::{DecisionRequest, DecisionResponse, ModelId, Request, Response}; /// A boxed client-specific error preserved as the source of a routed call failure. pub type BoxError = Box; @@ -157,3 +155,11 @@ pub trait RoutedLlmClient: Send + Sync { /// Make a request async fn call(&self, request: Request) -> Result; } + +/// Serves typed decision calls over a host-owned transport. +/// Implementations may be shared across targets and called concurrently. +#[async_trait] +pub trait RoutedDecisionClient: Send + Sync { + /// Evaluate the request using its selected model. + async fn call(&self, request: DecisionRequest) -> Result; +} diff --git a/crates/switchyard-nemo-relay-plugin/src/runtime.rs b/crates/switchyard-nemo-relay-plugin/src/runtime.rs index 95794f0d8..0a282c3e6 100644 --- a/crates/switchyard-nemo-relay-plugin/src/runtime.rs +++ b/crates/switchyard-nemo-relay-plugin/src/runtime.rs @@ -336,7 +336,23 @@ impl SwitchyardRuntime { } RunObservation::LlmCall(call) => { call_index += 1; - self.routing_call_events(events, call, call_index, metadata); + self.routing_call_events( + events, + call, + call_index, + metadata, + "switchyard.routing.llm_call", + ); + } + RunObservation::DecisionCall(call) => { + call_index += 1; + self.routing_call_events( + events, + call, + call_index, + metadata, + "switchyard.routing.decision_call", + ); } RunObservation::RoutingOverhead(duration) => { let latency_ms = duration.as_secs_f64() * 1_000.0; @@ -377,12 +393,13 @@ impl SwitchyardRuntime { call: LlmCallObservation, call_index: usize, metadata: &Json, + mark_name: &str, ) { let outcome = if call.is_success { "ok" } else { "error" }; let latency_ms = call.duration.as_secs_f64() * 1_000.0; let token_metrics = token_usage_metrics("routing", &call, metadata); events.push(RoutingEvent::Mark(RoutingMark { - name: "switchyard.routing.llm_call".into(), + name: mark_name.into(), data: json!({ "call_index": call_index, "selected_model": call.selected_model.as_str(), diff --git a/crates/switchyard-server/src/lib.rs b/crates/switchyard-server/src/lib.rs index 0f1738936..3381710f1 100644 --- a/crates/switchyard-server/src/lib.rs +++ b/crates/switchyard-server/src/lib.rs @@ -445,7 +445,7 @@ fn stats_observer( stats.record_error(&call.selected_model); } } - RunObservation::LlmCall(call) => { + RunObservation::LlmCall(call) | RunObservation::DecisionCall(call) => { let latency_ms = call.duration.as_secs_f64() * 1_000.0; if call.is_success { if let (Some((log, context)), Some(usage)) = From 2a03667b67a1d04cb46b7b87789dcca63bce4aee Mon Sep 17 00:00:00 2001 From: nachiketb Date: Mon, 5 Oct 2026 12:01:40 -0700 Subject: [PATCH 2/8] refactor(client): simplify the decision client public API Signed-off-by: nachiketb --- crates/libsy-llm-client/src/lib.rs | 2 +- crates/libsy-llm-client/src/observation.rs | 7 ++----- crates/libsy-llm-client/src/run.rs | 6 +++--- 3 files changed, 6 insertions(+), 9 deletions(-) diff --git a/crates/libsy-llm-client/src/lib.rs b/crates/libsy-llm-client/src/lib.rs index db0cf9c0c..2fc77a2bd 100644 --- a/crates/libsy-llm-client/src/lib.rs +++ b/crates/libsy-llm-client/src/lib.rs @@ -32,7 +32,7 @@ mod system_one; pub use backend::{Backend, DEFAULT_MAX_RETRIES, HttpBackendConfig}; pub use client::{AuxiliaryOperation, ModelConfig, TranslatingLlmClient}; pub use error::{LlmClientError, Result}; -pub use observation::{LlmCallObservation, ModelCallObservation, RunObservation, RunObserver}; +pub use observation::{LlmCallObservation, RunObservation, RunObserver}; pub use raw::RawResponse; pub use run::{ClientRouter, decide, run}; pub use switchyard_translation::RawEventStream; diff --git a/crates/libsy-llm-client/src/observation.rs b/crates/libsy-llm-client/src/observation.rs index 0fc97a5e5..2eb94debf 100644 --- a/crates/libsy-llm-client/src/observation.rs +++ b/crates/libsy-llm-client/src/observation.rs @@ -11,7 +11,7 @@ use switchyard_protocol::{ModelId, Usage}; /// One completed model call observed while serving an algorithm run. #[derive(Clone, Debug)] -pub struct ModelCallObservation { +pub struct LlmCallObservation { /// Model selected for the completed call. pub selected_model: ModelId, /// Whether the call completed successfully. @@ -22,9 +22,6 @@ pub struct ModelCallObservation { pub usage: Option, } -/// An LLM call's observation, sharing the fields used by decision calls. -pub type LlmCallObservation = ModelCallObservation; - /// Events emitted inline while [`crate::run()`] serves a routing request. #[derive(Clone, Debug)] pub enum RunObservation { @@ -33,7 +30,7 @@ pub enum RunObservation { /// A completed model call requested by the algorithm for routing work. LlmCall(LlmCallObservation), /// A completed decision call requested by the algorithm for routing work. - DecisionCall(ModelCallObservation), + DecisionCall(LlmCallObservation), /// A completed terminal model call made from the routing outcome. AnswerCall(LlmCallObservation), /// Routing time recorded by the `switchyard.routing_overhead_ms` metric. diff --git a/crates/libsy-llm-client/src/run.rs b/crates/libsy-llm-client/src/run.rs index f694fa05c..3d87f0906 100644 --- a/crates/libsy-llm-client/src/run.rs +++ b/crates/libsy-llm-client/src/run.rs @@ -37,7 +37,7 @@ use switchyard_protocol::{ }; use switchyard_translation::prepare_request_for_target; -use crate::observation::{LlmCallObservation, ModelCallObservation, RunObservation, RunObserver}; +use crate::observation::{LlmCallObservation, RunObservation, RunObserver}; use crate::{metrics, observability}; /// Run one request to completion, serving every offloaded model call with `client`. @@ -169,7 +169,7 @@ async fn serve_decision( if let Some(observations) = observations { observations .lock() - .push(RunObservation::DecisionCall(ModelCallObservation { + .push(RunObservation::DecisionCall(LlmCallObservation { selected_model: call.model.clone(), is_success: result.is_ok(), duration: started.elapsed(), @@ -715,7 +715,7 @@ impl ClientRouter { } /// Resolve a decision target without falling back to an LLM client. - pub fn route_decision( + fn route_decision( &self, model: &ModelId, ) -> std::result::Result<&Arc, LlmClientError> { From b0c0c236aa725b40a3b671d4b458ab96253e7722 Mon Sep 17 00:00:00 2001 From: nachiketb Date: Mon, 5 Oct 2026 12:04:19 -0700 Subject: [PATCH 3/8] refactor(client): use ModelCallObservation throughout Signed-off-by: nachiketb --- crates/libsy-llm-client/src/lib.rs | 2 +- crates/libsy-llm-client/src/observation.rs | 8 ++++---- crates/libsy-llm-client/src/run.rs | 12 ++++++------ crates/switchyard-nemo-relay-plugin/src/runtime.rs | 14 +++++++------- crates/switchyard-server/src/lib.rs | 4 ++-- 5 files changed, 20 insertions(+), 20 deletions(-) diff --git a/crates/libsy-llm-client/src/lib.rs b/crates/libsy-llm-client/src/lib.rs index 2fc77a2bd..69a57880a 100644 --- a/crates/libsy-llm-client/src/lib.rs +++ b/crates/libsy-llm-client/src/lib.rs @@ -32,7 +32,7 @@ mod system_one; pub use backend::{Backend, DEFAULT_MAX_RETRIES, HttpBackendConfig}; pub use client::{AuxiliaryOperation, ModelConfig, TranslatingLlmClient}; pub use error::{LlmClientError, Result}; -pub use observation::{LlmCallObservation, RunObservation, RunObserver}; +pub use observation::{ModelCallObservation, RunObservation, RunObserver}; pub use raw::RawResponse; pub use run::{ClientRouter, decide, run}; pub use switchyard_translation::RawEventStream; diff --git a/crates/libsy-llm-client/src/observation.rs b/crates/libsy-llm-client/src/observation.rs index 2eb94debf..53e660f14 100644 --- a/crates/libsy-llm-client/src/observation.rs +++ b/crates/libsy-llm-client/src/observation.rs @@ -11,7 +11,7 @@ use switchyard_protocol::{ModelId, Usage}; /// One completed model call observed while serving an algorithm run. #[derive(Clone, Debug)] -pub struct LlmCallObservation { +pub struct ModelCallObservation { /// Model selected for the completed call. pub selected_model: ModelId, /// Whether the call completed successfully. @@ -28,11 +28,11 @@ pub enum RunObservation { /// Metadata attached to the completed routing outcome. Outcome(OutcomeMetadata), /// A completed model call requested by the algorithm for routing work. - LlmCall(LlmCallObservation), + LlmCall(ModelCallObservation), /// A completed decision call requested by the algorithm for routing work. - DecisionCall(LlmCallObservation), + DecisionCall(ModelCallObservation), /// A completed terminal model call made from the routing outcome. - AnswerCall(LlmCallObservation), + AnswerCall(ModelCallObservation), /// Routing time recorded by the `switchyard.routing_overhead_ms` metric. RoutingOverhead(Duration), } diff --git a/crates/libsy-llm-client/src/run.rs b/crates/libsy-llm-client/src/run.rs index 3d87f0906..ad312277b 100644 --- a/crates/libsy-llm-client/src/run.rs +++ b/crates/libsy-llm-client/src/run.rs @@ -37,7 +37,7 @@ use switchyard_protocol::{ }; use switchyard_translation::prepare_request_for_target; -use crate::observation::{LlmCallObservation, RunObservation, RunObserver}; +use crate::observation::{ModelCallObservation, RunObservation, RunObserver}; use crate::{metrics, observability}; /// Run one request to completion, serving every offloaded model call with `client`. @@ -169,7 +169,7 @@ async fn serve_decision( if let Some(observations) = observations { observations .lock() - .push(RunObservation::DecisionCall(LlmCallObservation { + .push(RunObservation::DecisionCall(ModelCallObservation { selected_model: call.model.clone(), is_success: result.is_ok(), duration: started.elapsed(), @@ -249,7 +249,7 @@ async fn call_first_available( algorithm: &str, request: &Request, models: &[ModelId], - observe: &(dyn Fn(LlmCallObservation) + Send + Sync), + observe: &(dyn Fn(ModelCallObservation) + Send + Sync), ) -> Result { for (index, target) in models.iter().enumerate() { let request = clients.prepare_completion_request(request.clone(), target); @@ -326,7 +326,7 @@ async fn call_one( model_id: &ModelId, request: Request, algorithm: &str, - observe: &(dyn Fn(LlmCallObservation) + Send + Sync), + observe: &(dyn Fn(ModelCallObservation) + Send + Sync), // index is for span log index: usize, // count is for span log @@ -367,7 +367,7 @@ async fn call_one( } else { result }; - observe(LlmCallObservation { + observe(ModelCallObservation { selected_model: model_id.clone(), is_success: result.is_ok(), duration, @@ -1386,7 +1386,7 @@ mod tests { fn answer_observation_keeps_call_order() { let pending = Some(Arc::new(Mutex::new( ["answer", "judge"] - .map(|model| LlmCallObservation { + .map(|model| ModelCallObservation { selected_model: model.into(), is_success: true, duration: std::time::Duration::ZERO, diff --git a/crates/switchyard-nemo-relay-plugin/src/runtime.rs b/crates/switchyard-nemo-relay-plugin/src/runtime.rs index 0a282c3e6..4c1b7ab05 100644 --- a/crates/switchyard-nemo-relay-plugin/src/runtime.rs +++ b/crates/switchyard-nemo-relay-plugin/src/runtime.rs @@ -11,7 +11,7 @@ use nemo_relay_plugin::{ MetricValueType, PluginRuntime, }; use serde_json::{Map, json}; -use switchyard_llm_client::{LlmCallObservation, RunObservation, RunObserver}; +use switchyard_llm_client::{ModelCallObservation, RunObservation, RunObserver}; use switchyard_protocol::{ LlmClientError, LlmResponse, LlmResponseChunk, LlmStreamError, Metadata, ProviderExtensions, Request, Response, Usage, WireFormat, @@ -390,7 +390,7 @@ impl SwitchyardRuntime { fn routing_call_events( &self, events: &mut Vec, - call: LlmCallObservation, + call: ModelCallObservation, call_index: usize, metadata: &Json, mark_name: &str, @@ -734,7 +734,7 @@ fn routing_overhead_metric(latency_ms: f64, metadata: Json) -> RoutingEvent { fn token_usage_metrics( call_role: &str, - call: &LlmCallObservation, + call: &ModelCallObservation, metadata: &Json, ) -> Vec { let Some(usage) = call.usage.as_ref() else { @@ -1316,7 +1316,7 @@ mod tests { runtime.emit_observations( &mut events, vec![ - RunObservation::LlmCall(LlmCallObservation { + RunObservation::LlmCall(ModelCallObservation { selected_model: ModelId::from("routing-model"), is_success: false, duration: std::time::Duration::from_millis(12), @@ -1407,7 +1407,7 @@ mod tests { #[test] fn token_usage_metrics_distinguish_routing_and_answer_targets() { - let call = LlmCallObservation { + let call = ModelCallObservation { selected_model: ModelId::from("judge-model"), is_success: true, duration: std::time::Duration::from_millis(1), @@ -1469,13 +1469,13 @@ mod tests { runtime.emit_observations( &mut events, vec![ - RunObservation::AnswerCall(LlmCallObservation { + RunObservation::AnswerCall(ModelCallObservation { selected_model: ModelId::from("weak-target"), is_success: false, duration: std::time::Duration::from_millis(2), usage: None, }), - RunObservation::AnswerCall(LlmCallObservation { + RunObservation::AnswerCall(ModelCallObservation { selected_model: ModelId::from("strong-target"), is_success: true, duration: std::time::Duration::from_millis(3), diff --git a/crates/switchyard-server/src/lib.rs b/crates/switchyard-server/src/lib.rs index 3381710f1..842f77d92 100644 --- a/crates/switchyard-server/src/lib.rs +++ b/crates/switchyard-server/src/lib.rs @@ -1685,7 +1685,7 @@ fn endpoint_listing(has_routing_log: bool) -> String { #[cfg(test)] mod tests { - use switchyard_llm_client::LlmCallObservation; + use switchyard_llm_client::ModelCallObservation; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::sync::{Notify, oneshot}; @@ -1706,7 +1706,7 @@ mod tests { let observer = stats_observer(StatsAccumulator::default(), Some((log.clone(), context))); let call = |model: &str, answer: bool| { - let observation = LlmCallObservation { + let observation = ModelCallObservation { selected_model: ModelId::from(model), is_success: true, duration: Duration::from_millis(3), From 6f2f84c7a735577900589976ef7110234f4a5614 Mon Sep 17 00:00:00 2001 From: nachiketb Date: Mon, 5 Oct 2026 12:17:56 -0700 Subject: [PATCH 4/8] fix(client): reject mismatched System One answers Signed-off-by: nachiketb --- crates/libsy-llm-client/src/run.rs | 34 +++++++++++++++- crates/libsy-llm-client/src/system_one.rs | 48 +++++++++++++---------- 2 files changed, 61 insertions(+), 21 deletions(-) diff --git a/crates/libsy-llm-client/src/run.rs b/crates/libsy-llm-client/src/run.rs index ad312277b..654567369 100644 --- a/crates/libsy-llm-client/src/run.rs +++ b/crates/libsy-llm-client/src/run.rs @@ -1180,7 +1180,7 @@ mod tests { .expect(1) .mount(&server) .await; - let response = client.call(mixed).await?; + let response = client.call(mixed.clone()).await?; assert_eq!(response.model.as_deref(), Some("jev-1.13.0")); assert_eq!(response.id, None); assert_eq!(response.usage.input_tokens, Some(42)); @@ -1202,6 +1202,31 @@ mod tests { ); server.verify().await; + for (id, answer) in [ + ("unknown-question", json!({"type": "noul", "noul": 0.9})), + ("boolean", json!({"type": "score", "score": 0.0})), + ("score", json!({"type": "score", "score": -0.1})), + ( + "score", + json!({"type": "score", "score": 1.1, + "probabilities": {"0": 0.4, "1": 0.6}}), + ), + ] { + server.reset().await; + let mut invalid = body.clone(); + invalid["answers"][id] = answer; + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200).set_body_json(invalid)) + .expect(1) + .mount(&server) + .await; + assert!(matches!( + client.call(mixed.clone()).await, + Err(LlmClientError::ResponseTranslation(_)) + )); + server.verify().await; + } + body["answers"].as_object_mut().unwrap().remove("boolean"); body["answers"].as_object_mut().unwrap().remove("score"); let models = Arc::new(RuntimeModels::new(HashMap::from([ @@ -1228,12 +1253,19 @@ mod tests { }, )?)) }; + let mut undeclared_choice = body.clone(); + undeclared_choice["answers"]["route"]["choice"] = json!("unknown-option"); for (template, expected, is_success) in [ ( ResponseTemplate::new(200).set_body_json(&body), "efficient", true, ), + ( + ResponseTemplate::new(200).set_body_json(&undeclared_choice), + "capable", + false, + ), ( ResponseTemplate::new(200).set_body_string("invalid JSON"), "capable", diff --git a/crates/libsy-llm-client/src/system_one.rs b/crates/libsy-llm-client/src/system_one.rs index caf4881e5..649f2c23b 100644 --- a/crates/libsy-llm-client/src/system_one.rs +++ b/crates/libsy-llm-client/src/system_one.rs @@ -69,30 +69,33 @@ impl RoutedDecisionClient for SystemOneClient { .answers .into_iter() .map(|(id, answer)| { - let value = match answer.value { - WireValue::Noul { noul } => { + let value = match (request.questions.get(&id).map(|q| &q.kind), answer.value) { + (Some(DecisionKind::Boolean { .. }), WireValue::Noul { noul }) => { DecisionValue::Boolean(BooleanEstimate::ProbabilityTrue(noul)) } - WireValue::Choice { - choice, - probabilities, - } => DecisionValue::Choice { - selected: choice, - probabilities, - }, - WireValue::Score { - score, - probabilities, - } => { + ( + Some(DecisionKind::Choice { options }), + WireValue::Choice { + choice, + probabilities, + }, + ) if options.iter().any(|option| option.id == choice) => { + DecisionValue::Choice { + selected: choice, + probabilities, + } + } + ( + Some(DecisionKind::Score { levels }), + WireValue::Score { + score, + probabilities, + }, + ) if !levels.is_empty() + && (0.0..=(levels.len() - 1) as f64).contains(&score.0) => + { let probabilities = probabilities .map(|mut probabilities| { - let Some(DecisionKind::Score { levels }) = - request.questions.get(&id).map(|q| &q.kind) - else { - return Err(LlmClientError::ResponseTranslation(format!( - "score answer {id:?} has no matching score question" - ))); - }; let invalid_rubric = || { LlmClientError::ResponseTranslation(format!( "score probabilities for {id:?} do not match its rubric" @@ -116,6 +119,11 @@ impl RoutedDecisionClient for SystemOneClient { probabilities, } } + _ => { + return Err(LlmClientError::ResponseTranslation(format!( + "answer {id:?} has an unexpected question ID, kind, or value" + ))); + } }; Ok(( id, From 26077b7aae4b9f71035ebb87478ff98ffa712434 Mon Sep 17 00:00:00 2001 From: nachiketb Date: Mon, 5 Oct 2026 12:50:56 -0700 Subject: [PATCH 5/8] fix(client): tighten System One setup and response handling Signed-off-by: nachiketb --- crates/libsy-llm-client/src/run.rs | 32 +++++++++++++++++++++-- crates/libsy-llm-client/src/system_one.rs | 29 +++++++++++++++----- 2 files changed, 53 insertions(+), 8 deletions(-) diff --git a/crates/libsy-llm-client/src/run.rs b/crates/libsy-llm-client/src/run.rs index 654567369..f4bc3fe20 100644 --- a/crates/libsy-llm-client/src/run.rs +++ b/crates/libsy-llm-client/src/run.rs @@ -1135,10 +1135,18 @@ mod tests { let server = MockServer::start().await; let client = Arc::new(crate::SystemOneClient::new( - format!("{}/v1/systemone", server.uri()), + format!("{}/v1/systemone", server.uri()).parse()?, "test-key".into(), Duration::from_secs(2), )?); + assert!(matches!( + crate::SystemOneClient::new( + format!("{}/v1/systemone", server.uri()).parse()?, + "invalid\nkey".into(), + Duration::from_secs(2), + ), + Err(LlmClientError::Configuration { .. }) + )); let mixed: DecisionRequest = serde_json::from_value(json!({ "model": "judge", "context": {"task": "A simple task"}, "questions": { @@ -1202,6 +1210,12 @@ mod tests { ); server.verify().await; + let mut partial = body.clone(); + partial["answers"] + .as_object_mut() + .unwrap() + .remove("boolean"); + let mut invalid_responses = vec![json!({"answers": {}}), partial]; for (id, answer) in [ ("unknown-question", json!({"type": "noul", "noul": 0.9})), ("boolean", json!({"type": "score", "score": 0.0})), @@ -1212,9 +1226,18 @@ mod tests { "probabilities": {"0": 0.4, "1": 0.6}}), ), ] { - server.reset().await; let mut invalid = body.clone(); invalid["answers"][id] = answer; + if id == "unknown-question" { + invalid["answers"] + .as_object_mut() + .unwrap() + .remove("boolean"); + } + invalid_responses.push(invalid); + } + for invalid in invalid_responses { + server.reset().await; Mock::given(method("POST")) .respond_with(ResponseTemplate::new(200).set_body_json(invalid)) .expect(1) @@ -1266,6 +1289,11 @@ mod tests { "capable", false, ), + ( + ResponseTemplate::new(200).set_body_json(json!({"answers": {}})), + "capable", + false, + ), ( ResponseTemplate::new(200).set_body_string("invalid JSON"), "capable", diff --git a/crates/libsy-llm-client/src/system_one.rs b/crates/libsy-llm-client/src/system_one.rs index 649f2c23b..03a4827ac 100644 --- a/crates/libsy-llm-client/src/system_one.rs +++ b/crates/libsy-llm-client/src/system_one.rs @@ -7,6 +7,7 @@ use std::collections::BTreeMap; use std::time::Duration; use async_trait::async_trait; +use reqwest::{Url, header::HeaderValue}; use serde::Deserialize; use serde_json::{Value, json}; use switchyard_protocol::{ @@ -16,19 +17,24 @@ use switchyard_protocol::{ }; use crate::client::convert_reqwest_error; -use crate::{LlmClientError, Result}; +use crate::{LlmClientError, Result, metrics}; /// Serves decision requests through a System One endpoint, such as TypeSafe's Jev API. pub struct SystemOneClient { client: reqwest::Client, - endpoint: String, + endpoint: Url, api_key: String, } impl SystemOneClient { /// `endpoint` is the full URL, including `/v1/systemone`. Each call makes one /// attempt; `timeout` covers sending the request and reading the response body. - pub fn new(endpoint: impl Into, api_key: String, timeout: Duration) -> Result { + pub fn new(endpoint: Url, api_key: String, timeout: Duration) -> Result { + if HeaderValue::try_from(format!("Bearer {api_key}")).is_err() { + return Err(LlmClientError::Configuration { + message: "System One API key cannot be encoded as an HTTP header".into(), + }); + } let client = reqwest::Client::builder() .timeout(timeout) .redirect(reqwest::redirect::Policy::none()) @@ -36,7 +42,7 @@ impl SystemOneClient { .map_err(convert_reqwest_error)?; Ok(Self { client, - endpoint: endpoint.into(), + endpoint, api_key, }) } @@ -47,14 +53,20 @@ impl RoutedDecisionClient for SystemOneClient { async fn call(&self, request: DecisionRequest) -> Result { let response = self .client - .post(&self.endpoint) + .post(self.endpoint.clone()) .bearer_auth(&self.api_key) .json(&encode(&request)?) .send() .await + .inspect_err(|_| metrics::record_upstream_attempt(None)) .map_err(convert_reqwest_error)?; let status = response.status(); - let body = response.bytes().await.map_err(convert_reqwest_error)?; + let body = response + .bytes() + .await + .inspect_err(|_| metrics::record_upstream_attempt(None)) + .map_err(convert_reqwest_error)?; + metrics::record_upstream_attempt(Some(status.as_u16())); if !status.is_success() { return Err(LlmClientError::UpstreamHttp { status, @@ -65,6 +77,11 @@ impl RoutedDecisionClient for SystemOneClient { serde_json::from_slice(&body).map_err(|source| LlmClientError::InvalidResponse { source: Box::new(source), })?; + if !response.answers.keys().eq(request.questions.keys()) { + return Err(LlmClientError::ResponseTranslation( + "System One answer keys do not match the requested question keys".into(), + )); + } let answers = response .answers .into_iter() From a36603f434aff38225bb3cacdababe5c6d850c82 Mon Sep 17 00:00:00 2001 From: nachiketb Date: Mon, 5 Oct 2026 14:14:45 -0700 Subject: [PATCH 6/8] refactor(client): separate client registration from LLM policy Signed-off-by: nachiketb --- crates/libsy-llm-client/src/lib.rs | 6 +- crates/libsy-llm-client/src/run.rs | 319 +++++++++++++++++++++-------- 2 files changed, 231 insertions(+), 94 deletions(-) diff --git a/crates/libsy-llm-client/src/lib.rs b/crates/libsy-llm-client/src/lib.rs index 69a57880a..1236a49ee 100644 --- a/crates/libsy-llm-client/src/lib.rs +++ b/crates/libsy-llm-client/src/lib.rs @@ -1,7 +1,7 @@ // SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 -//! HTTP LLM client that speaks Switchyard's neutral IR directly. +//! HTTP clients for Switchyard's neutral LLM and decision requests. //! //! [`TranslatingLlmClient`] maps a model name (and the wire format resolved from //! the request) to a [`Backend`], @@ -12,7 +12,7 @@ //! streamed responses. //! //! [`SystemOneClient`] serves typed decision requests. Register decision targets -//! with [`ClientRouter::with_decision_clients`] to serve them alongside LLM calls. +//! with [`ClientRouterBuilder::decision_clients`] to serve them alongside LLM calls. //! //! [`run()`] pairs the client with a libsy algorithm: it drives //! [`switchyard_libsy::Algorithm::run_stream`], serves routing-time calls, and makes the terminal @@ -34,7 +34,7 @@ pub use client::{AuxiliaryOperation, ModelConfig, TranslatingLlmClient}; pub use error::{LlmClientError, Result}; pub use observation::{ModelCallObservation, RunObservation, RunObserver}; pub use raw::RawResponse; -pub use run::{ClientRouter, decide, run}; +pub use run::{ClientRouter, ClientRouterBuilder, decide, run}; pub use switchyard_translation::RawEventStream; pub use system_one::SystemOneClient; diff --git a/crates/libsy-llm-client/src/run.rs b/crates/libsy-llm-client/src/run.rs index f4bc3fe20..ac1da56d3 100644 --- a/crates/libsy-llm-client/src/run.rs +++ b/crates/libsy-llm-client/src/run.rs @@ -5,8 +5,8 @@ //! //! [`switchyard_libsy::Algorithm::run_stream`] is the whole libsy API: it yields a stream of //! steps and expects its consumer to serve every offloaded model call. [`run()`] is that -//! consumer — it drives the stream with [`switchyard_libsy::drive`], hands routing-time calls to -//! a [`RoutedLlmClient`], and serves the terminal routing outcome. +//! consumer — it drives the stream with [`switchyard_libsy::drive`], resolves a typed client +//! for each call, and serves the terminal routing outcome. //! //! libsy owns the stream mechanics; what this module adds is per-target request preparation, //! ordered candidate fallback, and the `libsy.client_call` span around each candidate. Each @@ -49,7 +49,7 @@ use crate::{metrics, observability}; /// `clients` resolves each offloaded call to the client for the target the algorithm /// selected — an algorithm may route among targets served by different providers, so this is /// a per-call lookup, not one client for the whole run. Use -/// [`ClientRouter::single`](ClientRouter::single) when one client serves every target. +/// [`ClientRouter::single`](ClientRouter::single) when one client serves every LLM target. /// /// Routing calls are buffered; client failures stop the request unless the call enables recovery. /// Once routing completes, non-timeout failures may try the outcome's ordered fallback candidates. @@ -66,7 +66,7 @@ pub async fn run( // This says if we have an observer, put Some(..) in routing_observations. // No observer means we don't want any routing_observations. let routing_observations = observer.as_ref().map(|_| Arc::new(Mutex::new(Vec::new()))); - let outcome = match clients.stored_state_owner(&request) { + let outcome = match clients.inner.response_state.stored_state_owner(&request) { Some(owner) => Ok(continue_on(owner, &algorithm_name, request)), None => { drive(algorithm, request, models, { @@ -134,7 +134,7 @@ pub async fn decide( models: Arc, ) -> Result { let routing_clients = clients.clone(); - let mut outcome = match clients.stored_state_owner(&request) { + let mut outcome = match clients.inner.response_state.stored_state_owner(&request) { Some(owner) => continue_on(owner, algorithm.name(), request), None => { drive(algorithm, request, models, move |call| { @@ -144,7 +144,10 @@ pub async fn decide( } }; let selected_model_id = outcome.selected_model_id()?.clone(); - outcome.request = clients.prepare_completion_request(outcome.request, &selected_model_id); + outcome.request = clients + .inner + .llm_policy + .prepare_completion_request(outcome.request, &selected_model_id); outcome.response = outcome .response .map(|response| clients.remember_state_owner(&outcome.request, response)) @@ -162,7 +165,7 @@ async fn serve_decision( call: CallDecision, observations: &Option>>>, ) -> Result<()> { - let client = clients.route_decision(&call.model); + let client = clients.inner.clients.route_decision(&call.model); let started = Instant::now(); let result = async { client?.call(call.request.clone()).await }.await; tracing::Span::current().record("outcome", if result.is_ok() { "ok" } else { "error" }); @@ -224,7 +227,10 @@ async fn serve( } }; let target = call.models.first().ok_or(LibsyError::NoTargets)?; - let request = clients.prepare_routing_request(call.request.clone(), target); + let request = clients + .inner + .llm_policy + .prepare_routing_request(call.request.clone(), target); match call_one( &clients, target, @@ -252,7 +258,10 @@ async fn call_first_available( observe: &(dyn Fn(ModelCallObservation) + Send + Sync), ) -> Result { for (index, target) in models.iter().enumerate() { - let request = clients.prepare_completion_request(request.clone(), target); + let request = clients + .inner + .llm_policy + .prepare_completion_request(request.clone(), target); match call_one( clients, target, @@ -584,34 +593,136 @@ fn conversation_id(fields: &serde_json::Map) -> Option<&str> { /// host's concern, and two targets in one run may sit on different providers. A router owns /// that mapping and resolves the client for each kind of call. /// -/// Cloning is cheap — the mapping is shared, so one router can serve every request. +/// Use [`Self::builder`] to register multiple call kinds. Cloning shares client mappings, +/// LLM request policy, and conversation state across runs. #[derive(Clone)] pub struct ClientRouter { inner: Arc, - decision_clients: Arc>>, } struct ClientRouting { - routing: Routing, + clients: ClientRegistry, + llm_policy: LlmRequestPolicy, + response_state: ResponseState, +} + +struct ClientRegistry { + llm: LlmClients, + decision: HashMap>, +} + +#[derive(Default)] +struct LlmRequestPolicy { target_prompts: HashMap, routing_answer_target: Option, +} + +struct ResponseState { /// Provider-owned routing pins and materialized cross-format history. - state_owners: Mutex, + owners: Mutex, /// Native Responses state only needs a routing pin when answer targets span clients. track_provider_state: bool, } -enum Routing { - /// One client serves every model. +enum LlmClients { + /// One client serves every LLM target. Single(Arc), /// Each model is served by the client configured for it. ByModel(HashMap>), } +/// Configure clients and LLM request policy before sharing a router across runs. +/// Registration ends at [`Self::build`]; cloned routers share the same clients and state. +pub struct ClientRouterBuilder { + clients: ClientRegistry, + llm_policy: LlmRequestPolicy, + completion_targets: Option>, +} + +impl ClientRouterBuilder { + /// Replace the LLM routing table with explicit model-to-client mappings. + pub fn llm_clients(mut self, clients: HashMap>) -> Self { + self.clients.llm = LlmClients::ByModel(clients); + self + } + + /// Use one LLM client for every model, leaving model lookup to that client. + pub fn single_llm_client(mut self, client: Arc) -> Self { + self.clients.llm = LlmClients::Single(client); + self + } + + /// Replace the decision routing table. Decision calls require an explicit mapping. + pub fn decision_clients( + mut self, + clients: HashMap>, + ) -> Self { + self.clients.decision = clients; + self + } + + /// Set LLM answer prompts keyed by resolved model ID, not deployment target name. + pub fn target_prompts(mut self, prompts: HashMap) -> Self { + self.llm_policy.target_prompts = prompts; + self + } + + /// Apply its target prompt to a routing-time LLM call whose response can be the answer. + pub fn routing_answer_target(mut self, target: Option) -> Self { + self.llm_policy.routing_answer_target = target; + self + } + + /// Use these completion targets to decide whether native Responses need a provider pin. + /// If omitted, every mapped LLM is considered a possible completion target. + pub fn completion_targets(mut self, targets: Vec) -> Self { + self.completion_targets = Some(targets); + self + } + + /// Finish registration and share the immutable configuration and conversation state. + pub fn build(self) -> ClientRouter { + let track_provider_state = match &self.clients.llm { + LlmClients::Single(_) => false, + LlmClients::ByModel(by_model) => { + let targets = self + .completion_targets + .unwrap_or_else(|| by_model.keys().cloned().collect()); + let mut clients = targets.iter().filter_map(|model| by_model.get(model)); + clients + .next() + .is_some_and(|first| clients.any(|client| !Arc::ptr_eq(first, client))) + } + }; + ClientRouter { + inner: Arc::new(ClientRouting { + clients: self.clients, + llm_policy: self.llm_policy, + response_state: ResponseState { + owners: Mutex::default(), + track_provider_state, + }, + }), + } + } +} + impl ClientRouter { + /// Start with no clients. Register each call kind before building the shared router. + pub fn builder() -> ClientRouterBuilder { + ClientRouterBuilder { + clients: ClientRegistry { + llm: LlmClients::ByModel(HashMap::new()), + decision: HashMap::new(), + }, + llm_policy: LlmRequestPolicy::default(), + completion_targets: None, + } + } + /// Build a router over `model name -> client`, for targets spread across providers. pub fn new(by_model: HashMap>) -> Self { - Self::new_with_target_prompts(by_model, HashMap::new(), None) + Self::builder().llm_clients(by_model).build() } /// Build a router with target prompts used by [`run`] and [`decide`]. @@ -627,13 +738,11 @@ impl ClientRouter { target_prompts: HashMap, routing_answer_target: Option, ) -> Self { - let completion_targets = by_model.keys().cloned().collect::>(); - Self::new_with_completion_targets( - by_model, - target_prompts, - routing_answer_target, - &completion_targets, - ) + Self::builder() + .llm_clients(by_model) + .target_prompts(target_prompts) + .routing_answer_target(routing_answer_target) + .build() } /// Build a router with an explicit list of models that can answer requests. @@ -649,40 +758,21 @@ impl ClientRouter { routing_answer_target: Option, completion_targets: &[ModelId], ) -> Self { - let mut clients = completion_targets - .iter() - .filter_map(|model| by_model.get(model)); - let spans_providers = clients - .next() - .is_some_and(|first| clients.any(|client| !Arc::ptr_eq(first, client))); - Self { - decision_clients: Arc::default(), - inner: Arc::new(ClientRouting { - routing: Routing::ByModel(by_model), - target_prompts, - routing_answer_target, - state_owners: Mutex::default(), - track_provider_state: spans_providers, - }), - } + Self::builder() + .llm_clients(by_model) + .target_prompts(target_prompts) + .routing_answer_target(routing_answer_target) + .completion_targets(completion_targets.to_vec()) + .build() } - /// A router that serves every model with one client — the single-provider case. + /// A router that serves every LLM target with one client — the single-provider case. /// /// [`TranslatingLlmClient`](crate::TranslatingLlmClient) already maps model names to /// backends internally and rejects ones it does not know, so enumerating them here would /// only duplicate that. pub fn single(client: Arc) -> Self { - Self { - decision_clients: Arc::default(), - inner: Arc::new(ClientRouting { - routing: Routing::Single(client), - target_prompts: HashMap::new(), - routing_answer_target: None, - state_owners: Mutex::default(), - track_provider_state: false, - }), - } + Self::builder().single_llm_client(client).build() } /// The client that serves `model`. @@ -693,9 +783,18 @@ impl ClientRouter { &self, model: &ModelId, ) -> std::result::Result<&Arc, LlmClientError> { - match &self.inner.routing { - Routing::Single(client) => Ok(client), - Routing::ByModel(by_model) => { + self.inner.clients.route_llm(model) + } +} + +impl ClientRegistry { + fn route_llm( + &self, + model: &ModelId, + ) -> std::result::Result<&Arc, LlmClientError> { + match &self.llm { + LlmClients::Single(client) => Ok(client), + LlmClients::ByModel(by_model) => { by_model .get(model) .ok_or_else(|| LlmClientError::Configuration { @@ -705,27 +804,20 @@ impl ClientRouter { } } - /// Register clients for decision targets independently of completion targets. - pub fn with_decision_clients( - mut self, - clients: HashMap>, - ) -> Self { - self.decision_clients = Arc::new(clients); - self - } - /// Resolve a decision target without falling back to an LLM client. fn route_decision( &self, model: &ModelId, ) -> std::result::Result<&Arc, LlmClientError> { - self.decision_clients + self.decision .get(model) .ok_or_else(|| LlmClientError::Configuration { message: format!("no decision client is configured for model {model:?}"), }) } +} +impl ResponseState { /// Return the recorded model for the requested response or conversation ID. fn stored_state_owner(&self, request: &Request) -> Option { let fields = &request.llm_request.extensions.fields; @@ -733,7 +825,7 @@ impl ClientRouter { Some(id) => ("previous_response_id", id), None => ("conversation", conversation_id(fields)?), }; - let state = self.inner.state_owners.lock().owner(id)?.clone(); + let state = self.owners.lock().owner(id)?.clone(); Some(StateOwner { model: state.model, trigger, @@ -748,7 +840,7 @@ impl ClientRouter { .fields .get("previous_response_id") .and_then(Value::as_str) - .and_then(|id| self.inner.state_owners.lock().owner(id)?.history.clone()); + .and_then(|id| self.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..) { Some(messages) => (parent, messages.to_vec()), @@ -759,7 +851,9 @@ impl ClientRouter { messages: Arc::from(messages), } } +} +impl ClientRouter { /// Record provider-owned Responses state or canonical history for a cross-format response. fn remember_state_owner(&self, request: &Request, mut response: Response) -> Result { let Some(model) = response.served_model().cloned() else { @@ -776,19 +870,22 @@ impl ClientRouter { .contains_key(&responses_format) || fields.contains_key("previous_response_id") || fields.contains_key("conversation"); - if !responses_request && !self.inner.track_provider_state { + let state = &self.inner.response_state; + if !responses_request && !state.track_provider_state { return Ok(response); } - let canonical_input = responses_request.then(|| self.canonical_input(request)); + let canonical_input = responses_request.then(|| state.canonical_input(request)); response.llm_response = match response.llm_response { LlmResponse::Agg(agg) => { if let Some(body) = agg.preservation.responses.get(&responses_format) { - if self.inner.track_provider_state { - self.remember_response(body, &model, store, conversation.as_deref()) + if state.track_provider_state { + state + .remember_response(body, &model, store, conversation.as_deref()) .map_err(|error| LibsyError::client_call(model.clone(), error))?; } } else if let Some(input) = &canonical_input { - self.remember_canonical_response(&agg, &model, store, input) + state + .remember_canonical_response(&agg, &model, store, input) .map_err(|error| LibsyError::client_call(model.clone(), error))?; } LlmResponse::Agg(agg) @@ -816,12 +913,13 @@ impl ClientRouter { mut native_responses, mut valid, )| async move { + let state = &router.inner.response_state; let Some(event) = stream.next().await else { if let Some(input) = &canonical_input && !native_responses && valid { - router.remember_canonical_response( + state.remember_canonical_response( &accumulator.finish(), &model, store, @@ -835,10 +933,10 @@ impl ClientRouter { && preserved.source().as_str() == WireFormat::OpenAiResponses.as_str() { native_responses = true; - if router.inner.track_provider_state + if state.track_provider_state && let Some(body) = preserved.raw().get("response") { - router.remember_response( + state.remember_response( body, &model, store, @@ -878,7 +976,9 @@ impl ClientRouter { }; Ok(response) } +} +impl ResponseState { /// Check both IDs from a response or stream event before recording either. The response's /// `store` value overrides the request value; `false` skips only the response ID. fn remember_response( @@ -888,7 +988,7 @@ impl ClientRouter { store: bool, conversation: Option<&str>, ) -> std::result::Result<(), LlmClientError> { - let mut owners = self.inner.state_owners.lock(); + let mut owners = self.owners.lock(); let response_id = body .get("id") .and_then(Value::as_str) @@ -925,11 +1025,10 @@ 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 + .owners + .lock() + .remember(response_id, None, 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"); @@ -940,18 +1039,20 @@ impl ClientRouter { } Ok(()) } +} +impl LlmRequestPolicy { /// Prepare a completion candidate with its configured target prompt. fn prepare_completion_request(&self, mut request: Request, target: &ModelId) -> Request { - let prompt = self.inner.target_prompts.get(target).map(String::as_str); + let prompt = self.target_prompts.get(target).map(String::as_str); prepare_request_for_target(&mut request.llm_request, target, prompt); request } /// Prepare a routing call, adding a target prompt only when it generates a candidate answer. fn prepare_routing_request(&self, mut request: Request, target: &ModelId) -> Request { - let prompt = if self.inner.routing_answer_target.as_ref() == Some(target) { - self.inner.target_prompts.get(target).map(String::as_str) + let prompt = if self.routing_answer_target.as_ref() == Some(target) { + self.target_prompts.get(target).map(String::as_str) } else { None }; @@ -1264,12 +1365,12 @@ mod tests { config: TaskClassifierConfig { judge: CapabilityJudgeConfig::Decision(DecisionJudgeConfig { cutoff: 0.4, - instructions: None, + instructions: Some(json!({"task": "Compare candidate outcomes"})), candidates: BTreeMap::from([ ("a".into(), "capable".into()), ("b".into(), "efficient".into()), ]), - evidence: json!({}), + evidence: json!({"cases": ["reference task"]}), }), ..TaskClassifierConfig::default() }, @@ -1308,6 +1409,12 @@ mod tests { server.reset().await; Mock::given(method("POST")) .and(path("/v1/systemone")) + .and(body_partial_json(json!({ + "state": {"evidence": {"cases": ["reference task"]}}, + "questions": {"route": { + "instructions": {"task": "Compare candidate outcomes"} + }} + }))) .respond_with(template) .expect(2) .mount(&server) @@ -1317,13 +1424,26 @@ mod tests { requests: Mutex::default(), first: FirstOutcome::Unauthorized, }); - let clients = - ClientRouter::single(llm.clone()).with_decision_clients(HashMap::from([( + let clients = ClientRouter::builder() + .single_llm_client(llm.clone()) + .decision_clients(HashMap::from([( ModelId::from("judge"), client.clone() as Arc, - )])); + )])) + .target_prompts(HashMap::from([ + ("judge".into(), "LLM judge prompt".into()), + ("efficient".into(), "Answer prompt".into()), + ("capable".into(), "Answer prompt".into()), + ])) + .routing_answer_target(Some("judge".into())) + .build(); + assert!(matches!( + clients.inner.clients.route_decision(&"efficient".into()), + Err(LlmClientError::Configuration { .. }) + )); let outcome = decide(algorithm()?, clients.clone(), request(), models.clone()).await?; assert_eq!(outcome.selected_model_id()?, expected); + assert_eq!(instruction_text(&outcome.request), ["Answer prompt"]); assert!(llm.calls.lock().is_empty()); let events = Arc::new(Mutex::new(Vec::new())); let captured = events.clone(); @@ -1337,6 +1457,7 @@ mod tests { .await?; assert_eq!(selected, expected); assert_eq!(&*llm.calls.lock(), &[ModelId::from(expected)]); + assert_eq!(instruction_text(&llm.requests.lock()[0]), ["Answer prompt"]); { let events = events.lock(); let RunObservation::DecisionCall(call) = &events[0] else { @@ -1764,6 +1885,8 @@ mod tests { metadata: None, }; let history = clients + .inner + .response_state .stored_state_owner(&third) .and_then(|owner| owner.history) .expect("chained canonical history"); @@ -1854,7 +1977,13 @@ mod tests { is_error: None, })], }]; - assert!(clients.stored_state_owner(&follow).is_none()); + assert!( + clients + .inner + .response_state + .stored_state_owner(&follow) + .is_none() + ); response .llm_response .into_agg() @@ -1863,6 +1992,8 @@ mod tests { let outcome = continue_on( clients + .inner + .response_state .stored_state_owner(&follow) .expect("completed stream state"), "passthrough", @@ -1916,7 +2047,13 @@ mod tests { .extensions .fields .insert("previous_response_id".to_string(), json!("msg_not_stored")); - assert!(clients.stored_state_owner(&follow).is_none()); + assert!( + clients + .inner + .response_state + .stored_state_owner(&follow) + .is_none() + ); Ok(()) } @@ -1932,7 +2069,7 @@ mod tests { client as Arc, )])); let model = ModelId::from("weak"); - clients.inner.state_owners.lock().by_id = (0..MAX_STATE_OWNERS) + clients.inner.response_state.owners.lock().by_id = (0..MAX_STATE_OWNERS) .map(|id| { ( format!("existing_{id}"), @@ -1982,7 +2119,7 @@ mod tests { .map_err(|error| LibsyError::client_call("weak", error))?; assert_eq!( - clients.inner.state_owners.lock().by_id.len(), + clients.inner.response_state.owners.lock().by_id.len(), MAX_STATE_OWNERS ); Ok(()) From c122a49df7f09494cfbdce700123b8d2d335cc65 Mon Sep 17 00:00:00 2001 From: nachiketb Date: Mon, 5 Oct 2026 14:20:53 -0700 Subject: [PATCH 7/8] refactor(client): simplify decision client registration Signed-off-by: nachiketb --- crates/libsy-llm-client/src/lib.rs | 7 +- crates/libsy-llm-client/src/run.rs | 339 ++++++++++------------------- 2 files changed, 115 insertions(+), 231 deletions(-) diff --git a/crates/libsy-llm-client/src/lib.rs b/crates/libsy-llm-client/src/lib.rs index 1236a49ee..ee1d9d616 100644 --- a/crates/libsy-llm-client/src/lib.rs +++ b/crates/libsy-llm-client/src/lib.rs @@ -1,7 +1,7 @@ // SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. // SPDX-License-Identifier: Apache-2.0 -//! HTTP clients for Switchyard's neutral LLM and decision requests. +//! HTTP LLM client that speaks Switchyard's neutral IR directly. //! //! [`TranslatingLlmClient`] maps a model name (and the wire format resolved from //! the request) to a [`Backend`], @@ -12,7 +12,8 @@ //! streamed responses. //! //! [`SystemOneClient`] serves typed decision requests. Register decision targets -//! with [`ClientRouterBuilder::decision_clients`] to serve them alongside LLM calls. +//! with [`ClientRouter::new_with_decision_clients`] or +//! [`ClientRouter::single_with_decision_clients`] to serve them alongside LLM calls. //! //! [`run()`] pairs the client with a libsy algorithm: it drives //! [`switchyard_libsy::Algorithm::run_stream`], serves routing-time calls, and makes the terminal @@ -34,7 +35,7 @@ pub use client::{AuxiliaryOperation, ModelConfig, TranslatingLlmClient}; pub use error::{LlmClientError, Result}; pub use observation::{ModelCallObservation, RunObservation, RunObserver}; pub use raw::RawResponse; -pub use run::{ClientRouter, ClientRouterBuilder, decide, run}; +pub use run::{ClientRouter, decide, run}; pub use switchyard_translation::RawEventStream; pub use system_one::SystemOneClient; diff --git a/crates/libsy-llm-client/src/run.rs b/crates/libsy-llm-client/src/run.rs index ac1da56d3..9f30991b1 100644 --- a/crates/libsy-llm-client/src/run.rs +++ b/crates/libsy-llm-client/src/run.rs @@ -5,8 +5,8 @@ //! //! [`switchyard_libsy::Algorithm::run_stream`] is the whole libsy API: it yields a stream of //! steps and expects its consumer to serve every offloaded model call. [`run()`] is that -//! consumer — it drives the stream with [`switchyard_libsy::drive`], resolves a typed client -//! for each call, and serves the terminal routing outcome. +//! consumer — it drives the stream with [`switchyard_libsy::drive`], hands routing-time calls to +//! a [`RoutedLlmClient`], and serves the terminal routing outcome. //! //! libsy owns the stream mechanics; what this module adds is per-target request preparation, //! ordered candidate fallback, and the `libsy.client_call` span around each candidate. Each @@ -49,7 +49,7 @@ use crate::{metrics, observability}; /// `clients` resolves each offloaded call to the client for the target the algorithm /// selected — an algorithm may route among targets served by different providers, so this is /// a per-call lookup, not one client for the whole run. Use -/// [`ClientRouter::single`](ClientRouter::single) when one client serves every LLM target. +/// [`ClientRouter::single`](ClientRouter::single) when one client serves every target. /// /// Routing calls are buffered; client failures stop the request unless the call enables recovery. /// Once routing completes, non-timeout failures may try the outcome's ordered fallback candidates. @@ -66,7 +66,7 @@ pub async fn run( // This says if we have an observer, put Some(..) in routing_observations. // No observer means we don't want any routing_observations. let routing_observations = observer.as_ref().map(|_| Arc::new(Mutex::new(Vec::new()))); - let outcome = match clients.inner.response_state.stored_state_owner(&request) { + let outcome = match clients.stored_state_owner(&request) { Some(owner) => Ok(continue_on(owner, &algorithm_name, request)), None => { drive(algorithm, request, models, { @@ -134,7 +134,7 @@ pub async fn decide( models: Arc, ) -> Result { let routing_clients = clients.clone(); - let mut outcome = match clients.inner.response_state.stored_state_owner(&request) { + let mut outcome = match clients.stored_state_owner(&request) { Some(owner) => continue_on(owner, algorithm.name(), request), None => { drive(algorithm, request, models, move |call| { @@ -144,10 +144,7 @@ pub async fn decide( } }; let selected_model_id = outcome.selected_model_id()?.clone(); - outcome.request = clients - .inner - .llm_policy - .prepare_completion_request(outcome.request, &selected_model_id); + outcome.request = clients.prepare_completion_request(outcome.request, &selected_model_id); outcome.response = outcome .response .map(|response| clients.remember_state_owner(&outcome.request, response)) @@ -165,7 +162,7 @@ async fn serve_decision( call: CallDecision, observations: &Option>>>, ) -> Result<()> { - let client = clients.inner.clients.route_decision(&call.model); + let client = clients.route_decision(&call.model); let started = Instant::now(); let result = async { client?.call(call.request.clone()).await }.await; tracing::Span::current().record("outcome", if result.is_ok() { "ok" } else { "error" }); @@ -227,10 +224,7 @@ async fn serve( } }; let target = call.models.first().ok_or(LibsyError::NoTargets)?; - let request = clients - .inner - .llm_policy - .prepare_routing_request(call.request.clone(), target); + let request = clients.prepare_routing_request(call.request.clone(), target); match call_one( &clients, target, @@ -258,10 +252,7 @@ async fn call_first_available( observe: &(dyn Fn(ModelCallObservation) + Send + Sync), ) -> Result { for (index, target) in models.iter().enumerate() { - let request = clients - .inner - .llm_policy - .prepare_completion_request(request.clone(), target); + let request = clients.prepare_completion_request(request.clone(), target); match call_one( clients, target, @@ -593,136 +584,34 @@ fn conversation_id(fields: &serde_json::Map) -> Option<&str> { /// host's concern, and two targets in one run may sit on different providers. A router owns /// that mapping and resolves the client for each kind of call. /// -/// Use [`Self::builder`] to register multiple call kinds. Cloning shares client mappings, -/// LLM request policy, and conversation state across runs. +/// Cloning is cheap — the mapping is shared, so one router can serve every request. #[derive(Clone)] pub struct ClientRouter { inner: Arc, } struct ClientRouting { - clients: ClientRegistry, - llm_policy: LlmRequestPolicy, - response_state: ResponseState, -} - -struct ClientRegistry { - llm: LlmClients, - decision: HashMap>, -} - -#[derive(Default)] -struct LlmRequestPolicy { + routing: Routing, + decision_clients: HashMap>, target_prompts: HashMap, routing_answer_target: Option, -} - -struct ResponseState { /// Provider-owned routing pins and materialized cross-format history. - owners: Mutex, + state_owners: Mutex, /// Native Responses state only needs a routing pin when answer targets span clients. track_provider_state: bool, } -enum LlmClients { - /// One client serves every LLM target. +enum Routing { + /// One client serves every model. Single(Arc), /// Each model is served by the client configured for it. ByModel(HashMap>), } -/// Configure clients and LLM request policy before sharing a router across runs. -/// Registration ends at [`Self::build`]; cloned routers share the same clients and state. -pub struct ClientRouterBuilder { - clients: ClientRegistry, - llm_policy: LlmRequestPolicy, - completion_targets: Option>, -} - -impl ClientRouterBuilder { - /// Replace the LLM routing table with explicit model-to-client mappings. - pub fn llm_clients(mut self, clients: HashMap>) -> Self { - self.clients.llm = LlmClients::ByModel(clients); - self - } - - /// Use one LLM client for every model, leaving model lookup to that client. - pub fn single_llm_client(mut self, client: Arc) -> Self { - self.clients.llm = LlmClients::Single(client); - self - } - - /// Replace the decision routing table. Decision calls require an explicit mapping. - pub fn decision_clients( - mut self, - clients: HashMap>, - ) -> Self { - self.clients.decision = clients; - self - } - - /// Set LLM answer prompts keyed by resolved model ID, not deployment target name. - pub fn target_prompts(mut self, prompts: HashMap) -> Self { - self.llm_policy.target_prompts = prompts; - self - } - - /// Apply its target prompt to a routing-time LLM call whose response can be the answer. - pub fn routing_answer_target(mut self, target: Option) -> Self { - self.llm_policy.routing_answer_target = target; - self - } - - /// Use these completion targets to decide whether native Responses need a provider pin. - /// If omitted, every mapped LLM is considered a possible completion target. - pub fn completion_targets(mut self, targets: Vec) -> Self { - self.completion_targets = Some(targets); - self - } - - /// Finish registration and share the immutable configuration and conversation state. - pub fn build(self) -> ClientRouter { - let track_provider_state = match &self.clients.llm { - LlmClients::Single(_) => false, - LlmClients::ByModel(by_model) => { - let targets = self - .completion_targets - .unwrap_or_else(|| by_model.keys().cloned().collect()); - let mut clients = targets.iter().filter_map(|model| by_model.get(model)); - clients - .next() - .is_some_and(|first| clients.any(|client| !Arc::ptr_eq(first, client))) - } - }; - ClientRouter { - inner: Arc::new(ClientRouting { - clients: self.clients, - llm_policy: self.llm_policy, - response_state: ResponseState { - owners: Mutex::default(), - track_provider_state, - }, - }), - } - } -} - impl ClientRouter { - /// Start with no clients. Register each call kind before building the shared router. - pub fn builder() -> ClientRouterBuilder { - ClientRouterBuilder { - clients: ClientRegistry { - llm: LlmClients::ByModel(HashMap::new()), - decision: HashMap::new(), - }, - llm_policy: LlmRequestPolicy::default(), - completion_targets: None, - } - } - /// Build a router over `model name -> client`, for targets spread across providers. pub fn new(by_model: HashMap>) -> Self { - Self::builder().llm_clients(by_model).build() + Self::new_with_target_prompts(by_model, HashMap::new(), None) } /// Build a router with target prompts used by [`run`] and [`decide`]. @@ -738,11 +627,13 @@ impl ClientRouter { target_prompts: HashMap, routing_answer_target: Option, ) -> Self { - Self::builder() - .llm_clients(by_model) - .target_prompts(target_prompts) - .routing_answer_target(routing_answer_target) - .build() + let completion_targets = by_model.keys().cloned().collect::>(); + Self::new_with_completion_targets( + by_model, + target_prompts, + routing_answer_target, + &completion_targets, + ) } /// Build a router with an explicit list of models that can answer requests. @@ -758,21 +649,66 @@ impl ClientRouter { routing_answer_target: Option, completion_targets: &[ModelId], ) -> Self { - Self::builder() - .llm_clients(by_model) - .target_prompts(target_prompts) - .routing_answer_target(routing_answer_target) - .completion_targets(completion_targets.to_vec()) - .build() + Self::new_with_decision_clients( + by_model, + HashMap::new(), + target_prompts, + routing_answer_target, + completion_targets, + ) + } + + /// Build a router with separate LLM and decision mappings, shared across clones. + /// LLM prompt and completion settings follow [`Self::new_with_completion_targets`]. + pub fn new_with_decision_clients( + by_model: HashMap>, + decision_clients: HashMap>, + target_prompts: HashMap, + routing_answer_target: Option, + completion_targets: &[ModelId], + ) -> Self { + let mut clients = completion_targets + .iter() + .filter_map(|model| by_model.get(model)); + let spans_providers = clients + .next() + .is_some_and(|first| clients.any(|client| !Arc::ptr_eq(first, client))); + Self { + inner: Arc::new(ClientRouting { + routing: Routing::ByModel(by_model), + decision_clients, + target_prompts, + routing_answer_target, + state_owners: Mutex::default(), + track_provider_state: spans_providers, + }), + } } - /// A router that serves every LLM target with one client — the single-provider case. + /// A router that serves every model with one client — the single-provider case. /// /// [`TranslatingLlmClient`](crate::TranslatingLlmClient) already maps model names to /// backends internally and rejects ones it does not know, so enumerating them here would /// only duplicate that. pub fn single(client: Arc) -> Self { - Self::builder().single_llm_client(client).build() + Self::single_with_decision_clients(client, HashMap::new()) + } + + /// Use one client for all LLM targets and explicit clients for decision targets. + pub fn single_with_decision_clients( + client: Arc, + decision_clients: HashMap>, + ) -> Self { + Self { + inner: Arc::new(ClientRouting { + routing: Routing::Single(client), + decision_clients, + target_prompts: HashMap::new(), + routing_answer_target: None, + state_owners: Mutex::default(), + track_provider_state: false, + }), + } } /// The client that serves `model`. @@ -783,18 +719,9 @@ impl ClientRouter { &self, model: &ModelId, ) -> std::result::Result<&Arc, LlmClientError> { - self.inner.clients.route_llm(model) - } -} - -impl ClientRegistry { - fn route_llm( - &self, - model: &ModelId, - ) -> std::result::Result<&Arc, LlmClientError> { - match &self.llm { - LlmClients::Single(client) => Ok(client), - LlmClients::ByModel(by_model) => { + match &self.inner.routing { + Routing::Single(client) => Ok(client), + Routing::ByModel(by_model) => { by_model .get(model) .ok_or_else(|| LlmClientError::Configuration { @@ -809,15 +736,14 @@ impl ClientRegistry { &self, model: &ModelId, ) -> std::result::Result<&Arc, LlmClientError> { - self.decision + self.inner + .decision_clients .get(model) .ok_or_else(|| LlmClientError::Configuration { message: format!("no decision client is configured for model {model:?}"), }) } -} -impl ResponseState { /// Return the recorded model for the requested response or conversation ID. fn stored_state_owner(&self, request: &Request) -> Option { let fields = &request.llm_request.extensions.fields; @@ -825,7 +751,7 @@ impl ResponseState { Some(id) => ("previous_response_id", id), None => ("conversation", conversation_id(fields)?), }; - let state = self.owners.lock().owner(id)?.clone(); + let state = self.inner.state_owners.lock().owner(id)?.clone(); Some(StateOwner { model: state.model, trigger, @@ -840,7 +766,7 @@ impl ResponseState { .fields .get("previous_response_id") .and_then(Value::as_str) - .and_then(|id| self.owners.lock().owner(id)?.history.clone()); + .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..) { Some(messages) => (parent, messages.to_vec()), @@ -851,9 +777,7 @@ impl ResponseState { messages: Arc::from(messages), } } -} -impl ClientRouter { /// Record provider-owned Responses state or canonical history for a cross-format response. fn remember_state_owner(&self, request: &Request, mut response: Response) -> Result { let Some(model) = response.served_model().cloned() else { @@ -870,22 +794,19 @@ impl ClientRouter { .contains_key(&responses_format) || fields.contains_key("previous_response_id") || fields.contains_key("conversation"); - let state = &self.inner.response_state; - if !responses_request && !state.track_provider_state { + if !responses_request && !self.inner.track_provider_state { return Ok(response); } - let canonical_input = responses_request.then(|| state.canonical_input(request)); + let canonical_input = responses_request.then(|| self.canonical_input(request)); response.llm_response = match response.llm_response { LlmResponse::Agg(agg) => { if let Some(body) = agg.preservation.responses.get(&responses_format) { - if state.track_provider_state { - state - .remember_response(body, &model, store, conversation.as_deref()) + if self.inner.track_provider_state { + self.remember_response(body, &model, store, conversation.as_deref()) .map_err(|error| LibsyError::client_call(model.clone(), error))?; } } else if let Some(input) = &canonical_input { - state - .remember_canonical_response(&agg, &model, store, input) + self.remember_canonical_response(&agg, &model, store, input) .map_err(|error| LibsyError::client_call(model.clone(), error))?; } LlmResponse::Agg(agg) @@ -913,13 +834,12 @@ impl ClientRouter { mut native_responses, mut valid, )| async move { - let state = &router.inner.response_state; let Some(event) = stream.next().await else { if let Some(input) = &canonical_input && !native_responses && valid { - state.remember_canonical_response( + router.remember_canonical_response( &accumulator.finish(), &model, store, @@ -933,10 +853,10 @@ impl ClientRouter { && preserved.source().as_str() == WireFormat::OpenAiResponses.as_str() { native_responses = true; - if state.track_provider_state + if router.inner.track_provider_state && let Some(body) = preserved.raw().get("response") { - state.remember_response( + router.remember_response( body, &model, store, @@ -976,9 +896,7 @@ impl ClientRouter { }; Ok(response) } -} -impl ResponseState { /// Check both IDs from a response or stream event before recording either. The response's /// `store` value overrides the request value; `false` skips only the response ID. fn remember_response( @@ -988,7 +906,7 @@ impl ResponseState { store: bool, conversation: Option<&str>, ) -> std::result::Result<(), LlmClientError> { - let mut owners = self.owners.lock(); + let mut owners = self.inner.state_owners.lock(); let response_id = body .get("id") .and_then(Value::as_str) @@ -1025,10 +943,11 @@ impl ResponseState { input.parent.clone(), Arc::from(segment), )); - let result = self - .owners - .lock() - .remember(response_id, None, model, Some(history)); + let result = + self.inner + .state_owners + .lock() + .remember(response_id, None, 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"); @@ -1039,20 +958,18 @@ impl ResponseState { } Ok(()) } -} -impl LlmRequestPolicy { /// Prepare a completion candidate with its configured target prompt. fn prepare_completion_request(&self, mut request: Request, target: &ModelId) -> Request { - let prompt = self.target_prompts.get(target).map(String::as_str); + let prompt = self.inner.target_prompts.get(target).map(String::as_str); prepare_request_for_target(&mut request.llm_request, target, prompt); request } /// Prepare a routing call, adding a target prompt only when it generates a candidate answer. fn prepare_routing_request(&self, mut request: Request, target: &ModelId) -> Request { - let prompt = if self.routing_answer_target.as_ref() == Some(target) { - self.target_prompts.get(target).map(String::as_str) + let prompt = if self.inner.routing_answer_target.as_ref() == Some(target) { + self.inner.target_prompts.get(target).map(String::as_str) } else { None }; @@ -1365,12 +1282,12 @@ mod tests { config: TaskClassifierConfig { judge: CapabilityJudgeConfig::Decision(DecisionJudgeConfig { cutoff: 0.4, - instructions: Some(json!({"task": "Compare candidate outcomes"})), + instructions: None, candidates: BTreeMap::from([ ("a".into(), "capable".into()), ("b".into(), "efficient".into()), ]), - evidence: json!({"cases": ["reference task"]}), + evidence: json!({}), }), ..TaskClassifierConfig::default() }, @@ -1409,12 +1326,6 @@ mod tests { server.reset().await; Mock::given(method("POST")) .and(path("/v1/systemone")) - .and(body_partial_json(json!({ - "state": {"evidence": {"cases": ["reference task"]}}, - "questions": {"route": { - "instructions": {"task": "Compare candidate outcomes"} - }} - }))) .respond_with(template) .expect(2) .mount(&server) @@ -1424,26 +1335,15 @@ mod tests { requests: Mutex::default(), first: FirstOutcome::Unauthorized, }); - let clients = ClientRouter::builder() - .single_llm_client(llm.clone()) - .decision_clients(HashMap::from([( + let clients = ClientRouter::single_with_decision_clients( + llm.clone(), + HashMap::from([( ModelId::from("judge"), client.clone() as Arc, - )])) - .target_prompts(HashMap::from([ - ("judge".into(), "LLM judge prompt".into()), - ("efficient".into(), "Answer prompt".into()), - ("capable".into(), "Answer prompt".into()), - ])) - .routing_answer_target(Some("judge".into())) - .build(); - assert!(matches!( - clients.inner.clients.route_decision(&"efficient".into()), - Err(LlmClientError::Configuration { .. }) - )); + )]), + ); let outcome = decide(algorithm()?, clients.clone(), request(), models.clone()).await?; assert_eq!(outcome.selected_model_id()?, expected); - assert_eq!(instruction_text(&outcome.request), ["Answer prompt"]); assert!(llm.calls.lock().is_empty()); let events = Arc::new(Mutex::new(Vec::new())); let captured = events.clone(); @@ -1457,7 +1357,6 @@ mod tests { .await?; assert_eq!(selected, expected); assert_eq!(&*llm.calls.lock(), &[ModelId::from(expected)]); - assert_eq!(instruction_text(&llm.requests.lock()[0]), ["Answer prompt"]); { let events = events.lock(); let RunObservation::DecisionCall(call) = &events[0] else { @@ -1885,8 +1784,6 @@ mod tests { metadata: None, }; let history = clients - .inner - .response_state .stored_state_owner(&third) .and_then(|owner| owner.history) .expect("chained canonical history"); @@ -1977,13 +1874,7 @@ mod tests { is_error: None, })], }]; - assert!( - clients - .inner - .response_state - .stored_state_owner(&follow) - .is_none() - ); + assert!(clients.stored_state_owner(&follow).is_none()); response .llm_response .into_agg() @@ -1992,8 +1883,6 @@ mod tests { let outcome = continue_on( clients - .inner - .response_state .stored_state_owner(&follow) .expect("completed stream state"), "passthrough", @@ -2047,13 +1936,7 @@ mod tests { .extensions .fields .insert("previous_response_id".to_string(), json!("msg_not_stored")); - assert!( - clients - .inner - .response_state - .stored_state_owner(&follow) - .is_none() - ); + assert!(clients.stored_state_owner(&follow).is_none()); Ok(()) } @@ -2069,7 +1952,7 @@ mod tests { client as Arc, )])); let model = ModelId::from("weak"); - clients.inner.response_state.owners.lock().by_id = (0..MAX_STATE_OWNERS) + clients.inner.state_owners.lock().by_id = (0..MAX_STATE_OWNERS) .map(|id| { ( format!("existing_{id}"), @@ -2119,7 +2002,7 @@ mod tests { .map_err(|error| LibsyError::client_call("weak", error))?; assert_eq!( - clients.inner.response_state.owners.lock().by_id.len(), + clients.inner.state_owners.lock().by_id.len(), MAX_STATE_OWNERS ); Ok(()) From a56b499e71dbc917d6ffec76c2149806af442917 Mon Sep 17 00:00:00 2001 From: nachiketb Date: Mon, 5 Oct 2026 14:32:05 -0700 Subject: [PATCH 8/8] refactor(client): trim System One content validation Signed-off-by: nachiketb --- crates/libsy-llm-client/src/run.rs | 32 +++++++++-------------- crates/libsy-llm-client/src/system_one.rs | 28 ++------------------ 2 files changed, 14 insertions(+), 46 deletions(-) diff --git a/crates/libsy-llm-client/src/run.rs b/crates/libsy-llm-client/src/run.rs index 9f30991b1..dcb4a6f7d 100644 --- a/crates/libsy-llm-client/src/run.rs +++ b/crates/libsy-llm-client/src/run.rs @@ -1237,6 +1237,11 @@ mod tests { for (id, answer) in [ ("unknown-question", json!({"type": "noul", "noul": 0.9})), ("boolean", json!({"type": "score", "score": 0.0})), + ( + "route", + json!({"type": "choice", "choice": "unknown-option", + "probabilities": {"advantage": 0.2, "no_advantage": 0.8}}), + ), ("score", json!({"type": "score", "score": -0.1})), ( "score", @@ -1254,16 +1259,20 @@ mod tests { } invalid_responses.push(invalid); } - for invalid in invalid_responses { + for template in invalid_responses + .into_iter() + .map(|body| ResponseTemplate::new(200).set_body_json(body)) + .chain([ResponseTemplate::new(200).set_body_string("invalid JSON")]) + { server.reset().await; Mock::given(method("POST")) - .respond_with(ResponseTemplate::new(200).set_body_json(invalid)) + .respond_with(template) .expect(1) .mount(&server) .await; assert!(matches!( client.call(mixed.clone()).await, - Err(LlmClientError::ResponseTranslation(_)) + Err(LlmClientError::ResponseTranslation(_) | LlmClientError::InvalidResponse { .. }) )); server.verify().await; } @@ -1294,29 +1303,12 @@ mod tests { }, )?)) }; - let mut undeclared_choice = body.clone(); - undeclared_choice["answers"]["route"]["choice"] = json!("unknown-option"); for (template, expected, is_success) in [ ( ResponseTemplate::new(200).set_body_json(&body), "efficient", true, ), - ( - ResponseTemplate::new(200).set_body_json(&undeclared_choice), - "capable", - false, - ), - ( - ResponseTemplate::new(200).set_body_json(json!({"answers": {}})), - "capable", - false, - ), - ( - ResponseTemplate::new(200).set_body_string("invalid JSON"), - "capable", - false, - ), ( ResponseTemplate::new(503).set_body_string("unavailable"), "capable", diff --git a/crates/libsy-llm-client/src/system_one.rs b/crates/libsy-llm-client/src/system_one.rs index 03a4827ac..611832ca8 100644 --- a/crates/libsy-llm-client/src/system_one.rs +++ b/crates/libsy-llm-client/src/system_one.rs @@ -167,15 +167,8 @@ fn encode(request: &DecisionRequest) -> Result { .ok_or_else(|| LlmClientError::InvalidRequest { message: "System One requires a selected model".into(), })?; - content(&request.context, false, "context")?; let mut questions = serde_json::Map::new(); for (id, question) in &request.questions { - let path = format!("questions.{id}"); - content( - &question.instructions, - true, - &format!("{path}.instructions"), - )?; let (kind, criteria) = match &question.kind { DecisionKind::Boolean { true_description, @@ -185,7 +178,6 @@ fn encode(request: &DecisionRequest) -> Result { for (key, description) in [("true", true_description), ("false", false_description)] { if let Some(description) = description { - content(description, true, &format!("{path}.{key}_description"))?; criteria.insert(key.into(), description.clone()); } } @@ -195,24 +187,18 @@ fn encode(request: &DecisionRequest) -> Result { let mut criteria = serde_json::Map::new(); for option in options { let description = option.description.as_ref().unwrap_or(&Value::Null); - content(description, true, &format!("{path}.options.{}", option.id))?; if criteria .insert(option.id.clone(), description.clone()) .is_some() { return Err(LlmClientError::InvalidRequest { - message: format!("duplicate option {:?} in {path}", option.id), + message: format!("duplicate option {:?} in questions.{id}", option.id), }); } } ("choice", Value::Object(criteria)) } - DecisionKind::Score { levels } => { - for (index, level) in levels.iter().enumerate() { - content(level, false, &format!("{path}.levels.{index}"))?; - } - ("score", json!(levels)) - } + DecisionKind::Score { levels } => ("score", json!(levels)), }; questions.insert( id.clone(), @@ -224,16 +210,6 @@ fn encode(request: &DecisionRequest) -> Result { Ok(json!({"model": model, "state": request.context, "questions": questions})) } -fn content(value: &Value, nullable: bool, path: &str) -> Result<()> { - if value.is_string() || value.is_object() || value.is_array() || (nullable && value.is_null()) { - return Ok(()); - } - Err(LlmClientError::RequestEncoding(format!( - "System One cannot represent {path} as {}", - value - ))) -} - #[derive(Deserialize)] struct WireResponse { id: Option,