newt-core 0.8.0

//! The OpenAI-compatible chat-completions SSE wire.
//!
//! The parsing half of #123, kept separate from the loop for the same reason
//! [`super::anthropic_wire`] is: a wire is a pure function over bytes, and the
//! cheapest honest test for one is a unit test with no HTTP in it.
//!
//! # Why this is not the Anthropic accumulator with different names
//!
//! The two protocols agree on FRAMING and on nothing else. Both are
//! `text/event-stream`, both put one JSON payload per `data:` line, and both
//! can split a line across two `reqwest` chunks — so both need the same
//! rolling line buffer, and that part is deliberately identical.
//!
//! Everything above the frame differs:
//!
//! | | Anthropic | OpenAI-compatible |
//! |---|---|---|
//! | terminator | a `message_stop` EVENT | the literal `data: [DONE]`, which is not JSON |
//! | text | `delta.text` under `content_block_delta` | `choices[0].delta.content` |
//! | reasoning | `delta.thinking` | `delta.reasoning_content`, or `delta.reasoning` |
//! | usage | split across `message_start` and `message_delta` | one object on a chunk whose `choices` is EMPTY |
//!
//! `[DONE]` is the one that matters most: it is a sentinel, not a document,
//! and a parser that hands every `data:` payload to `serde_json` drops the end
//! of the stream on the floor. Nothing else in this repository has ever parsed
//! it — the Anthropic wire terminates on an event, and `newt-mcp-client`'s
//! parser is whole-body rather than incremental — so this is new code rather
//! than a copy wearing a new name.

#[path = "openai_sse_response.rs"]
mod response;
use response::StrictResponse;

/// Whether a complete observed server error is present in an error chain.
/// Prefer that evidence over a later body-read failure; parsing failures alone
/// do not establish that the server rejected the request.
pub fn is_provider_error(error: &anyhow::Error) -> bool {
    error
        .chain()
        .any(|cause| cause.is::<response::ProviderError>())
}

/// The identity field (`response ID` / `response model`) that changed within
/// one response (#2334). The check stays a rejection; the retry policy may spend
/// one fresh attempt on it before any tool runs.
pub fn changed_identity(error: &anyhow::Error) -> Option<&'static str> {
    error
        .chain()
        .find_map(|cause| cause.downcast_ref::<response::IdentityChanged>())
        .map(response::IdentityChanged::field)
}

/// Interpret one provider response without replacing its observed wire bytes.
///
/// SSE uses the display parser's framing plus strict completion validation:
/// DONE, finish reason, stable complete tool calls, and intact JSON frames.
/// Complete JSON from a server ignoring `stream: true` is decoded once as-is.
/// The caller must record the original bytes before interpreting this value.
pub fn decode_response(bytes: &[u8]) -> anyhow::Result<serde_json::Value> {
    decode_response_reporting_usage(bytes, &mut None)
}

/// [`decode_response`], also leaving in `usage` what the stream reported. A
/// response strict validation rejects was still generated and billed (#2313).
pub(crate) fn decode_response_reporting_usage(
    bytes: &[u8],
    usage: &mut Option<crate::TokenUsage>,
) -> anyhow::Result<serde_json::Value> {
    // `Some(true)`: the bytes end inside a character, as a cut body can.
    let (body, invalid_utf8) = match std::str::from_utf8(bytes) {
        Ok(body) => (body, None),
        Err(error) => (
            std::str::from_utf8(&bytes[..error.valid_up_to()])?,
            Some(error.error_len().is_none()),
        ),
    };
    if body.trim_start().starts_with(['{', '[']) {
        let json: serde_json::Value = serde_json::from_str(body)?;
        return if response::is_error(&json) {
            Err(response::provider_error(&json))
        } else {
            anyhow::ensure!(invalid_utf8.is_none(), "OpenAI response has invalid UTF-8");
            Ok(json)
        };
    }
    let mut accumulator = SseAccumulator {
        strict: Some(StrictResponse::default()),
        ..Default::default()
    };
    let _ = accumulator.feed(body);
    let complete_lines_sound = !accumulator
        .strict
        .as_ref()
        .expect("decode_response enables strict validation")
        .has_problem();
    accumulator.flush_line();
    *usage = accumulator.round.usage;
    let mut strict = accumulator
        .strict
        .take()
        .expect("decode_response enables strict validation");
    if invalid_utf8.is_some() {
        strict.problem("OpenAI stream has invalid UTF-8");
    }
    // Ended before `[DONE]` with nothing wrong but a torn last line or
    // character: the body was cut, not malformed (#2318).
    let cut = !accumulator.round.done && complete_lines_sound && invalid_utf8 != Some(false);
    strict.finish(accumulator.round, cut)
}

