diff --git a/crates/llm/Cargo.toml b/crates/llm/Cargo.toml index 261e6118..2cdd2ae0 100644 --- a/crates/llm/Cargo.toml +++ b/crates/llm/Cargo.toml @@ -17,7 +17,7 @@ bytes.workspace = true serde.workspace = true serde_json.workspace = true thiserror.workspace = true -tokio = { version = "=1.53.1", default-features = false, features = ["rt", "fs", "io-util", "sync", "time"] } +tokio = { version = "=1.53.1", default-features = false, features = ["rt", "fs", "io-util", "macros", "sync", "time"] } oauth2 = { version = "5.0.0", default-features = false, features = ["timing-resistant-secret-traits"] } sha2.workspace = true jsonwebtoken = { version = "11.1.0", default-features = false, features = ["aws_lc_rs"] } diff --git a/crates/llm/README.org b/crates/llm/README.org index f7cc09c4..685e19d0 100644 --- a/crates/llm/README.org +++ b/crates/llm/README.org @@ -65,6 +65,13 @@ Use =Client::embed= with an explicit OpenAI-compatible embeddings URL for batches of text or token IDs. The client returns one float vector per input in input order and checks the returned dimensions. +For Anthropic Messages, a buffered assistant message carries its ordered +=anthropic_output= content blocks, including signed or redacted thinking. For +streaming turns, attach the terminal event's =anthropic_output= to the assistant +message before continuing. Weaver checks that the visible text and tool calls +still match those blocks. Debug formatting redacts the block contents. The +caller must retain this data with the conversation; Weaver does not compress it. + The client distinguishes rate limits, context length errors and temporary upstream failures. For HTTP rate limits it exposes integer seconds from =Retry-After= when available. These errors inform caller decisions. @@ -186,7 +193,16 @@ env = "OPENAI_API_KEY" Supported protocol names are =openai-chat-completions=, =openai-responses= and =anthropic-messages=. Both the profile and configuration path come from -trusted Harness startup settings. Supply the chosen model explicitly. +trusted Harness startup settings. For an Anthropic direct profile, explicit +=model= and =endpoint= win; missing fields use =ANTHROPIC_MODEL= and +=ANTHROPIC_BASE_URL=. The base URL is expanded to =/v1/messages=, preserving +an existing =/v1= or =/v1/messages= suffix. An omitted credential uses +=ANTHROPIC_API_KEY=, which is included in =credential_environment()=. Other +protocols still require explicit model, endpoint and credential. Environment +values are resolved at =prepare= and the resulting profile is immutable. +An Anthropic-compatible environment-backed profile with an optional OpenAI +alternative is provided in [[file:example.toml][example.toml]]; callers select =anthropic= +when invoking =prepare=. The client never compresses conversation context. The optional profile field =authentication = "bearer"= selects an explicit Authorization Bearer header. Its default, =provider_default=, uses Bearer for OpenAI and =x-api-key= for Anthropic API keys. Anthropic =sk-ant-oat= access @@ -665,3 +681,15 @@ These finite models and sampled traces check the declared abstraction; they are not an exhaustive proof of the Rust implementation or external IO atomicity. All five models run through the existing CI plan. Rust properties run in the normal client Cargo tests, with no separate test runner or cold dependency install. + +** Anthropic thinking controls and stream events + +=Options::anthropic_thinking= selects adaptive, manual budget or disabled +thinking, and =anthropic_effort= selects output effort. Availability varies by +model and provider. Manual thinking requires at least 1024 tokens and a larger +=max_tokens=; DeepSeek documents that it ignores the manual budget. Other +protocols reject these Anthropic-specific options. Omitted controls defer to the +provider. =TurnEvent::Thinking= yields provider-returned thinking separately +from final-answer text. Its debug representation omits the content; callers +choose whether to display or store it. Terminal =anthropic_output= remains +necessary for lossless continuation, including signatures and redacted blocks. diff --git a/crates/llm/benches/gateway.rs b/crates/llm/benches/gateway.rs index 3cf9e7b3..87e8c9a7 100644 --- a/crates/llm/benches/gateway.rs +++ b/crates/llm/benches/gateway.rs @@ -15,6 +15,7 @@ fn request_encoding(c: &mut Criterion) { tool_call_id: None, tool_calls: Vec::new(), responses_output: None, + anthropic_output: None, }]; let request = Request { model: "gpt-example", diff --git a/crates/llm/example.toml b/crates/llm/example.toml new file mode 100644 index 00000000..e7944b1a --- /dev/null +++ b/crates/llm/example.toml @@ -0,0 +1,28 @@ +# SPDX-FileCopyrightText: 2026 hardenedlinux Linux Team and Contributors +# +# SPDX-License-Identifier: LGPL-2.1-or-later AND Apache-2.0 + +# Anthropic-compatible route. Missing fields use ANTHROPIC_MODEL, +# ANTHROPIC_BASE_URL and ANTHROPIC_API_KEY at profile preparation. +[profiles.anthropic] +kind = "direct" +protocol = "anthropic-messages" +timeout_ms = 120000 +max_response_bytes = 8388608 + +# Explicit settings take precedence over environment variables. +# model = "deepseek-v4-pro[1m]" +# endpoint = "https://your-provider.example/v1/messages" +# [profiles.anthropic.credential] +# env = "ANTHROPIC_API_KEY" + +# Optional OpenAI profile: +# [profiles.openai] +# kind = "direct" +# protocol = "openai-responses" +# endpoint = "https://api.openai.com/v1/responses" +# model = "gpt-4.1-mini" +# timeout_ms = 120000 +# max_response_bytes = 8388608 +# [profiles.openai.credential] +# env = "OPENAI_API_KEY" diff --git a/crates/llm/src/anthropic_output.rs b/crates/llm/src/anthropic_output.rs new file mode 100644 index 00000000..bd3f017e --- /dev/null +++ b/crates/llm/src/anthropic_output.rs @@ -0,0 +1,99 @@ +// SPDX-FileCopyrightText: 2026 hardenedlinux Linux Team and Contributors +// +// SPDX-License-Identifier: LGPL-2.1-or-later AND Apache-2.0 + +//! Opaque Anthropic assistant output for lossless continuation. + +use crate::{Error, Message, Role, ToolCall}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use std::fmt; + +/// Ordered provider content blocks, including signed or redacted thinking. +#[derive(Clone, Deserialize, Eq, PartialEq, Serialize)] +#[serde(transparent)] +pub struct AnthropicOutput(pub(crate) Vec); + +impl fmt::Debug for AnthropicOutput { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("AnthropicOutput") + .field("block_count", &self.0.len()) + .finish() + } +} + +impl AnthropicOutput { + pub(crate) fn parse(blocks: Vec) -> Result<(Self, String, Vec), Error> { + let (text, calls) = Self::extract(&blocks)?; + Ok((Self(blocks), text, calls)) + } + + fn extract(blocks: &[Value]) -> Result<(String, Vec), Error> { + let mut text = String::new(); + let mut calls = Vec::new(); + for block in blocks { + match block["type"].as_str() { + Some("text") => text.push_str(block["text"].as_str().ok_or(Error::Response)?), + Some("thinking") + if block["thinking"].is_string() + && (block.get("signature").is_none() || block["signature"].is_string()) => { + } + Some("redacted_thinking") if block["data"].is_string() => {} + Some("tool_use") => { + let id = block["id"] + .as_str() + .filter(|s| !s.is_empty()) + .ok_or(Error::Response)?; + let name = block["name"] + .as_str() + .filter(|s| !s.is_empty()) + .ok_or(Error::Response)?; + let arguments = block["input"].as_object().ok_or(Error::Response)?; + calls.push(ToolCall { + id: id.into(), + name: name.into(), + arguments: Value::Object(arguments.clone()), + }); + } + _ => return Err(Error::Response), + } + } + Ok((text, calls)) + } + + /// Check observed answer and thinking against the retained provider blocks. + /// + /// This checks consistency, not the authenticity of a provider signature. + /// Redacted thinking remains opaque and is preserved in the continuation. + pub fn matches_observed(&self, text: &str, thinking: &str) -> bool { + let Ok((output_text, calls)) = Self::extract(&self.0) else { + return false; + }; + if output_text != text || !calls.is_empty() { + return false; + } + let mut remaining = thinking; + for block in &self.0 { + if block["type"].as_str() == Some("thinking") { + let Some(part) = block["thinking"].as_str() else { + return false; + }; + let Some(rest) = remaining.strip_prefix(part) else { + return false; + }; + remaining = rest; + } + } + remaining.is_empty() + } + + pub(crate) fn matches(&self, message: &Message) -> bool { + if message.role != Role::Assistant || !message.images.is_empty() { + return false; + } + match Self::extract(&self.0) { + Ok((text, calls)) => text == message.content && calls == message.tool_calls, + Err(_) => false, + } + } +} diff --git a/crates/llm/src/client.rs b/crates/llm/src/client.rs index b1b3b8e5..478ead65 100644 --- a/crates/llm/src/client.rs +++ b/crates/llm/src/client.rs @@ -62,6 +62,49 @@ impl Endpoint { }) } + /// Fingerprint the configured destination and credential for continuation binding. + /// + /// The fingerprint is sensitive: it permits guessing low-entropy credentials. + /// It is not proof of server identity or account identity. Credential rotation + /// changes it, and callers must not silently rebind existing continuations. + pub fn continuation_binding(&self, credential: &str) -> [u8; 32] { + use sha2::{Digest, Sha256}; + let mut hasher = Sha256::new(); + fn part(hasher: &mut Sha256, value: &[u8]) { + hasher.update((value.len() as u64).to_be_bytes()); + hasher.update(value); + } + part(&mut hasher, b"weaver.continuation.destination.v1"); + part(&mut hasher, self.url.as_str().as_bytes()); + part( + &mut hasher, + match self.protocol { + Protocol::OpenAi => b"openai-chat", + Protocol::OpenAiResponses => b"openai-responses", + Protocol::Anthropic => b"anthropic-messages", + }, + ); + part( + &mut hasher, + match self.authentication { + Authentication::ProviderDefault => b"provider-default", + Authentication::Bearer => b"bearer", + }, + ); + part(&mut hasher, credential.as_bytes()); + let mut headers: Vec<_> = self + .headers + .iter() + .map(|(name, value)| (name.as_str(), value.as_bytes())) + .collect(); + headers.sort_by(|left, right| left.0.cmp(right.0)); + for (name, value) in headers { + part(&mut hasher, name.as_bytes()); + part(&mut hasher, value); + } + hasher.finalize().into() + } + /// Bind an explicit authentication scheme without changing the wire protocol. pub fn with_authentication(mut self, authentication: Authentication) -> Self { self.authentication = authentication; diff --git a/crates/llm/src/config.rs b/crates/llm/src/config.rs index 232942f2..b47daf7e 100644 --- a/crates/llm/src/config.rs +++ b/crates/llm/src/config.rs @@ -46,11 +46,11 @@ enum Profile { protocol: WireProtocol, #[serde(default)] authentication: Authentication, - endpoint: String, - model: String, + endpoint: Option, + model: Option, timeout_ms: u64, max_response_bytes: usize, - credential: Credential, + credential: Option, }, Delegated { endpoint: String, @@ -128,11 +128,18 @@ impl Configuration { let mut names = Vec::new(); match profile { Profile::Direct { - credential: Credential::Static(source), + credential: Some(Credential::Static(source)), .. } => { names.extend(source.environment()); } + Profile::Direct { + protocol: WireProtocol::Anthropic, + credential: None, + .. + } => { + names.push("ANTHROPIC_API_KEY"); + } Profile::Delegated { exchange, operation, @@ -239,12 +246,27 @@ impl Configuration { else { return Err(Error::Configuration); }; + let (endpoint, model) = resolve_direct_target( + (*protocol).into(), + endpoint.as_deref(), + model.as_deref(), + |name| std::env::var(name).ok(), + )?; let mut endpoint = - Endpoint::new((*protocol).into(), endpoint)?.with_authentication(*authentication); + Endpoint::new((*protocol).into(), &endpoint)?.with_authentication(*authentication); if model.trim().is_empty() || model.len() > 1024 { return Err(Error::Configuration); } let client = Client::new(Duration::from_millis(*timeout_ms), *max_response_bytes)?; + let default_credential = matches!(protocol, WireProtocol::Anthropic).then(|| { + Credential::Static(SecretReference::Environment { + env: "ANTHROPIC_API_KEY".into(), + }) + }); + let credential = credential + .as_ref() + .or(default_credential.as_ref()) + .ok_or(Error::Configuration)?; let credential = match credential { Credential::Static(source) => { PreparedCredential::static_value(source.resolve(&self.directory).await?)? @@ -261,7 +283,7 @@ impl Configuration { oauth.as_ref().clone(), &mut endpoint, (*protocol).into(), - model, + &model, &client, ) .await? @@ -271,7 +293,7 @@ impl Configuration { openai.clone().into_oauth(), &mut endpoint, (*protocol).into(), - model, + &model, &client, ) .await? @@ -281,21 +303,21 @@ impl Configuration { copilot.clone().into_oauth(), &mut endpoint, (*protocol).into(), - model, + &model, &client, ) .await? } Credential::Oidc { oidc } => { let oauth = discover_oidc(oidc.clone(), client.http.clone()).await?; - self.prepare_oauth(oauth, &mut endpoint, (*protocol).into(), model, &client) + self.prepare_oauth(oauth, &mut endpoint, (*protocol).into(), &model, &client) .await? } }; Ok(PreparedProfile { client, endpoint, - model: model.clone(), + model, credential, }) } @@ -364,6 +386,51 @@ impl Configuration { } } +/// Resolve an explicit model and endpoint, filling absent Anthropic values from +/// the supplied environment. Explicit configuration always takes precedence. +/// +/// # Errors +/// Returns a configuration error when a required value is missing or invalid. +pub fn resolve_direct_target( + protocol: Protocol, + endpoint: Option<&str>, + model: Option<&str>, + read_env: impl Fn(&str) -> Option, +) -> Result<(String, String), Error> { + let model = match model { + Some(value) => value.to_owned(), + None if protocol == Protocol::Anthropic => { + read_env("ANTHROPIC_MODEL").ok_or(Error::Configuration)? + } + None => return Err(Error::Configuration), + }; + let endpoint = match endpoint { + Some(value) => value.to_owned(), + None if protocol == Protocol::Anthropic => { + let base = read_env("ANTHROPIC_BASE_URL").ok_or(Error::Configuration)?; + if base.len() > 4096 { + return Err(Error::Configuration); + } + let mut url = reqwest::Url::parse(&base).map_err(|_| Error::Configuration)?; + if url.query().is_some() || url.fragment().is_some() { + return Err(Error::Configuration); + } + let path = url.path().trim_end_matches('/'); + let path = if path.ends_with("/v1/messages") { + path.to_owned() + } else if path.ends_with("/v1") { + format!("{path}/messages") + } else { + format!("{path}/v1/messages") + }; + url.set_path(&path); + url.to_string() + } + None => return Err(Error::Configuration), + }; + Ok((endpoint, model)) +} + /// An immutable connection with a pooled client and a private credential. /// Reload trusted configuration to rotate or change its destination. pub struct PreparedProfile { @@ -409,6 +476,11 @@ impl std::fmt::Debug for PreparedProfile { } impl PreparedProfile { + /// Model identity bound to this prepared connection. + pub fn model(&self) -> &str { + &self.model + } + /// Access the shared login, refresh and logout owner of an OAuth profile. pub fn oauth(&self) -> Option<&OAuthSession> { match &self.credential { diff --git a/crates/llm/src/lib.rs b/crates/llm/src/lib.rs index 9e4a03f9..7c4e8cd3 100644 --- a/crates/llm/src/lib.rs +++ b/crates/llm/src/lib.rs @@ -3,6 +3,7 @@ // SPDX-License-Identifier: LGPL-2.1-or-later AND Apache-2.0 //! Direct, asynchronous model calls with caller-selected endpoints. +mod anthropic_output; mod client; mod config; mod embeddings; @@ -15,19 +16,20 @@ mod protocol; mod responses; mod stream; +pub use anthropic_output::AnthropicOutput; pub use client::{Authentication, Client, Endpoint, EventStream}; -pub use config::{Configuration, PreparedProfile}; +pub use config::{Configuration, PreparedProfile, resolve_direct_target}; pub use embeddings::{EmbeddingInput, EmbeddingRequest, EmbeddingUsage, Embeddings}; pub use error::Error; pub use images::{ImageInput, ImageMediaType}; pub use operation::PreparedOperation; pub use protocol::{ - Completion, CompletionChoice, Message, Options, Protocol, Request, ResponseFormat, Role, Tool, - ToolCall, ToolChoice, Usage, + AnthropicEffort, AnthropicSettings, AnthropicThinking, Completion, CompletionChoice, Message, + Options, Protocol, Request, ResponseFormat, Role, Tool, ToolCall, ToolChoice, Usage, }; pub use responses::ResponsesOutput; -pub use stream::{TurnEvent, TurnIdentity, TurnStream}; +pub use stream::{ThinkingDelta, TurnEvent, TurnIdentity, TurnStream}; #[cfg(test)] asp_rust::asp_rust_cargo_test_gate!(mode = deny, config = asp_rust::default_asp_rust_config()); diff --git a/crates/llm/src/protocol.rs b/crates/llm/src/protocol.rs index f3217f82..9614f3fa 100644 --- a/crates/llm/src/protocol.rs +++ b/crates/llm/src/protocol.rs @@ -61,6 +61,9 @@ pub struct Message { /// Stateless OpenAI Responses output to replay for this assistant turn. #[serde(default, skip_serializing_if = "Option::is_none")] pub responses_output: Option, + /// Anthropic content blocks required for lossless continuation. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub anthropic_output: Option, } /// A function the Harness offers to the model. @@ -164,6 +167,94 @@ impl ResponseFormat { } } +/// Anthropic Messages thinking mode; model support depends on the provider. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum AnthropicThinking { + /// Model decides when and how much to think. + Adaptive, + /// Manual thinking budget for models supporting extended thinking. + Enabled { budget_tokens: u32 }, + /// Explicitly disable thinking where the selected model supports it. + Disabled, +} + +impl<'de> Deserialize<'de> for AnthropicThinking { + fn deserialize>(deserializer: D) -> Result { + #[derive(Deserialize)] + #[serde(tag = "type", rename_all = "snake_case", deny_unknown_fields)] + enum Repr { + Adaptive {}, + Enabled { budget_tokens: u32 }, + Disabled {}, + } + Ok(match Repr::deserialize(deserializer)? { + Repr::Adaptive {} => Self::Adaptive, + Repr::Enabled { budget_tokens } => Self::Enabled { budget_tokens }, + Repr::Disabled {} => Self::Disabled, + }) + } +} + +/// Anthropic output effort; model support depends on the provider. +#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(rename_all = "lowercase")] +pub enum AnthropicEffort { + Low, + Medium, + High, + Max, +} + +/// Optional Anthropic generation settings for a configured provider route. +#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq)] +#[serde(deny_unknown_fields)] +pub struct AnthropicSettings { + /// Thinking mode, if explicitly configured. + pub thinking: Option, + /// Output effort, if explicitly configured. + pub effort: Option, + /// Maximum generated tokens, including any thinking budget. + pub max_tokens: Option, +} + +impl AnthropicSettings { + /// Validate and translate the settings into library request options. + /// + /// # Errors + /// Rejects a non-Anthropic route or an invalid thinking budget. + pub fn options(self, protocol: Protocol) -> Result { + if protocol != Protocol::Anthropic || self.max_tokens == Some(0) { + return Err(Error::Request("invalid Anthropic generation settings")); + } + validate_anthropic_options(protocol, self.thinking, self.effort, self.max_tokens)?; + Ok(Options { + anthropic_thinking: self.thinking, + anthropic_effort: self.effort, + max_tokens: self.max_tokens, + ..Options::default() + }) + } +} + +fn validate_anthropic_options( + protocol: Protocol, + thinking: Option, + effort: Option, + max_tokens: Option, +) -> Result<(), Error> { + if protocol != Protocol::Anthropic && (thinking.is_some() || effort.is_some()) { + return Err(Error::Request( + "Anthropic thinking options require Anthropic Messages", + )); + } + if let Some(AnthropicThinking::Enabled { budget_tokens }) = thinking + && (budget_tokens < 1024 || budget_tokens >= max_tokens.unwrap_or(1024)) + { + return Err(Error::Request("invalid Anthropic thinking budget")); + } + Ok(()) +} + /// Portable generation options. Unsupported provider options are rejected. #[derive(Clone, Debug, Default, Serialize)] pub struct Options { @@ -179,6 +270,12 @@ pub struct Options { /// Request terminal usage on OpenAI-compatible streams that support it. #[serde(skip)] pub stream_usage: bool, + /// Explicit thinking mode for Anthropic Messages. + #[serde(skip)] + pub anthropic_thinking: Option, + /// Explicit output effort for Anthropic Messages. + #[serde(skip)] + pub anthropic_effort: Option, /// Sampling temperature. #[serde(skip_serializing_if = "Option::is_none")] pub temperature: Option, @@ -238,6 +335,8 @@ impl Request<'_> { ((!message.tool_calls.is_empty() || message.responses_output.is_some()) && message.role != Role::Assistant) || (message.responses_output.is_some() && protocol != Protocol::OpenAiResponses) + || (message.anthropic_output.is_some() + && (protocol != Protocol::Anthropic || message.role != Role::Assistant)) || (message.role == Role::Tool && message.tool_call_id.as_deref().is_none_or(str::is_empty)) || (message.role != Role::Tool && message.tool_call_id.is_some()) @@ -248,6 +347,15 @@ impl Request<'_> { { return Err(Error::Request("invalid tool definition, call or result")); } + validate_anthropic_options( + protocol, + self.options.anthropic_thinking, + self.options.anthropic_effort, + self.options.max_tokens, + )?; + for message in self.messages { + crate::responses::validate_continuation(message)?; + } self.validate_images()?; let tool_choice = self.options.tool_choice.as_ref(); match tool_choice { @@ -368,14 +476,29 @@ impl Request<'_> { .collect(); let mut payload = json!({ "model": self.model, - "messages": anthropic_turns(turns), + "messages": anthropic_turns(turns)?, "stream": stream, "max_tokens": self.options.max_tokens.unwrap_or(1024), }); + let mut output_config = serde_json::Map::new(); + if let Some(format) = response_format { + output_config.insert("format".into(), format); + } + if let Some(effort) = self.options.anthropic_effort { + output_config.insert("effort".into(), json!(effort)); + } + let thinking = self.options.anthropic_thinking.map(|mode| match mode { + AnthropicThinking::Adaptive => json!({"type": "adaptive"}), + AnthropicThinking::Enabled { budget_tokens } => { + json!({"type": "enabled", "budget_tokens": budget_tokens}) + } + AnthropicThinking::Disabled => json!({"type": "disabled"}), + }); for (key, value) in [ + ("thinking", thinking), ("system", (!system.is_empty()).then(|| json!(system))), ("tool_choice", tool_choice), - ("output_config", response_format.map(|format| json!({"format": format}))), + ("output_config", (!output_config.is_empty()).then_some(Value::Object(output_config))), ("tools", (!self.options.tools.is_empty()).then(|| json!( self.options.tools.iter().map(|tool| json!({ "name": tool.name, "description": tool.description, "input_schema": tool.parameters, @@ -447,7 +570,7 @@ impl<'a> From<&'a Message> for OpenAiMessage<'a> { } } -fn anthropic_turns(turns: &[Message]) -> Vec { +fn anthropic_turns(turns: &[Message]) -> Result, Error> { let mut encoded = Vec::with_capacity(turns.len()); let mut results = Vec::new(); for message in turns { @@ -460,6 +583,15 @@ fn anthropic_turns(turns: &[Message]) -> Vec { if !results.is_empty() { encoded.push(json!({"role": "user", "content": std::mem::take(&mut results)})); } + if let Some(output) = &message.anthropic_output { + if !output.matches(message) { + return Err(Error::Request( + "Anthropic output does not match assistant message", + )); + } + encoded.push(json!({"role": "assistant", "content": output.0})); + continue; + } if !message.images.is_empty() { let content = content_blocks(&message.content, &message.images, Protocol::Anthropic); encoded.push(json!({"role": message.role, "content": content})); @@ -483,7 +615,7 @@ fn anthropic_turns(turns: &[Message]) -> Vec { if !results.is_empty() { encoded.push(json!({"role": "user", "content": results})); } - encoded + Ok(encoded) } /// Provider-reported token usage. @@ -564,6 +696,7 @@ fn openai_message<'de, D: serde::Deserializer<'de>>(value: D) -> Result content.push_str(block["text"].as_str().ok_or(Error::Response)?), - Some("tool_use") => tool_calls.push(ToolCall { - id: block["id"].as_str().ok_or(Error::Response)?.into(), - name: block["name"].as_str().ok_or(Error::Response)?.into(), - arguments: block["input"].clone(), - }), - _ => return Err(Error::Response), - } - } + let (anthropic_output, content, tool_calls) = + crate::AnthropicOutput::parse(response.content)?; Ok(Self { id: response.id, model: response.model, @@ -671,6 +793,7 @@ impl Completion { tool_call_id: None, tool_calls, responses_output: None, + anthropic_output: Some(anthropic_output), }, finish_reason: response.stop_reason, }], diff --git a/crates/llm/src/responses.rs b/crates/llm/src/responses.rs index ee552979..e018db4b 100644 --- a/crates/llm/src/responses.rs +++ b/crates/llm/src/responses.rs @@ -19,6 +19,26 @@ use crate::{ #[serde(transparent)] pub struct ResponsesOutput(Vec); +impl ResponsesOutput { + /// Check the provider's completed text against the exact streamed text. + pub(crate) fn matches_streamed_text(&self, expected: &[u8]) -> Result { + use sha2::{Digest, Sha256}; + let mut digest = Sha256::new(); + for item in &self.0 { + if item["type"] != "message" { + continue; + } + for part in item["content"].as_array().ok_or(Error::Response)? { + if part["type"] != "output_text" { + return Err(Error::Response); + } + digest.update(part["text"].as_str().ok_or(Error::Response)?.as_bytes()); + } + } + Ok(digest.finalize().as_slice() == expected) + } +} + impl fmt::Debug for ResponsesOutput { fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { formatter @@ -28,6 +48,58 @@ impl fmt::Debug for ResponsesOutput { } } +/// Ensure caller-visible assistant state still matches the provider output +/// that will be replayed. This prevents stale or tampered text/calls from +/// being silently replaced by a different opaque continuation. +pub(crate) fn validate_continuation(message: &Message) -> Result<(), Error> { + let Some(output) = &message.responses_output else { + return Ok(()); + }; + if message.role != Role::Assistant || !message.images.is_empty() || message.name.is_some() { + return Err(Error::Request("invalid Responses continuation message")); + } + let mut text = String::new(); + let mut calls = Vec::new(); + let mut ids = HashSet::new(); + for item in &output.0 { + match item["type"].as_str() { + Some("message") => append_text(item, &mut text) + .map_err(|_| Error::Request("invalid Responses continuation"))?, + Some("reasoning") => {} + Some("function_call") => { + let id = string(&item["call_id"]) + .map_err(|_| Error::Request("invalid Responses continuation"))?; + if !ids.insert(id) + || item["status"] + .as_str() + .is_some_and(|status| status != "completed") + { + return Err(Error::Request("invalid Responses continuation")); + } + let name = string(&item["name"]) + .map_err(|_| Error::Request("invalid Responses continuation"))?; + let arguments = item["arguments"] + .as_str() + .and_then(|json| serde_json::from_str::(json).ok()) + .filter(Value::is_object) + .ok_or(Error::Request("invalid Responses continuation"))?; + calls.push(ToolCall { + id: id.into(), + name: name.into(), + arguments, + }); + } + _ => return Err(Error::Request("invalid Responses continuation")), + } + } + if text != message.content || calls != message.tool_calls { + return Err(Error::Request( + "Responses continuation does not match assistant message", + )); + } + Ok(()) +} + pub(crate) fn encode( request: &Request<'_>, stream: bool, @@ -187,6 +259,7 @@ pub(crate) fn event(mut event: Value) -> Result, Error> { usage: final_response.usage, tool_calls: final_response.calls, responses_output: Some(ResponsesOutput(final_response.output)), + anthropic_output: None, })) } "response.failed" => { @@ -287,6 +360,7 @@ pub(crate) fn complete(bytes: &[u8]) -> Result { tool_call_id: None, tool_calls: response.calls, responses_output: Some(ResponsesOutput(response.output)), + anthropic_output: None, }, finish_reason: Some(reason.into()), }], diff --git a/crates/llm/src/stream.rs b/crates/llm/src/stream.rs index e9a71eac..a54a1fe1 100644 --- a/crates/llm/src/stream.rs +++ b/crates/llm/src/stream.rs @@ -6,17 +6,39 @@ use bytes::{Bytes, BytesMut}; use serde_json::Value; +use sha2::{Digest, Sha256}; use std::collections::{BTreeMap, HashSet}; -use tokio_sse_codec::{Frame, SseDecoder}; +use tokio_sse_codec::{Frame, SseDecodeError, SseDecoder}; use tokio_util::codec::Decoder as _; use crate::{Error, EventStream, Protocol, ToolCall, Usage}; +/// A provider-returned thinking fragment. Debug formatting hides its content. +#[derive(Eq, PartialEq)] +pub struct ThinkingDelta(String); + +impl ThinkingDelta { + /// Returns the provider-returned fragment for explicit presentation. + pub fn as_str(&self) -> &str { + &self.0 + } +} + +impl std::fmt::Debug for ThinkingDelta { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("ThinkingDelta") + .field("bytes", &self.0.len()) + .finish() + } +} + /// A provider-neutral event for a single assistant turn. #[derive(Debug, Eq, PartialEq)] pub enum TurnEvent { /// Text that can be displayed as it arrives. Text(String), + /// Provider-returned thinking, distinct from final-answer text. + Thinking(ThinkingDelta), /// Successful terminal event. The Harness can now execute these calls. Finished { /// Provider finish reason. @@ -29,6 +51,8 @@ pub enum TurnEvent { tool_calls: Vec, /// Opaque output for stateless Responses continuation. responses_output: Option, + /// Ordered Anthropic content blocks for continuation, including thinking. + anthropic_output: Option, }, } @@ -61,6 +85,8 @@ impl TurnStream { usage: None, anthropic_started: false, anthropic_block: None, + anthropic_output: BTreeMap::new(), + responses_text: Sha256::new(), }, } } @@ -154,7 +180,10 @@ impl SseFraming { fn next(&mut self) -> Result>, Error> { self.codec .decode(&mut self.pending) - .map_err(|_| Error::Response) + .map_err(|error| match error { + SseDecodeError::ExceededSizeLimit(_) => Error::Limit, + _ => Error::Response, + }) } } @@ -168,7 +197,8 @@ struct PartialCall { enum Block { Text, Tool, - Ignored, + Thinking, + RedactedThinking, } struct Decoder { @@ -179,6 +209,8 @@ struct Decoder { usage: Option, anthropic_started: bool, anthropic_block: Option<(u64, Block)>, + anthropic_output: BTreeMap, + responses_text: Sha256, } impl Decoder { @@ -193,7 +225,23 @@ impl Decoder { match self.protocol { Protocol::OpenAi => self.openai(&event), Protocol::Anthropic => self.anthropic(&event), - Protocol::OpenAiResponses => crate::responses::event(event), + Protocol::OpenAiResponses => { + let decoded = crate::responses::event(event)?; + match &decoded { + Some(TurnEvent::Text(text)) => self.responses_text.update(text.as_bytes()), + Some(TurnEvent::Finished { + responses_output: Some(output), + .. + }) => { + let expected = self.responses_text.clone().finalize(); + if !output.matches_streamed_text(&expected)? { + return Err(Error::Response); + } + } + _ => {} + } + Ok(decoded) + } } } @@ -235,12 +283,25 @@ impl Decoder { arguments, }); } + let anthropic_output = if self.protocol == Protocol::Anthropic { + let blocks = std::mem::take(&mut self.anthropic_output) + .into_values() + .collect(); + let (output, _, observed_calls) = crate::AnthropicOutput::parse(blocks)?; + if observed_calls != calls { + return Err(Error::Response); + } + Some(output) + } else { + None + }; Ok(TurnEvent::Finished { reason, identity: self.identity.take(), usage: self.usage.take(), tool_calls: calls, responses_output: None, + anthropic_output, }) } @@ -325,7 +386,7 @@ impl Decoder { .anthropic_block .is_some_and(|(active, _)| event["index"].as_u64() == Some(active)) => { - self.anthropic_block = None; + self.anthropic_stop()?; Ok(None) } "message_delta" if self.anthropic_block.is_none() => { @@ -360,14 +421,23 @@ impl Decoder { } "text" => ( Block::Text, - block["text"] - .as_str() + Some(block["text"].as_str().ok_or(Error::Response)?) .filter(|text| !text.is_empty()) .map(text_event), ), - "thinking" | "redacted_thinking" => (Block::Ignored, None), + "thinking" if block["thinking"].is_string() => ( + Block::Thinking, + block["thinking"] + .as_str() + .filter(|text| !text.is_empty()) + .map(|text| TurnEvent::Thinking(ThinkingDelta(text.to_owned()))), + ), + "redacted_thinking" if block["data"].is_string() => (Block::RedactedThinking, None), _ => return Err(Error::Response), }; + if self.anthropic_output.insert(index, block.clone()).is_some() { + return Err(Error::Response); + } self.anthropic_block = Some((index, kind)); Ok(text) } @@ -380,9 +450,11 @@ impl Decoder { } let delta = &event["delta"]; match (block, required_string(&delta["type"])?) { - (Block::Text, "text_delta") => Ok(Some(text_event( - delta["text"].as_str().ok_or(Error::Response)?, - ))), + (Block::Text, "text_delta") => { + let text = delta["text"].as_str().ok_or(Error::Response)?; + self.append_anthropic_field(index, "text", text)?; + Ok(Some(text_event(text))) + } (Block::Tool, "input_json_delta") => { let partial = self.tools.get_mut(&index).ok_or(Error::Response)?; partial @@ -390,11 +462,66 @@ impl Decoder { .push_str(delta["partial_json"].as_str().ok_or(Error::Response)?); Ok(None) } - (Block::Ignored, "thinking_delta" | "signature_delta") => Ok(None), + (Block::Thinking, "thinking_delta" | "signature_delta") => { + let field = if delta["type"] == "thinking_delta" { + "thinking" + } else { + "signature" + }; + let part = delta[field].as_str().ok_or(Error::Response)?; + self.append_anthropic_field(index, field, part)?; + if field == "thinking" && !part.is_empty() { + Ok(Some(TurnEvent::Thinking(ThinkingDelta(part.to_owned())))) + } else { + Ok(None) + } + } _ => Err(Error::Response), } } + fn append_anthropic_field( + &mut self, + index: u64, + field: &str, + value: &str, + ) -> Result<(), Error> { + let block = self + .anthropic_output + .get_mut(&index) + .and_then(Value::as_object_mut) + .ok_or(Error::Response)?; + let entry = block + .entry(field) + .or_insert_with(|| Value::String(String::new())); + let Value::String(text) = entry else { + return Err(Error::Response); + }; + text.push_str(value); + Ok(()) + } + + fn anthropic_stop(&mut self) -> Result<(), Error> { + let (index, kind) = self.anthropic_block.take().ok_or(Error::Response)?; + if kind == Block::Tool { + let call = self.tools.get(&index).ok_or(Error::Response)?; + let input: Value = if call.arguments.is_empty() { + serde_json::json!({}) + } else { + serde_json::from_str(&call.arguments).map_err(|_| Error::Response)? + }; + if !input.is_object() { + return Err(Error::Response); + } + self.anthropic_output + .get_mut(&index) + .and_then(Value::as_object_mut) + .ok_or(Error::Response)? + .insert("input".into(), input); + } + Ok(()) + } + fn start_tool(&mut self, index: u64, id: &str, name: &str) -> Result<(), Error> { use std::collections::btree_map::Entry; let Entry::Vacant(entry) = self.tools.entry(index) else { diff --git a/crates/llm/tests/integration/anthropic_reasoning.rs b/crates/llm/tests/integration/anthropic_reasoning.rs new file mode 100644 index 00000000..1892dd1e --- /dev/null +++ b/crates/llm/tests/integration/anthropic_reasoning.rs @@ -0,0 +1,179 @@ +// SPDX-FileCopyrightText: 2026 hardenedlinux Linux Team and Contributors +// +// SPDX-License-Identifier: LGPL-2.1-or-later AND Apache-2.0 + +use super::{ + Client, Endpoint, Error, Message, Options, Protocol, Request, Role, TestResult, messages, + upstream, +}; +use std::time::Duration; +use weaver_llm::TurnEvent; + +const REPLY: &str = concat!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nConnection: close\r\n\r\n", + r#"{"id":"id","model":"fixture","role":"assistant","content":[{"type":"thinking","thinking":"private-thought","signature":"private-signature"},{"type":"text","text":"OK"}],"stop_reason":"end_turn","usage":{"input_tokens":1,"output_tokens":1}}"#, +); +const STREAM: &str = concat!( + "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nConnection: close\r\n\r\n", + "data: {\"type\":\"message_start\",\"message\":{\"id\":\"id\",\"model\":\"fixture\",\"usage\":{\"input_tokens\":1,\"output_tokens\":0}}}\n\n", + "data: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"thinking\",\"thinking\":\"\"}}\n\n", + "data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"thinking_delta\",\"thinking\":\"private-thought\"}}\n\n", + "data: {\"type\":\"content_block_delta\",\"index\":0,\"delta\":{\"type\":\"signature_delta\",\"signature\":\"private-signature\"}}\n\n", + "data: {\"type\":\"content_block_stop\",\"index\":0}\n\n", + "data: {\"type\":\"content_block_start\",\"index\":1,\"content_block\":{\"type\":\"text\",\"text\":\"\"}}\n\n", + "data: {\"type\":\"content_block_delta\",\"index\":1,\"delta\":{\"type\":\"text_delta\",\"text\":\"OK\"}}\n\n", + "data: {\"type\":\"content_block_stop\",\"index\":1}\n\n", + "data: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"end_turn\"}}\n\n", + "data: {\"type\":\"message_stop\"}\n\n", +); + +#[tokio::test] +async fn buffered_thinking_is_preserved_for_the_next_request() -> TestResult { + let client = Client::new(Duration::from_secs(2), 8192)?; + let (url, first) = upstream(REPLY).await?; + let endpoint = Endpoint::new(Protocol::Anthropic, &url)?; + let initial = messages(); + let options = Options::default(); + let reply = client + .complete( + &endpoint, + "fixture", + &Request { + model: "fixture", + messages: &initial, + options: &options, + }, + ) + .await?; + first.await??; + let answer = reply.choices[0].message.clone(); + assert_eq!(answer.content, "OK"); + assert!(!format!("{:?}", answer.anthropic_output).contains("private-thought")); + verify_replay(&client, &initial, answer, &options).await +} + +#[tokio::test] +async fn streamed_thinking_is_preserved_for_the_next_request() -> TestResult { + let client = Client::new(Duration::from_secs(2), 8192)?; + let (url, first) = upstream(STREAM).await?; + let endpoint = Endpoint::new(Protocol::Anthropic, &url)?; + let initial = messages(); + let options = Options::default(); + let mut stream = client + .stream_turn( + &endpoint, + "fixture", + &Request { + model: "fixture", + messages: &initial, + options: &options, + }, + ) + .await?; + let Some(TurnEvent::Thinking(thinking)) = stream.next_event().await? else { + return Err("missing streamed thinking".into()); + }; + assert_eq!(thinking.as_str(), "private-thought"); + assert!(!format!("{thinking:?}").contains("private-thought")); + assert_eq!( + stream.next_event().await?, + Some(TurnEvent::Text("OK".into())) + ); + let Some(TurnEvent::Finished { + anthropic_output, + tool_calls, + .. + }) = stream.next_event().await? + else { + return Err("missing terminal event".into()); + }; + assert!(anthropic_output.is_some()); + first.await??; + let answer = Message { + role: Role::Assistant, + content: "OK".into(), + images: vec![], + name: None, + tool_call_id: None, + tool_calls, + responses_output: None, + anthropic_output, + }; + verify_replay(&client, &initial, answer, &options).await +} + +#[tokio::test] +async fn interrupted_tool_output_is_not_reported_as_replayable() -> TestResult { + const INTERRUPTED: &str = concat!( + "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nConnection: close\r\n\r\n", + "data: {\"type\":\"message_start\",\"message\":{\"id\":\"id\",\"model\":\"fixture\"}}\n\n", + "data: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"tool_use\",\"id\":\"call\",\"name\":\"tool\",\"input\":{}}}\n\n", + "data: {\"type\":\"content_block_stop\",\"index\":0}\n\n", + "data: {\"type\":\"message_delta\",\"delta\":{\"stop_reason\":\"max_tokens\"}}\n\n", + "data: {\"type\":\"message_stop\"}\n\n", + ); + let client = Client::new(Duration::from_secs(2), 8192)?; + let (url, first) = upstream(INTERRUPTED).await?; + let endpoint = Endpoint::new(Protocol::Anthropic, &url)?; + let initial = messages(); + let options = Options::default(); + let mut stream = client + .stream_turn( + &endpoint, + "fixture", + &Request { + model: "fixture", + messages: &initial, + options: &options, + }, + ) + .await?; + assert!(matches!(stream.next_event().await, Err(Error::Response))); + first.await??; + Ok(()) +} + +async fn verify_replay( + client: &Client, + initial: &[Message], + answer: Message, + options: &Options, +) -> TestResult { + let (url, observed) = upstream(REPLY).await?; + let endpoint = Endpoint::new(Protocol::Anthropic, &url)?; + let mut changed = answer.clone(); + changed.content.push('!'); + let invalid = [initial[1].clone(), changed]; + assert!(matches!( + client + .complete( + &endpoint, + "fixture", + &Request { + model: "fixture", + messages: &invalid, + options + } + ) + .await, + Err(Error::Request(_)) + )); + let conversation = [initial[1].clone(), answer, initial[1].clone()]; + client + .complete( + &endpoint, + "fixture", + &Request { + model: "fixture", + messages: &conversation, + options, + }, + ) + .await?; + let (_, body) = observed.await??; + let blocks = &body["messages"][1]["content"]; + assert_eq!(blocks[0]["thinking"], "private-thought"); + assert_eq!(blocks[0]["signature"], "private-signature"); + assert_eq!(blocks[1]["text"], "OK"); + Ok(()) +} diff --git a/crates/llm/tests/integration/client.rs b/crates/llm/tests/integration/client.rs index 25c1c948..6b74a2f9 100644 --- a/crates/llm/tests/integration/client.rs +++ b/crates/llm/tests/integration/client.rs @@ -63,6 +63,7 @@ fn messages() -> [Message; 2] { tool_call_id: None, tool_calls: Vec::new(), responses_output: None, + anthropic_output: None, }, Message { role: Role::User, @@ -72,6 +73,7 @@ fn messages() -> [Message; 2] { tool_call_id: None, tool_calls: Vec::new(), responses_output: None, + anthropic_output: None, }, ] } @@ -315,6 +317,7 @@ async fn harness_replays_tool_results_to_both_providers() -> TestResult { tool_call_id: Some(call.id.clone()), tool_calls: Vec::new(), responses_output: None, + anthropic_output: None, }) .collect(); conversation.push(Message { @@ -325,6 +328,7 @@ async fn harness_replays_tool_results_to_both_providers() -> TestResult { tool_call_id: None, tool_calls, responses_output: None, + anthropic_output: None, }); conversation.extend(results); @@ -406,6 +410,7 @@ async fn responses_stream_replays_reasoning_and_tool_results_without_server_stat "data: {\"type\":\"response.function_call_arguments.delta\",\"delta\":\"{\"}\n\n", "data: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\",\"id\":\"resp_1\",\"model\":\"gpt-fixture\",", "\"output\":[{\"type\":\"reasoning\",\"id\":\"rs_1\",\"summary\":[],\"encrypted_content\":\"opaque-fixture\"},", + "{\"type\":\"message\",\"content\":[{\"type\":\"output_text\",\"text\":\"Checking.\"}]},", "{\"type\":\"function_call\",\"id\":\"fc_1\",\"call_id\":\"call_1\",\"name\":\"read_file\",\"arguments\":\"{\\\"path\\\":\\\"README\\\"}\",\"status\":\"completed\"}],", "\"usage\":{\"input_tokens\":10,\"output_tokens\":4}}}\n\n", )).await?; @@ -468,6 +473,7 @@ async fn responses_stream_replays_reasoning_and_tool_results_without_server_stat tool_call_id: None, tool_calls, responses_output, + anthropic_output: None, }, Message { role: Role::Tool, @@ -477,8 +483,20 @@ async fn responses_stream_replays_reasoning_and_tool_results_without_server_stat tool_call_id: Some(call_id), tool_calls: Vec::new(), responses_output: None, + anthropic_output: None, }, ]); + let mut altered = conversation.clone(); + altered[2].content.push('!'); + assert!(matches!( + Request { + model: "gpt-fixture", + messages: &altered, + options: &options + } + .encode(Protocol::OpenAiResponses, true), + Err(Error::Request(_)) + )); let (url, captured) = upstream(concat!( "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nConnection: close\r\n\r\n", "{\"status\":\"completed\",\"id\":\"resp_2\",\"model\":\"gpt-fixture\",", @@ -498,9 +516,10 @@ async fn responses_stream_replays_reasoning_and_tool_results_without_server_stat assert_eq!(response.choices[0].message.content, "Done."); let payload = captured.await??.1; assert_eq!(payload["input"][2]["encrypted_content"], "opaque-fixture"); - assert_eq!(payload["input"][3]["call_id"], "call_1"); - assert_eq!(payload["input"][4]["type"], "function_call_output"); + assert_eq!(payload["input"][3]["type"], "message"); assert_eq!(payload["input"][4]["call_id"], "call_1"); + assert_eq!(payload["input"][5]["type"], "function_call_output"); + assert_eq!(payload["input"][5]["call_id"], "call_1"); assert!(payload.get("previous_response_id").is_none()); Ok(()) } @@ -558,6 +577,7 @@ async fn live_openai_responses_replays_reasoning_and_tool_output() -> TestResult while let Some(event) = turn.next_event().await? { match event { TurnEvent::Text(delta) => text.push_str(&delta), + TurnEvent::Thinking(_) => {} terminal @ TurnEvent::Finished { .. } => return Ok((text, terminal)), } } @@ -577,6 +597,7 @@ async fn live_openai_responses_replays_reasoning_and_tool_output() -> TestResult tool_call_id: None, tool_calls: Vec::new(), responses_output: None, + anthropic_output: None, }]; let options = Options { max_tokens: Some(512), @@ -638,6 +659,7 @@ async fn live_openai_responses_replays_reasoning_and_tool_output() -> TestResult tool_call_id: None, tool_calls, responses_output, + anthropic_output: None, }, Message { role: Role::Tool, @@ -647,6 +669,7 @@ async fn live_openai_responses_replays_reasoning_and_tool_output() -> TestResult tool_call_id: Some(call_id), tool_calls: Vec::new(), responses_output: None, + anthropic_output: None, }, ]); let (text, event) = terminal( @@ -767,3 +790,6 @@ mod oauth_lifecycle; mod openai_login; #[path = "provider_failures.rs"] mod provider_failures; + +#[path = "anthropic_reasoning.rs"] +mod anthropic_reasoning; diff --git a/crates/llm/tests/integration/configuration.rs b/crates/llm/tests/integration/configuration.rs index 00c1aca8..ad7a2fdf 100644 --- a/crates/llm/tests/integration/configuration.rs +++ b/crates/llm/tests/integration/configuration.rs @@ -45,6 +45,7 @@ async fn prepared_profile_streams_after_inputs_are_removed() -> TestResult { while let Some(event) = stream.next_event().await? { match event { weaver_llm::TurnEvent::Text(delta) => text.push_str(&delta), + weaver_llm::TurnEvent::Thinking(_) => {}, weaver_llm::TurnEvent::Finished { .. } => complete = true, } } @@ -431,3 +432,90 @@ async fn dropping_a_turn_closes_an_incomplete_exchange() -> TestResult { }) .await? } + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn anthropic_config_precedes_environment_and_missing_fields_use_it() -> TestResult { + const CHILD: &str = "WEAVER_ANTHROPIC_CONFIG_CHILD"; + const PATH: &str = "WEAVER_ANTHROPIC_CONFIG_PATH"; + const REPLY: &str = concat!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nConnection: close\r\n\r\n", + r#"{"id":"one","model":"fixture","role":"assistant","content":[{"type":"text","text":"Hi"}],"stop_reason":"end_turn","usage":{"input_tokens":1,"output_tokens":1}}"#, + ); + timeout(WATCHDOG, async { + if std::env::var_os(CHILD).is_some() { + let config = Configuration::load(std::env::var(PATH)?).await?; + let names: Vec<_> = config.credential_environment().collect(); + let expected = std::env::var("WEAVER_EXPECTED_CREDENTIAL_NAME")?; + assert!(names.contains(&expected.as_str())); + let prepared = config.prepare("local").await?; + let result = prepared.complete(&messages(), &Options::default()).await?; + assert_eq!(result.choices[0].message.content, "Hi"); + return Ok(()); + } + let dir = tempfile::tempdir()?; + let config_path = dir.path().join("weaver.toml"); + for explicit in [true, false] { + let (url, observed) = upstream(REPLY).await?; + let mut document = String::from( + "[profiles.local]\nkind = \"direct\"\nprotocol = \"anthropic-messages\"\ntimeout_ms = 1000\nmax_response_bytes = 4096\n", + ); + if explicit { + document.push_str(&format!("endpoint = \"{url}\"\nmodel = \"configured-model\"\n[profiles.local.credential]\nenv = \"WEAVER_EXPLICIT_KEY\"\n")); + } + tokio::fs::write(&config_path, document).await?; + let base = if explicit { "invalid://must-not-be-used".to_owned() } else { url.trim_end_matches("/model").to_owned() }; + let output = Command::new(std::env::current_exe()?) + .args(["--exact", "configuration::anthropic_config_precedes_environment_and_missing_fields_use_it"]) + .env(CHILD, "1").env(PATH, &config_path) + .env("ANTHROPIC_BASE_URL", base).env("ANTHROPIC_MODEL", "environment-model") + .env("ANTHROPIC_API_KEY", "fallback-secret").env("WEAVER_EXPLICIT_KEY", "explicit-secret") + .env("WEAVER_EXPECTED_CREDENTIAL_NAME", if explicit { "WEAVER_EXPLICIT_KEY" } else { "ANTHROPIC_API_KEY" }) + .kill_on_drop(true).output().await?; + assert!(output.status.success(), "child configuration failed"); + assert!(!String::from_utf8_lossy(&output.stdout).contains("secret")); + assert!(!String::from_utf8_lossy(&output.stderr).contains("secret")); + let (headers, body) = observed.await??; + let (path, model, secret) = if explicit { ("/model", "configured-model", "explicit-secret") } + else { ("/v1/messages", "environment-model", "fallback-secret") }; + assert!(headers.starts_with(&format!("post {path} http/1.1\r\n"))); + assert!(headers.contains(&format!("x-api-key: {secret}\r\n"))); + assert_eq!(body["model"], model); + } + Ok::<_, Box>(()) + }).await? +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn anthropic_invalid_explicit_values_do_not_fall_back() -> TestResult { + const CHILD: &str = "WEAVER_ANTHROPIC_INVALID_CHILD"; + const PATH: &str = "WEAVER_ANTHROPIC_INVALID_PATH"; + timeout(WATCHDOG, async { + if std::env::var_os(CHILD).is_some() { + let config = Configuration::load(std::env::var(PATH)?).await?; + assert!(matches!(config.prepare("local").await, Err(Error::Configuration))); + return Ok(()); + } + let dir = tempfile::tempdir()?; + let path = dir.path().join("weaver.toml"); + let base = "[profiles.local]\nkind = \"direct\"\nprotocol = \"anthropic-messages\"\ntimeout_ms = 1000\nmax_response_bytes = 4096\n"; + for fields in [ + "model = \"\"\nendpoint = \"https://configured.example/v1/messages\"\n", + "model = \"configured\"\nendpoint = \"\"\n", + "model = \"configured\"\nendpoint = \"https://configured.example/v1/messages\"\n[profiles.local.credential]\nenv = \"WEAVER_MISSING_EXPLICIT_CREDENTIAL\"\n", + ] { + tokio::fs::write(&path, format!("{base}{fields}")).await?; + let output = Command::new(std::env::current_exe()?) + .args(["--exact", "configuration::anthropic_invalid_explicit_values_do_not_fall_back"]) + .env(CHILD, "1").env(PATH, &path) + .env("ANTHROPIC_BASE_URL", "https://environment.example") + .env("ANTHROPIC_MODEL", "environment-model") + .env("ANTHROPIC_API_KEY", "fallback-private-token") + .env_remove("WEAVER_MISSING_EXPLICIT_CREDENTIAL") + .kill_on_drop(true).output().await?; + assert!(output.status.success(), "child rejection check failed"); + assert!(!String::from_utf8_lossy(&output.stdout).contains("fallback-private-token")); + assert!(!String::from_utf8_lossy(&output.stderr).contains("fallback-private-token")); + } + Ok::<_, Box>(()) + }).await? +} diff --git a/crates/llm/tests/integration/generation_controls.rs b/crates/llm/tests/integration/generation_controls.rs index 272b568e..c08a1336 100644 --- a/crates/llm/tests/integration/generation_controls.rs +++ b/crates/llm/tests/integration/generation_controls.rs @@ -105,3 +105,54 @@ fn generation_controls_follow_each_provider_contract() -> TestResult { } Ok(()) } + +#[test] +fn anthropic_thinking_and_effort_are_typed_and_compose_with_format() -> TestResult { + use serde_json::json; + use weaver_llm::{AnthropicEffort, AnthropicThinking, ResponseFormat}; + let messages = messages(); + let mut options = Options { + max_tokens: Some(2048), + anthropic_thinking: Some(AnthropicThinking::Adaptive), + anthropic_effort: Some(AnthropicEffort::High), + response_format: Some(ResponseFormat::JsonSchema { + name: "marker".into(), + schema: json!({"type": "object"}), + }), + ..Options::default() + }; + let encode = |options: &Options, protocol| { + Request { + model: "fixture", + messages: &messages, + options, + } + .encode(protocol, true) + }; + let body: Value = serde_json::from_slice(&encode(&options, Protocol::Anthropic)?)?; + assert_eq!(body["thinking"], json!({"type": "adaptive"})); + assert_eq!(body["output_config"]["effort"], "high"); + assert_eq!(body["output_config"]["format"]["type"], "json_schema"); + assert!(encode(&options, Protocol::OpenAi).is_err()); + assert!(encode(&options, Protocol::OpenAiResponses).is_err()); + options.anthropic_thinking = Some(AnthropicThinking::Enabled { + budget_tokens: 1024, + }); + let body: Value = serde_json::from_slice(&encode(&options, Protocol::Anthropic)?)?; + assert_eq!( + body["thinking"], + json!({"type": "enabled", "budget_tokens": 1024}) + ); + options.anthropic_thinking = Some(AnthropicThinking::Enabled { + budget_tokens: 1023, + }); + assert!(encode(&options, Protocol::Anthropic).is_err()); + options.anthropic_thinking = Some(AnthropicThinking::Enabled { + budget_tokens: 2048, + }); + assert!(encode(&options, Protocol::Anthropic).is_err()); + options.anthropic_thinking = Some(AnthropicThinking::Disabled); + let body: Value = serde_json::from_slice(&encode(&options, Protocol::Anthropic)?)?; + assert_eq!(body["thinking"], json!({"type": "disabled"})); + Ok(()) +} diff --git a/crates/llm/tests/integration/openai_login.rs b/crates/llm/tests/integration/openai_login.rs index f840a917..7fda962b 100644 --- a/crates/llm/tests/integration/openai_login.rs +++ b/crates/llm/tests/integration/openai_login.rs @@ -175,6 +175,7 @@ async fn dynamic_registration_restart_refresh_and_logout_use_the_issued_client() while let Some(event) = stream.next_event().await? { match event { weaver_llm::TurnEvent::Text(delta) => text.push_str(&delta), + weaver_llm::TurnEvent::Thinking(_) => {} weaver_llm::TurnEvent::Finished { .. } => finished = true, } } diff --git a/crates/llm/tests/unit/http2.rs b/crates/llm/tests/unit/http2.rs index be870dd9..3c9b221a 100644 --- a/crates/llm/tests/unit/http2.rs +++ b/crates/llm/tests/unit/http2.rs @@ -120,6 +120,7 @@ async fn open_turn( tool_call_id: None, tool_calls: Vec::new(), responses_output: None, + anthropic_output: None, }]; Ok(client .stream_turn( diff --git a/crates/llm/tests/unit/stream.rs b/crates/llm/tests/unit/stream.rs index 3c990433..901ea60f 100644 --- a/crates/llm/tests/unit/stream.rs +++ b/crates/llm/tests/unit/stream.rs @@ -26,3 +26,82 @@ fn sse_events_are_invariant_under_chunk_boundaries() -> Result<(), crate::Error> } Ok(()) } + +#[test] +fn responses_terminal_must_match_streamed_text() { + use super::{Decoder, TurnEvent}; + use crate::Protocol; + use sha2::Digest; + use std::collections::BTreeMap; + + fn decoder() -> Decoder { + Decoder { + protocol: Protocol::OpenAiResponses, + tools: BTreeMap::new(), + finish: None, + identity: None, + usage: None, + anthropic_started: false, + anthropic_block: None, + anthropic_output: BTreeMap::new(), + responses_text: sha2::Sha256::new(), + } + } + fn terminal(output: serde_json::Value) -> Vec { + serde_json::to_vec(&serde_json::json!({ + "type": "response.completed", + "response": {"id": "response", "model": "gpt-fixture", "status": "completed", "output": output} + })).unwrap() + } + for output in [ + serde_json::json!([]), + serde_json::json!([{"type": "message", "content": [{"type": "output_text", "text": "different"}]}]), + ] { + let mut turn = decoder(); + assert!(matches!( + turn.accept(br#"{"type":"response.output_text.delta","delta":"shown"}"#), + Ok(Some(TurnEvent::Text(_))) + )); + assert!(turn.accept(&terminal(output)).is_err()); + } + let mut turn = decoder(); + turn.accept(br#"{"type":"response.output_text.delta","delta":"sho"}"#) + .unwrap(); + turn.accept(br#"{"type":"response.output_text.delta","delta":"wn"}"#) + .unwrap(); + let output = serde_json::json!([{"type": "message", "content": [{"type": "output_text", "text": "shown"}]}]); + assert!(matches!( + turn.accept(&terminal(output)), + Ok(Some(TurnEvent::Finished { .. })) + )); +} + +#[test] +fn oversized_sse_event_reports_the_provider_byte_limit() { + let wire = b"data: 012345678901234567890123456789\n\n"; + for split in 0..=wire.len() { + let mut frame = SseFraming::new(16); + frame.push(&wire[..split]); + match frame.next() { + Err(error) => assert!( + matches!(error, crate::Error::Limit), + "split {split}: {error:?}" + ), + Ok(None) => { + frame.push(&wire[split..]); + assert!( + matches!(frame.next(), Err(crate::Error::Limit)), + "split {split}" + ); + } + Ok(Some(_)) => panic!("oversized SSE event admitted at split {split}"), + } + } + let mut valid = SseFraming::new(128); + valid.push(wire); + assert!(matches!(valid.next(), Ok(Some(Frame::Event(_))))); + + let mut malformed = SseFraming::new(128); + malformed.push(b"event: \xff\ndata: x\n\n"); + assert!(matches!(malformed.next(), Err(crate::Error::Response))); +}