diff --git a/crates/tinyinference-llm/src/error.rs b/crates/tinyinference-llm/src/error.rs index d366ee8..3bfaef7 100644 --- a/crates/tinyinference-llm/src/error.rs +++ b/crates/tinyinference-llm/src/error.rs @@ -25,4 +25,7 @@ pub enum Error { /// A provider model catalog used an invalid response envelope. #[error("catalog error: {0}")] Catalog(String), + /// The requested operation is not supported by this adapter. + #[error("unsupported operation: {0}")] + Unsupported(String), } diff --git a/crates/tinyinference-llm/src/lib.rs b/crates/tinyinference-llm/src/lib.rs index 086812e..6d745a1 100644 --- a/crates/tinyinference-llm/src/lib.rs +++ b/crates/tinyinference-llm/src/lib.rs @@ -12,6 +12,7 @@ pub mod error; pub mod failure; pub mod message; pub mod model; +mod network_guard; pub mod prompt_tools; pub mod providers; pub mod sentiment; @@ -28,6 +29,7 @@ pub use model::{ ChatModel, ModelRequest, ModelResponse, ModelStream, ModelStreamItem, context_window_for_model_id, model_id_supports_vision, }; +pub use network_guard::{allow_network_models, deny_network_models, network_models_denied}; pub use providers::{MockModel, ProviderKind, ProviderSpec}; pub use tool::{ToolCall, ToolDelta, ToolFormat, ToolSchema}; pub use usage::{Usage, UsageTotals}; diff --git a/crates/tinyinference-llm/src/message/mod.rs b/crates/tinyinference-llm/src/message/mod.rs index d57d904..97c514c 100644 --- a/crates/tinyinference-llm/src/message/mod.rs +++ b/crates/tinyinference-llm/src/message/mod.rs @@ -86,6 +86,13 @@ impl ContentBlock { ContentBlock::Thinking { text, .. } => text.chars().count(), ContentBlock::RedactedThinking { data } => data.chars().count(), ContentBlock::ProviderExtension(value) => value.to_string().chars().count(), + // Non-text media has no character-based weight of its own; + // charge the same flat weight as an image so budgeting/compaction + // does not under-count a transcript dominated by audio, video, or + // document attachments. + ContentBlock::Audio(_) | ContentBlock::Video(_) | ContentBlock::Document(_) => { + IMAGE_CHAR_WEIGHT + } } } } @@ -102,9 +109,7 @@ fn concat_text(content: &[ContentBlock]) -> String { impl Message { /// Creates a system message from text. pub fn system(content: impl Into) -> Self { - Message::System(SystemMessage { - content: vec![ContentBlock::Text(content.into())], - }) + Message::System(SystemMessage::text(content)) } /// Creates a user message from text. @@ -121,6 +126,7 @@ impl Message { content: vec![ContentBlock::Text(content.into())], tool_calls: Vec::new(), usage: None, + origin: None, }) } @@ -135,12 +141,16 @@ impl Message { } /// Returns the concatenated text of all text content blocks. + /// + /// [`Message::Custom`] carries no [`ContentBlock`]s; this returns its + /// `display` rendering (or an empty string when none was set). pub fn text(&self) -> String { match self { Message::System(m) => concat_text(&m.content), Message::User(m) => concat_text(&m.content), Message::Assistant(m) => concat_text(&m.content), Message::Tool(m) => concat_text(&m.content), + Message::Custom(m) => m.display.clone().unwrap_or_default(), } } @@ -159,17 +169,27 @@ impl Message { /// `String` allocation, which matters on hot paths such as token estimation /// over a whole transcript. pub fn char_len(&self) -> usize { - let content = match self { + match self { + Message::Custom(m) => m.display.as_deref().map_or(0, |d| d.chars().count()), + other => other + .content_blocks() + .iter() + .filter_map(ContentBlock::as_text) + .map(|t| t.chars().count()) + .sum(), + } + } + + /// Returns the ordered [`ContentBlock`]s for message kinds that carry + /// them. [`Message::Custom`] carries none. + fn content_blocks(&self) -> &[ContentBlock] { + match self { Message::System(m) => &m.content, Message::User(m) => &m.content, Message::Assistant(m) => &m.content, Message::Tool(m) => &m.content, - }; - content - .iter() - .filter_map(ContentBlock::as_text) - .map(|t| t.chars().count()) - .sum() + Message::Custom(_) => &[], + } } /// Approximate character weight of provider-visible content and structural @@ -181,13 +201,11 @@ impl Message { /// silently never trigger even as the real context window overflows. See /// [`ContentBlock::estimated_char_weight`]. pub fn estimated_char_weight(&self) -> usize { - let content = match self { - Message::System(m) => &m.content, - Message::User(m) => &m.content, - Message::Assistant(m) => &m.content, - Message::Tool(m) => &m.content, - }; - let content_weight: usize = content + if let Message::Custom(m) = self { + return m.display.as_deref().map_or(0, |d| d.chars().count()); + } + let content_weight: usize = self + .content_blocks() .iter() .map(ContentBlock::estimated_char_weight) .sum(); diff --git a/crates/tinyinference-llm/src/message/test.rs b/crates/tinyinference-llm/src/message/test.rs index b5703a1..cbe72ac 100644 --- a/crates/tinyinference-llm/src/message/test.rs +++ b/crates/tinyinference-llm/src/message/test.rs @@ -6,7 +6,7 @@ //! messages preserve their call id, and the [`MessageDelta`] default. use super::*; -use crate::tool::ToolCall; +use crate::tool::{ToolCall, ToolSchema}; use crate::usage::Usage; use serde_json::json; @@ -96,6 +96,7 @@ fn assistant_holds_tool_calls_and_usage() { content: vec![ContentBlock::Text("calling".into())], tool_calls: vec![ToolCall::new("c-1", "lookup", json!({}))], usage: Some(Usage::new(5, 5)), + origin: None, }); if let Message::Assistant(a) = &msg { assert_eq!(a.tool_calls.len(), 1); @@ -137,6 +138,7 @@ fn text_ignores_thinking_blocks() { ], tool_calls: Vec::new(), usage: None, + origin: None, }); // Reasoning blocks must never leak into visible text. assert_eq!(msg.text(), "the answer is 42"); @@ -195,6 +197,80 @@ fn thinking_block_serde_round_trips() { assert_eq!(back, redacted); } +#[test] +fn media_ref_constructors_and_media_type_accessor() { + let url = MediaRef::url("https://example.com/a.wav"); + assert_eq!(url.media_type(), None); + + let base64 = MediaRef::base64("AAAA", "audio/wav"); + assert_eq!(base64.media_type(), Some("audio/wav")); + + let path = MediaRef::path("/tmp/a.pdf"); + assert_eq!(path.media_type(), None); +} + +#[test] +fn audio_video_document_blocks_round_trip_through_json() { + let blocks = vec![ + ContentBlock::Audio(MediaRef::base64("AAAA", "audio/wav")), + ContentBlock::Video(MediaRef::url("https://example.com/v.mp4")), + ContentBlock::Document(MediaRef::path("/tmp/doc.pdf")), + ]; + for block in blocks { + let wire = serde_json::to_value(&block).unwrap(); + let back: ContentBlock = serde_json::from_value(wire).unwrap(); + assert_eq!(back, block); + } +} + +#[test] +fn non_text_media_blocks_are_not_reasoning_and_carry_no_visible_text() { + let block = ContentBlock::Audio(MediaRef::url("https://example.com/a.wav")); + assert!(!block.is_reasoning()); + assert_eq!(block.as_text(), None); +} + +#[test] +fn custom_message_text_uses_display_and_carries_no_content_blocks() { + let custom = Message::Custom(CustomMessage { + kind: "compaction".into(), + payload: json!({"summary": "..."}), + display: Some("Compacted 40 turns".into()), + }); + assert_eq!(custom.text(), "Compacted 40 turns"); + assert_eq!(custom.char_len(), "Compacted 40 turns".chars().count()); + assert_eq!( + custom.estimated_char_weight(), + "Compacted 40 turns".chars().count() + ); + assert!(custom.artifact().is_none()); + + let no_display = Message::Custom(CustomMessage { + kind: "label".into(), + payload: json!({"name": "checkpoint"}), + display: None, + }); + assert_eq!(no_display.text(), ""); + assert_eq!(no_display.char_len(), 0); + assert_eq!(no_display.estimated_char_weight(), 0); +} + +#[test] +fn custom_message_round_trips_through_serde() { + let custom = Message::Custom(CustomMessage { + kind: "audit".into(), + payload: json!({"note": "reviewed"}), + display: None, + }); + let wire = serde_json::to_value(&custom).unwrap(); + assert_eq!( + wire, + json!({ "custom": { "kind": "audit", "payload": { "note": "reviewed" } } }) + ); + let back: Message = serde_json::from_value(wire).unwrap(); + assert_eq!(back, custom); +} + #[test] fn legacy_content_without_thinking_still_parses() { // Additive tagging: transcripts serialized before thinking blocks existed @@ -208,3 +284,187 @@ fn legacy_content_without_thinking_still_parses() { assert_eq!(blocks[0].as_text(), Some("hello")); assert!(!blocks[1].is_reasoning()); } + +// --------------------------------------------------------------------------- +// SystemMessage sections/tool deltas, replay_system_state (B6) +// --------------------------------------------------------------------------- + +fn tool(name: &str) -> ToolSchema { + ToolSchema::new(name, format!("{name} tool"), json!({"type": "object"})) +} + +#[test] +fn legacy_system_message_without_new_fields_deserializes_as_a_no_op_patch() { + // A transcript persisted before SystemMessage gained sections/tool deltas + // must still deserialize: the new fields are all `#[serde(default)]`. + let legacy = json!({ "content": [{ "text": "you are a helpful assistant" }] }); + let msg: SystemMessage = serde_json::from_value(legacy).unwrap(); + assert_eq!( + msg.content, + vec![ContentBlock::Text("you are a helpful assistant".into())] + ); + assert!(msg.sections.is_empty()); + assert!(msg.tools_added.is_empty()); + assert!(msg.tools_removed.is_empty()); +} + +#[test] +fn system_message_text_constructor_is_a_no_op_patch() { + let msg = SystemMessage::text("hello"); + assert!(!msg.is_empty_patch()); + assert!(msg.sections.is_empty()); + assert!(msg.tools_added.is_empty()); + assert!(msg.tools_removed.is_empty()); + + assert!(SystemMessage::default().is_empty_patch()); +} + +#[test] +fn replay_system_state_folds_sections_and_tool_deltas_in_order() { + // Turn 1: baseline persona plus two tools. + let base = SystemMessage { + tools_added: vec![tool("search"), tool("read_file")], + ..SystemMessage::text("You are Aria.") + }; + let turn1 = vec![Message::System(base), Message::user("hi")]; + + // Turn 2: a patch adds a "browse" tool, drops "read_file", and adds a + // named instructions section. + let mut sections = std::collections::BTreeMap::new(); + sections.insert( + "tool_changes".to_string(), + Some("browse is now available.".to_string()), + ); + let patch = SystemMessage { + tools_added: vec![tool("browse")], + tools_removed: vec!["read_file".to_string()], + sections, + ..SystemMessage::default() + }; + let mut messages = turn1; + messages.push(Message::assistant("ok")); + messages.push(Message::System(patch)); + messages.push(Message::user("go")); + + let (prompt, tools) = replay_system_state(&messages); + + let names: Vec<&str> = tools.iter().map(|t| t.name.as_str()).collect(); + assert_eq!(names, vec!["search", "browse"]); + assert!(prompt.contains("You are Aria.")); + assert!(prompt.contains("tool_changes")); + assert!(prompt.contains("browse is now available.")); +} + +#[test] +fn replay_system_state_section_removal_drops_it_from_the_effective_prompt() { + let mut add_sections = std::collections::BTreeMap::new(); + add_sections.insert("scratch".to_string(), Some("temporary note".to_string())); + let add = SystemMessage { + sections: add_sections, + ..SystemMessage::default() + }; + let mut remove_sections = std::collections::BTreeMap::new(); + remove_sections.insert("scratch".to_string(), None); + let remove = SystemMessage { + sections: remove_sections, + ..SystemMessage::default() + }; + + let messages = vec![Message::System(add), Message::System(remove)]; + let (prompt, tools) = replay_system_state(&messages); + assert!(!prompt.contains("temporary note")); + assert!(tools.is_empty()); +} + +#[test] +fn replay_system_state_later_tool_schema_for_same_name_wins() { + let first = SystemMessage { + tools_added: vec![ToolSchema::new("search", "v1", json!({"type": "object"}))], + ..SystemMessage::default() + }; + let second = SystemMessage { + tools_added: vec![ToolSchema::new("search", "v2", json!({"type": "object"}))], + ..SystemMessage::default() + }; + + let messages = vec![Message::System(first), Message::System(second)]; + let (_, tools) = replay_system_state(&messages); + assert_eq!(tools.len(), 1); + assert_eq!(tools[0].description, "v2"); +} + +#[test] +fn replay_system_state_ignores_non_system_messages() { + let messages = vec![ + Message::user("hello"), + Message::assistant("hi"), + Message::tool("c-1", "result"), + ]; + let (prompt, tools) = replay_system_state(&messages); + assert!(prompt.is_empty()); + assert!(tools.is_empty()); +} + +#[test] +fn model_profile_default_disallows_mid_conversation_system_messages() { + // Conservative default: a caller must opt in per-provider (see + // `providers::openai::transport::derive_profile`), because folding into + // the leading system message is always correct while inserting one where + // the provider does not actually honor it silently loses content. + assert!(!crate::model::ModelProfile::default().mid_conversation_system_messages); +} + +// --------------------------------------------------------------------------- +// AssistantMessage origin metadata +// --------------------------------------------------------------------------- + +#[test] +fn message_origin_round_trips_through_serde() { + let assistant = AssistantMessage { + id: Some("msg_1".into()), + content: vec![ContentBlock::Text("hi".into())], + tool_calls: Vec::new(), + usage: None, + origin: Some(MessageOrigin { + provider: "anthropic".into(), + api: "messages".into(), + model: "claude-sonnet-4-6".into(), + }), + }; + let wire = serde_json::to_value(&assistant).unwrap(); + assert_eq!( + wire["origin"], + json!({ "provider": "anthropic", "api": "messages", "model": "claude-sonnet-4-6" }) + ); + let back: AssistantMessage = serde_json::from_value(wire).unwrap(); + assert_eq!(back, assistant); +} + +#[test] +fn message_origin_is_none_by_default_and_omitted_from_wire() { + let assistant = AssistantMessage { + id: None, + content: vec![ContentBlock::Text("hi".into())], + tool_calls: Vec::new(), + usage: None, + origin: None, + }; + let wire = serde_json::to_value(&assistant).unwrap(); + assert!( + wire.get("origin").is_none(), + "a None origin must not appear on the wire" + ); +} + +#[test] +fn legacy_assistant_message_without_origin_field_deserializes_with_none() { + // Additive tagging: a journal serialized before `origin` existed must + // still deserialize, with the field defaulting to `None`. + let legacy = json!({ + "id": "msg_1", + "content": [{ "text": "hi" }], + "tool_calls": [], + }); + let assistant: AssistantMessage = serde_json::from_value(legacy).unwrap(); + assert_eq!(assistant.origin, None); +} diff --git a/crates/tinyinference-llm/src/message/types.rs b/crates/tinyinference-llm/src/message/types.rs index b444ef8..beed646 100644 --- a/crates/tinyinference-llm/src/message/types.rs +++ b/crates/tinyinference-llm/src/message/types.rs @@ -9,10 +9,12 @@ //! Ergonomic constructors ([`Message::system`], [`Message::user`], …) and a //! [`Message::text`] accessor keep the public surface easy to use. +use std::collections::BTreeMap; + use serde::{Deserialize, Serialize}; use serde_json::Value; -use crate::tool::ToolCall; +use crate::tool::{ToolCall, ToolSchema}; use crate::usage::Usage; /// A typed unit of message content. @@ -50,6 +52,12 @@ pub enum ContentBlock { }, /// An opaque provider-specific block preserved verbatim. ProviderExtension(Value), + /// A reference to an audio clip. + Audio(MediaRef), + /// A reference to a video clip. + Video(MediaRef), + /// A reference to a document (PDF and similar). + Document(MediaRef), } /// A reference to an image, either by URL or inline base64 data. @@ -62,11 +70,222 @@ pub struct ImageRef { pub mime_type: Option, } -/// A system/developer instruction message. +/// A reference to a non-text media asset (audio, video, or document), +/// carried by [`ContentBlock::Audio`]/[`ContentBlock::Video`]/ +/// [`ContentBlock::Document`]. +/// +/// The harness itself never fetches [`MediaRef::Url`] or +/// [`MediaRef::Path`] content; a host that needs to resolve those into bytes +/// (for example to inline them for a provider whose wire format requires +/// base64) owns that fetch and its safety policy. #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case", tag = "source")] +pub enum MediaRef { + /// A remote or data URL. + Url { + /// The URL (or data URI) to fetch. + url: String, + /// Optional MIME type (for example `audio/wav`), when known. + #[serde(default, skip_serializing_if = "Option::is_none")] + media_type: Option, + }, + /// Inline base64-encoded content. + Base64 { + /// Base64-encoded bytes. + data: String, + /// MIME type of the decoded content (for example + /// `application/pdf`). + media_type: String, + }, + /// A local filesystem path. Only meaningful to a host that has + /// filesystem access and chooses to resolve it; providers never see raw + /// paths and a host must inline the file's bytes before sending it. + Path { + /// The filesystem path. + path: String, + /// Optional MIME type, when known. + #[serde(default, skip_serializing_if = "Option::is_none")] + media_type: Option, + }, +} + +impl MediaRef { + /// Creates a [`MediaRef::Url`] reference. + #[must_use] + pub fn url(url: impl Into) -> Self { + Self::Url { + url: url.into(), + media_type: None, + } + } + + /// Creates a [`MediaRef::Base64`] reference. + #[must_use] + pub fn base64(data: impl Into, media_type: impl Into) -> Self { + Self::Base64 { + data: data.into(), + media_type: media_type.into(), + } + } + + /// Creates a [`MediaRef::Path`] reference. + #[must_use] + pub fn path(path: impl Into) -> Self { + Self::Path { + path: path.into(), + media_type: None, + } + } + + /// Returns the MIME type, when known. + #[must_use] + pub fn media_type(&self) -> Option<&str> { + match self { + Self::Url { media_type, .. } | Self::Path { media_type, .. } => media_type.as_deref(), + Self::Base64 { media_type, .. } => Some(media_type.as_str()), + } + } +} + +/// A system/developer instruction message. +/// +/// A transcript may carry more than one `SystemMessage`: the leading one +/// establishes the run's baseline instructions, and any later one is a +/// **patch** that layers additively onto everything before it (see +/// [`replay_system_state`]). This is what lets a mid-run change — a toolset +/// gaining or losing a tool, an instructions section being added or revised — +/// be expressed as a small delta appended to (or inserted into) the +/// transcript instead of rewriting the leading system message and busting a +/// provider's cached prefix. +/// +/// The additional fields are all `#[serde(default)]` so a transcript +/// persisted before this type gained them deserializes unchanged (every +/// existing `SystemMessage` reads back with empty `sections`, +/// `tools_added`, and `tools_removed` — i.e. a no-op patch). +#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)] pub struct SystemMessage { /// Ordered content blocks. + #[serde(default)] pub content: Vec, + /// Named content sections this message contributes. + /// + /// Keyed by a stable section name (for example `"tool_changes"` or + /// `"persona"`). `Some(text)` sets or replaces the section's content; + /// `None` removes a section a prior message in the transcript + /// established. [`BTreeMap`] keeps replay deterministic regardless of + /// insertion order. + #[serde(default, skip_serializing_if = "BTreeMap::is_empty")] + pub sections: BTreeMap>, + /// Tool schemas this message adds to the effective tool set. + /// + /// A name already present is replaced (the newer declaration wins), so a + /// patch can both add a brand-new tool and republish a changed schema for + /// an existing one. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub tools_added: Vec, + /// Names of tools this message removes from the effective tool set. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub tools_removed: Vec, +} + +impl SystemMessage { + /// Creates a plain-text system message with no sections or tool deltas. + pub fn text(content: impl Into) -> Self { + Self { + content: vec![ContentBlock::Text(content.into())], + sections: BTreeMap::new(), + tools_added: Vec::new(), + tools_removed: Vec::new(), + } + } + + /// Returns `true` when this message carries no content, sections, or tool + /// deltas — i.e. replaying it would be a pure no-op. + pub fn is_empty_patch(&self) -> bool { + self.content.is_empty() + && self.sections.is_empty() + && self.tools_added.is_empty() + && self.tools_removed.is_empty() + } +} + +/// Walks `messages` and folds every [`SystemMessage`] in order into one +/// effective `SystemState`: the reconstructed named sections and tool set a +/// live run would have after processing the same sequence of patches. +/// +/// This is the read-side counterpart to a `declare_tool_changes`-style +/// writer (see `tinyagents-harness::agent_loop`): given only the transcript, +/// it answers "what system prompt and tool set was actually in effect" — +/// which is what makes the transcript itself the durable record of tool +/// loadout changes, rather than requiring an out-of-band log. +/// +/// Non-system messages are ignored. The leading free-form `content` text of +/// every `SystemMessage` (patches included) is concatenated in order, +/// separated by blank lines, ahead of the rendered named sections — a patch +/// that only adds `content` (no `sections`) still contributes its text. +pub fn replay_system_state(messages: &[Message]) -> (String, Vec) { + let mut leading_content = String::new(); + let mut section_order: Vec = Vec::new(); + let mut section_text: BTreeMap = BTreeMap::new(); + let mut tool_order: Vec = Vec::new(); + let mut tool_by_name: BTreeMap = BTreeMap::new(); + + for message in messages { + let Message::System(system) = message else { + continue; + }; + let text = super::concat_text(&system.content); + if !text.is_empty() { + if !leading_content.is_empty() { + leading_content.push_str("\n\n"); + } + leading_content.push_str(&text); + } + for (name, value) in &system.sections { + match value { + Some(text) => { + if !section_text.contains_key(name) { + section_order.push(name.clone()); + } + section_text.insert(name.clone(), text.clone()); + } + None => { + section_text.remove(name); + section_order.retain(|existing| existing != name); + } + } + } + for tool in &system.tools_added { + if !tool_by_name.contains_key(&tool.name) { + tool_order.push(tool.name.clone()); + } + tool_by_name.insert(tool.name.clone(), tool.clone()); + } + for name in &system.tools_removed { + tool_by_name.remove(name); + tool_order.retain(|existing| existing != name); + } + } + + let tools: Vec = tool_order + .into_iter() + .filter_map(|name| tool_by_name.remove(&name)) + .collect(); + + let mut parts = Vec::new(); + if !leading_content.is_empty() { + parts.push(leading_content); + } + for name in section_order { + if let Some(text) = section_text.remove(&name) + && !text.is_empty() + { + parts.push(format!("{name}\n\n{text}")); + } + } + let prompt = parts.join("\n\n"); + + (prompt, tools) } /// A user/human input message. @@ -90,6 +309,35 @@ pub struct AssistantMessage { /// Token usage reported for this message, when known. #[serde(default, skip_serializing_if = "Option::is_none")] pub usage: Option, + /// The provider/api/model that produced this message, when known. + /// + /// Stamped by the provider adapter that built the response (unary or the + /// terminal item of a stream). Absent for messages authored by the host + /// (for example a synthesized system/user turn) or replayed from a + /// journal written before this field existed — both are `None` rather + /// than a guessed value. A cross-provider harness compares this against + /// the target model's origin before replaying the message on a different + /// provider; see `tinyagents_harness::agent_loop::handoff_transform`. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub origin: Option, +} + +/// The provider/api/model that produced an [`AssistantMessage`]. +/// +/// Used to detect a mid-session provider or model switch so a cross-provider +/// handoff transform can drop or rewrite content the new target cannot +/// replay (signed thinking, provider-specific tool-call id shapes, and so +/// on). Equality is structural: two origins are the same only when all three +/// fields match exactly. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct MessageOrigin { + /// Provider family identifier (for example `openai`, `anthropic`). + pub provider: String, + /// API surface used for the call (for example `chat_completions`, + /// `responses`, `messages`). + pub api: String, + /// Provider model id that produced the message. + pub model: String, } /// A tool result message correlated to a prior tool call. @@ -123,6 +371,27 @@ pub enum Message { Assistant(AssistantMessage), /// Tool result. Tool(ToolMessage), + /// Host-defined out-of-band record (e.g. a compaction marker, a label, or + /// an audit note) that rides in the same message stream as ordinary + /// conversation turns but is never sent to a provider. + /// + /// Every request-building path (provider `convert`/`request` modules) + /// filters these out before serializing a provider payload; see the + /// `sanitize_history` / request-conversion call sites in each provider + /// module for the enforcement point. + Custom(CustomMessage), +} + +/// Payload for [`Message::Custom`]. +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct CustomMessage { + /// Host-defined discriminator, e.g. `"compaction"` or `"label"`. + pub kind: String, + /// Host-defined structured payload. + pub payload: Value, + /// Optional human-readable rendering for transcript display. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub display: Option, } /// An incremental message update used for streaming model output. diff --git a/crates/tinyinference-llm/src/model/mod.rs b/crates/tinyinference-llm/src/model/mod.rs index 115c664..f543807 100644 --- a/crates/tinyinference-llm/src/model/mod.rs +++ b/crates/tinyinference-llm/src/model/mod.rs @@ -84,6 +84,37 @@ fn matches_context_pattern(lower: &str, pattern: &str, mode: ContextPatternMatch } } +/// Derives the compatibility [`crate::message::MessageDelta`] for a block-aware +/// [`ModelStreamItem::BlockDelta`] fragment. +/// +/// Block-aware adapters (Anthropic and, incrementally, the OpenAI adapters) +/// emit both channels for the same fragment: the block-indexed item for +/// consumers that track block boundaries, and the flat delta this helper +/// builds for consumers (including [`StreamAccumulator`]) that only +/// understand the pre-existing shape. `call_id` and `tool_name` are only +/// meaningful for [`BlockDelta::ToolArgs`] and are ignored otherwise. +#[must_use] +pub fn block_delta_to_message_delta( + delta: &BlockDelta, + call_id: &str, + tool_name: Option<&str>, +) -> crate::message::MessageDelta { + match delta { + BlockDelta::Text(text) => crate::message::MessageDelta::text(text.clone()), + BlockDelta::Thinking(text) => crate::message::MessageDelta::reasoning(text.clone()), + BlockDelta::ToolArgs(content) => crate::message::MessageDelta { + text: String::new(), + reasoning: String::new(), + tool_call: Some(crate::tool::ToolDelta { + call_id: call_id.to_string(), + content: content.clone(), + tool_name: tool_name.map(str::to_string), + content_index: None, + }), + }, + } +} + /// Returns a generic context-window hint for a raw provider model id. /// /// Returns `None` for unknown ids rather than guessing. Hosts with product tier @@ -339,6 +370,9 @@ impl ModelProfile { image_out: true, audio_in: true, audio_out: true, + video_in: true, + video_out: true, + document_in: true, }, tool_calling: true, parallel_tool_calls: true, @@ -550,6 +584,7 @@ impl ModelResponse { content: vec![ContentBlock::Text(content.into())], tool_calls: Vec::new(), usage: None, + origin: None, }, usage: None, finish_reason: None, @@ -661,6 +696,8 @@ pub struct StreamAccumulator { /// [`crate::Error::Provider`] and preserve the /// status/code/`retryable` classification the retry layer needs. failed_provider: Option, + /// Terminal deferral, when a [`ModelStreamItem::Deferred`] item was seen. + deferred: Option, } impl StreamAccumulator { @@ -690,6 +727,13 @@ impl StreamAccumulator { ModelStreamItem::UsageDelta(usage) => { self.usage = Some(*usage); } + // Block-boundary items carry no information the accumulator needs: + // every fragment a block-aware adapter emits as `BlockDelta` is + // also folded into the compatibility `MessageDelta`/`ToolCallDelta` + // channel handled above, so reconstruction here is unaffected. + ModelStreamItem::BlockStart { .. } + | ModelStreamItem::BlockDelta { .. } + | ModelStreamItem::BlockEnd { .. } => {} ModelStreamItem::Completed(response) => { self.completed = Some(response.clone()); } @@ -703,9 +747,19 @@ impl StreamAccumulator { // `insufficient_quota` / 400 must not be retried as transient). self.failed_provider = Some(error.clone()); } + ModelStreamItem::Deferred(handle) => { + self.deferred = Some(handle.clone()); + } } } + /// Returns the [`DeferredHandle`] folded in by a + /// [`ModelStreamItem::Deferred`] item, when one was seen. + #[must_use] + pub fn deferred(&self) -> Option<&DeferredHandle> { + self.deferred.as_ref() + } + /// Appends a tool-call argument fragment for `call_id`, preserving /// first-seen ordering across calls. Records the first non-empty `tool_name` /// seen for the call (call-opening deltas carry it; argument fragments do @@ -757,6 +811,13 @@ impl StreamAccumulator { return Err(crate::Error::Model(message)); } + if let Some(handle) = self.deferred { + return Err(crate::Error::Unsupported(format!( + "stream deferred call {handle:?}; call `deferred()` before `finish()` and \ + resolve it via `ChatModel::fetch_deferred`" + ))); + } + if let Some(mut response) = self.completed { // Reconcile the response and message usage with any streamed // `UsageDelta`, preferring an already-present value and never @@ -801,6 +862,7 @@ impl StreamAccumulator { content, tool_calls, usage: self.usage, + origin: None, }; Ok(ModelResponse { message, diff --git a/crates/tinyinference-llm/src/model/test.rs b/crates/tinyinference-llm/src/model/test.rs index eba820d..51c4fc9 100644 --- a/crates/tinyinference-llm/src/model/test.rs +++ b/crates/tinyinference-llm/src/model/test.rs @@ -457,11 +457,13 @@ fn finish_names_reconstructed_tool_call_from_the_call_opening_delta_name() { call_id: "call-1".into(), content: String::new(), tool_name: Some("search".into()), + ..Default::default() })); acc.push(&ModelStreamItem::ToolCallDelta(ToolDelta { call_id: "call-1".into(), content: r#"{"q":"rust"}"#.into(), tool_name: None, + ..Default::default() })); let finished = acc.finish().unwrap(); @@ -479,6 +481,7 @@ fn finish_marks_malformed_reconstructed_tool_arguments_invalid() { call_id: "call-1".into(), content: "{broken".into(), tool_name: Some("search".into()), + ..Default::default() })); let response = accumulator.finish().unwrap(); let call = &response.message.tool_calls[0]; @@ -506,8 +509,28 @@ fn model_stream_item_roundtrips_every_variant() { call_id: "call-1".into(), content: "{\"q\":1}".into(), tool_name: None, + ..Default::default() })); roundtrip_stream_item(ModelStreamItem::UsageDelta(Usage::new(3, 5))); + roundtrip_stream_item(ModelStreamItem::BlockStart { + index: 0, + kind: crate::model::BlockKind::Text, + }); + roundtrip_stream_item(ModelStreamItem::BlockStart { + index: 1, + kind: crate::model::BlockKind::ToolCall { + id: "call-1".into(), + name: "search".into(), + }, + }); + roundtrip_stream_item(ModelStreamItem::BlockDelta { + index: 0, + delta: crate::model::BlockDelta::ToolArgs("{}".into()), + }); + roundtrip_stream_item(ModelStreamItem::BlockEnd { + index: 0, + block: crate::message::ContentBlock::Text("done".into()), + }); roundtrip_stream_item(ModelStreamItem::Completed(ModelResponse::assistant("done"))); // The scalar-carrying variant an internally tagged enum could not encode. roundtrip_stream_item(ModelStreamItem::Failed("boom".to_string())); @@ -516,6 +539,40 @@ fn model_stream_item_roundtrips_every_variant() { message: "nope".into(), ..ProviderError::default() })); + roundtrip_stream_item(ModelStreamItem::ProviderFailed(ProviderError { + provider: "anthropic".into(), + message: "overloaded".into(), + stop_reason: Some("pause_turn".into()), + partial_message: Some(crate::message::AssistantMessage { + id: Some("msg_1".into()), + content: vec![crate::message::ContentBlock::Text("partial".into())], + tool_calls: Vec::new(), + usage: None, + origin: None, + }), + ..ProviderError::default() + })); +} + +#[test] +fn block_delta_to_message_delta_maps_each_channel() { + use crate::model::{BlockDelta, block_delta_to_message_delta}; + + let text = block_delta_to_message_delta(&BlockDelta::Text("hi".into()), "", None); + assert_eq!(text.text, "hi"); + assert!(text.reasoning.is_empty()); + assert!(text.tool_call.is_none()); + + let thinking = block_delta_to_message_delta(&BlockDelta::Thinking("plan".into()), "", None); + assert_eq!(thinking.reasoning, "plan"); + assert!(thinking.text.is_empty()); + + let args = + block_delta_to_message_delta(&BlockDelta::ToolArgs("{}".into()), "call-1", Some("s")); + let tool_call = args.tool_call.expect("tool_call fragment"); + assert_eq!(tool_call.call_id, "call-1"); + assert_eq!(tool_call.content, "{}"); + assert_eq!(tool_call.tool_name.as_deref(), Some("s")); } #[test] @@ -569,3 +626,147 @@ fn stream_accumulator_reconstruct_without_reasoning_has_no_thinking_block() { vec![ContentBlock::Text("hi".into())] ); } + +#[test] +fn model_profile_new_fields_default_to_none_or_false() { + let profile = ModelProfile::default(); + assert!(profile.schema_transform.is_none()); + assert!(profile.default_structured_mode.is_none()); + assert!(profile.prompted_output_template.is_none()); + assert!(profile.thinking_tags.is_none()); + assert!(!profile.ignore_streamed_leading_whitespace); + assert!(profile.thinking_level_map.is_empty()); + assert_eq!(profile.compat, ProviderCompat::default()); +} + +#[test] +fn model_profile_round_trips_new_fields_through_json() { + let mut profile = ModelProfile { + schema_transform: Some(SchemaTransform::Chain(vec![ + SchemaTransform::InlineRefs, + SchemaTransform::NoAdditionalProperties, + ])), + default_structured_mode: Some(StructuredMode::Prompted), + prompted_output_template: Some("Respond as JSON matching: {schema}".into()), + thinking_tags: Some(("".into(), "".into())), + ignore_streamed_leading_whitespace: true, + ..ModelProfile::default() + }; + profile + .thinking_level_map + .insert("low".into(), ReasoningConfig::effort(ReasoningEffort::Low)); + profile.compat = ProviderCompat { + mid_conversation_system_messages: true, + strict_tools: true, + cache_retention: false, + session_affinity: true, + max_tool_name_length: Some(64), + tool_id_pattern: Some("^[a-z0-9_]+$".into()), + }; + + let json = serde_json::to_string(&profile).unwrap(); + let round_tripped: ModelProfile = serde_json::from_str(&json).unwrap(); + assert_eq!(round_tripped, profile); +} + +#[test] +fn schema_transform_strip_defs_removes_top_level_defs() { + let schema = json!({ + "type": "object", + "$defs": {"Foo": {"type": "string"}}, + "properties": {"a": {"$ref": "#/$defs/Foo"}} + }); + let out = SchemaTransform::StripDefs.apply(&schema); + assert!(out.get("$defs").is_none()); + // Ref itself is left untouched by StripDefs (that's InlineRefs' job). + assert_eq!(out["properties"]["a"]["$ref"], "#/$defs/Foo"); +} + +#[test] +fn schema_transform_inline_refs_resolves_and_drops_defs() { + let schema = json!({ + "type": "object", + "$defs": {"Foo": {"type": "string", "minLength": 1}}, + "properties": {"a": {"$ref": "#/$defs/Foo"}} + }); + let out = SchemaTransform::InlineRefs.apply(&schema); + assert!(out.get("$defs").is_none()); + assert_eq!(out["properties"]["a"]["type"], "string"); + assert_eq!(out["properties"]["a"]["minLength"], 1); +} + +#[test] +fn schema_transform_no_additional_properties_sets_false_recursively() { + let schema = json!({ + "type": "object", + "properties": { + "nested": {"type": "object", "properties": {"x": {"type": "string"}}} + } + }); + let out = SchemaTransform::NoAdditionalProperties.apply(&schema); + assert_eq!(out["additionalProperties"], false); + assert_eq!(out["properties"]["nested"]["additionalProperties"], false); +} + +#[test] +fn schema_transform_no_additional_properties_respects_existing_value() { + let schema = json!({"type": "object", "additionalProperties": true}); + let out = SchemaTransform::NoAdditionalProperties.apply(&schema); + assert_eq!(out["additionalProperties"], true); +} + +#[test] +fn schema_transform_gemini_compat_strips_unsupported_keywords() { + let schema = json!({ + "type": "object", + "additionalProperties": false, + "$schema": "http://json-schema.org/draft-07/schema#", + "default": {}, + "properties": {"a": {"type": "string", "examples": ["x"]}} + }); + let out = SchemaTransform::GeminiCompat.apply(&schema); + assert!(out.get("additionalProperties").is_none()); + assert!(out.get("$schema").is_none()); + assert!(out.get("default").is_none()); + assert!(out["properties"]["a"].get("examples").is_none()); +} + +#[test] +fn schema_transform_openai_strict_inlines_forbids_extra_and_requires_all() { + let schema = json!({ + "type": "object", + "$defs": {"Foo": {"type": "string"}}, + "properties": { + "a": {"$ref": "#/$defs/Foo"}, + "b": {"type": "number"} + } + }); + let out = SchemaTransform::OpenAiStrict.apply(&schema); + assert!(out.get("$defs").is_none()); + assert_eq!(out["additionalProperties"], false); + assert_eq!(out["properties"]["a"]["type"], "string"); + let required: Vec = out["required"] + .as_array() + .unwrap() + .iter() + .map(|v| v.as_str().unwrap().to_string()) + .collect(); + assert_eq!(required, vec!["a".to_string(), "b".to_string()]); +} + +#[test] +fn schema_transform_chain_applies_steps_in_order() { + let schema = json!({ + "type": "object", + "$defs": {"Foo": {"type": "string"}}, + "properties": {"a": {"$ref": "#/$defs/Foo"}} + }); + let chained = SchemaTransform::Chain(vec![ + SchemaTransform::InlineRefs, + SchemaTransform::NoAdditionalProperties, + ]) + .apply(&schema); + assert!(chained.get("$defs").is_none()); + assert_eq!(chained["properties"]["a"]["type"], "string"); + assert_eq!(chained["additionalProperties"], false); +} diff --git a/crates/tinyinference-llm/src/model/types.rs b/crates/tinyinference-llm/src/model/types.rs index 4e51699..e3b27a9 100644 --- a/crates/tinyinference-llm/src/model/types.rs +++ b/crates/tinyinference-llm/src/model/types.rs @@ -20,7 +20,7 @@ use serde_json::Value; use crate::Result; use crate::cache::CachePolicy; -use crate::message::{AssistantMessage, Message, MessageDelta}; +use crate::message::{AssistantMessage, ContentBlock, Message, MessageDelta}; use crate::tool::{ToolDelta, ToolSchema}; use crate::usage::Usage; @@ -163,6 +163,12 @@ pub struct Modalities { pub audio_in: bool, /// Produces audio output. pub audio_out: bool, + /// Accepts video input. + pub video_in: bool, + /// Produces video output. + pub video_out: bool, + /// Accepts document input (PDF and similar). + pub document_in: bool, } impl Default for Modalities { @@ -174,6 +180,9 @@ impl Default for Modalities { image_out: false, audio_in: false, audio_out: false, + video_in: false, + video_out: false, + document_in: false, } } } @@ -240,6 +249,285 @@ pub struct ModelProfile { /// Maximum output tokens, when known. #[serde(default, skip_serializing_if = "Option::is_none")] pub max_output_tokens: Option, + /// Whether the provider accepts a `system`/developer-role message + /// anywhere in the transcript's message list, not only as a single + /// leading block. + /// + /// This is `false` by default and must be opted into explicitly, because + /// getting it wrong silently drops content: an OpenAI-style chat API + /// (`Message::System` translated 1:1 to a `role: "system"` wire message + /// at its transcript position — see `providers::openai::convert:: + /// translate_message`) genuinely honors a system message wherever it + /// appears, so `true` is correct there. Anthropic's Messages API has no + /// such slot: every `Message::System` in the transcript, wherever it + /// occurs, is collected into one top-level `system` array ahead of the + /// `messages` list (see `providers::anthropic::request`), so a "mid- + /// conversation" system message is actually hoisted to the front on the + /// wire — `false` here is what tells a caller (the transcript-carried + /// system-patch mechanism, `docs/runtime-comparison/plan.md`'s B6) to + /// fold a patch into the leading system message instead of inserting it + /// in place. + #[serde(default)] + pub mid_conversation_system_messages: bool, + /// JSON-schema transform this model's adapter must apply before sending a + /// schema to the provider (for example stripping `$defs` a provider + /// rejects, or forcing `additionalProperties: false`). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub schema_transform: Option, + /// Structured-output strategy the harness should default to for this + /// model when the caller does not pin one explicitly. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub default_structured_mode: Option, + /// Prompt template used when `default_structured_mode` (or an explicit + /// override) resolves to [`StructuredMode::Prompted`]. Implementations + /// should substitute the target schema into this template. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub prompted_output_template: Option, + /// Open/close tag pair (for example `("", "")`) that this + /// model emits around chain-of-thought text. Response normalization + /// should extract tagged spans into a thinking content block rather than + /// leaving them inline in the visible text. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub thinking_tags: Option<(String, String)>, + /// Whether leading whitespace on the first streamed text delta is an + /// artifact of this provider's wire format and should be dropped rather + /// than surfaced to the caller. + #[serde(default)] + pub ignore_streamed_leading_whitespace: bool, + /// Maps a named reasoning/thinking level (for example `"low"`, + /// `"high"`, or a provider-specific label) to the [`ReasoningConfig`] it + /// expands to for this model. + #[serde(default, skip_serializing_if = "std::collections::BTreeMap::is_empty")] + pub thinking_level_map: std::collections::BTreeMap, + /// Provider-family compatibility quirks that do not fit the capability + /// model above. + #[serde(default)] + pub compat: ProviderCompat, + /// Regex a tool-call id must match to be accepted by this provider, when + /// the provider constrains the shape (for example Anthropic's + /// `^[a-zA-Z0-9_-]{1,64}$`). `None` means the provider imposes no shape + /// constraint beyond an opaque string. + /// + /// Consulted by a cross-provider handoff transform + /// (`tinyagents_harness::agent_loop::handoff_transform`) to decide + /// whether a tool-call id minted by a different provider needs + /// normalizing before replay; not enforced by this crate's own request + /// building. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub tool_call_id_pattern: Option, + /// Maximum accepted tool-call id length, when the provider constrains it. + /// Often redundant with a bound already encoded in + /// [`tool_call_id_pattern`](Self::tool_call_id_pattern) (as it is for + /// Anthropic's `{1,64}`), but kept separate so a normalizer can truncate + /// without needing to parse the regex. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub max_tool_call_id_len: Option, +} + +/// A named, serializable JSON-schema transform applied before a schema is +/// sent to a provider. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case", tag = "kind", content = "value")] +pub enum SchemaTransform { + /// Removes the top-level `$defs`/`definitions` map from the schema. + StripDefs, + /// Resolves `$ref` pointers into `$defs`/`definitions` inline, then + /// drops the now-unused definitions map. + InlineRefs, + /// Recursively sets `additionalProperties: false` on every object + /// schema that does not already specify it. + NoAdditionalProperties, + /// Applies the subset of adjustments Gemini's schema dialect requires: + /// strips `additionalProperties`, `$schema`, `default`, and `examples` + /// keywords it does not accept. + GeminiCompat, + /// Applies OpenAI "strict" JSON-schema mode: inlines refs, forces + /// `additionalProperties: false`, and marks every property required. + OpenAiStrict, + /// Applies a sequence of transforms in order. + Chain(Vec), +} + +impl SchemaTransform { + /// Applies this transform to `schema`, returning the transformed value. + /// The input is never mutated in place. + #[must_use] + pub fn apply(&self, schema: &Value) -> Value { + let mut out = schema.clone(); + match self { + Self::StripDefs => { + strip_defs(&mut out); + } + Self::InlineRefs => { + let defs = collect_defs(&out); + inline_refs(&mut out, &defs); + strip_defs(&mut out); + } + Self::NoAdditionalProperties => { + set_no_additional_properties(&mut out); + } + Self::GeminiCompat => { + strip_keys( + &mut out, + &["additionalProperties", "$schema", "default", "examples"], + ); + } + Self::OpenAiStrict => { + let defs = collect_defs(&out); + inline_refs(&mut out, &defs); + strip_defs(&mut out); + set_no_additional_properties(&mut out); + require_all_properties(&mut out); + } + Self::Chain(steps) => { + for step in steps { + out = step.apply(&out); + } + } + } + out + } +} + +fn strip_defs(value: &mut Value) { + if let Value::Object(map) = value { + map.remove("$defs"); + map.remove("definitions"); + for v in map.values_mut() { + strip_defs(v); + } + } else if let Value::Array(items) = value { + for v in items { + strip_defs(v); + } + } +} + +fn strip_keys(value: &mut Value, keys: &[&str]) { + if let Value::Object(map) = value { + for key in keys { + map.remove(*key); + } + for v in map.values_mut() { + strip_keys(v, keys); + } + } else if let Value::Array(items) = value { + for v in items { + strip_keys(v, keys); + } + } +} + +fn collect_defs(value: &Value) -> serde_json::Map { + let mut defs = serde_json::Map::new(); + if let Value::Object(map) = value { + if let Some(Value::Object(d)) = map.get("$defs") { + defs.extend(d.clone()); + } + if let Some(Value::Object(d)) = map.get("definitions") { + defs.extend(d.clone()); + } + } + defs +} + +fn inline_refs(value: &mut Value, defs: &serde_json::Map) { + match value { + Value::Object(map) => { + if let Some(Value::String(reference)) = map.get("$ref").cloned() { + let name = reference.rsplit('/').next().unwrap_or(reference.as_str()); + if let Some(resolved) = defs.get(name) { + let mut resolved = resolved.clone(); + inline_refs(&mut resolved, defs); + *value = resolved; + return; + } + } + for v in map.values_mut() { + inline_refs(v, defs); + } + } + Value::Array(items) => { + for v in items { + inline_refs(v, defs); + } + } + _ => {} + } +} + +fn set_no_additional_properties(value: &mut Value) { + if let Value::Object(map) = value { + let is_object_schema = matches!(map.get("type"), Some(Value::String(t)) if t == "object") + || map.contains_key("properties"); + if is_object_schema && !map.contains_key("additionalProperties") { + map.insert("additionalProperties".into(), Value::Bool(false)); + } + for v in map.values_mut() { + set_no_additional_properties(v); + } + } else if let Value::Array(items) = value { + for v in items { + set_no_additional_properties(v); + } + } +} + +fn require_all_properties(value: &mut Value) { + if let Value::Object(map) = value { + if let Some(Value::Object(props)) = map.get("properties").cloned() { + let required: Vec = props.keys().cloned().map(Value::String).collect(); + map.insert("required".into(), Value::Array(required)); + } + for v in map.values_mut() { + require_all_properties(v); + } + } else if let Value::Array(items) = value { + for v in items { + require_all_properties(v); + } + } +} + +/// The structured-output extraction strategy the harness should use for a +/// model call. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum StructuredMode { + /// Extract structured output via a synthetic tool call. + Tool, + /// Use the provider's native constrained-JSON output mode. + Native, + /// Ask for structured output via a prompt template and parse the + /// resulting text. + Prompted, +} + +/// Provider-family compatibility quirks that affect how the harness builds a +/// request, independent of the model's advertised capabilities. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct ProviderCompat { + /// Supports a `system` message appearing anywhere in the conversation, + /// not only as the first message. + #[serde(default)] + pub mid_conversation_system_messages: bool, + /// Supports strict tool-schema validation (every property required, + /// `additionalProperties: false` enforced by the provider). + #[serde(default)] + pub strict_tools: bool, + /// Supports explicit prompt-cache retention control. + #[serde(default)] + pub cache_retention: bool, + /// Requires calls for one logical session to land on the same backend + /// instance (sticky routing) to benefit from caching. + #[serde(default)] + pub session_affinity: bool, + /// Maximum length, in bytes, of a tool name this provider accepts. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub max_tool_name_length: Option, + /// Regex pattern tool-call ids from this provider must match. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub tool_id_pattern: Option, } /// A set of required capabilities used to validate a request against a @@ -598,6 +886,50 @@ pub struct ProviderError { /// Raw provider payload, when available. #[serde(default, skip_serializing_if = "Option::is_none")] pub raw: Option, + /// The assistant message accumulated from stream items before the + /// failure, when any content had arrived. Lets a caller keep (or discard) + /// partial work instead of losing every block a mid-stream failure + /// interrupted. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub partial_message: Option, + /// The stop/finish reason reported before the failure, when the provider + /// sent one (for example Anthropic's `message_delta.stop_reason`) prior + /// to the error that ended the stream. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub stop_reason: Option, +} + +/// The syntactic category of a streamed content block, established when the +/// block opens ([`ModelStreamItem::BlockStart`]). +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case", tag = "kind")] +pub enum BlockKind { + /// Visible assistant text. + Text, + /// Model reasoning/thinking content. + Thinking, + /// A tool call. The id and name are known as soon as the block opens + /// (Anthropic's `content_block_start`; OpenAI's first `tool_calls[]` + /// fragment for a wire index), before any argument fragments arrive. + ToolCall { + /// Provider-assigned call identifier. + id: String, + /// Tool name. + name: String, + }, +} + +/// An incremental fragment belonging to the open block named in the +/// accompanying [`ModelStreamItem::BlockDelta::index`]. +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case", tag = "type", content = "content")] +pub enum BlockDelta { + /// Visible text fragment. + Text(String), + /// Thinking/reasoning fragment. + Thinking(String), + /// Incremental tool-call argument JSON fragment. + ToolArgs(String), } /// A single item produced by a real, asynchronous model stream. @@ -633,18 +965,105 @@ pub enum ModelStreamItem { /// The stream has opened; no content has arrived yet. Started, /// An incremental message fragment (text and/or a tool-call fragment). + /// + /// Kept alongside [`ModelStreamItem::BlockDelta`] for backward + /// compatibility: every fragment a block-aware adapter emits as a + /// `BlockDelta` is also folded into a `MessageDelta` on the same channel + /// (see [`crate::model::block_delta_to_message_delta`]), so consumers that + /// only understand the flat delta shape keep working unchanged. MessageDelta(MessageDelta), - /// An incremental tool-call argument fragment correlated by call id. + /// An incremental tool-call argument fragment correlated by call id. When + /// the provider reports block-indexed content, [`ToolDelta::content_index`] + /// names the block this fragment belongs to. ToolCallDelta(ToolDelta), /// A usage update. Providers may send cumulative usage; the accumulator /// keeps the most recent value. UsageDelta(Usage), + /// A new content block has opened at `index`. Anthropic's + /// `content_block_start` maps to this 1:1; the OpenAI adapters derive it + /// from a delta's shape changing (text starting, or a new `tool_calls[]` + /// wire index appearing). + BlockStart { + /// Zero-based position of this block within the assistant message, + /// stable for the life of the block. + index: usize, + /// The block's syntactic category. + kind: BlockKind, + }, + /// An incremental fragment for the open block at `index`. + BlockDelta { + /// Position of the block this fragment belongs to. + index: usize, + /// The fragment payload. + delta: BlockDelta, + }, + /// The block at `index` has closed; `block` is its fully assembled + /// content, ready to append to an [`AssistantMessage::content`] in order. + BlockEnd { + /// Position of the closed block. + index: usize, + /// The finished content block. + block: ContentBlock, + }, /// Terminal success: the fully merged response. Completed(ModelResponse), /// Terminal failure with a human-readable error message. Failed(String), /// Terminal failure with normalized provider details. ProviderFailed(ProviderError), + /// Terminal deferral: the provider accepted the request but will finish + /// it asynchronously (for example an OpenAI batch or background + /// response). The caller polls or otherwise resolves the response later + /// via [`ChatModel::fetch_deferred`]. + Deferred(DeferredHandle), +} + +/// An opaque, provider-issued handle to a model call whose response is not +/// yet available (for example a queued batch job or a background response). +/// +/// The handle is deliberately provider-neutral and serializable so a host can +/// persist it and resume polling after a process restart. `id` is the only +/// field callers must treat as meaningful to the provider; `kind` and +/// `metadata` are informational. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct DeferredHandle { + /// Provider family identifier (for example `openai`). + pub provider: String, + /// Provider-issued identifier for the deferred call (batch id, response + /// id, or similar). + pub id: String, + /// Provider-specific deferral kind (for example `"batch"` or + /// `"background"`), when known. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub kind: Option, + /// Additional provider-specific metadata needed to resolve the handle. + #[serde(default, skip_serializing_if = "serde_json::Map::is_empty")] + pub metadata: serde_json::Map, +} + +impl DeferredHandle { + /// Creates a handle for `provider`/`id` with no kind or metadata. + #[must_use] + pub fn new(provider: impl Into, id: impl Into) -> Self { + Self { + provider: provider.into(), + id: id.into(), + kind: None, + metadata: serde_json::Map::new(), + } + } +} + +/// The current status of a previously deferred model call. +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(rename_all = "snake_case", tag = "status")] +pub enum DeferredStatus { + /// Still queued or in progress; not yet ready. + Pending, + /// Finished successfully. + Completed(Box), + /// Finished with a failure. + Failed(String), } /// A cancellation guard owned by a model stream. @@ -864,4 +1283,17 @@ pub trait ChatModel: Send + Sync { None => stream, }) } + + /// Resolves a previously issued [`DeferredHandle`] (see + /// [`ModelStreamItem::Deferred`]), returning the current + /// [`DeferredStatus`]. + /// + /// The default implementation returns [`crate::Error::Unsupported`]; only + /// adapters that can actually issue deferred calls (for example an + /// OpenAI batch/background adapter) should override this. + async fn fetch_deferred(&self, _handle: &DeferredHandle) -> Result { + Err(crate::Error::Unsupported( + "this model adapter does not support deferred calls".to_string(), + )) + } } diff --git a/crates/tinyinference-llm/src/network_guard.rs b/crates/tinyinference-llm/src/network_guard.rs new file mode 100644 index 0000000..f3c1a9c --- /dev/null +++ b/crates/tinyinference-llm/src/network_guard.rs @@ -0,0 +1,90 @@ +//! Process-wide guard that lets a host forbid network-backed model calls. +//! +//! Tests and evaluation harnesses often want a hard guarantee that a run +//! never reaches the network, regardless of which model a caller happened to +//! configure. [`deny_network_models`] sets a process-wide flag that +//! network-backed [`crate::model::ChatModel`] adapters (the OpenAI and +//! Anthropic providers) check before issuing any HTTP request, returning +//! [`crate::Error::Validation`] instead of dialing out. [`MockModel`](crate::providers::MockModel) +//! and other in-process adapters are unaffected because they never reach this +//! check. +//! +//! The guard is a single process-wide `AtomicBool`. It is intended for test +//! setup (for example a `#[ctor]`-style fixture or the first line of a test +//! module) rather than per-request policy. + +use std::sync::atomic::{AtomicBool, Ordering}; + +static NETWORK_MODELS_DENIED: AtomicBool = AtomicBool::new(false); + +/// Forbids network-backed model providers from issuing HTTP requests for the +/// remainder of the process. +/// +/// Idempotent: calling this more than once has no additional effect. Use +/// [`allow_network_models`] to lift the restriction (primarily useful for +/// resetting shared test state between cases). +pub fn deny_network_models() { + NETWORK_MODELS_DENIED.store(true, Ordering::SeqCst); +} + +/// Lifts a restriction previously installed by [`deny_network_models`]. +pub fn allow_network_models() { + NETWORK_MODELS_DENIED.store(false, Ordering::SeqCst); +} + +/// Returns whether [`deny_network_models`] is currently in effect. +#[must_use] +pub fn network_models_denied() -> bool { + NETWORK_MODELS_DENIED.load(Ordering::SeqCst) +} + +/// Returns [`crate::Error::Validation`] when network models are denied, +/// otherwise `Ok(())`. Network-backed provider adapters call this at the top +/// of every request-issuing path. +pub(crate) fn ensure_network_models_allowed() -> crate::Result<()> { + if network_models_denied() { + return Err(crate::Error::Validation( + "network-backed model calls are denied for this process; call \ + tinyinference_llm::allow_network_models() to lift the restriction" + .to_string(), + )); + } + Ok(()) +} + +#[cfg(test)] +mod test { + use super::*; + use std::sync::Mutex; + + // The guard is process-wide `static` state, so tests that flip it must + // not interleave with each other; serialize them with a mutex rather + // than relying on cargo test's default single-process, multi-thread + // execution to happen to avoid collisions. + static GUARD_TEST_LOCK: Mutex<()> = Mutex::new(()); + + #[test] + fn denies_and_allows_round_trip() { + let _lock = GUARD_TEST_LOCK.lock().unwrap(); + allow_network_models(); + assert!(!network_models_denied()); + assert!(ensure_network_models_allowed().is_ok()); + + deny_network_models(); + assert!(network_models_denied()); + assert!(ensure_network_models_allowed().is_err()); + + allow_network_models(); + assert!(!network_models_denied()); + assert!(ensure_network_models_allowed().is_ok()); + } + + #[test] + fn deny_is_idempotent() { + let _lock = GUARD_TEST_LOCK.lock().unwrap(); + deny_network_models(); + deny_network_models(); + assert!(network_models_denied()); + allow_network_models(); + } +} diff --git a/crates/tinyinference-llm/src/prompt_tools/mod.rs b/crates/tinyinference-llm/src/prompt_tools/mod.rs index 9199cdb..5eaca90 100644 --- a/crates/tinyinference-llm/src/prompt_tools/mod.rs +++ b/crates/tinyinference-llm/src/prompt_tools/mod.rs @@ -177,7 +177,8 @@ pub fn coalesce_tool_results(messages: &[Message]) -> Vec { /// [`TOOL_RESULTS_PREFIX`], and templates that look for a user query want a /// request to answer, not the transcript of a tool the model itself invoked — /// Qwen 3's template makes the same distinction. Neither does an empty or -/// whitespace-only turn. Non-text content (JSON, an image) does count. +/// whitespace-only turn. Non-text content (JSON, an image, audio, video, or a +/// document) does count. fn is_resolvable_user_query(message: &Message) -> bool { let Message::User(user) = message else { return false; @@ -191,7 +192,11 @@ fn is_resolvable_user_query(message: &Message) -> bool { } user.content.iter().any(|block| match block { ContentBlock::Text(text) => !text.trim().is_empty(), - ContentBlock::Json(_) | ContentBlock::Image(_) => true, + ContentBlock::Json(_) + | ContentBlock::Image(_) + | ContentBlock::Audio(_) + | ContentBlock::Video(_) + | ContentBlock::Document(_) => true, ContentBlock::Thinking { .. } | ContentBlock::RedactedThinking { .. } | ContentBlock::ProviderExtension(_) => false, diff --git a/crates/tinyinference-llm/src/providers/anthropic/mod.rs b/crates/tinyinference-llm/src/providers/anthropic/mod.rs index ed419f1..e9ff512 100644 --- a/crates/tinyinference-llm/src/providers/anthropic/mod.rs +++ b/crates/tinyinference-llm/src/providers/anthropic/mod.rs @@ -72,6 +72,15 @@ const DEFAULT_CONNECT_TIMEOUT_SECS: u64 = 30; /// prose — the previous 1,024 truncated real tool calls mid-argument. const DEFAULT_MAX_TOKENS: u32 = 4096; const PROVIDER: &str = "anthropic"; +/// [`crate::message::MessageOrigin::api`] value stamped on every response +/// this adapter builds (unary and streamed terminal). +pub(super) const MESSAGES_API: &str = "messages"; +/// Anthropic's accepted `tool_use.id`/`tool_result.tool_use_id` shape. +const TOOL_CALL_ID_PATTERN: &str = "^[a-zA-Z0-9_-]{1,64}$"; +/// Anthropic's accepted tool-call id length ceiling (also encoded in +/// [`TOOL_CALL_ID_PATTERN`], but kept as a plain number for callers that +/// truncate without parsing the regex). +const TOOL_CALL_ID_MAX_LEN: usize = 64; /// A chat model backed by Anthropic's native Messages API. pub struct AnthropicModel { @@ -85,6 +94,7 @@ pub struct AnthropicModel { temperature_override: Option, temperature_unsupported: Vec, allow_insecure_http: bool, + request_options: crate::providers::ProviderRequestOptions, } impl std::fmt::Debug for AnthropicModel { @@ -99,6 +109,7 @@ impl std::fmt::Debug for AnthropicModel { .field("temperature_override", &self.temperature_override) .field("temperature_unsupported", &self.temperature_unsupported) .field("allow_insecure_http", &self.allow_insecure_http) + .field("request_options", &self.request_options) .finish() } } @@ -134,15 +145,32 @@ impl AnthropicModel { streaming: true, streaming_tool_chunks: true, reasoning: true, + // Anthropic rejects a `tool_use`/`tool_result` id outside + // this shape with a 400. + tool_call_id_pattern: Some(TOOL_CALL_ID_PATTERN.to_string()), + max_tool_call_id_len: Some(TOOL_CALL_ID_MAX_LEN), ..ModelProfile::default() }, model, temperature_override: None, temperature_unsupported: Vec::new(), allow_insecure_http: false, + request_options: crate::providers::ProviderRequestOptions::default(), } } + /// Sets host-supplied request hooks and an optional HTTP client override + /// applied around every call this adapter makes. See + /// [`crate::providers::ProviderRequestOptions`]. + #[must_use] + pub fn with_request_options( + mut self, + options: crate::providers::ProviderRequestOptions, + ) -> Self { + self.request_options = options; + self + } + /// Overrides the default model id used when a request does not specify one. pub fn with_model(mut self, model: impl Into) -> Self { self.model = model.into(); @@ -211,6 +239,18 @@ impl AnthropicModel { request.model.as_deref().unwrap_or(&self.model) } + /// Builds the [`crate::message::MessageOrigin`] to stamp on a response to + /// this request: this adapter's fixed provider/API plus the model that + /// actually served the call (a request-level override, when set, else + /// the instance default). + pub(super) fn origin_for(&self, request: &ModelRequest) -> crate::message::MessageOrigin { + crate::message::MessageOrigin { + provider: PROVIDER.to_string(), + api: MESSAGES_API.to_string(), + model: self.request_model(request).to_string(), + } + } + fn request_body(&self, request: &ModelRequest) -> Value { let mut body = request_body(request, &self.model); match effective_temperature( @@ -232,6 +272,7 @@ impl AnthropicModel { } async fn post(&self, request: &ModelRequest, streaming: bool) -> Result { + crate::network_guard::ensure_network_models_allowed()?; let endpoint = reqwest::Url::parse(&self.endpoint()) .map_err(|error| Error::Validation(format!("invalid Anthropic base URL: {error}")))?; match endpoint.scheme() { @@ -253,8 +294,9 @@ impl AnthropicModel { if streaming { body["stream"] = Value::Bool(true); } - let request_builder = self - .client + self.request_options.apply_payload(&mut body); + let client = self.request_options.http.as_ref().unwrap_or(&self.client); + let request_builder = client .post(endpoint) .header("x-api-key", &self.api_key) .header("anthropic-version", ANTHROPIC_VERSION) @@ -312,6 +354,8 @@ impl AnthropicModel { retryable, retry_after_ms: tinyinference_core::parse_retry_after_ms(retry_after), raw, + partial_message: None, + stop_reason: None, } } @@ -361,7 +405,12 @@ impl ChatModel for AnthropicModel { .json() .await .map_err(|error| Error::Model(format!("anthropic response was not JSON: {error}")))?; - parse_response(body).map(|response| response.inherit_correlation(request.correlation)) + self.request_options.observe_response(&body); + let origin = self.origin_for(&request); + parse_response(body).map(|mut response| { + response.message.origin = Some(origin); + response.inherit_correlation(request.correlation) + }) } async fn stream(&self, _state: &State, request: ModelRequest) -> Result { diff --git a/crates/tinyinference-llm/src/providers/anthropic/request.rs b/crates/tinyinference-llm/src/providers/anthropic/request.rs index 53d6cfe..7c76edb 100644 --- a/crates/tinyinference-llm/src/providers/anthropic/request.rs +++ b/crates/tinyinference-llm/src/providers/anthropic/request.rs @@ -71,6 +71,8 @@ pub(crate) fn request_body(request: &ModelRequest, default_model: &str) -> Value })], ); } + // Host-side out-of-band record; never sent to the provider. + Message::Custom(_) => {} } } @@ -207,14 +209,19 @@ fn text_only_blocks(content: &[ContentBlock]) -> Vec { ContentBlock::Thinking { .. } | ContentBlock::RedactedThinking { .. } | ContentBlock::Image(_) - | ContentBlock::ProviderExtension(_) => None, + | ContentBlock::ProviderExtension(_) + | ContentBlock::Audio(_) + | ContentBlock::Video(_) + | ContentBlock::Document(_) => None, }) .collect() } -/// User-side content: text and images. Thinking blocks never appear in user -/// content; provider extensions have no faithful representation and are -/// dropped. +/// User-side content: text, images, and documents (Anthropic's native +/// `document` block). Audio and video have no Messages API representation +/// and are rendered as placeholder text rather than silently dropped. +/// Thinking blocks never appear in user content; provider extensions have no +/// faithful representation and are dropped. fn content_blocks(content: &[ContentBlock]) -> Vec { content .iter() @@ -222,6 +229,9 @@ fn content_blocks(content: &[ContentBlock]) -> Vec { ContentBlock::Text(text) => text_block(text), ContentBlock::Json(value) => text_block(&value.to_string()), ContentBlock::Image(image) => Some(image_block(image)), + ContentBlock::Document(media) => Some(document_block(media)), + ContentBlock::Audio(media) => Some(unsupported_media_placeholder("audio", media)), + ContentBlock::Video(media) => Some(unsupported_media_placeholder("video", media)), ContentBlock::Thinking { .. } | ContentBlock::RedactedThinking { .. } | ContentBlock::ProviderExtension(_) => None, @@ -254,7 +264,10 @@ fn assistant_blocks(content: &[ContentBlock]) -> Vec { signature: None, .. } | ContentBlock::Image(_) - | ContentBlock::ProviderExtension(_) => None, + | ContentBlock::ProviderExtension(_) + | ContentBlock::Audio(_) + | ContentBlock::Video(_) + | ContentBlock::Document(_) => None, }) .collect() } @@ -282,3 +295,41 @@ fn image_block(image: &ImageRef) -> Value { "source": { "type": "url", "url": image.url }, }) } + +/// Renders a document reference as Anthropic's `document` content block. +/// `MediaRef::Path` has no wire representation (the harness never reads +/// local files) and falls back to a placeholder text block instead of being +/// silently dropped. +fn document_block(media: &crate::message::MediaRef) -> Value { + use crate::message::MediaRef; + match media { + MediaRef::Base64 { data, media_type } => json!({ + "type": "document", + "source": { "type": "base64", "media_type": media_type, "data": data }, + }), + MediaRef::Url { url, .. } => json!({ + "type": "document", + "source": { "type": "url", "url": url }, + }), + MediaRef::Path { path, .. } => json!({ + "type": "text", + "text": format!("[document attachment omitted: local path {path} was not resolved]"), + }), + } +} + +/// Renders an audio or video reference as a placeholder text block: neither +/// has a wire representation in Anthropic's Messages API. +fn unsupported_media_placeholder(kind: &str, media: &crate::message::MediaRef) -> Value { + let descriptor = match media { + crate::message::MediaRef::Url { url, .. } => url.clone(), + crate::message::MediaRef::Base64 { media_type, .. } => { + format!("inline {media_type} data") + } + crate::message::MediaRef::Path { path, .. } => path.clone(), + }; + json!({ + "type": "text", + "text": format!("[{kind} attachment omitted: {descriptor}]"), + }) +} diff --git a/crates/tinyinference-llm/src/providers/anthropic/response.rs b/crates/tinyinference-llm/src/providers/anthropic/response.rs index 67c1497..94b2cb7 100644 --- a/crates/tinyinference-llm/src/providers/anthropic/response.rs +++ b/crates/tinyinference-llm/src/providers/anthropic/response.rs @@ -83,6 +83,9 @@ pub(crate) fn parse_response(body: Value) -> Result { content, tool_calls, usage: Some(usage), + // Stamped by the caller (`AnthropicModel::invoke`), which knows + // the configured provider/model. + origin: None, }, usage: Some(usage), finish_reason: Some(stop_reason), diff --git a/crates/tinyinference-llm/src/providers/anthropic/stream.rs b/crates/tinyinference-llm/src/providers/anthropic/stream.rs index 175056d..aacd59a 100644 --- a/crates/tinyinference-llm/src/providers/anthropic/stream.rs +++ b/crates/tinyinference-llm/src/providers/anthropic/stream.rs @@ -26,7 +26,9 @@ use futures::{Stream, StreamExt}; use serde_json::Value; use crate::message::{AssistantMessage, ContentBlock, MessageDelta}; -use crate::model::{ModelResponse, ModelStream, ModelStreamItem, ProviderError}; +use crate::model::{ + BlockDelta, BlockKind, ModelResponse, ModelStream, ModelStreamItem, ProviderError, +}; use crate::tool::{ToolCall, ToolDelta}; use crate::usage::Usage; use crate::{Error, Result}; @@ -55,6 +57,40 @@ enum OpenBlock { Redacted(String), } +impl OpenBlock { + /// Converts a closed block into the [`ContentBlock`] carried on + /// [`ModelStreamItem::BlockEnd`]. + /// + /// [`ContentBlock`] has no dedicated tool-call variant (tool calls live on + /// [`AssistantMessage::tool_calls`]), so a closed tool-use block is + /// represented as [`ContentBlock::Json`] carrying `{id, name, arguments}`; + /// consumers that want the parsed [`crate::tool::ToolCall`] already saw the + /// id and name on the matching [`ModelStreamItem::BlockStart`]. + fn into_content_block(self) -> ContentBlock { + match self { + OpenBlock::Text(text) => ContentBlock::Text(text), + OpenBlock::Thinking { text, signature } => ContentBlock::Thinking { text, signature }, + OpenBlock::Redacted(data) => ContentBlock::RedactedThinking { data }, + OpenBlock::ToolUse { + id, + name, + partial_json, + } => { + let arguments = if partial_json.trim().is_empty() { + Value::Object(Default::default()) + } else { + serde_json::from_str(&partial_json).unwrap_or(Value::String(partial_json)) + }; + ContentBlock::Json(serde_json::json!({ + "id": id, + "name": name, + "arguments": arguments, + })) + } + } + } +} + /// Provider-side accumulator rebuilding the terminal [`ModelResponse`]. #[derive(Debug, Default)] struct AnthropicStreamAcc { @@ -119,10 +155,18 @@ impl AnthropicStreamAcc { Some("tool_use") => { let id = block["id"].as_str().unwrap_or_default().to_string(); let name = block["name"].as_str().unwrap_or_default().to_string(); + pending.push_back(ModelStreamItem::BlockStart { + index, + kind: BlockKind::ToolCall { + id: id.clone(), + name: name.clone(), + }, + }); pending.push_back(ModelStreamItem::ToolCallDelta(ToolDelta { call_id: id.clone(), content: String::new(), tool_name: Some(name.clone()), + content_index: Some(index), })); OpenBlock::ToolUse { id, @@ -130,16 +174,44 @@ impl AnthropicStreamAcc { partial_json: String::new(), } } - Some("thinking") => OpenBlock::Thinking { - text: block["thinking"].as_str().unwrap_or_default().to_string(), - signature: None, - }, + Some("thinking") => { + pending.push_back(ModelStreamItem::BlockStart { + index, + kind: BlockKind::Thinking, + }); + let text = block["thinking"].as_str().unwrap_or_default().to_string(); + if !text.is_empty() { + pending.push_back(ModelStreamItem::BlockDelta { + index, + delta: BlockDelta::Thinking(text.clone()), + }); + pending.push_back(ModelStreamItem::MessageDelta( + MessageDelta::reasoning(text.clone()), + )); + } + OpenBlock::Thinking { + text, + signature: None, + } + } Some("redacted_thinking") => { + pending.push_back(ModelStreamItem::BlockStart { + index, + kind: BlockKind::Thinking, + }); OpenBlock::Redacted(block["data"].as_str().unwrap_or_default().to_string()) } _ => { + pending.push_back(ModelStreamItem::BlockStart { + index, + kind: BlockKind::Text, + }); let text = block["text"].as_str().unwrap_or_default().to_string(); if !text.is_empty() { + pending.push_back(ModelStreamItem::BlockDelta { + index, + delta: BlockDelta::Text(text.clone()), + }); pending.push_back(ModelStreamItem::MessageDelta(MessageDelta::text( text.clone(), ))); @@ -157,6 +229,10 @@ impl AnthropicStreamAcc { (Some("text_delta"), Some(OpenBlock::Text(text))) => { let fragment = delta["text"].as_str().unwrap_or_default(); text.push_str(fragment); + pending.push_back(ModelStreamItem::BlockDelta { + index, + delta: BlockDelta::Text(fragment.to_string()), + }); pending.push_back(ModelStreamItem::MessageDelta(MessageDelta::text( fragment.to_string(), ))); @@ -171,15 +247,24 @@ impl AnthropicStreamAcc { ) => { let fragment = delta["partial_json"].as_str().unwrap_or_default(); partial_json.push_str(fragment); + pending.push_back(ModelStreamItem::BlockDelta { + index, + delta: BlockDelta::ToolArgs(fragment.to_string()), + }); pending.push_back(ModelStreamItem::ToolCallDelta(ToolDelta { call_id: id.clone(), content: fragment.to_string(), tool_name: Some(name.clone()), + content_index: Some(index), })); } (Some("thinking_delta"), Some(OpenBlock::Thinking { text, .. })) => { let fragment = delta["thinking"].as_str().unwrap_or_default(); text.push_str(fragment); + pending.push_back(ModelStreamItem::BlockDelta { + index, + delta: BlockDelta::Thinking(fragment.to_string()), + }); pending.push_back(ModelStreamItem::MessageDelta(MessageDelta { text: String::new(), reasoning: fragment.to_string(), @@ -197,6 +282,15 @@ impl AnthropicStreamAcc { _ => {} } } + Some("content_block_stop") => { + let index = event_index(&event)?; + if let Some(block) = self.slot(index).clone() { + pending.push_back(ModelStreamItem::BlockEnd { + index, + block: block.into_content_block(), + }); + } + } Some("message_delta") => { if let Some(stop_reason) = event["delta"]["stop_reason"].as_str() { self.stop_reason = Some(stop_reason.to_string()); @@ -289,6 +383,10 @@ impl AnthropicStreamAcc { content, tool_calls, usage, + // Stamped by the `sse_next` call site (which owns + // `SseState::model`); this accumulator has no provider/model + // context of its own. + origin: None, }, usage, finish_reason: self.stop_reason, @@ -351,6 +449,11 @@ impl SseState { }; provider_error.provider = PROVIDER.to_string(); provider_error.model = Some(self.model.clone()); + provider_error.stop_reason = self.acc.stop_reason.clone(); + let partial = std::mem::take(&mut self.acc).into_response().message; + if !partial.content.is_empty() || !partial.tool_calls.is_empty() { + provider_error.partial_message = Some(partial); + } ModelStreamItem::ProviderFailed(provider_error) } @@ -417,7 +520,12 @@ async fn sse_next(mut state: SseState) -> Option<(ModelStreamItem, SseState)> { return None; } state.terminal_emitted = true; - let response = std::mem::take(&mut state.acc).into_response(); + let mut response = std::mem::take(&mut state.acc).into_response(); + response.message.origin = Some(crate::message::MessageOrigin { + provider: PROVIDER.to_string(), + api: super::MESSAGES_API.to_string(), + model: state.model.clone(), + }); return Some((ModelStreamItem::Completed(response), state)); } match state.bytes.next().await { diff --git a/crates/tinyinference-llm/src/providers/anthropic/test.rs b/crates/tinyinference-llm/src/providers/anthropic/test.rs index 209bca1..a91a8e1 100644 --- a/crates/tinyinference-llm/src/providers/anthropic/test.rs +++ b/crates/tinyinference-llm/src/providers/anthropic/test.rs @@ -5,7 +5,8 @@ use super::*; use crate::cache::CachePolicy; use crate::message::{ContentBlock, ImageRef, Message, ToolMessage}; use crate::model::{ - ModelStreamItem, PromptSegment, ReasoningConfig, ReasoningEffort, SegmentRole, ToolChoice, + BlockDelta, BlockKind, ModelStreamItem, PromptSegment, ReasoningConfig, ReasoningEffort, + SegmentRole, ToolChoice, }; use crate::tool::{ToolCall, ToolSchema}; @@ -210,6 +211,24 @@ fn tool_results_use_anthropic_tool_result_blocks() { ); } +#[test] +fn custom_messages_are_never_sent_to_the_provider() { + let request = ModelRequest::new(vec![ + Message::user("hi"), + Message::Custom(crate::message::CustomMessage { + kind: "compaction".into(), + payload: serde_json::json!({"summary": "..."}), + display: Some("Compacted".into()), + }), + Message::assistant("hello"), + ]); + let body = request_body(&request, "test-model"); + let messages = body["messages"].as_array().unwrap(); + assert_eq!(messages.len(), 2); + assert_eq!(messages[0]["role"], "user"); + assert_eq!(messages[1]["role"], "assistant"); +} + /// Parallel tool calls answer as consecutive tool messages; the Messages API /// requires them merged into one user turn. #[test] @@ -272,6 +291,7 @@ fn signed_thinking_is_replayed_and_unsigned_thinking_is_dropped() { ], tool_calls: vec![], usage: None, + origin: None, }); let body = request_body(&ModelRequest::new(vec![Message::user("q"), assistant]), "m"); let content = body["messages"][1]["content"].as_array().unwrap(); @@ -315,6 +335,63 @@ fn images_become_base64_or_url_sources() { ); } +#[test] +fn document_blocks_render_as_document_source_or_placeholder() { + use crate::message::MediaRef; + + let request = ModelRequest::new(vec![Message::User(crate::message::UserMessage { + content: vec![ + ContentBlock::Document(MediaRef::base64("QQ==", "application/pdf")), + ContentBlock::Document(MediaRef::url("https://example.com/a.pdf")), + ContentBlock::Document(MediaRef::path("/tmp/local.pdf")), + ], + })]); + let body = request_body(&request, "m"); + let content = body["messages"][0]["content"].as_array().unwrap(); + assert_eq!(content[0]["type"], "document"); + assert_eq!( + content[0]["source"], + json!({ "type": "base64", "media_type": "application/pdf", "data": "QQ==" }) + ); + assert_eq!(content[1]["type"], "document"); + assert_eq!( + content[1]["source"], + json!({ "type": "url", "url": "https://example.com/a.pdf" }) + ); + // A local path has no wire representation; it becomes a placeholder text + // block rather than being silently dropped. + assert_eq!(content[2]["type"], "text"); + assert!( + content[2]["text"] + .as_str() + .unwrap() + .contains("/tmp/local.pdf") + ); +} + +#[test] +fn audio_and_video_blocks_become_placeholder_text() { + use crate::message::MediaRef; + + let request = ModelRequest::new(vec![Message::User(crate::message::UserMessage { + content: vec![ + ContentBlock::Audio(MediaRef::url("https://example.com/a.wav")), + ContentBlock::Video(MediaRef::base64("AAAA", "video/mp4")), + ], + })]); + let body = request_body(&request, "m"); + let content = body["messages"][0]["content"].as_array().unwrap(); + assert_eq!(content[0]["type"], "text"); + assert!( + content[0]["text"] + .as_str() + .unwrap() + .contains("https://example.com/a.wav") + ); + assert_eq!(content[1]["type"], "text"); + assert!(content[1]["text"].as_str().unwrap().contains("video")); +} + #[test] fn provider_options_flatten_except_reserved_and_the_routing_hint() { let request = ModelRequest::new(vec![Message::user("hi")]).with_provider_options(json!({ @@ -686,3 +763,230 @@ async fn oversized_sse_content_block_index_is_rejected() { Some(ModelStreamItem::ProviderFailed(error)) if error.message.contains("exceeds limit") )); } + +#[tokio::test] +async fn streaming_emits_block_boundaries_for_interleaved_thinking_text_and_tool_call() { + // A recorded-shape SSE fixture with three content blocks in wire order: + // thinking (0), text (1), tool_use (2). Asserts `BlockStart`/`BlockDelta`/ + // `BlockEnd` map 1:1 onto `content_block_start`/`_delta`/`_stop` and that + // `ToolCallDelta` fragments carry the wire `content_index`. + let events = [ + json!({"type":"message_start","message":{"id":"msg_b","usage":{"input_tokens":2,"output_tokens":0}}}), + json!({"type":"content_block_start","index":0,"content_block":{"type":"thinking","thinking":""}}), + json!({"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":"plan it"}}), + json!({"type":"content_block_stop","index":0}), + json!({"type":"content_block_start","index":1,"content_block":{"type":"text","text":""}}), + json!({"type":"content_block_delta","index":1,"delta":{"type":"text_delta","text":"Sure, "}}), + json!({"type":"content_block_delta","index":1,"delta":{"type":"text_delta","text":"let me check."}}), + json!({"type":"content_block_stop","index":1}), + json!({"type":"content_block_start","index":2,"content_block":{"type":"tool_use","id":"toolu_1","name":"lookup","input":{}}}), + json!({"type":"content_block_delta","index":2,"delta":{"type":"input_json_delta","partial_json":"{\"q\":"}}), + json!({"type":"content_block_delta","index":2,"delta":{"type":"input_json_delta","partial_json":"1}"}}), + json!({"type":"content_block_stop","index":2}), + json!({"type":"message_delta","delta":{"stop_reason":"tool_use"},"usage":{"output_tokens":9}}), + json!({"type":"message_stop"}), + ]; + let items: Vec = stream::stream_from_bytes(vec![sse(&events)], "m") + .collect() + .await; + + let starts: Vec<(usize, &BlockKind)> = items + .iter() + .filter_map(|item| match item { + ModelStreamItem::BlockStart { index, kind } => Some((*index, kind)), + _ => None, + }) + .collect(); + assert_eq!(starts.len(), 3); + assert_eq!(starts[0], (0, &BlockKind::Thinking)); + assert_eq!(starts[1], (1, &BlockKind::Text)); + assert_eq!( + starts[2], + ( + 2, + &BlockKind::ToolCall { + id: "toolu_1".to_string(), + name: "lookup".to_string(), + } + ) + ); + + let deltas: Vec<(usize, &BlockDelta)> = items + .iter() + .filter_map(|item| match item { + ModelStreamItem::BlockDelta { index, delta } => Some((*index, delta)), + _ => None, + }) + .collect(); + assert_eq!( + deltas, + vec![ + (0, &BlockDelta::Thinking("plan it".to_string())), + (1, &BlockDelta::Text("Sure, ".to_string())), + (1, &BlockDelta::Text("let me check.".to_string())), + (2, &BlockDelta::ToolArgs("{\"q\":".to_string())), + (2, &BlockDelta::ToolArgs("1}".to_string())), + ] + ); + + let ends: Vec = items + .iter() + .filter_map(|item| match item { + ModelStreamItem::BlockEnd { index, .. } => Some(*index), + _ => None, + }) + .collect(); + assert_eq!(ends, vec![0, 1, 2]); + let Some(ModelStreamItem::BlockEnd { + block: ContentBlock::Thinking { text, .. }, + .. + }) = items + .iter() + .find(|item| matches!(item, ModelStreamItem::BlockEnd { index: 0, .. })) + else { + panic!("expected thinking BlockEnd at index 0"); + }; + assert_eq!(text, "plan it"); + let Some(ModelStreamItem::BlockEnd { + block: ContentBlock::Text(text), + .. + }) = items + .iter() + .find(|item| matches!(item, ModelStreamItem::BlockEnd { index: 1, .. })) + else { + panic!("expected text BlockEnd at index 1"); + }; + assert_eq!(text, "Sure, let me check."); + let Some(ModelStreamItem::BlockEnd { + block: ContentBlock::Json(value), + .. + }) = items + .iter() + .find(|item| matches!(item, ModelStreamItem::BlockEnd { index: 2, .. })) + else { + panic!("expected tool-call BlockEnd at index 2"); + }; + assert_eq!(value["id"], "toolu_1"); + assert_eq!(value["name"], "lookup"); + assert_eq!(value["arguments"], json!({"q": 1})); + + // Every ToolCallDelta for the tool-use block carries its wire index. + let tool_indices: Vec> = items + .iter() + .filter_map(|item| match item { + ModelStreamItem::ToolCallDelta(delta) => Some(delta.content_index), + _ => None, + }) + .collect(); + assert_eq!(tool_indices, vec![Some(2), Some(2), Some(2)]); + + // Compatibility: MessageDelta still carries the flat text/reasoning. + let text: String = items + .iter() + .filter_map(|item| match item { + ModelStreamItem::MessageDelta(delta) => Some(delta.text.as_str()), + _ => None, + }) + .collect(); + assert_eq!(text, "Sure, let me check."); + let reasoning: String = items + .iter() + .filter_map(|item| match item { + ModelStreamItem::MessageDelta(delta) => Some(delta.reasoning.as_str()), + _ => None, + }) + .collect(); + assert_eq!(reasoning, "plan it"); +} + +#[tokio::test] +async fn provider_failed_terminal_carries_partial_message_and_stop_reason() { + // A mid-stream `error` event after some content has already arrived must + // surface the partial assistant message and last-known stop reason, so a + // caller can decide whether to keep or discard the partial turn instead + // of losing it outright. + let events = [ + json!({"type":"message_start","message":{"id":"msg_e","usage":{"input_tokens":1,"output_tokens":0}}}), + json!({"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}), + json!({"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"partial answer"}}), + json!({"type":"message_delta","delta":{"stop_reason":"pause_turn"},"usage":{"output_tokens":2}}), + json!({"type":"error","error":{"type":"overloaded_error","message":"the server is overloaded"}}), + ]; + let items: Vec = stream::stream_from_bytes(vec![sse(&events)], "m") + .collect() + .await; + let Some(ModelStreamItem::ProviderFailed(error)) = items.last() else { + panic!("expected ProviderFailed, got {:?}", items.last()); + }; + assert_eq!(error.stop_reason.as_deref(), Some("pause_turn")); + let partial = error + .partial_message + .as_ref() + .expect("partial message must be present"); + assert_eq!( + partial.content, + vec![ContentBlock::Text("partial answer".to_string())] + ); +} + +#[test] +fn origin_for_records_provider_api_and_effective_model() { + let model = AnthropicModel::new("key").with_model("claude-opus-4-6"); + let request = ModelRequest::default(); + let origin = model.origin_for(&request); + assert_eq!(origin.provider, "anthropic"); + assert_eq!(origin.api, "messages"); + assert_eq!(origin.model, "claude-opus-4-6"); +} + +#[test] +fn origin_for_prefers_a_per_request_model_override() { + let model = AnthropicModel::new("key").with_model("claude-opus-4-6"); + let request = ModelRequest { + model: Some("claude-sonnet-4-6".to_string()), + ..Default::default() + }; + let origin = model.origin_for(&request); + assert_eq!(origin.model, "claude-sonnet-4-6"); +} + +#[test] +fn default_profile_advertises_the_tool_call_id_shape() { + let model = AnthropicModel::new("key"); + let profile = &model.profile; + assert_eq!( + profile.tool_call_id_pattern.as_deref(), + Some("^[a-zA-Z0-9_-]{1,64}$") + ); + assert_eq!(profile.max_tool_call_id_len, Some(64)); +} + +#[tokio::test] +async fn streamed_terminal_response_carries_origin() { + let events = [ + json!({"type":"message_start","message":{"id":"msg_o","usage":{"input_tokens":1,"cache_read_input_tokens":0,"cache_creation_input_tokens":0,"output_tokens":1}}}), + json!({"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}), + json!({"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"hi"}}), + json!({"type":"content_block_stop","index":0}), + json!({"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":1}}), + json!({"type":"message_stop"}), + ]; + let items: Vec = + stream::stream_from_bytes(vec![sse(&events)], "claude-opus-4-6") + .collect() + .await; + let completed = items + .into_iter() + .find_map(|item| match item { + ModelStreamItem::Completed(response) => Some(response), + _ => None, + }) + .expect("a terminal Completed item"); + let origin = completed + .message + .origin + .expect("origin stamped on stream terminal"); + assert_eq!(origin.provider, "anthropic"); + assert_eq!(origin.api, "messages"); + assert_eq!(origin.model, "claude-opus-4-6"); +} diff --git a/crates/tinyinference-llm/src/providers/mock.rs b/crates/tinyinference-llm/src/providers/mock.rs index f9c3268..ecfd30d 100644 --- a/crates/tinyinference-llm/src/providers/mock.rs +++ b/crates/tinyinference-llm/src/providers/mock.rs @@ -239,6 +239,7 @@ impl ChatModel for MockModel { content: Vec::new(), tool_calls: vec![tool_call], usage: Some(usage), + origin: None, }; ModelResponse { message, @@ -354,6 +355,7 @@ impl MockModel { content: vec![ContentBlock::Text(s)], tool_calls: Vec::new(), usage: Some(Usage::new(10, output_tokens)), + origin: None, }, usage: Some(Usage::new(10, output_tokens)), finish_reason: Some("stop".to_string()), diff --git a/crates/tinyinference-llm/src/providers/openai/convert.rs b/crates/tinyinference-llm/src/providers/openai/convert.rs index af3afac..69493ab 100644 --- a/crates/tinyinference-llm/src/providers/openai/convert.rs +++ b/crates/tinyinference-llm/src/providers/openai/convert.rs @@ -67,6 +67,14 @@ pub(super) fn translate_message(message: &Message) -> Result { tool_calls: Vec::new(), tool_call_id: Some(tool.tool_call_id.clone()), }, + // Callers filter `Message::Custom` out of the messages slice before + // calling this function; it is a host-side out-of-band record that is + // never sent to a provider. + Message::Custom(_) => { + return Err(Error::Validation( + "Message::Custom must be filtered before wire translation".to_string(), + )); + } }; Ok(wire) } @@ -78,7 +86,11 @@ fn translate_text_content(blocks: &[ContentBlock]) -> Result { ContentBlock::Text(value) => text.push_str(value), ContentBlock::Json(value) => text.push_str(&value.to_string()), ContentBlock::Thinking { .. } | ContentBlock::RedactedThinking { .. } => {} - ContentBlock::Image(_) | ContentBlock::ProviderExtension(_) => { + ContentBlock::Image(_) + | ContentBlock::ProviderExtension(_) + | ContentBlock::Audio(_) + | ContentBlock::Video(_) + | ContentBlock::Document(_) => { return Err(unrepresentable_block_error()); } } @@ -95,25 +107,32 @@ fn translate_text_content(blocks: &[ContentBlock]) -> Result { /// representation, so it fails closed with a validation error rather than being /// silently dropped. pub(super) fn translate_user_content(blocks: &[ContentBlock]) -> Result { - let has_image = blocks + let has_media = blocks .iter() - .any(|block| matches!(block, ContentBlock::Image(_))); + .any(|block| matches!(block, ContentBlock::Image(_) | ContentBlock::Audio(_))); - if !has_image { - // No image: render as a single string, but still fail closed on blocks - // that cannot be represented. + if !has_media { + // No image/audio: render as a single string, but still fail closed on + // blocks that cannot be represented. let mut text = String::new(); for block in blocks { match block { ContentBlock::Text(t) => text.push_str(t), ContentBlock::Json(value) => text.push_str(&value.to_string()), - ContentBlock::Image(_) => unreachable!("guarded by has_image"), + ContentBlock::Image(_) | ContentBlock::Audio(_) => { + unreachable!("guarded by has_media") + } // OpenAI-compatible requests have no representation for // reasoning blocks; they are dropped rather than failing the // request (matching the assistant path, which serializes via // `Message::text` and drops them naturally). ContentBlock::Thinking { .. } | ContentBlock::RedactedThinking { .. } => {} - ContentBlock::ProviderExtension(_) => { + // Chat Completions has no video or document input; fail + // closed rather than silently drop an attachment the caller + // expected to be sent. + ContentBlock::ProviderExtension(_) + | ContentBlock::Video(_) + | ContentBlock::Document(_) => { return Err(unrepresentable_block_error()); } } @@ -137,10 +156,13 @@ pub(super) fn translate_user_content(blocks: &[ContentBlock]) -> Result parts.push(input_audio_part(media)?), // See the string-rendering arm above: reasoning blocks have no // OpenAI representation and are dropped, not failed. ContentBlock::Thinking { .. } | ContentBlock::RedactedThinking { .. } => {} - ContentBlock::ProviderExtension(_) => { + ContentBlock::ProviderExtension(_) + | ContentBlock::Video(_) + | ContentBlock::Document(_) => { return Err(unrepresentable_block_error()); } } @@ -195,6 +217,41 @@ pub(super) fn unrepresentable_block_error() -> Error { ) } +/// Renders an audio [`crate::message::MediaRef`] as an OpenAI Chat +/// Completions `input_audio` content part. +/// +/// The wire format requires inline base64 data, so a [`MediaRef::Url`] or +/// [`MediaRef::Path`] reference — which the harness never fetches or reads +/// itself — fails closed rather than being silently dropped or sent +/// malformed. +fn input_audio_part(media: &crate::message::MediaRef) -> Result { + use crate::message::MediaRef; + match media { + MediaRef::Base64 { data, media_type } => Ok(ContentPartWire::InputAudio { + input_audio: InputAudioWire { + data: data.clone(), + format: audio_format_from_media_type(media_type), + }, + }), + MediaRef::Url { .. } | MediaRef::Path { .. } => Err(Error::Validation( + "OpenAI input_audio requires inline base64 data; resolve the \ + audio reference to bytes before sending it" + .to_string(), + )), + } +} + +/// Derives the OpenAI `input_audio.format` token (`"wav"`, `"mp3"`, …) from a +/// MIME type such as `audio/wav`, defaulting to `"wav"` when unrecognized. +fn audio_format_from_media_type(media_type: &str) -> String { + media_type + .rsplit('/') + .next() + .filter(|format| !format.is_empty()) + .unwrap_or("wav") + .to_string() +} + /// Translates a [`ToolChoice`] into the OpenAI `tool_choice` JSON value. pub(super) fn translate_tool_choice(choice: &ToolChoice) -> Value { match choice { @@ -357,6 +414,9 @@ pub(super) fn parse_chat_response( content, tool_calls, usage, + // Stamped by the transport call site, which knows the configured + // provider/model; this parser is provider-agnostic wire decoding. + origin: None, }; Ok(ModelResponse { diff --git a/crates/tinyinference-llm/src/providers/openai/mod.rs b/crates/tinyinference-llm/src/providers/openai/mod.rs index f222f95..8fabb90 100644 --- a/crates/tinyinference-llm/src/providers/openai/mod.rs +++ b/crates/tinyinference-llm/src/providers/openai/mod.rs @@ -53,8 +53,8 @@ use serde_json::{Map, Value, json}; use crate::message::{AssistantMessage, ContentBlock, Message, MessageDelta}; use crate::model::{ - ChatModel, Modalities, ModelProfile, ModelRequest, ModelResponse, ModelStatus, ModelStream, - ModelStreamItem, ProviderError, ResponseFormat, ToolChoice, + BlockDelta, BlockKind, ChatModel, Modalities, ModelProfile, ModelRequest, ModelResponse, + ModelStatus, ModelStream, ModelStreamItem, ProviderError, ResponseFormat, ToolChoice, }; use crate::tool::{ToolCall, ToolDelta}; use crate::usage::Usage; @@ -62,6 +62,14 @@ use crate::{Error, Result}; use super::ProviderSpec; +/// [`crate::message::MessageOrigin::api`] value stamped on responses built +/// from the Chat Completions endpoint (including every OpenAI-compatible +/// local-runtime preset, which shares this transport). +pub(super) const CHAT_COMPLETIONS_API: &str = "chat_completions"; +/// [`crate::message::MessageOrigin::api`] value stamped on responses built +/// from the `/v1/responses` endpoint. +pub(super) const RESPONSES_API: &str = "responses"; + /// Default model id used when neither the request nor the builder override it. const DEFAULT_MODEL: &str = "gpt-4.1-mini"; /// Default OpenAI API base URL. diff --git a/crates/tinyinference-llm/src/providers/openai/responses.rs b/crates/tinyinference-llm/src/providers/openai/responses.rs index fe7413e..84b580c 100644 --- a/crates/tinyinference-llm/src/providers/openai/responses.rs +++ b/crates/tinyinference-llm/src/providers/openai/responses.rs @@ -334,6 +334,8 @@ pub(super) fn build_responses_input(messages: &[Message]) -> (Option, Ve format!("[tool_result id={} ]\n{body}", m.tool_call_id) } } + // Host-side out-of-band record; never sent to the provider. + Message::Custom(_) => continue, }; if text.trim().is_empty() { continue; @@ -561,6 +563,9 @@ pub(super) fn parse_responses_response(value: Value) -> ModelResponse { content, tool_calls: Vec::new(), usage, + // Stamped by the transport call site (`invoke_responses`), which + // knows the configured provider/model. + origin: None, }, usage, finish_reason, @@ -600,6 +605,23 @@ mod tests { assert_eq!(input[1].content[0].text, "hello"); } + #[test] + fn build_input_skips_custom_messages() { + let messages = vec![ + Message::user("hi"), + Message::Custom(crate::message::CustomMessage { + kind: "compaction".into(), + payload: json!({"summary": "..."}), + display: Some("Compacted".into()), + }), + Message::assistant("hello"), + ]; + let (_, input) = build_responses_input(&messages); + assert_eq!(input.len(), 2); + assert_eq!(input[0].content[0].text, "hi"); + assert_eq!(input[1].content[0].text, "hello"); + } + #[test] fn extract_text_prefers_output_text_then_scans_content() { let with_convenience = ResponsesResponse { diff --git a/crates/tinyinference-llm/src/providers/openai/sse.rs b/crates/tinyinference-llm/src/providers/openai/sse.rs index 9b7d322..d8114b0 100644 --- a/crates/tinyinference-llm/src/providers/openai/sse.rs +++ b/crates/tinyinference-llm/src/providers/openai/sse.rs @@ -17,6 +17,88 @@ pub(super) struct ToolCallBuild { args: String, } +/// The block-channel's notion of "which content stream is currently open", +/// used to detect a switch (and thus a `BlockEnd`/`BlockStart` pair) between +/// text, reasoning, and each individual tool call. +/// +/// A tool call is identified by its slot in [`OpenAiStreamAcc::tool_calls`] +/// (not the block index): two tool calls always occupy distinct slots, so +/// comparing slots is enough to tell fragments for different calls apart. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum OpenKind { + Text, + Reasoning, + ToolCall(usize), +} + +/// What the caller of [`OpenAiStreamAcc::ensure_block`] wants opened, carrying +/// whatever data a fresh [`BlockKind`] needs to be constructed. Text and +/// reasoning blocks need nothing beyond their kind; a tool-call block needs +/// the id/name known at the moment it first opens. +enum BlockRequest { + Text, + Reasoning, + ToolCall { + slot: usize, + id: String, + name: String, + }, +} + +impl BlockRequest { + fn kind(&self) -> OpenKind { + match self { + BlockRequest::Text => OpenKind::Text, + BlockRequest::Reasoning => OpenKind::Reasoning, + BlockRequest::ToolCall { slot, .. } => OpenKind::ToolCall(*slot), + } + } +} + +/// The block channel's accumulated content for one opened block, mirrored +/// independently of [`OpenAiStreamAcc::text`]/`reasoning`/`tool_calls` (which +/// remain the sole source of truth for [`OpenAiStreamAcc::into_response`]). +/// This exists only to assemble the [`ModelStreamItem::BlockEnd`] payload. +#[derive(Clone, Debug)] +enum BlockBuf { + Text(String), + Reasoning(String), + ToolCall { + id: String, + name: String, + args: String, + }, +} + +impl BlockBuf { + /// Converts a closed block into the [`ContentBlock`] carried on + /// [`ModelStreamItem::BlockEnd`]. Mirrors the Anthropic adapter's + /// `OpenBlock::into_content_block`: a tool-call block has no dedicated + /// [`ContentBlock`] variant, so it is represented as [`ContentBlock::Json`] + /// carrying `{id, name, arguments}`. + fn into_content_block(self) -> ContentBlock { + match self { + BlockBuf::Text(text) => ContentBlock::Text(text), + BlockBuf::Reasoning(text) => ContentBlock::Thinking { + text, + signature: None, + }, + BlockBuf::ToolCall { id, name, args } => { + let arguments = if args.trim().is_empty() { + Value::Object(Default::default()) + } else { + serde_json::from_str(&args).unwrap_or(Value::String(args)) + }; + ContentBlock::Json(serde_json::json!({ + "id": id, + "name": name, + "arguments": arguments, + })) + } + } + } +} + /// Provider-side accumulator that rebuilds the authoritative [`ModelResponse`] /// from streamed chunks. Distinct from the generic /// [`StreamAccumulator`][crate::model::StreamAccumulator]: it tracks @@ -50,6 +132,20 @@ pub(super) struct OpenAiStreamAcc { /// opened slot, so id-less argument continuations for that index keep /// following the call most recently opened there. index_slots: std::collections::HashMap, + /// Block-channel content, one entry per opened block (text, reasoning, or + /// a tool call), in first-open order. `content_index` on the flat + /// [`ToolDelta`]/compatibility channel is this vector's index, so it + /// matches [`ModelStreamItem::BlockStart`]/`BlockDelta`/`BlockEnd`. + blocks: Vec, + /// The currently open block (its index into `blocks` and its kind), or + /// `None` between blocks and before the first one opens. OpenAI chat + /// completions gives no explicit "block closed" signal the way Anthropic's + /// `content_block_stop` does, so a block is considered closed exactly when + /// a fragment for a *different* block arrives, or when `finish_reason`/the + /// stream ends — never reopened once closed (a later fragment for the same + /// conceptual channel opens a fresh block instead, matching how densely + /// each block gets its own [`ModelStreamItem::BlockStart`]). + current_block: Option<(usize, OpenKind)>, } impl OpenAiStreamAcc { @@ -78,12 +174,14 @@ impl OpenAiStreamAcc { pending.push_back(ModelStreamItem::UsageDelta(usage)); } for mut choice in chunk.choices.into_iter().filter(|choice| choice.index == 0) { + let finished = choice.finish_reason.is_some(); if let Some(reason) = choice.finish_reason { self.finish_reason = Some(reason); } let reasoning = delta_reasoning_text(&mut choice.delta); if !reasoning.is_empty() { self.reasoning.push_str(&reasoning); + self.push_block_delta(BlockRequest::Reasoning, &reasoning, pending); pending.push_back(ModelStreamItem::MessageDelta(MessageDelta { text: String::new(), reasoning, @@ -103,6 +201,7 @@ impl OpenAiStreamAcc { let mut reasoning = String::new(); extractor.push(&content, &mut visible, &mut reasoning); if !reasoning.is_empty() { + self.push_block_delta(BlockRequest::Reasoning, &reasoning, pending); pending.push_back(ModelStreamItem::MessageDelta(MessageDelta { text: String::new(), reasoning, @@ -110,6 +209,7 @@ impl OpenAiStreamAcc { })); } if !visible.is_empty() { + self.push_block_delta(BlockRequest::Text, &visible, pending); pending.push_back(ModelStreamItem::MessageDelta(MessageDelta { text: visible, reasoning: String::new(), @@ -118,6 +218,7 @@ impl OpenAiStreamAcc { } } None => { + self.push_block_delta(BlockRequest::Text, &content, pending); pending.push_back(ModelStreamItem::MessageDelta(MessageDelta { text: content, reasoning: String::new(), @@ -128,31 +229,156 @@ impl OpenAiStreamAcc { } for fragment in choice.delta.tool_calls { let idx = self.resolve_slot(&fragment); - let slot = &mut self.tool_calls[idx]; + let id_present = fragment.id.as_deref().is_some_and(|id| !id.is_empty()); + let name_present = fragment + .function + .as_ref() + .and_then(|function| function.name.as_deref()) + .is_some_and(|name| !name.is_empty()); if let Some(id) = fragment.id.filter(|id| !id.is_empty()) { - slot.id = id; + self.tool_calls[idx].id = id; } if let Some(function) = fragment.function { if let Some(name) = function.name.filter(|n| !n.is_empty()) { - slot.name = name; + self.tool_calls[idx].name = name; + } + // Open the tool-call's block as soon as either its id or + // its name is known, even if no argument fragment has + // arrived yet, mirroring the Anthropic adapter's + // `content_block_start` for `tool_use`. + if id_present || name_present { + if self.tool_calls[idx].id.is_empty() { + self.tool_calls[idx].id = tool_call_id(idx, ""); + } + let call_id = tool_call_id(idx, &self.tool_calls[idx].id); + let name = self.tool_calls[idx].name.clone(); + self.ensure_block( + BlockRequest::ToolCall { + slot: idx, + id: call_id, + name, + }, + pending, + ); } if let Some(args) = function.arguments.filter(|a| !a.is_empty()) { - slot.args.push_str(&args); - if slot.id.is_empty() { - slot.id = tool_call_id(idx, ""); + self.tool_calls[idx].args.push_str(&args); + if self.tool_calls[idx].id.is_empty() { + self.tool_calls[idx].id = tool_call_id(idx, ""); + } + let call_id = tool_call_id(idx, &self.tool_calls[idx].id); + let name = self.tool_calls[idx].name.clone(); + let block_index = self.ensure_block( + BlockRequest::ToolCall { + slot: idx, + id: call_id.clone(), + name: name.clone(), + }, + pending, + ); + if let Some(BlockBuf::ToolCall { args: buf, .. }) = + self.blocks.get_mut(block_index) + { + buf.push_str(&args); } - let call_id = tool_call_id(idx, &slot.id); + pending.push_back(ModelStreamItem::BlockDelta { + index: block_index, + delta: BlockDelta::ToolArgs(args.clone()), + }); pending.push_back(ModelStreamItem::ToolCallDelta(ToolDelta { call_id, content: args, - // Surface the tool name (captured into `slot.name` from - // the call-opening fragment) so consumers can label the - // call as it streams; the accumulator keeps the first. - tool_name: Some(slot.name.clone()).filter(|n| !n.is_empty()), + // Surface the tool name (captured into `slot.name` + // from the call-opening fragment) so consumers can + // label the call as it streams; the accumulator + // keeps the first. + tool_name: Some(name).filter(|n| !n.is_empty()), + content_index: Some(block_index), })); } } } + if finished { + self.close_current_block(pending); + } + } + } + + /// Routes a text or reasoning fragment onto the block channel: opens (or + /// continues) the matching block via [`Self::ensure_block`], appends the + /// fragment into its buffer, and emits the corresponding + /// [`ModelStreamItem::BlockDelta`]. + fn push_block_delta( + &mut self, + request: BlockRequest, + fragment: &str, + pending: &mut VecDeque, + ) { + let delta = match &request { + BlockRequest::Text => BlockDelta::Text(fragment.to_string()), + BlockRequest::Reasoning => BlockDelta::Thinking(fragment.to_string()), + BlockRequest::ToolCall { .. } => { + debug_assert!(false, "push_block_delta is only used for text/reasoning"); + return; + } + }; + let index = self.ensure_block(request, pending); + match self.blocks.get_mut(index) { + Some(BlockBuf::Text(buf)) => buf.push_str(fragment), + Some(BlockBuf::Reasoning(buf)) => buf.push_str(fragment), + _ => {} + } + pending.push_back(ModelStreamItem::BlockDelta { index, delta }); + } + + /// Returns the index of the block matching `request`, opening a new one + /// (emitting [`ModelStreamItem::BlockStart`] and closing whatever block + /// was previously open) if the requested kind is not already the current + /// block. + fn ensure_block( + &mut self, + request: BlockRequest, + pending: &mut VecDeque, + ) -> usize { + let kind = request.kind(); + if let Some((index, open)) = self.current_block + && open == kind + { + return index; + } + self.close_current_block(pending); + let (buf, start_kind) = match request { + BlockRequest::Text => (BlockBuf::Text(String::new()), BlockKind::Text), + BlockRequest::Reasoning => (BlockBuf::Reasoning(String::new()), BlockKind::Thinking), + BlockRequest::ToolCall { id, name, .. } => ( + BlockBuf::ToolCall { + id: id.clone(), + name: name.clone(), + args: String::new(), + }, + BlockKind::ToolCall { id, name }, + ), + }; + let index = self.blocks.len(); + self.blocks.push(buf); + pending.push_back(ModelStreamItem::BlockStart { + index, + kind: start_kind, + }); + self.current_block = Some((index, kind)); + index + } + + /// Closes whichever block is currently open (if any), emitting its + /// [`ModelStreamItem::BlockEnd`]. A no-op when no block is open. + fn close_current_block(&mut self, pending: &mut VecDeque) { + if let Some((index, _)) = self.current_block.take() + && let Some(buf) = self.blocks.get(index) + { + pending.push_back(ModelStreamItem::BlockEnd { + index, + block: buf.clone().into_content_block(), + }); } } @@ -270,6 +496,10 @@ impl OpenAiStreamAcc { content, tool_calls, usage: self.usage, + // Stamped by the `sse_next` call site (which owns + // `SseState::provider`/`model`); this accumulator has no + // provider/model context of its own. + origin: None, }; ModelResponse { message, @@ -377,21 +607,22 @@ impl SseState { if payload == "[DONE]" { self.completion_seen = true; self.finished = true; + // Defensive: normally already closed when the finish-reason chunk + // was ingested; a no-op if so. + let mut pending = std::mem::take(&mut self.pending); + self.acc.close_current_block(&mut pending); + self.pending = pending; return; } let Ok(value) = serde_json::from_str::(payload) else { if payload.starts_with('{') || payload.starts_with('[') { - self.pending - .push_back(ModelStreamItem::ProviderFailed(ProviderError { - provider: self.provider.clone(), - model: Some(self.model.clone()), - message: "provider returned malformed SSE JSON".into(), - retryable: false, - raw: Some(Value::String(payload.into())), - ..ProviderError::default() - })); - self.finished = true; - self.terminal_emitted = true; + let item = self.provider_failure(ProviderError { + message: "provider returned malformed SSE JSON".into(), + retryable: false, + raw: Some(Value::String(payload.into())), + ..ProviderError::default() + }); + self.pending.push_back(item); } return; }; @@ -400,10 +631,9 @@ impl SseState { // `ChatCompletionChunk`, so it must be detected first and surfaced as a // terminal failure rather than folded in as an empty chunk and swallowed. if let Some(error) = value.get("error") { - self.pending - .push_back(ModelStreamItem::ProviderFailed(self.stream_error(error))); - self.finished = true; - self.terminal_emitted = true; + let error = self.stream_error(error); + let item = self.provider_failure(error); + self.pending.push_back(item); return; } if let Ok(chunk) = serde_json::from_value::(value) { @@ -441,6 +671,26 @@ impl SseState { ..ProviderError::default() } } + + /// Builds the terminal failure item, discarding any block/message deltas + /// still queued: a failure must be the last item a consumer sees, not + /// followed by fragments parsed before it surfaced (mirrors the + /// Anthropic adapter's `provider_failure`). Populates + /// `partial_message`/`stop_reason` from whatever had accumulated so a + /// mid-stream failure does not discard already-streamed content. + fn provider_failure(&mut self, mut error: ProviderError) -> ModelStreamItem { + self.pending.clear(); + self.finished = true; + self.terminal_emitted = true; + error.provider = self.provider.clone(); + error.model = Some(self.model.clone()); + error.stop_reason = self.acc.finish_reason.clone(); + let partial = std::mem::take(&mut self.acc).into_response().message; + if !partial.content.is_empty() || !partial.tool_calls.is_empty() { + error.partial_message = Some(partial); + } + ModelStreamItem::ProviderFailed(error) + } } /// Advances the SSE [`SseState`] by one item for [`futures::stream::unfold`]. @@ -461,7 +711,12 @@ pub(super) async fn sse_next(mut state: SseState) -> Option<(ModelStreamItem, Ss // Reconstruction is infallible: malformed tool arguments become an // `ToolCall::invalid` call inside the response (not a stream // failure), so the agent loop recovers instead of aborting the run. - let response = std::mem::take(&mut state.acc).into_response(); + let mut response = std::mem::take(&mut state.acc).into_response(); + response.message.origin = Some(crate::message::MessageOrigin { + provider: state.provider.clone(), + api: CHAT_COMPLETIONS_API.to_string(), + model: state.model.clone(), + }); return Some((ModelStreamItem::Completed(response), state)); } match state.bytes.next().await { @@ -470,16 +725,12 @@ pub(super) async fn sse_next(mut state: SseState) -> Option<(ModelStreamItem, Ss state.drain_lines(); } Some(Err(error)) => { - state.finished = true; - state.terminal_emitted = true; - let provider_error = ProviderError { - provider: state.provider.clone(), - model: Some(state.model.clone()), + let item = state.provider_failure(ProviderError { message: error.to_string(), retryable: true, ..ProviderError::default() - }; - return Some((ModelStreamItem::ProviderFailed(provider_error), state)); + }); + return Some((item, state)); } None => { // Drain any final `data:` line the provider sent without a @@ -488,18 +739,12 @@ pub(super) async fn sse_next(mut state: SseState) -> Option<(ModelStreamItem, Ss if state.terminal_emitted || state.completion_seen { state.finished = true; } else { - state.finished = true; - state.terminal_emitted = true; - return Some(( - ModelStreamItem::ProviderFailed(ProviderError { - provider: state.provider.clone(), - model: Some(state.model.clone()), - message: "provider stream ended before a completion signal".into(), - retryable: true, - ..ProviderError::default() - }), - state, - )); + let item = state.provider_failure(ProviderError { + message: "provider stream ended before a completion signal".into(), + retryable: true, + ..ProviderError::default() + }); + return Some((item, state)); } } } diff --git a/crates/tinyinference-llm/src/providers/openai/test.rs b/crates/tinyinference-llm/src/providers/openai/test.rs index 400e2eb..45aeb7a 100644 --- a/crates/tinyinference-llm/src/providers/openai/test.rs +++ b/crates/tinyinference-llm/src/providers/openai/test.rs @@ -8,10 +8,10 @@ use serde_json::json; use super::*; -use crate::message::Message; +use crate::message::{ContentBlock, Message}; use crate::model::{ - ChatModel, ModelRequest, ModelStreamItem, ProviderError, ResponseFormat, StreamAccumulator, - ToolChoice, + BlockDelta, BlockKind, ChatModel, ModelRequest, ModelStreamItem, ProviderError, ResponseFormat, + StreamAccumulator, ToolChoice, }; use crate::providers::{ProviderKind, ProviderSpec}; use crate::tool::ToolSchema; @@ -109,6 +109,25 @@ fn translates_request_to_openai_json_shape() { assert_eq!(value["seed"], json!(7)); } +#[test] +fn custom_messages_are_never_sent_to_the_provider() { + let request = ModelRequest::new(vec![ + Message::system("sys"), + Message::Custom(crate::message::CustomMessage { + kind: "compaction".into(), + payload: json!({"summary": "..."}), + display: Some("Compacted 12 turns".into()), + }), + Message::user("hi"), + ]); + let body = model().translate_request(&request).unwrap(); + let value = serde_json::to_value(&body).unwrap(); + let messages = value["messages"].as_array().unwrap(); + assert_eq!(messages.len(), 2); + assert_eq!(messages[0]["role"], json!("system")); + assert_eq!(messages[1]["role"], json!("user")); +} + #[test] fn translates_provider_options_for_local_openai_compatible_models() { let request = ModelRequest::new(vec![Message::user("hi")]) @@ -196,6 +215,7 @@ fn translates_assistant_tool_calls_to_stringified_arguments() { invalid: None, }], usage: None, + origin: None, }), Message::tool("call-1", "sunny, 21C"), ]); @@ -244,6 +264,7 @@ fn translates_structured_tool_result_content() { fn translates_structured_system_content_and_rejects_images() { let request = ModelRequest::new(vec![Message::System(crate::message::SystemMessage { content: vec![ContentBlock::Json(json!({"policy": "strict"}))], + ..Default::default() })]); let value = serde_json::to_value(model().translate_request(&request).unwrap()).unwrap(); assert_eq!( @@ -256,6 +277,7 @@ fn translates_structured_system_content_and_rejects_images() { url: "https://example.test/image.png".into(), mime_type: None, })], + ..Default::default() })]); assert!(model().translate_request(&invalid).is_err()); } @@ -671,6 +693,8 @@ fn provider_failed_stream_item_finishes_as_provider_error() { retryable: true, retry_after_ms: None, raw: None, + partial_message: None, + stop_reason: None, })); match accumulator.finish().unwrap_err() { @@ -1461,6 +1485,54 @@ fn user_image_blocks_render_as_content_parts() { ); } +#[test] +fn user_audio_blocks_render_as_input_audio_parts() { + use crate::message::{ContentBlock, MediaRef, UserMessage}; + + let request = ModelRequest::new(vec![Message::User(UserMessage { + content: vec![ + ContentBlock::Text("transcribe this".to_string()), + ContentBlock::Audio(MediaRef::base64("AAAA", "audio/wav")), + ], + })]); + + let value = serde_json::to_value(model().translate_request(&request).unwrap()).unwrap(); + let content = &value["messages"][0]["content"]; + assert!(content.is_array(), "expected content parts, got {content}"); + assert_eq!(content[1]["type"], json!("input_audio")); + assert_eq!(content[1]["input_audio"]["data"], json!("AAAA")); + assert_eq!(content[1]["input_audio"]["format"], json!("wav")); +} + +#[test] +fn user_audio_block_by_url_fails_closed() { + use crate::message::{ContentBlock, MediaRef, UserMessage}; + + let request = ModelRequest::new(vec![Message::User(UserMessage { + content: vec![ContentBlock::Audio(MediaRef::url( + "https://example.test/a.wav", + ))], + })]); + + let error = model().translate_request(&request).unwrap_err(); + assert!(matches!(error, Error::Validation(_))); +} + +#[test] +fn user_document_block_fails_closed() { + use crate::message::{ContentBlock, MediaRef, UserMessage}; + + let request = ModelRequest::new(vec![Message::User(UserMessage { + content: vec![ContentBlock::Document(MediaRef::base64( + "QQ==", + "application/pdf", + ))], + })]); + + let error = model().translate_request(&request).unwrap_err(); + assert!(matches!(error, Error::Validation(_))); +} + #[test] fn text_only_user_message_stays_a_plain_string() { // The common text-only case keeps its historical plain-string wire shape. @@ -2556,3 +2628,374 @@ fn a_null_tool_calls_delta_is_read_as_no_fragments() { assert!(chunk.choices[0].delta.tool_calls.is_empty()); assert_eq!(chunk.choices[0].delta.content.as_deref(), Some("Hi")); } + +/// Unwraps a [`ModelStreamItem::BlockStart`], panicking with the actual item +/// on a mismatch (used by the block-derivation tests below). +fn expect_block_start(item: &ModelStreamItem) -> (usize, &BlockKind) { + match item { + ModelStreamItem::BlockStart { index, kind } => (*index, kind), + other => panic!("expected BlockStart, got {other:?}"), + } +} + +/// Unwraps a [`ModelStreamItem::BlockDelta`], panicking with the actual item +/// on a mismatch. +fn expect_block_delta(item: &ModelStreamItem) -> (usize, &BlockDelta) { + match item { + ModelStreamItem::BlockDelta { index, delta } => (*index, delta), + other => panic!("expected BlockDelta, got {other:?}"), + } +} + +/// Unwraps a [`ModelStreamItem::BlockEnd`], panicking with the actual item on +/// a mismatch. +fn expect_block_end(item: &ModelStreamItem) -> (usize, &ContentBlock) { + match item { + ModelStreamItem::BlockEnd { index, block } => (*index, block), + other => panic!("expected BlockEnd, got {other:?}"), + } +} + +#[tokio::test] +async fn sse_stream_derives_block_boundaries_for_reasoning_text_and_tool_calls() { + // Interleaved reasoning → text → two tool calls (the second split so a + // block continues across chunks, matching how OpenAI streams parallel + // calls), then `finish_reason`. Exercises the OpenAI chat-completions + // block derivation: `BlockStart`/`BlockDelta`/`BlockEnd` should appear in + // first-open order, sharing one dense index space across text, reasoning, + // and every tool call — mirroring the Anthropic adapter's + // `content_block_start`/`_delta`/`_stop` triplets. + let raw: Vec> = vec![ + format!( + "data: {}\n\n", + json!({ "choices": [{ "delta": { "reasoning_content": "Let me think. " } }] }) + ) + .into_bytes(), + format!( + "data: {}\n\n", + json!({ "choices": [{ "delta": { "content": "The answer is 42." } }] }) + ) + .into_bytes(), + format!( + "data: {}\n\n", + json!({ + "choices": [{ + "delta": { + "tool_calls": [{ + "index": 0, + "id": "call-1", + "function": { "name": "get_weather", "arguments": "{\"city\":" } + }] + } + }] + }) + ) + .into_bytes(), + format!( + "data: {}\n\n", + json!({ + "choices": [{ + "delta": { + "tool_calls": [{ + "index": 0, + "function": { "arguments": "\"NYC\"}" } + }] + } + }] + }) + ) + .into_bytes(), + format!( + "data: {}\n\n", + json!({ + "choices": [{ + "delta": { + "tool_calls": [{ + "index": 1, + "id": "call-2", + "function": { "name": "get_time", "arguments": "{\"tz\":\"UTC\"}" } + }] + }, + "finish_reason": "tool_calls" + }] + }) + ) + .into_bytes(), + b"data: [DONE]\n\n".to_vec(), + ]; + + let items = collect_sse(raw).await; + + let blocks: Vec<&ModelStreamItem> = items + .iter() + .filter(|item| { + matches!( + item, + ModelStreamItem::BlockStart { .. } + | ModelStreamItem::BlockDelta { .. } + | ModelStreamItem::BlockEnd { .. } + ) + }) + .collect(); + assert_eq!(blocks.len(), 13, "unexpected block sequence: {blocks:#?}"); + + // Reasoning block (index 0): opens first, one delta, closes when the text + // block takes over. + let (index, kind) = expect_block_start(blocks[0]); + assert_eq!(index, 0); + assert_eq!(*kind, BlockKind::Thinking); + let (index, delta) = expect_block_delta(blocks[1]); + assert_eq!(index, 0); + assert_eq!(*delta, BlockDelta::Thinking("Let me think. ".to_string())); + let (index, block) = expect_block_end(blocks[2]); + assert_eq!(index, 0); + assert_eq!( + *block, + ContentBlock::Thinking { + text: "Let me think. ".to_string(), + signature: None, + } + ); + + // Text block (index 1): opens, one delta, closes when the first tool + // call's id/name arrives. + let (index, kind) = expect_block_start(blocks[3]); + assert_eq!(index, 1); + assert_eq!(*kind, BlockKind::Text); + let (index, delta) = expect_block_delta(blocks[4]); + assert_eq!(index, 1); + assert_eq!(*delta, BlockDelta::Text("The answer is 42.".to_string())); + let (index, block) = expect_block_end(blocks[5]); + assert_eq!(index, 1); + assert_eq!(*block, ContentBlock::Text("The answer is 42.".to_string())); + + // First tool call (index 2): opens on the fragment carrying its id/name, + // gets two argument deltas split across chunks, closes when the second + // tool call's id/name arrives. + let (index, kind) = expect_block_start(blocks[6]); + assert_eq!(index, 2); + assert_eq!( + *kind, + BlockKind::ToolCall { + id: "call-1".to_string(), + name: "get_weather".to_string(), + } + ); + let (index, delta) = expect_block_delta(blocks[7]); + assert_eq!(index, 2); + assert_eq!(*delta, BlockDelta::ToolArgs("{\"city\":".to_string())); + let (index, delta) = expect_block_delta(blocks[8]); + assert_eq!(index, 2); + assert_eq!(*delta, BlockDelta::ToolArgs("\"NYC\"}".to_string())); + let (index, block) = expect_block_end(blocks[9]); + assert_eq!(index, 2); + assert_eq!( + *block, + ContentBlock::Json(json!({ + "id": "call-1", + "name": "get_weather", + "arguments": { "city": "NYC" }, + })) + ); + + // Second tool call (index 3): opens, one argument delta, closes on + // `finish_reason`. + let (index, kind) = expect_block_start(blocks[10]); + assert_eq!(index, 3); + assert_eq!( + *kind, + BlockKind::ToolCall { + id: "call-2".to_string(), + name: "get_time".to_string(), + } + ); + let (index, delta) = expect_block_delta(blocks[11]); + assert_eq!(index, 3); + assert_eq!(*delta, BlockDelta::ToolArgs("{\"tz\":\"UTC\"}".to_string())); + let (index, block) = expect_block_end(blocks[12]); + assert_eq!(index, 3); + assert_eq!( + *block, + ContentBlock::Json(json!({ + "id": "call-2", + "name": "get_time", + "arguments": { "tz": "UTC" }, + })) + ); + + // Reducing just the block items — the `BlockEnd` payloads, in index + // order — reproduces the terminal `AssistantMessage`: the content blocks + // in order, and the tool calls with their reassembled arguments. + let mut reduced_content = Vec::new(); + let mut reduced_calls: Vec<(String, String, Value)> = Vec::new(); + for item in &blocks { + if let ModelStreamItem::BlockEnd { block, .. } = item { + match block { + ContentBlock::Json(value) => { + reduced_calls.push(( + value["id"].as_str().unwrap().to_string(), + value["name"].as_str().unwrap().to_string(), + value["arguments"].clone(), + )); + } + other => reduced_content.push(other.clone()), + } + } + } + + let response = items + .iter() + .find_map(|item| match item { + ModelStreamItem::Completed(response) => Some(response.clone()), + _ => None, + }) + .expect("stream must complete"); + + assert_eq!(response.message.content, reduced_content); + assert_eq!(response.message.tool_calls.len(), reduced_calls.len()); + for (call, (id, name, arguments)) in response.message.tool_calls.iter().zip(&reduced_calls) { + assert_eq!(&call.id, id); + assert_eq!(&call.name, name); + assert_eq!(&call.arguments, arguments); + } +} + +#[tokio::test] +async fn sse_stream_mid_stream_error_preserves_partial_message() { + // Text and a tool call arrive, then the provider streams a mid-stream + // `{"error": ...}` payload instead of a completion. + // `ProviderError::partial_message` must carry whatever had already + // accumulated (mirrors the Anthropic adapter's contract) instead of + // discarding it. + let raw: Vec> = vec![ + b"data: {\"choices\":[{\"delta\":{\"content\":\"partial answer\"}}]}\n\n".to_vec(), + format!( + "data: {}\n\n", + json!({ + "choices": [{ + "delta": { + "tool_calls": [{ + "index": 0, + "id": "call-1", + "function": { "name": "lookup", "arguments": "{\"q\":1}" } + }] + } + }] + }) + ) + .into_bytes(), + b"data: {\"error\":{\"message\":\"upstream exploded\"}}\n\n".to_vec(), + b"data: [DONE]\n\n".to_vec(), + ]; + + let items = collect_sse(raw).await; + let failed = items + .iter() + .find_map(|item| match item { + ModelStreamItem::ProviderFailed(error) => Some(error), + _ => None, + }) + .expect("mid-stream error should emit ProviderFailed"); + assert!(failed.message.contains("upstream exploded")); + + let partial = failed + .partial_message + .as_ref() + .expect("partial content accumulated before the failure must not be discarded"); + assert_eq!( + partial.content, + vec![ContentBlock::Text("partial answer".to_string())] + ); + assert_eq!(partial.tool_calls.len(), 1); + assert_eq!(partial.tool_calls[0].id, "call-1"); + assert_eq!(partial.tool_calls[0].name, "lookup"); + assert_eq!(partial.tool_calls[0].arguments, json!({ "q": 1 })); + + // No terminal Completed follows the failure. + assert!(matches!( + items.last(), + Some(ModelStreamItem::ProviderFailed(_)) + )); +} + +#[test] +fn stamp_origin_records_provider_api_and_effective_model() { + let model = OpenAiModel::new("key").with_model("gpt-4.1"); + let request = ModelRequest::default(); + let mut response = ModelResponse { + message: crate::message::AssistantMessage { + id: None, + content: Vec::new(), + tool_calls: Vec::new(), + usage: None, + origin: None, + }, + usage: None, + finish_reason: None, + raw: None, + resolved_model: None, + continue_turn: None, + served_from_cache: false, + correlation: None, + resolved_route: None, + }; + model.stamp_origin(&mut response, &request, CHAT_COMPLETIONS_API); + let origin = response.message.origin.expect("origin stamped"); + assert_eq!(origin.provider, "openai"); + assert_eq!(origin.api, "chat_completions"); + assert_eq!(origin.model, "gpt-4.1"); +} + +#[test] +fn stamp_origin_prefers_a_per_request_model_override() { + let model = OpenAiModel::new("key").with_model("gpt-4.1"); + let request = ModelRequest { + model: Some("gpt-4.1-mini".to_string()), + ..Default::default() + }; + let mut response = ModelResponse { + message: crate::message::AssistantMessage { + id: None, + content: Vec::new(), + tool_calls: Vec::new(), + usage: None, + origin: None, + }, + usage: None, + finish_reason: None, + raw: None, + resolved_model: None, + continue_turn: None, + served_from_cache: false, + correlation: None, + resolved_route: None, + }; + model.stamp_origin(&mut response, &request, RESPONSES_API); + let origin = response.message.origin.expect("origin stamped"); + assert_eq!(origin.api, "responses"); + assert_eq!(origin.model, "gpt-4.1-mini"); +} + +#[tokio::test] +async fn streamed_terminal_response_carries_origin() { + let raw: Vec> = vec![ + b"data: {\"choices\":[{\"delta\":{\"content\":\"hi\"},\"finish_reason\":\"stop\"}]}\n\n" + .to_vec(), + b"data: [DONE]\n\n".to_vec(), + ]; + let items = collect_sse(raw).await; + let completed = items + .into_iter() + .find_map(|item| match item { + ModelStreamItem::Completed(response) => Some(response), + _ => None, + }) + .expect("a terminal Completed item"); + let origin = completed + .message + .origin + .expect("origin stamped on stream terminal"); + assert_eq!(origin.provider, "openai"); + assert_eq!(origin.api, "chat_completions"); + assert_eq!(origin.model, "gpt-4.1-mini"); +} diff --git a/crates/tinyinference-llm/src/providers/openai/transport.rs b/crates/tinyinference-llm/src/providers/openai/transport.rs index 7dc64f1..cf17789 100644 --- a/crates/tinyinference-llm/src/providers/openai/transport.rs +++ b/crates/tinyinference-llm/src/providers/openai/transport.rs @@ -125,6 +125,10 @@ pub struct OpenAiModel { /// default: hosted OpenAI rejects unknown part fields, and its own cache is /// automatic. See [`Self::with_explicit_cache_control`]. pub(super) explicit_cache_control: bool, + /// Host-supplied request hooks and HTTP client override. See + /// [`crate::providers::ProviderRequestOptions`]. Currently applied to the + /// Chat Completions transport path only ([`Self::post_json`]). + request_options: crate::providers::ProviderRequestOptions, } impl std::fmt::Debug for OpenAiModel { @@ -287,6 +291,18 @@ pub(super) fn derive_profile(provider: &str, model: &str) -> ModelProfile { reasoning, reasoning_effort: reasoning, max_input_tokens: crate::model::context_window_for_model_id(model), + // The Chat Completions wire format translates every `Message::System` + // independently, at its transcript position, into a `role: "system"` + // wire message (`convert::translate_message`) — there is no single + // "the" system slot the way Anthropic's Messages API has one. A + // system message placed mid-transcript is therefore genuinely + // effective where it sits, not silently dropped or hoisted, so this + // defaults `true` for the OpenAI-compatible chat path. It is turned + // back off by `with_merge_system_into_user`, which folds every system + // message into the leading user turn and drops the role entirely — + // at that point there is no wire position for a mid-transcript patch + // to occupy. + mid_conversation_system_messages: true, ..ModelProfile::default() } } @@ -330,9 +346,22 @@ impl OpenAiModel { json_schema_strict: AtomicBool::new(true), native_tools_on_wire: AtomicBool::new(true), responses_requires_stream: AtomicBool::new(false), + request_options: crate::providers::ProviderRequestOptions::default(), } } + /// Sets host-supplied request hooks and an optional HTTP client override + /// applied around Chat Completions calls this adapter makes. See + /// [`crate::providers::ProviderRequestOptions`]. + #[must_use] + pub fn with_request_options( + mut self, + options: crate::providers::ProviderRequestOptions, + ) -> Self { + self.request_options = options; + self + } + /// Routes calls to the OpenAI **Responses API** (`/v1/responses`) instead of /// Chat Completions. Required for the OpenAI Codex OAuth backend; pair with /// [`with_extra_query_param`](Self::with_extra_query_param) + @@ -408,6 +437,12 @@ impl OpenAiModel { /// role, for OpenAI-compatible endpoints that reject a `system` role. pub fn with_merge_system_into_user(mut self) -> Self { self.merge_system_into_user = true; + // Once every system message is folded into the leading user turn, + // the wire request carries no `system`-role message at all, so + // there is no transcript position for a mid-conversation system + // patch to occupy — a caller must fold such a patch into the + // leading system message before this transform runs instead. + self.profile.mid_conversation_system_messages = false; self } @@ -999,6 +1034,31 @@ impl OpenAiModel { Ok(()) } + /// Stamps [`crate::message::AssistantMessage::origin`] on a freshly built + /// response with this instance's configured provider/model and the given + /// API surface (see [`CHAT_COMPLETIONS_API`]/[`RESPONSES_API`]). + /// + /// Every response-building path (unary, the non-streaming SSE fallback, + /// and the Responses API) funnels through here (or the analogous + /// `sse_next` streamed-terminal site) so a later cross-provider handoff + /// transform can detect when a message was produced by a different + /// provider/model than the one it is about to be replayed against. Local + /// OpenAI-compatible runtimes (Ollama, LM Studio, …) share this transport + /// and are stamped with their own `provider` (e.g. `"ollama"`), not + /// `"openai"`. + pub(super) fn stamp_origin( + &self, + response: &mut ModelResponse, + request: &ModelRequest, + api: &str, + ) { + response.message.origin = Some(crate::message::MessageOrigin { + provider: self.provider.clone(), + api: api.to_string(), + model: request.model.clone().unwrap_or_else(|| self.model.clone()), + }); + } + /// Returns the default model id this instance will request. pub fn model(&self) -> &str { &self.model @@ -1102,6 +1162,9 @@ impl OpenAiModel { }; let mut messages = source_messages .iter() + // `Message::Custom` is a host-side out-of-band record; never sent + // to the provider. + .filter(|message| !matches!(message, Message::Custom(_))) .map(translate_message) .collect::>>()?; if self.explicit_cache_control && request.wants_prompt_cache_breakpoints() { @@ -1393,7 +1456,9 @@ impl OpenAiModel { ) })?, }; - Ok(responses::parse_responses_response(value)) + let mut response = responses::parse_responses_response(value); + self.stamp_origin(&mut response, request, RESPONSES_API); + Ok(response) } /// Shared `POST {responses_url}` with auth, query params, and timeout, mapped @@ -1454,8 +1519,12 @@ impl OpenAiModel { streaming: bool, what: &str, ) -> Result { + crate::network_guard::ensure_network_models_allowed()?; let url = format!("{}/chat/completions", self.base_url); - let mut builder = self.authorized(self.client.post(&url)).json(body); + let mut payload = serde_json::to_value(body)?; + self.request_options.apply_payload(&mut payload); + let client = self.request_options.http.as_ref().unwrap_or(&self.client); + let mut builder = self.authorized(client.post(&url)).json(&payload); if let Some(timeout) = request_timeout(timeout_ms, streaming) { builder = builder.timeout(timeout); } @@ -1540,6 +1609,8 @@ impl OpenAiModel { retryable, raw, retry_after_ms: None, + partial_message: None, + stop_reason: None, } } @@ -1775,7 +1846,9 @@ impl ChatModel for OpenAiModel { .map_err(|e| Error::Model(format!("openai response body read failed: {e}")))?; let value: Value = serde_json::from_str(&text)?; - let response = parse_chat_response(value, self.effective_reasoning_tags())?; + self.request_options.observe_response(&value); + let mut response = parse_chat_response(value, self.effective_reasoning_tags())?; + self.stamp_origin(&mut response, &request, CHAT_COMPLETIONS_API); // Prompt-guided tools: recover text-mode tool calls into // `message.tool_calls` when native tool calling was suppressed. if self.prompt_guided_for(&request) { @@ -1852,6 +1925,7 @@ impl ChatModel for OpenAiModel { })?; let value: Value = serde_json::from_str(&text)?; let mut parsed = parse_chat_response(value, self.effective_reasoning_tags())?; + self.stamp_origin(&mut parsed, &request, CHAT_COMPLETIONS_API); if self.prompt_guided_for(&request) { parsed = crate::prompt_tools::recover_tool_calls(parsed, &request.tools); } diff --git a/crates/tinyinference-llm/src/providers/openai/types.rs b/crates/tinyinference-llm/src/providers/openai/types.rs index 6d241da..c28dde1 100644 --- a/crates/tinyinference-llm/src/providers/openai/types.rs +++ b/crates/tinyinference-llm/src/providers/openai/types.rs @@ -249,6 +249,11 @@ pub enum ContentPartWire { /// The `image_url` object. image_url: ImageUrlWire, }, + /// Inline base64 audio, as OpenAI's Chat Completions `input_audio` part. + InputAudio { + /// The `input_audio` object. + input_audio: InputAudioWire, + }, } /// The `image_url` payload of a [`ContentPartWire::ImageUrl`]. @@ -258,6 +263,16 @@ pub struct ImageUrlWire { pub url: String, } +/// The `input_audio` payload of a [`ContentPartWire::InputAudio`]. +#[derive(Clone, Debug, Serialize)] +pub struct InputAudioWire { + /// Base64-encoded audio bytes. + pub data: String, + /// Audio container format (for example `"wav"` or `"mp3"`), derived + /// from the media type. + pub format: String, +} + /// A single message in the request `messages` array. #[derive(Clone, Debug, Serialize)] pub struct ChatMessageWire { diff --git a/crates/tinyinference-llm/src/providers/test.rs b/crates/tinyinference-llm/src/providers/test.rs index 04a3595..5e72347 100644 --- a/crates/tinyinference-llm/src/providers/test.rs +++ b/crates/tinyinference-llm/src/providers/test.rs @@ -469,3 +469,81 @@ fn provider_spec_defaults_and_overrides_are_normalized() { assert_eq!(tinyhumans.base_url, "https://api.tinyhumans.ai/openai/v1"); assert!(tinyhumans.model.is_empty()); } + +// --------------------------------------------------------------------------- +// ProviderRequestOptions +// --------------------------------------------------------------------------- + +#[test] +fn provider_request_options_default_has_no_hooks_or_client_override() { + use crate::providers::ProviderRequestOptions; + + let options = ProviderRequestOptions::default(); + let mut payload = json!({"a": 1}); + options.apply_payload(&mut payload); + assert_eq!(payload, json!({"a": 1}), "no on_payload hook: unchanged"); + // observe_response must not panic when unset. + options.observe_response(&json!({"b": 2})); + assert!(options.http.is_none()); +} + +#[test] +fn provider_request_options_on_payload_mutates_the_wire_body() { + use crate::providers::ProviderRequestOptions; + use std::sync::Arc; + + let options = ProviderRequestOptions { + on_payload: Some(Arc::new(|payload: &mut serde_json::Value| { + payload["injected"] = json!(true); + })), + ..ProviderRequestOptions::default() + }; + let mut payload = json!({"model": "m"}); + options.apply_payload(&mut payload); + assert_eq!(payload, json!({"model": "m", "injected": true})); +} + +#[test] +fn provider_request_options_on_response_observes_without_mutating() { + use crate::providers::ProviderRequestOptions; + use std::sync::{Arc, Mutex}; + + let seen = Arc::new(Mutex::new(None)); + let seen_clone = seen.clone(); + let options = ProviderRequestOptions { + on_response: Some(Arc::new(move |response: &serde_json::Value| { + *seen_clone.lock().unwrap() = Some(response.clone()); + })), + ..ProviderRequestOptions::default() + }; + options.observe_response(&json!({"id": "resp_1"})); + assert_eq!(*seen.lock().unwrap(), Some(json!({"id": "resp_1"}))); +} + +#[test] +fn provider_request_options_debug_redacts_closures() { + use crate::providers::ProviderRequestOptions; + use std::sync::Arc; + + let options = ProviderRequestOptions { + on_payload: Some(Arc::new(|_: &mut serde_json::Value| {})), + ..ProviderRequestOptions::default() + }; + let rendered = format!("{options:?}"); + assert!(rendered.contains("")); + assert!(!rendered.contains("closure")); +} + +#[test] +fn anthropic_and_openai_adapters_accept_request_options() { + use crate::providers::ProviderRequestOptions; + use crate::providers::anthropic::AnthropicModel; + use crate::providers::openai::OpenAiModel; + + // Builder methods compile and construct successfully; the hooks + // themselves are exercised in isolation above since these adapters have + // no network-free way to observe an outbound request body. + let _anthropic = + AnthropicModel::new("key").with_request_options(ProviderRequestOptions::default()); + let _openai = OpenAiModel::new("key").with_request_options(ProviderRequestOptions::default()); +} diff --git a/crates/tinyinference-llm/src/providers/types.rs b/crates/tinyinference-llm/src/providers/types.rs index 4e9f48a..0f08fb2 100644 --- a/crates/tinyinference-llm/src/providers/types.rs +++ b/crates/tinyinference-llm/src/providers/types.rs @@ -3,13 +3,78 @@ //! All public and internal types for the `providers` module live here. //! Implementations and trait-impls are in `mod.rs`. -use std::sync::Mutex; +use std::sync::{Arc, Mutex}; use serde::{Deserialize, Serialize}; use serde_json::Value; use crate::model::ModelResponse; +// --------------------------------------------------------------------------- +// Provider request options +// --------------------------------------------------------------------------- + +/// Host-supplied hooks and transport override applied around a single +/// provider adapter's HTTP call. +/// +/// Set on an adapter at construction time (for example +/// `OpenAiModel::with_request_options`) rather than per [`crate::model::ModelRequest`]: +/// [`ModelRequest`](crate::model::ModelRequest) is a serializable, provider-neutral value and +/// cannot carry closures. `on_payload` runs immediately before the adapter +/// serializes and sends the wire request body, so a host can inject +/// provider-specific fields (for example a beta header's JSON companion, or +/// an organization id) without the harness knowing about them. `on_response` +/// runs after a successful response is parsed into JSON, before the adapter +/// normalizes it, so a host can log or inspect the raw payload. `http` +/// overrides the adapter's own [`reqwest::Client`] (for a custom proxy, +/// timeout, or TLS configuration) when set. +/// +/// Neither hook may fail: they observe or mutate in place. A hook that needs +/// to reject a request should be implemented as request validation before the +/// call is made instead. +#[derive(Clone, Default)] +pub struct ProviderRequestOptions { + /// Invoked with the mutable wire payload immediately before it is sent. + pub on_payload: Option, + /// Invoked with the raw response payload after a successful call. + pub on_response: Option, + /// HTTP client to use in place of the adapter's own, when set. + pub http: Option, +} + +/// A payload-mutation hook: see [`ProviderRequestOptions::on_payload`]. +pub type PayloadHook = Arc; + +/// A response-observation hook: see [`ProviderRequestOptions::on_response`]. +pub type ResponseHook = Arc; + +impl ProviderRequestOptions { + /// Runs [`Self::on_payload`], when set. + pub fn apply_payload(&self, payload: &mut Value) { + if let Some(hook) = &self.on_payload { + hook(payload); + } + } + + /// Runs [`Self::on_response`], when set. + pub fn observe_response(&self, response: &Value) { + if let Some(hook) = &self.on_response { + hook(response); + } + } +} + +impl std::fmt::Debug for ProviderRequestOptions { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("ProviderRequestOptions") + .field("on_payload", &self.on_payload.as_ref().map(|_| "")) + .field("on_response", &self.on_response.as_ref().map(|_| "")) + .field("http", &self.http.as_ref().map(|_| "")) + .finish() + } +} + // --------------------------------------------------------------------------- // Provider selection types // --------------------------------------------------------------------------- diff --git a/crates/tinyinference-llm/src/tool.rs b/crates/tinyinference-llm/src/tool.rs index f4f6e43..6601d43 100644 --- a/crates/tinyinference-llm/src/tool.rs +++ b/crates/tinyinference-llm/src/tool.rs @@ -132,7 +132,7 @@ impl ToolCall { } /// An incremental tool-call fragment emitted by a model stream. -#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)] pub struct ToolDelta { /// Call identifier this fragment belongs to. pub call_id: String, @@ -141,6 +141,11 @@ pub struct ToolDelta { /// Tool name when the provider supplies it. #[serde(default, skip_serializing_if = "Option::is_none")] pub tool_name: Option, + /// Position of this tool call's content block within the assistant + /// message, when the provider reports block-indexed content (Anthropic's + /// `content_block` index, or the OpenAI `tool_calls[].index`). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub content_index: Option, } fn validate_schema_value(schema: &Value, value: &Value, path: &str) -> crate::Result<()> { diff --git a/crates/tinyinference-local/src/service/model_rpc_tests.rs b/crates/tinyinference-local/src/service/model_rpc_tests.rs index 2007feb..221347a 100644 --- a/crates/tinyinference-local/src/service/model_rpc_tests.rs +++ b/crates/tinyinference-local/src/service/model_rpc_tests.rs @@ -41,6 +41,7 @@ fn model_outcome_enforces_empty_and_normalizes_usage() { content: vec![ContentBlock::Text(text.to_string())], tool_calls: Vec::new(), usage: Some(usage), + origin: None, }, usage: Some(usage), finish_reason: None, @@ -82,6 +83,7 @@ fn model_outcome_enforces_empty_and_normalizes_usage() { }], tool_calls: Vec::new(), usage: None, + origin: None, }, usage: None, finish_reason: None,