/// The outcome class of a strict-decode rejection, by its cause (#2318). `None`
/// for a provider's own error, whose class follows its text as an HTTP error
/// body's does. A cut body, and a response whose identity changed (#2334), are
/// transport failures; a shape this decoder does not support is ours; every
/// other rejection is a defect in the model's output.
pub(crate) fn rejection_class(error: &anyhow::Error) -> Option<super::observability::ErrorClass> {
    use super::observability::ErrorClass;
    let caused_by = |is: fn(&(dyn std::error::Error + 'static)) -> bool| error.chain().any(is);
    if is_provider_error(error) {
        None
    } else if caused_by(|c| c.is::<response::StreamCut>() || c.is::<response::IdentityChanged>()) {
        Some(ErrorClass::Transport)
    } else if caused_by(|c| c.is::<response::Unsupported>()) {
        Some(ErrorClass::Harness)
    } else {
        Some(ErrorClass::Model)
    }
}

/// One display-affecting thing a chunk produced.
///
/// Two arms and no more, because those are the two channels a turn shows: the
/// answer, and the thinking above it. A `finish_reason`, a role-only opening
/// delta and an empty string change nothing on screen and so produce nothing.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum StreamAction {
    /// A piece of the answer.
    TextDelta(String),
    /// A piece of the model's reasoning, for the trickle above the answer.
    ReasoningDelta(String),
}

/// What a completed streaming round amounts to.
#[derive(Debug, Default, Clone, PartialEq, Eq)]
pub struct OpenAiStreamRound {
    /// The whole answer, concatenated — the same text the non-streaming wire
    /// would have returned in `message.content`.
    pub text: String,
    /// The whole reasoning body, concatenated. Empty when the model sent none.
    pub reasoning: String,
    /// Token usage, when the server sent it. Requires
    /// `stream_options.include_usage` on the request; a server that ignores
    /// that option leaves this `None` rather than inventing a number.
    pub usage: Option<crate::TokenUsage>,
    /// True once `data: [DONE]` arrived — the stream ENDED rather than being
    /// cut. A caller that streamed a partial answer and never saw this knows
    /// the difference between "the model finished" and "the socket did".
    pub done: bool,
}

/// Incremental parser over an OpenAI-compatible `text/event-stream` body.
///
/// Feed it raw chunks in arrival order; it returns the display actions each
/// chunk completed. It never blocks, never allocates per byte, and holds only
/// the bytes of one unterminated line.
#[derive(Debug, Default)]
pub struct SseAccumulator {
    /// The tail of a `data:` line that arrived without its newline. An SSE
    /// line routinely straddles a chunk boundary; a per-chunk `lines()` split
    /// silently drops the halves.
    line_buf: String,
    round: OpenAiStreamRound,
    strict: Option<StrictResponse>,
}

impl SseAccumulator {
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    /// True once `data: [DONE]` has been seen.
    #[must_use]
    pub fn is_done(&self) -> bool {
        self.round.done
    }

    /// Feed one raw HTTP chunk; returns the display actions it completed.
    pub fn feed(&mut self, chunk: &str) -> Vec<StreamAction> {
        let mut actions = Vec::new();
        self.line_buf.push_str(chunk);
        while let Some(pos) = self.line_buf.find('\n') {
            let line: String = self.line_buf.drain(..=pos).collect();
            self.apply_line(line.trim_end_matches(['\n', '\r']), &mut actions);
        }
        actions
    }

    /// Flush an unterminated final line and take the round.
    ///
    /// A server that ends the body without a trailing newline still gets its
    /// last event applied, which is how a short reply can otherwise vanish.
    #[must_use]
    pub fn finish(mut self) -> OpenAiStreamRound {
        self.flush_line();
        self.round
    }

    fn flush_line(&mut self) {
        if !self.line_buf.is_empty() {
            let line = std::mem::take(&mut self.line_buf);
            let mut actions = Vec::new();
            self.apply_line(line.trim_end_matches(['\n', '\r']), &mut actions);
        }
    }

    fn apply_line(&mut self, line: &str, actions: &mut Vec<StreamAction>) {
        if let Some(strict) = self.strict.as_mut() {
            if line.is_empty() {
                strict.event(None);
            } else if let Some(event) = line.strip_prefix("event:") {
                strict.event(Some(event.trim()));
            }
        }
        // Display consumes data only. Strict completion validation also tracks
        // explicit error events above; comments and other metadata are ignored.
        let Some(data) = line.strip_prefix("data:") else {
            return;
        };
        let data = data.trim();

        // The sentinel, checked BEFORE parsing. `[DONE]` is not JSON, so a
        // parser that reaches for serde first throws away the one token that
        // says the stream ended on purpose.
        if self.round.done {
            if let Some(strict) = self.strict.as_mut() {
                strict.problem("OpenAI stream sent data after [DONE]");
            }
        }
        if data == "[DONE]" {
            self.round.done = true;
            return;
        }

        let Ok(json) = serde_json::from_str::<serde_json::Value>(data) else {
            if let Some(strict) = self.strict.as_mut() {
                strict.problem("OpenAI stream has a malformed data frame");
            }
            return;
        };
        if let Some(strict) = self.strict.as_mut() {
            strict.observe(&json);
        }

        // Usage rides its own chunk when `stream_options.include_usage` is set:
        // `choices` is empty and `usage` is populated. Read it before touching
        // choices so an empty-choices chunk is not mistaken for a dead event.
        if let Some(usage) = super::trim::openai_usage(&json["usage"]) {
            self.round.usage = Some(usage);
        }

        let delta = &json["choices"][0]["delta"];

        // Reasoning first: it precedes the answer, and two spellings are in the
        // wild — `reasoning_content` (DeepSeek, vLLM, llama.cpp) and
        // `reasoning` (several gateways). Neither is in the OpenAI spec, so
        // both are accepted and neither is required.
        for key in ["reasoning_content", "reasoning"] {
            if let Some(text) = delta[key].as_str() {
                if !text.is_empty() {
                    self.round.reasoning.push_str(text);
                    actions.push(StreamAction::ReasoningDelta(text.to_string()));
                }
            }
        }

        if let Some(text) = delta["content"].as_str() {
            // The opening delta is `{"role":"assistant"}` with no content, and
            // keep-alive chunks carry `content: ""`. Both are silence.
            if !text.is_empty() {
                self.round.text.push_str(text);
                actions.push(StreamAction::TextDelta(text.to_string()));
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    /// Frame payloads the way a server does: one `data:` line per event,
    /// blank-line separated.
    fn body(frames: &[&str]) -> String {
        frames
            .iter()
            .map(|f| format!("data: {f}\n\n"))
            .collect::<String>()
    }

    fn feed_all(acc: &mut SseAccumulator, s: &str) -> Vec<StreamAction> {
        acc.feed(s)
    }

    #[test]
    fn deltas_accumulate_into_one_answer() {
        let mut acc = SseAccumulator::new();
        let actions = feed_all(
            &mut acc,
            &body(&[
                r#"{"choices":[{"delta":{"role":"assistant"}}]}"#,
                r#"{"choices":[{"delta":{"content":"Hel"}}]}"#,
                r#"{"choices":[{"delta":{"content":"lo"}}]}"#,
                "[DONE]",
            ]),
        );
        assert_eq!(
            actions,
            vec![
                StreamAction::TextDelta("Hel".into()),
                StreamAction::TextDelta("lo".into())
            ],
            "the role-only opening delta must produce nothing"
        );
        assert!(acc.is_done());
        let round = acc.finish();
        assert_eq!(round.text, "Hello");
        assert!(round.done);
    }

    /// The reason this parser holds a buffer at all. `reqwest` splits where the
    /// socket did, not where the protocol did.
    #[test]
    fn an_event_split_across_two_chunks_is_not_lost() {
        let whole = body(&[r#"{"choices":[{"delta":{"content":"split me"}}]}"#]);
        let cut = whole.len() / 2;

        let mut acc = SseAccumulator::new();
        let first = acc.feed(&whole[..cut]);
        assert!(first.is_empty(), "half an event is not an event: {first:?}");
        let second = acc.feed(&whole[cut..]);
        assert_eq!(second, vec![StreamAction::TextDelta("split me".into())]);
        assert_eq!(acc.finish().text, "split me");
    }

    /// Byte-at-a-time is the adversarial form of the same property.
    #[test]
    fn one_byte_at_a_time_produces_the_same_answer() {
        let whole = body(&[
            r#"{"choices":[{"delta":{"content":"a"}}]}"#,
            r#"{"choices":[{"delta":{"content":"b"}}]}"#,
            "[DONE]",
        ]);
        let mut acc = SseAccumulator::new();
        let mut actions = Vec::new();
        for ch in whole.chars() {
            actions.extend(acc.feed(&ch.to_string()));
        }
        assert_eq!(
            actions,
            vec![
                StreamAction::TextDelta("a".into()),
                StreamAction::TextDelta("b".into())
            ]
        );
        assert!(acc.is_done());
    }

    /// `[DONE]` is a sentinel, not a document. A parser that hands it to serde
    /// first never learns the stream ended.
    #[test]
    fn the_done_sentinel_is_recognized_and_is_not_json() {
        assert!(
            serde_json::from_str::<serde_json::Value>("[DONE]").is_err(),
            "if this ever parses, the check order in apply_line stops mattering"
        );
        let mut acc = SseAccumulator::new();
        acc.feed("data: [DONE]\n\n");
        assert!(acc.is_done());
    }

    /// The usage chunk carries an EMPTY `choices` array. Reading choices first
    /// and bailing would drop the only token counts the wire ever sends.
    #[test]
    fn usage_arrives_on_a_chunk_with_no_choices() {
        let mut acc = SseAccumulator::new();
        acc.feed(&body(&[
            r#"{"choices":[{"delta":{"content":"hi"}}]}"#,
            r#"{"choices":[],"usage":{"prompt_tokens":11,"completion_tokens":22}}"#,
            "[DONE]",
        ]));
        let round = acc.finish();
        assert_eq!(round.text, "hi");
        let usage = round.usage.expect("usage chunk was sent");
        assert_eq!(usage.input_tokens, 11);
        assert_eq!(usage.output_tokens, 22);
    }

    /// A server that ignores `stream_options` leaves usage absent. Absent is a
    /// fact; zero would be a fabrication.
    #[test]
    fn no_usage_chunk_leaves_usage_none() {
        let mut acc = SseAccumulator::new();
        acc.feed(&body(&[
            r#"{"choices":[{"delta":{"content":"hi"}}]}"#,
            "[DONE]",
        ]));
        assert!(acc.finish().usage.is_none());
    }

    #[test]
    fn both_reasoning_spellings_reach_the_trickle() {
        for key in ["reasoning_content", "reasoning"] {
            let mut acc = SseAccumulator::new();
            let actions = acc.feed(&body(&[&format!(
                r#"{{"choices":[{{"delta":{{"{key}":"because"}}}}]}}"#
            )]));
            assert_eq!(
                actions,
                vec![StreamAction::ReasoningDelta("because".into())],
                "spelling {key} was not recognized"
            );
            assert_eq!(acc.finish().reasoning, "because");
        }
    }

    /// Reasoning and answer are different channels and must not merge — the
    /// whole point of the trickle is that thinking renders above the answer.
    #[test]
    fn reasoning_never_lands_in_the_answer() {
        let mut acc = SseAccumulator::new();
        acc.feed(&body(&[
            r#"{"choices":[{"delta":{"reasoning_content":"thinking"}}]}"#,
            r#"{"choices":[{"delta":{"content":"answer"}}]}"#,
            "[DONE]",
        ]));
        let round = acc.finish();
        assert_eq!(round.text, "answer");
        assert_eq!(round.reasoning, "thinking");
    }

    /// A body that ends without its final newline still gets its last event.
    #[test]
    fn an_unterminated_final_line_is_flushed_by_finish() {
        let mut acc = SseAccumulator::new();
        let actions = acc.feed(r#"data: {"choices":[{"delta":{"content":"tail"}}]}"#);
        assert!(actions.is_empty(), "no newline yet, so no completed line");
        assert_eq!(acc.finish().text, "tail");
    }

    /// Garbage does not panic and does not become the answer.
    #[test]
    fn unparseable_and_foreign_lines_are_ignored() {
        let mut acc = SseAccumulator::new();
        let actions = acc.feed(concat!(
            "event: message\n",
            ": a comment\n",
            "data: {not json\n",
            "\n",
            "data: {\"choices\":[{\"delta\":{\"content\":\"ok\"}}]}\n\n",
        ));
        assert_eq!(actions, vec![StreamAction::TextDelta("ok".into())]);
        assert_eq!(acc.finish().text, "ok");
    }

    /// A cut stream is distinguishable from a finished one. The caller needs
    /// that to decide whether to fall back to the probe content.
    #[test]
    fn a_cut_stream_is_not_done() {
        let mut acc = SseAccumulator::new();
        acc.feed(&body(&[r#"{"choices":[{"delta":{"content":"partial"}}]}"#]));
        assert!(!acc.is_done());
        let round = acc.finish();
        assert_eq!(round.text, "partial");
        assert!(!round.done, "no [DONE] arrived, so the stream was cut");
    }
}

#[cfg(test)]
#[path = "openai_sse_strict_tests.rs"]
mod strict_tests;