diff --git a/crates/harness-app/src/lib.rs b/crates/harness-app/src/lib.rs index 999bcb3..d81b365 100644 --- a/crates/harness-app/src/lib.rs +++ b/crates/harness-app/src/lib.rs @@ -19,7 +19,7 @@ use harness_core::tool::ToolRegistry; use harness_core::types::{ Message, MessageId, ModelRef, Part, PartBody, PartId, Session, SessionId, }; -use harness_providers::{AnthropicProvider, ProviderRegistry}; +use harness_providers::{AnthropicProvider, OpenAiProvider, ProviderRegistry}; use tokio::task::JoinHandle; use tokio_util::sync::CancellationToken; @@ -122,6 +122,21 @@ impl EngineHandle { { providers.register(Arc::new(AnthropicProvider::new(key))); } + if let Some(key) = config + .providers + .get("openai") + .and_then(|p| p.api_key.clone()) + { + let provider = match config + .providers + .get("openai") + .and_then(|p| p.base_url.clone()) + { + Some(base_url) => OpenAiProvider::with_base_url(key, base_url), + None => OpenAiProvider::new(key), + }; + providers.register(Arc::new(provider)); + } let data_dir = dirs::data_dir() .unwrap_or_else(std::env::temp_dir) diff --git a/crates/harness-providers/src/codec/mod.rs b/crates/harness-providers/src/codec/mod.rs index e529997..daffdf8 100644 --- a/crates/harness-providers/src/codec/mod.rs +++ b/crates/harness-providers/src/codec/mod.rs @@ -1 +1,3 @@ pub mod anthropic; +pub mod openai_chat; +pub mod openai_responses; diff --git a/crates/harness-providers/src/codec/openai_chat.rs b/crates/harness-providers/src/codec/openai_chat.rs new file mode 100644 index 0000000..a9310eb --- /dev/null +++ b/crates/harness-providers/src/codec/openai_chat.rs @@ -0,0 +1,502 @@ +//! Request builder + SSE decoder for OpenAI's `/chat/completions` streaming API. +//! +//! Also used as the fallback codec for OpenAI-compatible endpoints (including Copilot models +//! whose `supported_endpoints` list `/chat/completions`). + +use std::collections::HashMap; + +use async_stream::try_stream; +use eventsource_stream::Eventsource; +use futures::Stream; +use harness_core::llm::{ + FinishReason, LlmEvent, LlmEventStream, LlmRequest, ProviderError, ReasoningEffort, Role, + WireContent, +}; +use harness_core::types::TokenUsage; +use serde_json::{json, Value}; + +fn effort_str(effort: ReasoningEffort) -> &'static str { + match effort { + ReasoningEffort::Low => "low", + ReasoningEffort::Medium => "medium", + ReasoningEffort::High => "high", + } +} + +fn role_str(role: Role) -> &'static str { + match role { + Role::System => "system", + Role::User => "user", + Role::Assistant => "assistant", + Role::Tool => "tool", + } +} + +/// Flattens our grouped `WireMessage`s into the flat chat-completions message list. Tool +/// results become their own `role: "tool"` messages (one per result), which is what the API +/// expects regardless of how the engine grouped them. +fn build_messages(system: &[String], messages: &[harness_core::llm::WireMessage]) -> Vec { + let mut out: Vec = Vec::new(); + + if !system.is_empty() { + out.push(json!({"role": "system", "content": system.join("\n\n")})); + } + + for m in messages { + // Tool results are always emitted as standalone `tool` messages. + for c in &m.content { + if let WireContent::ToolResult { + call_id, output, .. + } = c + { + out.push(json!({ + "role": "tool", + "tool_call_id": call_id, + "content": output, + })); + } + } + + let mut text = String::new(); + let mut images: Vec = Vec::new(); + let mut tool_calls: Vec = Vec::new(); + for c in &m.content { + match c { + WireContent::Text { text: t } => { + if !text.is_empty() { + text.push('\n'); + } + text.push_str(t); + } + WireContent::ToolCall { + call_id, + name, + input, + } => tool_calls.push(json!({ + "id": call_id, + "type": "function", + "function": {"name": name, "arguments": input.to_string()}, + })), + WireContent::Image { mime_type, data } => images.push(json!({ + "type": "image_url", + "image_url": {"url": format!("data:{mime_type};base64,{data}")}, + })), + WireContent::ToolResult { .. } => {} // handled above + } + } + + // A message that carried nothing but tool results contributes no further entry. + if text.is_empty() && images.is_empty() && tool_calls.is_empty() { + continue; + } + + let mut msg = json!({"role": role_str(m.role)}); + if images.is_empty() { + msg["content"] = json!(text); + } else { + let mut parts = vec![json!({"type": "text", "text": text})]; + parts.extend(images); + msg["content"] = json!(parts); + } + if !tool_calls.is_empty() { + msg["tool_calls"] = json!(tool_calls); + } + out.push(msg); + } + + out +} + +pub fn build_request(req: &LlmRequest) -> Value { + let mut body = json!({ + "model": req.model, + "messages": build_messages(&req.system, &req.messages), + "stream": true, + "stream_options": {"include_usage": true}, + }); + + if !req.tools.is_empty() { + body["tools"] = Value::Array( + req.tools + .iter() + .map(|t| { + json!({ + "type": "function", + "function": { + "name": t.name, + "description": t.description, + "parameters": t.parameters, + }, + }) + }) + .collect(), + ); + } + if let Some(temp) = req.temperature { + body["temperature"] = json!(temp); + } + if let Some(max) = req.max_tokens { + body["max_completion_tokens"] = json!(max); + } + if let Some(effort) = req.reasoning.as_ref().and_then(|r| r.effort) { + body["reasoning_effort"] = json!(effort_str(effort)); + } + body +} + +fn map_finish_reason(reason: &str) -> FinishReason { + match reason { + "stop" => FinishReason::Stop, + "tool_calls" | "function_call" => FinishReason::ToolCalls, + "length" => FinishReason::Length, + "content_filter" => FinishReason::ContentFilter, + other => FinishReason::Unknown(other.to_string()), + } +} + +#[derive(Default)] +struct ToolAccum { + call_id: String, + name: String, + args: String, +} + +/// Decodes a chat-completions SSE byte stream into normalized `LlmEvent`s. Tool-call deltas +/// are accumulated by their `index` and flushed as `ToolCall`s once the stream ends. +pub fn decode(byte_stream: S) -> LlmEventStream +where + S: Stream> + Send + 'static, + E: std::error::Error + Send + Sync + 'static, +{ + let events = byte_stream.eventsource(); + Box::pin(try_stream! { + futures::pin_mut!(events); + let mut usage = TokenUsage::default(); + let mut reason = FinishReason::Stop; + let mut text_open = false; + let mut reasoning_open = false; + let mut tools: HashMap = HashMap::new(); + let mut tool_order: Vec = Vec::new(); + + while let Some(item) = futures::StreamExt::next(&mut events).await { + let event = item.map_err(|e| ProviderError::Decode(e.to_string()))?; + let data = event.data.trim(); + if data.is_empty() { + continue; + } + if data == "[DONE]" { + break; + } + let value: Value = serde_json::from_str(data) + .map_err(|e| ProviderError::Decode(format!("{e}: {data}")))?; + + if let Some(u) = value.get("usage").filter(|u| u.is_object()) { + usage.input = u["prompt_tokens"].as_u64().unwrap_or(usage.input); + usage.output = u["completion_tokens"].as_u64().unwrap_or(usage.output); + usage.reasoning = u["completion_tokens_details"]["reasoning_tokens"] + .as_u64() + .unwrap_or(usage.reasoning); + usage.cache_read = u["prompt_tokens_details"]["cached_tokens"] + .as_u64() + .unwrap_or(usage.cache_read); + } + + let choice = &value["choices"][0]; + let delta = &choice["delta"]; + + if let Some(rc) = delta["reasoning_content"].as_str().filter(|s| !s.is_empty()) { + if !reasoning_open { + reasoning_open = true; + yield LlmEvent::ReasoningStart { id: "reasoning".into() }; + } + yield LlmEvent::ReasoningDelta { id: "reasoning".into(), text: rc.to_string() }; + } + + if let Some(text) = delta["content"].as_str().filter(|s| !s.is_empty()) { + if reasoning_open { + reasoning_open = false; + yield LlmEvent::ReasoningEnd { id: "reasoning".into(), signature: None }; + } + if !text_open { + text_open = true; + yield LlmEvent::TextStart { id: "0".into() }; + } + yield LlmEvent::TextDelta { id: "0".into(), text: text.to_string() }; + } + + if let Some(calls) = delta["tool_calls"].as_array() { + for call in calls { + let index = call["index"].as_u64().unwrap_or(0); + let entry = tools.entry(index).or_insert_with(|| { + tool_order.push(index); + ToolAccum::default() + }); + if let Some(id) = call["id"].as_str().filter(|s| !s.is_empty()) { + entry.call_id = id.to_string(); + } + if let Some(name) = call["function"]["name"].as_str().filter(|s| !s.is_empty()) { + entry.name = name.to_string(); + yield LlmEvent::ToolInputStart { + call_id: entry.call_id.clone(), + name: entry.name.clone(), + }; + } + if let Some(args) = call["function"]["arguments"].as_str().filter(|s| !s.is_empty()) { + entry.args.push_str(args); + yield LlmEvent::ToolInputDelta { + call_id: entry.call_id.clone(), + json: args.to_string(), + }; + } + } + } + + if let Some(fr) = choice["finish_reason"].as_str() { + reason = map_finish_reason(fr); + } + } + + if reasoning_open { + yield LlmEvent::ReasoningEnd { id: "reasoning".into(), signature: None }; + } + if text_open { + yield LlmEvent::TextEnd { id: "0".into() }; + } + for index in tool_order { + if let Some(entry) = tools.remove(&index) { + let input: Value = if entry.args.trim().is_empty() { + json!({}) + } else { + serde_json::from_str(&entry.args).unwrap_or(Value::Null) + }; + yield LlmEvent::ToolCall { call_id: entry.call_id, name: entry.name, input }; + } + } + yield LlmEvent::Finish { reason, usage }; + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use futures::StreamExt; + use harness_core::llm::{Initiator, ReasoningOpts, ToolSchema, WireMessage}; + + fn sse_stream(raw: &'static str) -> LlmEventStream { + let chunks: Vec> = + vec![Ok(bytes::Bytes::from_static(raw.as_bytes()))]; + decode(futures::stream::iter(chunks)) + } + + fn req(model: &str) -> LlmRequest { + LlmRequest { + model: model.into(), + system: vec![], + messages: vec![], + tools: vec![], + temperature: None, + max_tokens: None, + reasoning: None, + initiator: Initiator::User, + } + } + + #[tokio::test] + async fn decodes_text_only_response() { + let raw = concat!( + "data: {\"choices\":[{\"delta\":{\"content\":\"Hel\"},\"finish_reason\":null}]}\n\n", + "data: {\"choices\":[{\"delta\":{\"content\":\"lo\"},\"finish_reason\":null}]}\n\n", + "data: {\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n", + "data: {\"choices\":[],\"usage\":{\"prompt_tokens\":10,\"completion_tokens\":5}}\n\n", + "data: [DONE]\n\n", + ); + let events: Vec = sse_stream(raw).map(|e| e.unwrap()).collect().await; + assert_eq!( + events, + vec![ + LlmEvent::TextStart { id: "0".into() }, + LlmEvent::TextDelta { + id: "0".into(), + text: "Hel".into() + }, + LlmEvent::TextDelta { + id: "0".into(), + text: "lo".into() + }, + LlmEvent::TextEnd { id: "0".into() }, + LlmEvent::Finish { + reason: FinishReason::Stop, + usage: TokenUsage { + input: 10, + output: 5, + ..Default::default() + }, + }, + ] + ); + } + + #[tokio::test] + async fn decodes_tool_call_accumulated_by_index() { + let raw = concat!( + "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_1\",\"function\":{\"name\":\"read\",\"arguments\":\"\"}}]}}]}\n\n", + "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\"{\\\"file\\\"\"}}]}}]}\n\n", + "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"function\":{\"arguments\":\":\\\"a.txt\\\"}\"}}]}}]}\n\n", + "data: {\"choices\":[{\"delta\":{},\"finish_reason\":\"tool_calls\"}],\"usage\":{\"prompt_tokens\":1,\"completion_tokens\":8}}\n\n", + "data: [DONE]\n\n", + ); + let events: Vec = sse_stream(raw).map(|e| e.unwrap()).collect().await; + assert_eq!( + events, + vec![ + LlmEvent::ToolInputStart { + call_id: "call_1".into(), + name: "read".into() + }, + LlmEvent::ToolInputDelta { + call_id: "call_1".into(), + json: "{\"file\"".into() + }, + LlmEvent::ToolInputDelta { + call_id: "call_1".into(), + json: ":\"a.txt\"}".into() + }, + LlmEvent::ToolCall { + call_id: "call_1".into(), + name: "read".into(), + input: json!({"file": "a.txt"}), + }, + LlmEvent::Finish { + reason: FinishReason::ToolCalls, + usage: TokenUsage { + input: 1, + output: 8, + ..Default::default() + }, + }, + ] + ); + } + + #[tokio::test] + async fn decodes_reasoning_content_before_text() { + let raw = concat!( + "data: {\"choices\":[{\"delta\":{\"reasoning_content\":\"hmm\"}}]}\n\n", + "data: {\"choices\":[{\"delta\":{\"content\":\"answer\"}}]}\n\n", + "data: {\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n", + "data: [DONE]\n\n", + ); + let events: Vec = sse_stream(raw).map(|e| e.unwrap()).collect().await; + assert_eq!( + events, + vec![ + LlmEvent::ReasoningStart { + id: "reasoning".into() + }, + LlmEvent::ReasoningDelta { + id: "reasoning".into(), + text: "hmm".into() + }, + LlmEvent::ReasoningEnd { + id: "reasoning".into(), + signature: None + }, + LlmEvent::TextStart { id: "0".into() }, + LlmEvent::TextDelta { + id: "0".into(), + text: "answer".into() + }, + LlmEvent::TextEnd { id: "0".into() }, + LlmEvent::Finish { + reason: FinishReason::Stop, + usage: TokenUsage::default() + }, + ] + ); + } + + #[test] + fn build_request_wraps_tools_in_function_envelope() { + let mut r = req("gpt-4o"); + r.tools = vec![ToolSchema { + name: "read".into(), + description: "reads a file".into(), + parameters: json!({"type": "object"}), + }]; + let body = build_request(&r); + assert_eq!(body["tools"][0]["type"], "function"); + assert_eq!(body["tools"][0]["function"]["name"], "read"); + assert_eq!( + body["tools"][0]["function"]["parameters"], + json!({"type": "object"}) + ); + assert_eq!(body["stream_options"]["include_usage"], true); + } + + #[test] + fn build_request_prepends_system_message() { + let mut r = req("gpt-4o"); + r.system = vec!["env".into(), "agent".into()]; + r.messages = vec![WireMessage { + role: Role::User, + content: vec![WireContent::Text { text: "hi".into() }], + }]; + let body = build_request(&r); + let msgs = body["messages"].as_array().unwrap(); + assert_eq!(msgs[0]["role"], "system"); + assert_eq!(msgs[0]["content"], "env\n\nagent"); + assert_eq!(msgs[1]["role"], "user"); + assert_eq!(msgs[1]["content"], "hi"); + } + + #[test] + fn build_request_expands_tool_results_to_tool_messages() { + let mut r = req("gpt-4o"); + r.messages = vec![WireMessage { + role: Role::Tool, + content: vec![WireContent::ToolResult { + call_id: "call_1".into(), + output: "contents".into(), + is_error: false, + }], + }]; + let body = build_request(&r); + let msgs = body["messages"].as_array().unwrap(); + assert_eq!(msgs.len(), 1); + assert_eq!(msgs[0]["role"], "tool"); + assert_eq!(msgs[0]["tool_call_id"], "call_1"); + assert_eq!(msgs[0]["content"], "contents"); + } + + #[test] + fn build_request_emits_assistant_tool_calls() { + let mut r = req("gpt-4o"); + r.messages = vec![WireMessage { + role: Role::Assistant, + content: vec![WireContent::ToolCall { + call_id: "call_1".into(), + name: "read".into(), + input: json!({"file": "a.txt"}), + }], + }]; + let body = build_request(&r); + let msgs = body["messages"].as_array().unwrap(); + assert_eq!(msgs[0]["tool_calls"][0]["id"], "call_1"); + assert_eq!(msgs[0]["tool_calls"][0]["function"]["name"], "read"); + assert_eq!( + msgs[0]["tool_calls"][0]["function"]["arguments"], + "{\"file\":\"a.txt\"}" + ); + } + + #[test] + fn build_request_sets_reasoning_effort() { + let mut r = req("o3"); + r.reasoning = Some(ReasoningOpts { + effort: Some(ReasoningEffort::High), + budget_tokens: None, + }); + let body = build_request(&r); + assert_eq!(body["reasoning_effort"], "high"); + } +} diff --git a/crates/harness-providers/src/codec/openai_responses.rs b/crates/harness-providers/src/codec/openai_responses.rs new file mode 100644 index 0000000..fc5de23 --- /dev/null +++ b/crates/harness-providers/src/codec/openai_responses.rs @@ -0,0 +1,402 @@ +//! Request builder + SSE decoder for OpenAI's `/responses` streaming API. +//! +//! The responses API is the preferred surface for `gpt-*` / `o-*` models: it carries +//! reasoning items natively and reports reasoning-token usage separately. + +use std::collections::HashMap; + +use async_stream::try_stream; +use eventsource_stream::Eventsource; +use futures::Stream; +use harness_core::llm::{ + FinishReason, LlmEvent, LlmEventStream, LlmRequest, ProviderError, ReasoningEffort, Role, + WireContent, +}; +use harness_core::types::TokenUsage; +use serde_json::{json, Value}; + +fn effort_str(effort: ReasoningEffort) -> &'static str { + match effort { + ReasoningEffort::Low => "low", + ReasoningEffort::Medium => "medium", + ReasoningEffort::High => "high", + } +} + +/// Builds the `input` item list. Text/image content become `message` items; tool calls and +/// their results become `function_call` / `function_call_output` items (the responses API +/// keeps these as top-level items rather than nesting them inside messages). +fn build_input(messages: &[harness_core::llm::WireMessage]) -> Vec { + let mut out: Vec = Vec::new(); + for m in messages { + let (role, text_type) = match m.role { + Role::Assistant => ("assistant", "output_text"), + _ => ("user", "input_text"), + }; + let mut content_parts: Vec = Vec::new(); + for c in &m.content { + match c { + WireContent::Text { text } => { + content_parts.push(json!({"type": text_type, "text": text})); + } + WireContent::Image { mime_type, data } => { + content_parts.push(json!({ + "type": "input_image", + "image_url": format!("data:{mime_type};base64,{data}"), + })); + } + WireContent::ToolCall { + call_id, + name, + input, + } => out.push(json!({ + "type": "function_call", + "call_id": call_id, + "name": name, + "arguments": input.to_string(), + })), + WireContent::ToolResult { + call_id, output, .. + } => out.push(json!({ + "type": "function_call_output", + "call_id": call_id, + "output": output, + })), + } + } + if !content_parts.is_empty() { + out.push(json!({"type": "message", "role": role, "content": content_parts})); + } + } + out +} + +pub fn build_request(req: &LlmRequest) -> Value { + let mut body = json!({ + "model": req.model, + "input": build_input(&req.messages), + "stream": true, + "store": false, + }); + + if !req.system.is_empty() { + body["instructions"] = json!(req.system.join("\n\n")); + } + if !req.tools.is_empty() { + body["tools"] = Value::Array( + req.tools + .iter() + .map(|t| { + json!({ + "type": "function", + "name": t.name, + "description": t.description, + "parameters": t.parameters, + }) + }) + .collect(), + ); + } + if let Some(temp) = req.temperature { + body["temperature"] = json!(temp); + } + if let Some(max) = req.max_tokens { + body["max_output_tokens"] = json!(max); + } + if let Some(effort) = req.reasoning.as_ref().and_then(|r| r.effort) { + body["reasoning"] = json!({"effort": effort_str(effort), "summary": "auto"}); + } + body +} + +fn map_status(status: Option<&str>, had_tool_call: bool) -> FinishReason { + match status { + Some("completed") if had_tool_call => FinishReason::ToolCalls, + Some("completed") => FinishReason::Stop, + Some("incomplete") => FinishReason::Length, + Some(other) => FinishReason::Unknown(other.to_string()), + None if had_tool_call => FinishReason::ToolCalls, + None => FinishReason::Stop, + } +} + +/// Tracks which normalized stream an `item_id` belongs to so deltas route correctly. +enum ItemKind { + Text, + Reasoning, + FunctionCall { call_id: String, name: String }, +} + +/// Decodes a `/responses` SSE byte stream into normalized `LlmEvent`s. +pub fn decode(byte_stream: S) -> LlmEventStream +where + S: Stream> + Send + 'static, + E: std::error::Error + Send + Sync + 'static, +{ + let events = byte_stream.eventsource(); + Box::pin(try_stream! { + futures::pin_mut!(events); + let mut items: HashMap = HashMap::new(); + let mut fn_args: HashMap = HashMap::new(); + let mut usage = TokenUsage::default(); + let mut had_tool_call = false; + + while let Some(item) = futures::StreamExt::next(&mut events).await { + let event = item.map_err(|e| ProviderError::Decode(e.to_string()))?; + if event.data.trim().is_empty() { + continue; + } + let value: Value = serde_json::from_str(&event.data) + .map_err(|e| ProviderError::Decode(format!("{e}: {}", event.data)))?; + let kind = value["type"].as_str().unwrap_or_default(); + + match kind { + "response.output_item.added" => { + let item = &value["item"]; + let id = item["id"].as_str().unwrap_or_default().to_string(); + match item["type"].as_str().unwrap_or_default() { + "message" => { + items.insert(id.clone(), ItemKind::Text); + yield LlmEvent::TextStart { id }; + } + "reasoning" => { + items.insert(id.clone(), ItemKind::Reasoning); + yield LlmEvent::ReasoningStart { id }; + } + "function_call" => { + let call_id = item["call_id"].as_str().unwrap_or_default().to_string(); + let name = item["name"].as_str().unwrap_or_default().to_string(); + fn_args.insert(id.clone(), String::new()); + items.insert(id, ItemKind::FunctionCall { call_id: call_id.clone(), name: name.clone() }); + had_tool_call = true; + yield LlmEvent::ToolInputStart { call_id, name }; + } + _ => {} + } + } + "response.output_text.delta" => { + let id = value["item_id"].as_str().unwrap_or_default().to_string(); + let text = value["delta"].as_str().unwrap_or_default().to_string(); + yield LlmEvent::TextDelta { id, text }; + } + "response.reasoning_summary_text.delta" => { + let id = value["item_id"].as_str().unwrap_or_default().to_string(); + let text = value["delta"].as_str().unwrap_or_default().to_string(); + yield LlmEvent::ReasoningDelta { id, text }; + } + "response.function_call_arguments.delta" => { + let id = value["item_id"].as_str().unwrap_or_default().to_string(); + let delta = value["delta"].as_str().unwrap_or_default(); + if let (Some(buf), Some(ItemKind::FunctionCall { call_id, .. })) = + (fn_args.get_mut(&id), items.get(&id)) + { + buf.push_str(delta); + yield LlmEvent::ToolInputDelta { call_id: call_id.clone(), json: delta.to_string() }; + } + } + "response.output_item.done" => { + let id = value["item"]["id"].as_str().unwrap_or_default().to_string(); + match items.remove(&id) { + Some(ItemKind::Text) => yield LlmEvent::TextEnd { id }, + Some(ItemKind::Reasoning) => yield LlmEvent::ReasoningEnd { id, signature: None }, + Some(ItemKind::FunctionCall { call_id, name }) => { + let raw = fn_args.remove(&id).unwrap_or_default(); + let input: Value = if raw.trim().is_empty() { + json!({}) + } else { + serde_json::from_str(&raw).unwrap_or(Value::Null) + }; + yield LlmEvent::ToolCall { call_id, name, input }; + } + None => {} + } + } + "response.completed" | "response.incomplete" => { + let response = &value["response"]; + let u = &response["usage"]; + usage.input = u["input_tokens"].as_u64().unwrap_or(0); + usage.output = u["output_tokens"].as_u64().unwrap_or(0); + usage.reasoning = u["output_tokens_details"]["reasoning_tokens"].as_u64().unwrap_or(0); + usage.cache_read = u["input_tokens_details"]["cached_tokens"].as_u64().unwrap_or(0); + let status = response["status"].as_str(); + yield LlmEvent::Finish { reason: map_status(status, had_tool_call), usage }; + } + "error" | "response.failed" => { + let message = value["response"]["error"]["message"] + .as_str() + .or_else(|| value["message"].as_str()) + .unwrap_or("unknown error") + .to_string(); + Err(ProviderError::Http { status: 0, body: message })?; + } + _ => {} + } + } + }) +} + +#[cfg(test)] +mod tests { + use super::*; + use futures::StreamExt; + use harness_core::llm::{Initiator, ReasoningOpts, ToolSchema, WireMessage}; + + fn sse_stream(raw: &'static str) -> LlmEventStream { + let chunks: Vec> = + vec![Ok(bytes::Bytes::from_static(raw.as_bytes()))]; + decode(futures::stream::iter(chunks)) + } + + fn req(model: &str) -> LlmRequest { + LlmRequest { + model: model.into(), + system: vec![], + messages: vec![], + tools: vec![], + temperature: None, + max_tokens: None, + reasoning: None, + initiator: Initiator::User, + } + } + + #[tokio::test] + async fn decodes_text_response() { + let raw = concat!( + "data: {\"type\":\"response.output_item.added\",\"item\":{\"id\":\"msg_1\",\"type\":\"message\"}}\n\n", + "data: {\"type\":\"response.output_text.delta\",\"item_id\":\"msg_1\",\"delta\":\"Hello\"}\n\n", + "data: {\"type\":\"response.output_item.done\",\"item\":{\"id\":\"msg_1\"}}\n\n", + "data: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\",\"usage\":{\"input_tokens\":10,\"output_tokens\":5}}}\n\n", + ); + let events: Vec = sse_stream(raw).map(|e| e.unwrap()).collect().await; + assert_eq!( + events, + vec![ + LlmEvent::TextStart { id: "msg_1".into() }, + LlmEvent::TextDelta { + id: "msg_1".into(), + text: "Hello".into() + }, + LlmEvent::TextEnd { id: "msg_1".into() }, + LlmEvent::Finish { + reason: FinishReason::Stop, + usage: TokenUsage { + input: 10, + output: 5, + ..Default::default() + }, + }, + ] + ); + } + + #[tokio::test] + async fn decodes_function_call_with_reasoning_tokens() { + let raw = concat!( + "data: {\"type\":\"response.output_item.added\",\"item\":{\"id\":\"fc_1\",\"type\":\"function_call\",\"call_id\":\"call_1\",\"name\":\"read\"}}\n\n", + "data: {\"type\":\"response.function_call_arguments.delta\",\"item_id\":\"fc_1\",\"delta\":\"{\\\"file\\\":\"}\n\n", + "data: {\"type\":\"response.function_call_arguments.delta\",\"item_id\":\"fc_1\",\"delta\":\"\\\"a.txt\\\"}\"}\n\n", + "data: {\"type\":\"response.output_item.done\",\"item\":{\"id\":\"fc_1\"}}\n\n", + "data: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\",\"usage\":{\"input_tokens\":3,\"output_tokens\":9,\"output_tokens_details\":{\"reasoning_tokens\":4}}}}\n\n", + ); + let events: Vec = sse_stream(raw).map(|e| e.unwrap()).collect().await; + assert_eq!( + events, + vec![ + LlmEvent::ToolInputStart { + call_id: "call_1".into(), + name: "read".into() + }, + LlmEvent::ToolInputDelta { + call_id: "call_1".into(), + json: "{\"file\":".into() + }, + LlmEvent::ToolInputDelta { + call_id: "call_1".into(), + json: "\"a.txt\"}".into() + }, + LlmEvent::ToolCall { + call_id: "call_1".into(), + name: "read".into(), + input: json!({"file": "a.txt"}), + }, + LlmEvent::Finish { + reason: FinishReason::ToolCalls, + usage: TokenUsage { + input: 3, + output: 9, + reasoning: 4, + ..Default::default() + }, + }, + ] + ); + } + + #[tokio::test] + async fn failed_response_surfaces_as_err() { + let raw = "data: {\"type\":\"response.failed\",\"response\":{\"error\":{\"message\":\"boom\"}}}\n\n"; + let events: Vec> = sse_stream(raw).collect().await; + assert!( + matches!(events.last(), Some(Err(ProviderError::Http { body, .. })) if body == "boom") + ); + } + + #[test] + fn build_request_puts_system_in_instructions() { + let mut r = req("gpt-5"); + r.system = vec!["env".into(), "agent".into()]; + let body = build_request(&r); + assert_eq!(body["instructions"], "env\n\nagent"); + assert!(body.get("messages").is_none()); + } + + #[test] + fn build_request_flattens_tool_calls_and_results_to_items() { + let mut r = req("gpt-5"); + r.messages = vec![ + WireMessage { + role: Role::Assistant, + content: vec![WireContent::ToolCall { + call_id: "call_1".into(), + name: "read".into(), + input: json!({"file": "a.txt"}), + }], + }, + WireMessage { + role: Role::Tool, + content: vec![WireContent::ToolResult { + call_id: "call_1".into(), + output: "contents".into(), + is_error: false, + }], + }, + ]; + let body = build_request(&r); + let input = body["input"].as_array().unwrap(); + assert_eq!(input[0]["type"], "function_call"); + assert_eq!(input[0]["call_id"], "call_1"); + assert_eq!(input[0]["arguments"], "{\"file\":\"a.txt\"}"); + assert_eq!(input[1]["type"], "function_call_output"); + assert_eq!(input[1]["output"], "contents"); + } + + #[test] + fn build_request_uses_flat_function_tool_shape_and_reasoning() { + let mut r = req("o3"); + r.tools = vec![ToolSchema { + name: "read".into(), + description: "reads a file".into(), + parameters: json!({"type": "object"}), + }]; + r.reasoning = Some(ReasoningOpts { + effort: Some(ReasoningEffort::Medium), + budget_tokens: None, + }); + let body = build_request(&r); + assert_eq!(body["tools"][0]["type"], "function"); + assert_eq!(body["tools"][0]["name"], "read"); + assert_eq!(body["reasoning"]["effort"], "medium"); + assert_eq!(body["reasoning"]["summary"], "auto"); + } +} diff --git a/crates/harness-providers/src/lib.rs b/crates/harness-providers/src/lib.rs index 63c8f2d..07cdffb 100644 --- a/crates/harness-providers/src/lib.rs +++ b/crates/harness-providers/src/lib.rs @@ -1,6 +1,8 @@ pub mod anthropic; pub mod codec; +pub mod openai; pub mod registry; pub use anthropic::AnthropicProvider; +pub use openai::OpenAiProvider; pub use registry::ProviderRegistry; diff --git a/crates/harness-providers/src/openai.rs b/crates/harness-providers/src/openai.rs new file mode 100644 index 0000000..9961d17 --- /dev/null +++ b/crates/harness-providers/src/openai.rs @@ -0,0 +1,173 @@ +use std::time::Duration; + +use async_trait::async_trait; +use harness_core::llm::{LlmEventStream, LlmRequest, Provider, ProviderError}; +use harness_core::types::ModelInfo; +use tokio_util::sync::CancellationToken; + +use crate::codec::{openai_chat, openai_responses}; + +const DEFAULT_BASE_URL: &str = "https://api.openai.com/v1"; + +/// Which OpenAI wire format to speak for a given model. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ApiFlavor { + Responses, + Chat, +} + +/// `gpt-*` and `o-*` models use the responses API; everything else (including +/// OpenAI-compatible third-party endpoints) falls back to chat completions. +fn flavor_for(model: &str) -> ApiFlavor { + let is_native = model.starts_with("gpt-") + || model.starts_with("o1") + || model.starts_with("o3") + || model.starts_with("o4") + || model == "o1" + || model == "o3"; + if is_native { + ApiFlavor::Responses + } else { + ApiFlavor::Chat + } +} + +pub struct OpenAiProvider { + api_key: String, + base_url: String, + client: reqwest::Client, +} + +impl OpenAiProvider { + pub fn new(api_key: impl Into) -> Self { + Self { + api_key: api_key.into(), + base_url: DEFAULT_BASE_URL.to_string(), + client: reqwest::Client::new(), + } + } + + pub fn with_base_url(api_key: impl Into, base_url: impl Into) -> Self { + Self { + api_key: api_key.into(), + base_url: base_url.into(), + client: reqwest::Client::new(), + } + } + + fn classify_error( + status: reqwest::StatusCode, + body: String, + retry_after: Option, + ) -> ProviderError { + match status.as_u16() { + 400 if body.contains("context_length_exceeded") + || body.contains("maximum context length") => + { + ProviderError::ContextOverflow + } + 401 | 403 => ProviderError::Auth(body), + 429 => ProviderError::RateLimited { retry_after }, + s if (500..600).contains(&s) => ProviderError::Overloaded, + s => ProviderError::Http { status: s, body }, + } + } +} + +#[async_trait] +impl Provider for OpenAiProvider { + fn id(&self) -> &str { + "openai" + } + + async fn list_models(&self) -> Result, ProviderError> { + // models.dev metadata is layered on top by the caller (see `modelsdev.rs`). + Ok(Vec::new()) + } + + async fn stream( + &self, + req: LlmRequest, + cancel: CancellationToken, + ) -> Result { + let (path, body) = match flavor_for(&req.model) { + ApiFlavor::Responses => ("/responses", openai_responses::build_request(&req)), + ApiFlavor::Chat => ("/chat/completions", openai_chat::build_request(&req)), + }; + let url = format!("{}{}", self.base_url, path); + + let send = self + .client + .post(&url) + .header("authorization", format!("Bearer {}", self.api_key)) + .json(&body) + .send(); + + let response = tokio::select! { + result = send => result.map_err(|e| ProviderError::Network(e.to_string()))?, + _ = cancel.cancelled() => return Err(ProviderError::Cancelled), + }; + + if !response.status().is_success() { + let status = response.status(); + let retry_after = response + .headers() + .get("retry-after") + .and_then(|v| v.to_str().ok()) + .and_then(|s| s.parse::().ok()) + .map(Duration::from_secs); + let body = response.text().await.unwrap_or_default(); + return Err(Self::classify_error(status, body, retry_after)); + } + + Ok(match flavor_for(&req.model) { + ApiFlavor::Responses => openai_responses::decode(response.bytes_stream()), + ApiFlavor::Chat => openai_chat::decode(response.bytes_stream()), + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn routes_gpt_and_o_series_to_responses() { + assert_eq!(flavor_for("gpt-4o"), ApiFlavor::Responses); + assert_eq!(flavor_for("gpt-5"), ApiFlavor::Responses); + assert_eq!(flavor_for("o1"), ApiFlavor::Responses); + assert_eq!(flavor_for("o3-mini"), ApiFlavor::Responses); + assert_eq!(flavor_for("o4-mini"), ApiFlavor::Responses); + } + + #[test] + fn routes_other_models_to_chat() { + assert_eq!(flavor_for("llama-3.1-70b"), ApiFlavor::Chat); + assert_eq!(flavor_for("deepseek-chat"), ApiFlavor::Chat); + } + + #[test] + fn id_is_openai() { + assert_eq!(OpenAiProvider::new("k").id(), "openai"); + } + + #[test] + fn classifies_context_overflow_from_400_body() { + assert!(matches!( + OpenAiProvider::classify_error( + reqwest::StatusCode::BAD_REQUEST, + "context_length_exceeded".into(), + None, + ), + ProviderError::ContextOverflow + )); + assert!(matches!( + OpenAiProvider::classify_error( + reqwest::StatusCode::BAD_REQUEST, + "some other error".into(), + None, + ), + ProviderError::Http { status: 400, .. } + )); + } +}