The Provider Abstraction and a Real HTTP Call

In Chapter 7 you read the Provider trait signature line by line; this chapter plugs it into real HTTP. By the end you should be able to answer:

• A connection timeout and an HTTP 400 — which becomes which ProviderError variant, and which one justifies switching to the next model? Which lines decide each case?
• At what moment does the code read the API key from its environment variable? And why does a streaming request deliberately get no total timeout?
• A streaming response arrives split between Hel and lo — which struct reassembles the complete line?

The through-line is one request: body {"model":"group/coder","messages":[…]} (the group from Chapter 11's config), headed for an OpenAI-compatible endpoint. Its model field gets rewritten to the upstream id first — the routing layer does that (Chapter 13); this chapter only covers sending it out and getting an honest answer back.

The contract: three methods, one provider

Without this abstraction, ask and the routing loop would each hand-roll their own HTTP; and to test failover in Chapter 13 you need a fake provider that never touches the network. The trait Provider pins "one upstream" down as a contract. First, the thing it carries:

/// A chat request body, passed through untouched except for `model`.
#[derive(Debug, Clone)]
pub struct ChatRequest(pub Value);

impl ChatRequest {
    // ...
    /// A copy of the body with `model` set to the upstream model id.
    ///
    /// One deliberate `clone` per candidate: the original request must stay
    /// intact so the failover loop can rewrite it for the next model.
    pub fn with_model(&self, model_id: &str) -> ChatRequest {
        let mut body = self.0.clone();
        // Insert through the object map (not `body["model"] = ...`, which
        // panics on non-objects); callers guarantee an object body.
        if let Value::Object(map) = &mut body {
            map.insert("model".to_string(), Value::String(model_id.to_string()));
        }
        ChatRequest(body)
    }
    // ...
}

Read it: start at pub struct ChatRequest(pub Value) — a newtype over serde_json::Value (Chapter 6); .0 is the raw JSON. Leaving messages/tools untyped is deliberate: the module doc says the gateway forwards tools and temperature faithfully without understanding them, and the only field it ever rewrites is model. with_model(&self, …) takes a read-only borrow (Chapter 5) and opens with .0.clone() — one deliberate clone, because the original request must stay intact for the failover loop to rewrite it for the next candidate; the comment also explains the object-map insert instead of index assignment (the latter panics on non-objects). The ChatRequest it returns is the one that actually travels. The contract itself:

/// Byte chunks of a streaming (SSE) response body, in arrival order.
pub type ByteStream = Pin<Box<dyn Stream<Item = Result<Bytes, ProviderError>> + Send>>;

/// One upstream LLM API. [`crate::openai`] implements it for any
/// OpenAI-compatible endpoint; unit tests implement it with in-memory fakes.
#[async_trait::async_trait]
pub trait Provider: Send + Sync {
    /// Provider name, as used in `[providers.<name>]` and model references.
    fn name(&self) -> &str;

    /// Send the request and wait for the full JSON response.
    async fn complete(&self, request: ChatRequest) -> Result<ChatResponse, ProviderError>;

    /// Send the request and return the raw SSE byte stream. A successful
    /// return means the upstream already *accepted* the request (the HTTP
    /// status was checked); failures after the first byte are delivered
    /// through the stream itself — at that point failover is impossible.
    async fn complete_stream(&self, request: ChatRequest) -> Result<ByteStream, ProviderError>;
}

Read it: meet ByteStream first — read Pin<Box<dyn Stream<…>>> as "a pointer holding some byte stream". The three methods each answer one question: name() reports the [providers.X] table name for log correlation; complete waits for the full JSON; complete_stream hands back a raw byte stream. #[async_trait] and Send + Sync are old friends from Chapter 7; that doc line about failover being impossible after the first byte comes back in Section 4.

TIP

Pin is a self-reference safety marker, there for the compiler's benefit. On a first read, "a pointer holding a stream" is all you need.

A real POST: reqwest and timeouts

OpenAiProvider has four fields: name; base_url (trailing slash trimmed at construction, so endpoint() is just one format!("{}/chat/completions", …)); api_key_env — the environment variable's name, not the value, per Chapter 11's rule; and one http: Client built once and reused, with the 10-second CONNECT_TIMEOUT set at construction. Both paths share a single private method, send:

/// How long to wait for a TCP connection to the upstream.
const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
/// Total budget for one non-streaming round trip. Streams deliberately have
/// no overall timeout — a long answer is not a stuck one.
const REQUEST_TIMEOUT: Duration = Duration::from_secs(120);
// ...
/// Fire the POST. Network-level failures (DNS, connect, timeout) are
/// retryable by definition: nothing was learned about the request.
async fn send(
    &self,
    request: &ChatRequest,
    streaming: bool,
) -> Result<reqwest::Response, ProviderError> {
    let mut builder = self
        .http
        .post(self.endpoint())
        .bearer_auth(self.api_key()?)
        .json(&request.0);
    if !streaming {
        builder = builder.timeout(REQUEST_TIMEOUT);
    }
    builder.send().await.map_err(|error| {
        ProviderError::Retryable(format!("request to {} failed: {error}", self.endpoint()))
    })
}

Read it: follow the data from self.http.post(self.endpoint()) — the URL is built, bearer_auth(self.api_key()?) puts the key into the Authorization: Bearer header (and ? returns early right there with "environment variable not set", the Retryable error api_key() produces), then .json(&request.0) serializes the through-line body into the POST body. The key is resolved at call time, not startup: keys rotate, says the comment, and a missing key is one provider's problem, not a reason for the whole process to die. The timeout branch covers non-streaming only: 120 seconds budgets one full round trip, while streams deliberately get none — "a long answer is not a stuck one". The final map_err files DNS, connect, and timeout failures all under Retryable: nothing has tested the request itself yet, so another route might just work.

With that skeleton, complete reads itself: send(&request, false) → a non-success status goes to classify (next section) → success means response.json() into a ChatResponse, and even "upstream sent invalid JSON" counts as Retryable. complete_stream is symmetric — just streaming = true, and a chunk-by-chunk byte stream comes back.

Retryable or Fatal: sorting failures

Think courier delivery: a wrong address fails with every company (Fatal); a warehouse backlog might clear at the next one (Retryable). Where the analogy breaks: on Fatal, jackdaw never "fixes the address and reships" — it passes the upstream verdict through untouched. This split is the only input to failover, so the boundary lives in the type:

/// How an upstream failure is classified. The router's failover decision
/// rests entirely on this split, so the boundary is drawn carefully:
///
/// * **Retryable** — network errors, timeouts, HTTP 429, HTTP 5xx. The
///   request may be perfectly fine; a different model might succeed.
/// * **Fatal** — every other 4xx. The request itself is the problem
///   (unknown model, malformed call, bad key); switching models cannot
///   help, so the gateway passes the upstream verdict through unchanged.
#[derive(Debug, Clone, thiserror::Error)]
pub enum ProviderError {
    #[error("retryable: {0}")]
    Retryable(String),
    #[error("upstream rejected the request with HTTP {status}")]
    Fatal { status: u16, body: String },
}

Read it: the two variants' shapes are the semantics — Retryable(String) carries a single sentence, since we're switching anyway; Fatal { status, body } keeps the upstream status code and body verbatim, because they go back to the caller untouched. thiserror's #[error(…)] generates Display (Chapter 11 drew its division of labor with anyhow), and the doc comment nails the boundary: network errors, timeouts, 429, 5xx → Retryable; every other 4xx means the request itself is the problem, so switching models cannot help. The decision happens in classify:

/// Read the error body once, then classify 429/5xx as retryable and
/// every other non-2xx as fatal.
async fn classify(&self, response: reqwest::Response) -> ProviderError {
    let status = response.status().as_u16();
    let body = response.text().await.unwrap_or_default();
    if status == 429 || (500..600).contains(&status) {
        // Keep a short excerpt: enough to act on, not enough to drown logs.
        let excerpt: String = body.chars().take(200).collect();
        ProviderError::Retryable(format!("HTTP {status}: {excerpt}"))
    } else {
        ProviderError::Fatal { status, body }
    }
}

Read it: start at let status = …as_u16(), and note that response.text() reads the error body first — classification needs both, and once read, the response is consumed. The if line is the heart of this chapter: 429 and 5xx become Retryable, truncated to 200 characters so logs don't drown; everything else falls to Fatal. That answers the first question from the opening: a connection timeout turns Retryable at send's map_err; an HTTP 400 reaches the else here and becomes Fatal.

SSE streams: a line exists only once it ends

The wire format for streaming, verbatim from the source comment:

data: {"choices":[{"delta":{"content":"Hel"}}]}
data: {"choices":[{"delta":{"content":"lo"}}]}
data: [DONE]

data: lines carry payloads; data: [DONE] is the closing sentinel. The transport layer does no parsing — the complete_stream comment says bytes pass through untouched, "SSE parsing belongs to whoever needs the text (ask), not to the transport". The parsing side:

/// Parse a single SSE line. Blank lines, comments and non-`data` lines
/// carry nothing jackdaw needs, so they map to `None`.
pub fn parse_sse_line(line: &str) -> Option<SseData> {
    // Accept both "data: x" and "data:x" spellings.
    let payload = line.strip_prefix("data:")?.trim_start();
    if payload == "[DONE]" {
        return Some(SseData::Done);
    }
    serde_json::from_str(payload).ok().map(SseData::Payload)
}

/// The delta text inside one stream chunk: `choices[0].delta.content`.
pub fn delta_content(payload: &Value) -> Option<&str> {
    payload
        .pointer("/choices/0/delta/content")
        .and_then(Value::as_str)
}

Read it: strip_prefix("data:")? goes first — a non-data line returns None right there (? operating on Option, Chapter 6), and the comment says both data: x and data:x spellings are accepted. [DONE] maps to SseData::Done; everything else goes to serde_json, and a parse failure becomes None via .ok() — a bad line is simply dropped. delta_content walks the JSON Pointer /choices/0/delta/content straight to the delta text, returning None when it's absent (the first chunk may carry only a role). Network chunks don't respect line boundaries — a split can land mid UTF-8 character, and the third opening question lands here:

#[derive(Default)]
pub struct SseAggregator {
    /// Complete-but-unparsed bytes of the current line.
    pending: Vec<u8>,
    /// Assistant text assembled so far.
    text: String,
    /// How much of `text` has already been emitted by [`Self::take_recent`].
    emitted: usize,
}

impl SseAggregator {
    /// Feed one raw chunk from the stream.
    pub fn push(&mut self, chunk: &[u8]) {
        self.pending.extend_from_slice(chunk);
        while let Some(newline) = self.pending.iter().position(|byte| *byte == b'\n') {
            let line_bytes: Vec<u8> = self.pending.drain(..=newline).collect();
            let line = String::from_utf8_lossy(&line_bytes);
            self.consume(line.trim_end_matches(['\r', '\n']));
        }
    }
    // ...
    /// Everything assembled since the previous call — this is what lets
    /// `ask` print deltas live while `--no-stream` waits for the whole text.
    pub fn take_recent(&mut self) -> String {
        let start = self.emitted;
        self.emitted = self.text.len();
        self.text[start..].to_string()
    }
    // ...
}

Read it: push appends the new chunk to pending, then while let hunts for newlines — drain(..=newline) takes the whole line plus the newline, converts it to a string, trims the \r\n tail, and hands it to consume, which parses it and appends the delta text to text; the newline byte never appears inside a multi-byte UTF-8 character, so buffering until a newline also guarantees a line never breaks mid-character. take_recent uses emitted to remember what it already handed out and returns only the new part — that's how ask prints as chunks arrive, while --no-stream waits for all of text().

INFO

One simplification: the full SSE spec defines event:, retry:, and other lines, but these endpoints never send them. The source comment states that jackdaw reads only data: lines — "that is all these endpoints send".

WARNING

If the stream breaks halfway, the error arrives inside the stream as Retryable("stream interrupted…") — and the gateway does not switch to the next model: part of the answer may already be forwarded, and resending would duplicate it. That is the trait doc's "failover is impossible" line; the gateway-side semantics get the full treatment in Chapter 14.

Wrapping up

A timeout turns Retryable at send's map_err; a 400 turns Fatal in classify's else. The key is read at call time; streams get no total timeout. SseAggregator::push holds bytes until a newline proves the line complete. Next chapter wires the candidate list and this two-way split into the group/<id> failover loop.

Self-Check

1. Adding a provider that speaks the Anthropic protocol: which files change, and does the ProviderError classification change with it?
2. parse_sse_line receives, in order, "data: [DONE]", "event: ping", "data: not-json" — what does each call return?
3. Why does a missing environment variable count as Retryable rather than Fatal?

Answers

1. A new module with impl Provider for …; provider.rs and openai.rs stay untouched. The classification asks "is the problem the request or the upstream?", which is protocol-independent — no change needed. What does change is assembly: build_providers currently builds one OpenAiProvider per [providers.X] table, so telling protocols apart needs a type marker in the config — the entry point for Chapter 16's extension exercise "Anthropic protocol translation".
2. Some(SseData::Done), None, None: the second line lacks the data: prefix; the third has the prefix but isn't valid JSON, so .ok() folds the failure into None. The tests parses_the_done_sentinel and ignores_non_data_lines assert exactly these cases.
3. What's missing is this one provider's key: inside a routing group, the next member may read a different, properly set environment variable, so switching might just succeed — which is precisely what Retryable means. The message also states that the config stores the variable's name, so whoever reads the log knows what to export.