fix(client): enforce response deadlines and stop routing on errors

CHANGELOG.md

@@ -8,6 +8,11 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ### Added
 
+- **`timeout_ms` on `[llm_clients.<name>]`** — one deadline covers all attempts,
+  retry delays, and the complete response, including stream reads. Unset leaves
+  the wait unbounded; `0` is rejected. A timeout returns `504` without trying another
+  model, or a framed error if the final answer has already started streaming.
+  Timed-out attempts are counted in metrics.
 - **Per-target `reasoning_effort`** — a target can force the reasoning effort
   of every request it serves, replacing the caller's value (`reasoning.effort`
   on the Responses wire, `reasoning_effort` on Chat Completions), so a strong
@@ -89,6 +94,11 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ### Changed
 
+- **HTTP client errors stop routing** — after the configured retries, the Rust
+  runner stops the request instead of letting the routing algorithm choose a
+  fallback. This also applies when `timeout_ms` is unset or an advisor has
+  `fail_open = true`. The runner collects streams used during routing and preserves
+  provider events for replay. Invalid judge verdicts keep their existing fallback.
 - **`Algorithm::route` returns `Result<RoutingOutcome>`** — instead of the
   bare final `Result`, so callers observe the full routing outcome (see #458
   for the design). (#459)

crates/libsy-llm-client/README.md

@@ -36,8 +36,8 @@ It depends on `switchyard-libsy`, `switchyard-protocol`, and
   `anthropic-version`).
 - **Model rewrite.** The resolved [`ModelId`] is both the map key and the model id
   sent upstream — it overwrites whatever `model` the request arrived with.
-- **Streaming is chosen by the request.** If the encoded body has `stream: true`
-  (i.e. `request.llm_request.stream`), you get `LlmResponse::Stream`; otherwise
+- **Streaming is chosen by the encoded request body.** If the body has
+  `stream: true` after `extra_body` is applied, you get `LlmResponse::Stream`; otherwise
   `LlmResponse::Agg`. OpenAI Chat streaming requests default
   `stream_options.include_usage` to `true`; an explicit caller value is preserved.
 
@@ -69,7 +69,9 @@ fn build_client() -> switchyard_llm_client::Result<TranslatingLlmClient> {
         forward_auth: false,
         extra_headers: BTreeMap::new(),
         extra_body: BTreeMap::new(),
+        reasoning_effort: None,
         max_retries: 2,
+        timeout: None,
     };
 
     let models = [ModelConfig::new(
@@ -247,13 +249,23 @@ fn build_multi_format_client(
   transport failures, timeouts, HTTP 408/429, and 5xx responses. Buffered body
   transport failures are retried; streaming body failures are not replayed after
   the response has been returned.
-
-Retries replay the same upstream request to the same model. Each candidate's
-`max_retries` budget is exhausted before candidate fallback advances to the next
-model. The worst case is `candidates × (max_retries + 1)` upstream requests, and
-total latency includes every candidate's capped `Retry-After` backoff. A transport
-failure can duplicate a request that the provider processed but did not finish
-returning.
+- `HttpBackendConfig::timeout` bounds one complete response, including retries,
+  retry delays, and every stream read. Expiry returns `LlmClientError::Timeout`,
+  either from the call or from the returned stream, which then ends. `None` leaves
+  the wait unbounded.
+
+`run` and `decide` collect streams used during routing, including answers that an
+algorithm must inspect, before returning them to the algorithm. They retain the
+provider events for replay. After the configured retries, a client failure stops
+`run` or `decide` before the algorithm can choose another routing candidate. This
+also applies when `timeout` is `None` or an advisor has `fail_open = true`.
+`libsy` and custom hosts that drive it directly are unchanged.
+
+Retries replay the same upstream request to the same model. After routing completes,
+non-timeout failures may try another completion candidate. A timeout stops the call.
+A transport failure can duplicate a request that the provider processed but did not
+finish returning. Each attempt is counted once; expiry during a retry delay or
+after a stream has started does not count another attempt.
 
 ## Errors
 

crates/libsy-llm-client/src/backend.rs

@@ -3,6 +3,7 @@
 
 //! Per-provider backend configuration: wire format, upstream URL, and auth.
 
+use std::time::Duration;
 use std::{collections::BTreeMap, fmt};
 
 use reqwest::RequestBuilder;
@@ -62,6 +63,9 @@ pub struct HttpBackendConfig {
     pub reasoning_effort: Option<String>,
     /// Additional attempts after the initial upstream request.
     pub max_retries: u32,
+    /// Deadline for one complete response, including retries, retry delays, and stream reads.
+    /// `None` leaves the wait unbounded.
+    pub timeout: Option<Duration>,
 }
 
 impl fmt::Debug for HttpBackendConfig {
@@ -74,6 +78,7 @@ impl fmt::Debug for HttpBackendConfig {
             .field("extra_body_keys", &self.extra_body.keys())
             .field("reasoning_effort", &self.reasoning_effort)
             .field("max_retries", &self.max_retries)
+            .field("timeout", &self.timeout)
             .finish()
     }
 }
@@ -256,6 +261,11 @@ impl Backend {
         self.config().max_retries
     }
 
+    /// Deadline for all attempts and the complete response; `None` leaves the wait unbounded.
+    pub fn timeout(&self) -> Option<Duration> {
+        self.config().timeout
+    }
+
     /// Whether this backend speaks the Anthropic Messages wire format — the only
     /// one with a `count_tokens` endpoint.
     pub fn is_anthropic(&self) -> bool {
@@ -348,6 +358,7 @@ mod tests {
             extra_body: BTreeMap::new(),
             reasoning_effort: None,
             max_retries: 0,
+            timeout: None,
         }
     }
 

crates/libsy-llm-client/src/client.rs

@@ -224,8 +224,9 @@ impl TranslatingLlmClient {
     /// forwarded headers plus the backend's static headers and auth, and return the
     /// successful upstream response. A
     /// buffered response is fully collected within the retry boundary; a streamed
-    /// response has its first decoded event checked within that boundary. A non-success
-    /// status maps to a typed error — a 400 is classified as a context-window
+    /// response has its first decoded event checked within that boundary. The backend's
+    /// deadline, when set, spans every attempt, retry delay, and subsequent stream read. A
+    /// non-success status maps to a typed error — a 400 is classified as a context-window
     /// overflow via the backend's provider rules. Shared by
     /// [`call_rewrite_model`](Self::call_rewrite_model) (which POSTs to the
     /// backend's completion URL and decodes a response) and
@@ -269,15 +270,47 @@ impl TranslatingLlmClient {
         let url = endpoint.url(backend);
         record_gen_ai_request(&url, model, streaming);
 
+        self.send_with_retries(&url, backend, &body, metadata, model, streaming)
+            .await
+    }
+
+    // Sends the encoded body, retrying retryable failures within the backend's retry budget.
+    async fn send_with_retries(
+        &self,
+        url: &str,
+        backend: &Backend,
+        body: &Value,
+        metadata: Option<&Metadata>,
+        model: &ModelId,
+        streaming: bool,
+    ) -> Result<EncodedResponse> {
         let max_retries = u64::from(backend.max_retries());
         let max_attempts = max_retries + 1;
+        let expires_at = backend
+            .timeout()
+            .map(|timeout| (tokio::time::sleep(timeout).deadline(), timeout));
+        let deadline = async {
+            match expires_at {
+                Some((at, timeout)) => {
+                    tokio::time::sleep_until(at).await;
+                    timeout
+                }
+                None => std::future::pending().await,
+            }
+        };
+        tokio::pin!(deadline);
         let mut attempt = 0_u64;
         loop {
+            if let Some((at, timeout)) = expires_at
+                && tokio::time::Instant::now() >= at
+            {
+                return Err(deadline_error(timeout));
+            }
             let span = tracing::debug_span!(
                 target: "libsy",
                 "libsy.upstream_attempt",
                 model = %model,
-                wire_format = %wire_format,
+                wire_format = %backend.wire_format(),
                 attempt = attempt + 1,
                 max_attempts,
                 retry = attempt > 0,
@@ -287,10 +320,22 @@ impl TranslatingLlmClient {
                 will_retry = tracing::field::Empty,
                 retry_delay_ms = tracing::field::Empty,
             );
-            let result = self
-                .send_once(&url, backend, &body, metadata, model, streaming)
-                .instrument(span.clone())
-                .await;
+            let mut attempt_started = false;
+            let result = tokio::select! {
+                biased;
+                timeout = &mut deadline => {
+                    if attempt_started {
+                        metrics::record_upstream_attempt(None);
+                    }
+                    span.record("outcome", "error");
+                    span.record("will_retry", false);
+                    return Err(deadline_error(timeout));
+                }
+                result = async {
+                    attempt_started = true;
+                    self.send_once(url, backend, body, metadata, model, streaming).await
+                }.instrument(span.clone()) => result,
+            };
             // The retained handle updates this same attempt span with its outcome.
             match result {
                 Ok(response) => {
@@ -300,7 +345,18 @@ impl TranslatingLlmClient {
                     if attempt > 0 {
                         metrics::record_retry_recovered();
                     }
-                    return Ok(response);
+                    return Ok(match response {
+                        EncodedResponse::Streaming {
+                            status,
+                            chunks,
+                            upstream_headers,
+                        } => EncodedResponse::Streaming {
+                            status,
+                            chunks: stream_with_deadline(chunks, expires_at),
+                            upstream_headers,
+                        },
+                        buffered => buffered,
+                    });
                 }
                 Err(failure) => {
                     let will_retry = attempt < max_retries && failure.is_retryable();
@@ -317,7 +373,11 @@ impl TranslatingLlmClient {
                     span.record("retry_delay_ms", duration_millis(delay));
                     // Close the attempt span before sleeping so backoff is not attempt latency.
                     drop(span);
-                    tokio::time::sleep(delay).await;
+                    tokio::select! {
+                        biased;
+                        timeout = &mut deadline => return Err(deadline_error(timeout)),
+                        _ = tokio::time::sleep(delay) => {}
+                    }
                     attempt += 1;
                 }
             }
@@ -653,6 +713,18 @@ struct AttemptFailure {
     retry_after: Option<Duration>,
 }
 
+fn deadline_error(timeout: Duration) -> LlmClientError {
+    LlmClientError::Timeout {
+        source: Box::new(std::io::Error::new(
+            std::io::ErrorKind::TimedOut,
+            format!(
+                "response did not finish within {} ms, retries included",
+                duration_millis(timeout)
+            ),
+        )),
+    }
+}
+
 impl AttemptFailure {
     fn is_retryable(&self) -> bool {
         match &self.error {
@@ -665,6 +737,26 @@ impl AttemptFailure {
     }
 }
 
+fn stream_with_deadline(
+    chunks: LlmResponseStream,
+    deadline: Option<(tokio::time::Instant, Duration)>,
+) -> LlmResponseStream {
+    let Some((at, timeout)) = deadline else {
+        return chunks;
+    };
+    stream::unfold(Some(chunks), move |chunks| async move {
+        let mut chunks = chunks?;
+        let chunk = tokio::select! {
+            biased;
+            _ = tokio::time::sleep_until(at) => return Some((Err(deadline_error(timeout)), None)),
+            chunk = chunks.next() => chunk?,
+        };
+        let next = chunk.is_ok().then_some(chunks);
+        Some((chunk, next))
+    })
+    .boxed()
+}
+
 async fn prepare_response_stream(
     response: reqwest::Response,
     backend: &Backend,
@@ -1199,6 +1291,7 @@ mod tests {
             extra_body: BTreeMap::new(),
             reasoning_effort: None,
             max_retries: 0,
+            timeout: None,
         }
     }
 
@@ -2212,6 +2305,52 @@ mod tests {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn deadline_expires_before_send_or_stream_poll()
+    -> std::result::Result<(), Box<dyn Error + Sync + Send + 'static>> {
+        let server = MockServer::start().await;
+        Mock::given(method("POST"))
+            .respond_with(ResponseTemplate::new(200).set_body_raw(
+                "data: {\"id\":\"test\",\"model\":\"gpt\",\"choices\":[{\"index\":0,\"delta\":{\"content\":\"ok\"},\"finish_reason\":\"stop\"}]}\n\ndata: [DONE]\n\n",
+                "text/event-stream",
+            ))
+            .mount(&server)
+            .await;
+        for timeout in [Duration::ZERO, Duration::from_millis(200)] {
+            let mut backend = config(&server.uri());
+            backend.timeout = Some(timeout);
+            let client = TranslatingLlmClient::new(&[ModelConfig::new(
+                "gpt",
+                Backend::OpenAiChat(backend),
+                None,
+            )])?;
+            let result = client
+                .call_rewrite_model(request_for(Some("gpt"), true), None)
+                .await;
+            if timeout.is_zero() {
+                assert!(matches!(result, Err(LlmClientError::Timeout { .. })));
+                assert!(
+                    server
+                        .received_requests()
+                        .await
+                        .expect("recorded requests")
+                        .is_empty()
+                );
+            } else {
+                let LlmResponse::Stream(mut chunks) = result?.llm_response else {
+                    return Err("expected a streaming response".into());
+                };
+                tokio::time::sleep(Duration::from_millis(300)).await;
+                assert!(matches!(
+                    chunks.next().await,
+                    Some(Err(LlmClientError::Timeout { .. }))
+                ));
+                assert!(chunks.next().await.is_none());
+            }
+        }
+        Ok(())
+    }
+
     #[tokio::test]
     async fn timeout_is_retried_before_a_response_is_returned()
     -> std::result::Result<(), Box<dyn Error + Sync + Send + 'static>> {