1use serde::{Deserialize, Serialize};
2use tokio::sync::broadcast;
3use tokio_util::sync::CancellationToken;
4
5use crate::error::RuntimeError;
6use crate::event::{NodeEvent, Observable};
7use crate::message::{Message, MessageOrigin, MessagePart, MessageRole};
8use crate::provider::{
9 AssistantMessage, CallTiming, DEFAULT_STREAM_BUFFER, LlmRequest, Provider, ReasoningEffort,
10 ReasoningSelection, ReasoningWireProfile, StopReason, TokenUsage, estimate_tokens,
11};
12use crate::providers::classify_attachment_error;
13use crate::tool::BoxFut;
14
15pub struct OpenAiProvider {
16 name: String,
17 api_key: String,
18 base_url: String,
19 client: reqwest::Client,
20 max_tokens: Option<u32>,
21 reasoning_format: OpenAiReasoningFormat,
22 prompt_cache_key: bool,
23}
24
25#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
26pub enum OpenAiReasoningFormat {
27 #[serde(rename = "reasoning-effort", alias = "official")]
28 Official,
29 #[default]
30 #[serde(rename = "thinking-toggle", alias = "compatible-thinking")]
31 CompatibleThinking,
32}
33
34impl OpenAiReasoningFormat {
35 pub fn for_provider_kind(kind: &str) -> Self {
36 if kind == "openai" {
37 Self::Official
38 } else {
39 Self::CompatibleThinking
40 }
41 }
42
43 pub fn as_str(self) -> &'static str {
44 match self {
45 Self::Official => "reasoning-effort",
46 Self::CompatibleThinking => "thinking-toggle",
47 }
48 }
49}
50
51impl std::fmt::Display for OpenAiReasoningFormat {
52 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
53 f.write_str(self.as_str())
54 }
55}
56
57impl std::str::FromStr for OpenAiReasoningFormat {
58 type Err = String;
59
60 fn from_str(value: &str) -> Result<Self, Self::Err> {
61 match value.trim().to_ascii_lowercase().as_str() {
62 "reasoning-effort" | "official" => Ok(Self::Official),
63 "thinking-toggle" | "compatible-thinking" => Ok(Self::CompatibleThinking),
64 _ => Err(format!(
65 "invalid reasoning format `{value}`; expected `reasoning-effort` or `thinking-toggle`"
66 )),
67 }
68 }
69}
70
71impl OpenAiProvider {
72 pub fn new(name: impl Into<String>, api_key: impl Into<String>) -> Self {
73 Self {
74 name: name.into(),
75 api_key: api_key.into(),
76 base_url: "https://api.openai.com/v1".into(),
77 client: reqwest::Client::new(),
78 max_tokens: None,
79 reasoning_format: OpenAiReasoningFormat::default(),
80 prompt_cache_key: true,
81 }
82 }
83
84 pub fn with_base_url(mut self, url: impl Into<String>) -> Self {
85 self.base_url = url.into();
86 self
87 }
88
89 pub fn with_max_tokens(mut self, n: u32) -> Self {
90 self.max_tokens = Some(n);
91 self
92 }
93
94 pub fn with_reasoning_format(mut self, format: OpenAiReasoningFormat) -> Self {
95 self.reasoning_format = format;
96 self
97 }
98
99 pub fn with_prompt_cache_key(mut self, enabled: bool) -> Self {
100 self.prompt_cache_key = enabled;
101 self
102 }
103
104 fn validate_reasoning(&self, selection: &ReasoningSelection) -> Result<(), RuntimeError> {
105 let profile = match self.reasoning_format {
106 OpenAiReasoningFormat::Official => ReasoningWireProfile::OpenAiOfficial,
107 OpenAiReasoningFormat::CompatibleThinking => ReasoningWireProfile::CompatibleThinking,
108 };
109 profile
110 .validate(selection, self.max_tokens)
111 .map_err(|error| RuntimeError::ToolFailed(format!("invalid request: {error}")))
112 }
113
114 fn build_body(
115 &self,
116 req: &LlmRequest,
117 stream: bool,
118 ) -> Result<ChatCompletionsRequest, RuntimeError> {
119 let mut wire_messages: Vec<ChatMessage> = Vec::new();
120 if let Some(sys) = req.system.as_ref().filter(|system| !system.is_empty()) {
121 wire_messages.push(ChatMessage {
122 role: "system",
123 content: Some(ChatContent::Text(sys.clone())),
124 reasoning_content: None,
125 tool_calls: None,
126 tool_call_id: None,
127 });
128 }
129 for m in &req.messages {
130 if let Some(message) = build_wire_message(m, &req.tools, self.reasoning_format)? {
131 wire_messages.push(message);
132 }
133 }
134 let tools: Vec<WireToolSpec> = req
135 .tools
136 .iter()
137 .map(|t| WireToolSpec {
138 kind: "function",
139 function: WireToolFunction {
140 name: crate::tool_naming::to_wire(&t.name),
141 description: t.description.clone(),
142 parameters: t.input_schema.clone(),
143 },
144 })
145 .collect();
146 let (reasoning_effort, thinking) = match self.reasoning_format {
147 OpenAiReasoningFormat::Official => (official_reasoning_effort(&req.reasoning), None),
148 OpenAiReasoningFormat::CompatibleThinking => {
149 (None, compatible_thinking(&req.reasoning))
150 }
151 };
152 Ok(ChatCompletionsRequest {
153 model: req.model.clone(),
154 stream,
155 max_tokens: (self.reasoning_format == OpenAiReasoningFormat::CompatibleThinking)
156 .then_some(self.max_tokens)
157 .flatten(),
158 max_completion_tokens: (self.reasoning_format == OpenAiReasoningFormat::Official)
159 .then_some(self.max_tokens)
160 .flatten(),
161 messages: wire_messages,
162 tools,
163 stream_options: if stream {
164 Some(StreamOptions {
165 include_usage: true,
166 })
167 } else {
168 None
169 },
170 reasoning_effort,
171 thinking,
172 prompt_cache_key: req.prompt_cache_key.clone(),
173 })
174 }
175
176 fn build_request(
177 &self,
178 req: &LlmRequest,
179 stream: bool,
180 ) -> Result<reqwest::RequestBuilder, RuntimeError> {
181 let body = self.build_body(req, stream)?;
182 Ok(self
183 .client
184 .post(format!("{}/chat/completions", self.base_url))
185 .bearer_auth(&self.api_key)
186 .json(&body))
187 }
188
189 #[doc(hidden)]
190 pub fn wire_body_bytes(&self, req: &LlmRequest, stream: bool) -> Vec<u8> {
191 serde_json::to_vec(
192 &self
193 .build_body(req, stream)
194 .expect("build OpenAI wire body"),
195 )
196 .expect("serialize wire body")
197 }
198}
199
200fn official_reasoning_effort(selection: &ReasoningSelection) -> Option<String> {
201 match selection {
202 ReasoningSelection::Disabled => Some(ReasoningEffort::None.to_string()),
203 ReasoningSelection::Effort { effort, .. } => Some(effort.to_string()),
204 ReasoningSelection::ProviderDefault
205 | ReasoningSelection::Auto { .. }
206 | ReasoningSelection::BudgetTokens { .. } => None,
207 }
208}
209
210fn compatible_thinking(selection: &ReasoningSelection) -> Option<ThinkingConfig> {
211 match selection {
212 ReasoningSelection::ProviderDefault => None,
213 ReasoningSelection::Disabled
214 | ReasoningSelection::Effort {
215 effort: ReasoningEffort::None,
216 ..
217 } => Some(ThinkingConfig { kind: "disabled" }),
218 ReasoningSelection::Auto { .. }
219 | ReasoningSelection::Effort { .. }
220 | ReasoningSelection::BudgetTokens { .. } => Some(ThinkingConfig { kind: "enabled" }),
221 }
222}
223
224fn build_wire_message(
225 m: &Message,
226 tools: &[crate::tool::ToolSpec],
227 reasoning_format: OpenAiReasoningFormat,
228) -> Result<Option<ChatMessage>, RuntimeError> {
229 let message = match m.role {
230 MessageRole::System => {
231 let text = m.text_concat();
232 if text.is_empty() {
233 return Ok(None);
234 }
235 ChatMessage {
236 role: "system",
237 content: Some(ChatContent::Text(text)),
238 reasoning_content: None,
239 tool_calls: None,
240 tool_call_id: None,
241 }
242 }
243 MessageRole::Tool => {
244 let (id, content) = extract_tool_result(m);
245 ChatMessage {
246 role: "tool",
247 content: Some(ChatContent::Text(content)),
248 reasoning_content: None,
249 tool_calls: None,
250 tool_call_id: Some(id),
251 }
252 }
253 MessageRole::Assistant => {
254 let (text_parts, tool_uses) = split_assistant_parts(&m.parts, tools);
255 let text = text_parts.join("");
256 let content = (!text.is_empty()).then_some(ChatContent::Text(text));
257 let reasoning = (reasoning_format == OpenAiReasoningFormat::CompatibleThinking)
258 .then(|| m.thinking_concat())
259 .filter(|thinking| !thinking.is_empty());
260 let tool_calls = if tool_uses.is_empty() {
261 None
262 } else {
263 Some(tool_uses)
264 };
265 if content.is_none() && reasoning.is_none() && tool_calls.is_none() {
266 return Ok(None);
267 }
268 ChatMessage {
269 role: "assistant",
270 content,
271 reasoning_content: reasoning,
272 tool_calls,
273 tool_call_id: None,
274 }
275 }
276 MessageRole::User => {
277 let parts = build_user_parts(&m.parts)?;
278 if parts.is_empty() {
279 return Ok(None);
280 }
281 let content = if parts.iter().all(|p| matches!(p, ChatPart::Text { .. })) {
282 let joined: String = parts
283 .iter()
284 .filter_map(|p| match p {
285 ChatPart::Text { text } => Some(text.as_str()),
286 _ => None,
287 })
288 .collect();
289 if joined.is_empty() {
290 return Ok(None);
291 }
292 Some(ChatContent::Text(joined))
293 } else {
294 Some(ChatContent::Parts(parts))
295 };
296 ChatMessage {
297 role: "user",
298 content,
299 reasoning_content: None,
300 tool_calls: None,
301 tool_call_id: None,
302 }
303 }
304 };
305 Ok(Some(message))
306}
307
308fn build_user_parts(parts: &[MessagePart]) -> Result<Vec<ChatPart>, RuntimeError> {
309 let mut out = Vec::with_capacity(parts.len());
310 for p in parts {
311 match p {
312 MessagePart::ContextRecord(record) => out.push(ChatPart::Text {
313 text: record.render_for_model(),
314 }),
315 MessagePart::CompactSummary { summary, .. } => out.push(ChatPart::Text {
316 text: summary.clone(),
317 }),
318 MessagePart::Text { text } if !text.is_empty() => {
319 out.push(ChatPart::Text { text: text.clone() })
320 }
321 MessagePart::Image { source } => {
322 let data = crate::attachment_store::image_base64(source)?;
323 let url = format!("data:{};base64,{}", source.media_type, data);
324 out.push(ChatPart::ImageUrl {
325 image_url: ImageUrl {
326 url,
327 detail: (!matches!(source.detail, crate::provider::ImageDetail::Auto))
328 .then(|| source.detail.as_str()),
329 },
330 });
331 }
332 _ => {}
333 }
334 }
335 Ok(out)
336}
337
338fn extract_tool_result(m: &Message) -> (String, String) {
339 for p in &m.parts {
340 if let MessagePart::ToolResult {
341 tool_use_id,
342 content,
343 ..
344 } = p
345 {
346 return (tool_use_id.clone(), content.clone());
347 }
348 }
349 (String::new(), m.text_concat())
350}
351
352fn split_assistant_parts(
353 parts: &[MessagePart],
354 tool_specs: &[crate::tool::ToolSpec],
355) -> (Vec<String>, Vec<WireToolCall>) {
356 let mut text = Vec::new();
357 let mut tools = Vec::new();
358 for p in parts {
359 match p {
360 MessagePart::ContextRecord(record) => text.push(record.render_for_model()),
361 MessagePart::Text { text: t } => text.push(t.clone()),
362 MessagePart::ToolUse {
363 id,
364 name,
365 input,
366 intent,
367 } => tools.push(WireToolCall {
368 id: id.clone(),
369 kind: "function",
370 function: WireFunctionCall {
371 name: crate::tool_naming::to_wire(name),
372 arguments: crate::message::encode_tool_call_input(
373 input,
374 intent.as_ref(),
375 name,
376 tool_specs,
377 )
378 .to_string(),
379 },
380 }),
381 _ => {}
382 }
383 }
384 (text, tools)
385}
386
387impl Provider for OpenAiProvider {
388 fn name(&self) -> &str {
389 &self.name
390 }
391
392 fn capabilities(&self) -> crate::provider::ProviderCapabilities {
393 crate::provider::ProviderCapabilities {
394 prompt_cache_key: self.prompt_cache_key,
395 context_prefix_profile: crate::context_plan::ContextPrefixProfile::OpenAiChat,
396 }
397 }
398
399 fn context_prefix(
400 &self,
401 req: &LlmRequest,
402 ) -> Result<crate::context_plan::ContextPrefixSnapshot, RuntimeError> {
403 let body = self.build_body(req, true)?;
404 let mut builder = crate::context_plan::ContextPrefixSnapshot::builder(
405 crate::context_plan::ContextPrefixProfile::OpenAiChat,
406 req,
407 );
408 for tool in &body.tools {
409 builder.push(crate::context_plan::ContextPrefixLane::Tools, tool)?;
410 }
411 for (index, message) in body.messages.iter().enumerate() {
412 let lane = if index == 0 && req.system.is_some() {
413 crate::context_plan::ContextPrefixLane::Stable
414 } else if req
415 .messages
416 .get(index.saturating_sub(usize::from(req.system.is_some())))
417 .is_some_and(Message::contains_context_record)
418 {
419 crate::context_plan::ContextPrefixLane::Records
420 } else {
421 crate::context_plan::ContextPrefixLane::Messages
422 };
423 builder.push(lane, message)?;
424 }
425 Ok(builder.finish())
426 }
427
428 fn call<'a>(&'a self, req: LlmRequest) -> BoxFut<'a, Result<AssistantMessage, RuntimeError>> {
429 if let Err(error) = self.validate_reasoning(&req.reasoning) {
430 return Box::pin(async move { Err(error) });
431 }
432 let request = match self.build_request(&req, false) {
433 Ok(request) => request,
434 Err(error) => return Box::pin(async move { Err(error) }),
435 };
436 let turn_id = next_turn_id_from_req(&req);
437 Box::pin(async move {
438 let resp = request.send().await.map_err(net_err)?;
439 let status = resp.status();
440 let body: ChatCompletionsResponse = if status.is_success() {
441 resp.json().await.map_err(net_err)?
442 } else {
443 let body_text = resp.text().await.unwrap_or_default();
444 if let Some(reason) = classify_attachment_error(status.as_u16(), &body_text) {
445 return Err(RuntimeError::AttachmentError { reason });
446 }
447 return Err(RuntimeError::ToolFailed(format!(
448 "openai http {status}: {body_text}"
449 )));
450 };
451 Ok(response_to_assistant(body, turn_id, &req.tools))
452 })
453 }
454
455 fn call_streaming(&self, req: LlmRequest) -> Observable<AssistantMessage> {
456 let preflight = self
457 .validate_reasoning(&req.reasoning)
458 .and_then(|()| self.build_request(&req, true));
459 let turn_id = next_turn_id_from_req(&req);
460 let streaming_tools = req.tools.clone();
461 let (tx, events) = broadcast::channel(DEFAULT_STREAM_BUFFER);
462 let cancel = CancellationToken::new();
463 let cancel_for_task = cancel.clone();
464 let output: BoxFut<'static, Result<AssistantMessage, RuntimeError>> = Box::pin(
465 async move {
466 let request = preflight?;
467 use eventsource_stream::Eventsource;
468 use futures::StreamExt;
469
470 let resp = tokio::select! {
471 biased;
472 _ = cancel_for_task.cancelled() => return Err(RuntimeError::Cancelled("openai cancelled before send".into())),
473 r = request.send() => r.map_err(net_err)?,
474 };
475 let status = resp.status();
476 if !status.is_success() {
477 let body = resp.text().await.unwrap_or_default();
478 if let Some(reason) = classify_attachment_error(status.as_u16(), &body) {
479 return Err(RuntimeError::AttachmentError { reason });
480 }
481 return Err(RuntimeError::ToolFailed(format!(
482 "openai http {status}: {body}"
483 )));
484 }
485
486 let mut stream = resp.bytes_stream().eventsource();
487 let mut acc_text = String::new();
488 let mut acc_thinking = String::new();
489 let mut cumulative = 0u64;
490 let mut final_usage: Option<OpenAiUsage> = None;
491 let mut resp_model: Option<String> = None;
492 let mut resp_id: Option<String> = None;
493 let mut partial_tool_calls: Vec<PartialToolCall> = Vec::new();
494 let mut stop_reason = StopReason::End;
495 while let Some(event) = tokio::select! {
496 biased;
497 _ = cancel_for_task.cancelled() => None,
498 next = stream.next() => next,
499 } {
500 let event = event.map_err(|e| RuntimeError::ToolFailed(format!("sse: {e}")))?;
501 if event.data == "[DONE]" {
502 break;
503 }
504 if event.data.is_empty() {
505 continue;
506 }
507 let parsed: serde_json::Value = match serde_json::from_str(&event.data) {
508 Ok(v) => v,
509 Err(_) => continue,
510 };
511 if let Some(content) = parsed
512 .pointer("/choices/0/delta/content")
513 .and_then(|v| v.as_str())
514 && !content.is_empty()
515 {
516 acc_text.push_str(content);
517 cumulative += estimate_tokens(content);
518 let _ = tx.send(NodeEvent::LlmChunk {
519 text: content.to_string(),
520 cumulative_tokens: cumulative,
521 });
522 }
523 if let Some(reasoning) = parsed
524 .pointer("/choices/0/delta/reasoning_content")
525 .and_then(|v| v.as_str())
526 && !reasoning.is_empty()
527 {
528 acc_thinking.push_str(reasoning);
529 let _ = tx.send(NodeEvent::ThinkingChunk {
530 text: reasoning.to_string(),
531 });
532 }
533 if let Some(m) = parsed.get("model").and_then(|v| v.as_str()) {
534 resp_model = Some(m.to_string());
535 }
536 if let Some(id) = parsed.get("id").and_then(|v| v.as_str()) {
537 resp_id = Some(id.to_string());
538 }
539 if let Some(tcs) = parsed
540 .pointer("/choices/0/delta/tool_calls")
541 .and_then(|v| v.as_array())
542 {
543 for tc in tcs {
544 let idx =
545 tc.get("index").and_then(|v| v.as_u64()).unwrap_or(0) as usize;
546 while partial_tool_calls.len() <= idx {
547 partial_tool_calls.push(PartialToolCall::default());
548 }
549 let slot = &mut partial_tool_calls[idx];
550 if let Some(id) = tc.get("id").and_then(|v| v.as_str()) {
551 slot.id = id.to_string();
552 }
553 if let Some(name) = tc
554 .pointer("/function/name")
555 .and_then(|v| v.as_str())
556 .filter(|name| !name.is_empty())
557 {
558 slot.name = name.to_string();
559 }
560 if let Some(args) =
561 tc.pointer("/function/arguments").and_then(|v| v.as_str())
562 {
563 slot.arguments.push_str(args);
564 let _ = tx.send(NodeEvent::ToolCallDraft {
565 index: idx,
566 call_id: slot.id.clone(),
567 name: crate::tool_naming::from_wire(
568 &slot.name,
569 &streaming_tools,
570 ),
571 arguments_delta: args.to_string(),
572 });
573 }
574 }
575 }
576 if let Some(reason) = parsed
577 .pointer("/choices/0/finish_reason")
578 .and_then(|v| v.as_str())
579 {
580 stop_reason = parse_stop_reason(reason);
581 }
582 if let Some(usage_obj) = parsed.get("usage") {
583 if !usage_obj.is_null() {
584 final_usage =
585 serde_json::from_value::<OpenAiUsage>(usage_obj.clone()).ok();
586 }
587 }
588 }
589 if cancel_for_task.is_cancelled() {
590 let _ = tx.send(NodeEvent::LlmDone {
591 total_tokens: cumulative,
592 });
593 return Err(RuntimeError::Cancelled(
594 "openai cancelled mid-stream".into(),
595 ));
596 }
597 let total = final_usage
598 .as_ref()
599 .and_then(|u| u.completion_tokens)
600 .unwrap_or(cumulative);
601 let _ = tx.send(NodeEvent::LlmDone {
602 total_tokens: total,
603 });
604
605 let mut parts: Vec<MessagePart> = Vec::new();
606 if !acc_thinking.is_empty() {
607 parts.push(MessagePart::Thinking {
608 thinking: acc_thinking,
609 signature: None,
610 });
611 }
612 if !acc_text.is_empty() {
613 parts.push(MessagePart::Text { text: acc_text });
614 }
615 for tc in partial_tool_calls {
616 if tc.id.is_empty() && tc.name.is_empty() && tc.arguments.is_empty() {
617 continue;
618 }
619 let input: serde_json::Value = if tc.arguments.is_empty() {
620 serde_json::Value::Object(Default::default())
621 } else {
622 serde_json::from_str(&tc.arguments).unwrap_or(serde_json::Value::Null)
623 };
624 let name = crate::tool_naming::from_wire(&tc.name, &streaming_tools);
625 let (input, intent) =
626 crate::message::decode_tool_call_input(input, &name, &streaming_tools);
627 parts.push(MessagePart::ToolUse {
628 id: tc.id,
629 name,
630 input,
631 intent,
632 });
633 }
634
635 let token_usage = if let Some(u) = &final_usage {
636 let cached = u
637 .prompt_tokens_details
638 .as_ref()
639 .and_then(|d| d.cached_tokens)
640 .unwrap_or(0);
641 let cache_write = u
642 .prompt_tokens_details
643 .as_ref()
644 .and_then(|d| d.cache_write_tokens)
645 .unwrap_or(0);
646 let cache_read = cached.max(u.prompt_cache_hit_tokens.unwrap_or(0));
647 let reasoning = u
648 .completion_tokens_details
649 .as_ref()
650 .and_then(|d| d.reasoning_tokens)
651 .unwrap_or(0);
652 TokenUsage {
653 input: crate::provider::regular_input_tokens(
654 u.prompt_tokens.unwrap_or(0),
655 cache_read,
656 cache_write,
657 ),
658 cached_input: cache_read,
659 output: u.completion_tokens.unwrap_or(0),
660 cache_write,
661 reasoning_tokens: reasoning,
662 }
663 } else {
664 TokenUsage {
665 output: total,
666 ..Default::default()
667 }
668 };
669
670 Ok(AssistantMessage {
671 message: Message {
672 role: MessageRole::Assistant,
673 parts,
674 turn_id,
675 origin: MessageOrigin::User,
676 },
677 stop_reason,
678 token_usage,
679 timing: CallTiming::default(),
680 model: resp_model.unwrap_or_default(),
681 response_id: resp_id,
682 })
683 },
684 );
685 Observable {
686 output,
687 events,
688 cancel,
689 }
690 }
691
692 fn discover_models(
693 &self,
694 ) -> crate::tool::BoxFut<'static, Vec<crate::provider::DiscoveredModel>> {
695 let discovery = self.try_discover_models();
696 Box::pin(async move {
697 discovery
698 .await
699 .unwrap_or_default()
700 .into_iter()
701 .map(crate::provider::DiscoveredModel::from)
702 .collect()
703 })
704 }
705
706 fn try_discover_models(
707 &self,
708 ) -> crate::tool::BoxFut<
709 'static,
710 Result<Vec<crate::provider::DiscoveredModelDetails>, crate::provider::ModelDiscoveryError>,
711 > {
712 let base_url = self.base_url.clone();
713 let api_key = self.api_key.clone();
714 Box::pin(async move {
715 #[derive(serde::Deserialize)]
716 struct ModelsResponse {
717 data: Vec<ModelEntry>,
718 }
719 #[derive(serde::Deserialize)]
720 struct ModelEntry {
721 id: String,
722 }
723
724 let client = reqwest::Client::builder()
725 .timeout(std::time::Duration::from_secs(10))
726 .build()
727 .map_err(|error| {
728 crate::provider::ModelDiscoveryError::Transport(error.to_string())
729 })?;
730 let resp = client
731 .get(format!("{base_url}/models"))
732 .bearer_auth(&api_key)
733 .send()
734 .await
735 .map_err(|error| {
736 crate::provider::ModelDiscoveryError::Transport(error.to_string())
737 })?;
738 let status = resp.status();
739 let bytes = resp.bytes().await.map_err(|error| {
740 crate::provider::ModelDiscoveryError::Transport(error.to_string())
741 })?;
742 if !status.is_success() {
743 return Err(crate::provider::ModelDiscoveryError::Http {
744 status: status.as_u16(),
745 body: String::from_utf8_lossy(&bytes).chars().take(512).collect(),
746 });
747 }
748 let body: ModelsResponse = serde_json::from_slice(&bytes).map_err(|error| {
749 crate::provider::ModelDiscoveryError::InvalidResponse(error.to_string())
750 })?;
751 Ok(body
752 .data
753 .into_iter()
754 .filter(|m| !m.id.starts_with("ft:"))
755 .map(|m| {
756 let (budget, thinking) =
757 crate::model_registry::lookup_known_model(&m.id).unwrap_or((32_768, false));
758 crate::provider::DiscoveredModelDetails {
759 slug: m.id,
760 context_budget: Some(budget),
761 capability_knowledge: crate::provider::CapabilityKnowledge::Legacy {
762 thinking,
763 },
764 }
765 })
766 .collect())
767 })
768 }
769
770 fn test_connection(&self) -> BoxFut<'_, Result<String, String>> {
771 let base_url = self.base_url.clone();
772 let api_key = self.api_key.clone();
773 let name = self.name.clone();
774 Box::pin(async move {
775 let client = reqwest::Client::builder()
776 .timeout(std::time::Duration::from_secs(15))
777 .build()
778 .map_err(|e| e.to_string())?;
779 let resp = client
780 .get(format!("{}/models", base_url.trim_end_matches('/')))
781 .bearer_auth(&api_key)
782 .send()
783 .await
784 .map_err(|e| format!("connection failed — {e}"))?;
785 let status = resp.status();
786 if status.is_success() {
787 Ok(format!("\"{name}\" responded OK"))
788 } else {
789 let body = resp.text().await.unwrap_or_default();
790 Err(format!(
791 "returned {status} — {}",
792 crate::provider::bounded_utf8_prefix(&body, 200)
793 ))
794 }
795 })
796 }
797}
798
799#[derive(Default)]
800struct PartialToolCall {
801 id: String,
802 name: String,
803 arguments: String,
804}
805
806fn response_to_assistant(
807 body: ChatCompletionsResponse,
808 turn_id: crate::event::TurnId,
809 tools: &[crate::tool::ToolSpec],
810) -> AssistantMessage {
811 let mut parts: Vec<MessagePart> = Vec::new();
812 let mut stop_reason = StopReason::End;
813 if let Some(choice) = body.choices.into_iter().next() {
814 if let Some(msg) = choice.message {
815 if let Some(reasoning) = msg.reasoning_content {
816 parts.push(MessagePart::Thinking {
817 thinking: reasoning,
818 signature: None,
819 });
820 }
821 if let Some(content) = msg.content {
822 parts.push(MessagePart::Text { text: content });
823 }
824 if let Some(tool_calls) = msg.tool_calls {
825 for tc in tool_calls {
826 let input: serde_json::Value = if tc.function.arguments.is_empty() {
827 serde_json::Value::Object(Default::default())
828 } else {
829 serde_json::from_str(&tc.function.arguments)
830 .unwrap_or(serde_json::Value::Null)
831 };
832 let name = crate::tool_naming::from_wire(&tc.function.name, tools);
833 let (input, intent) =
834 crate::message::decode_tool_call_input(input, &name, tools);
835 parts.push(MessagePart::ToolUse {
836 id: tc.id,
837 name,
838 input,
839 intent,
840 });
841 }
842 }
843 }
844 if let Some(reason) = choice.finish_reason {
845 stop_reason = parse_stop_reason(&reason);
846 }
847 }
848 let usage = body.usage.map(|u| {
849 let cached = u
850 .prompt_tokens_details
851 .as_ref()
852 .and_then(|d| d.cached_tokens)
853 .unwrap_or(0);
854 let cache_write = u
855 .prompt_tokens_details
856 .as_ref()
857 .and_then(|d| d.cache_write_tokens)
858 .unwrap_or(0);
859 let cache_read = cached.max(u.prompt_cache_hit_tokens.unwrap_or(0));
860 let reasoning = u
861 .completion_tokens_details
862 .as_ref()
863 .and_then(|d| d.reasoning_tokens)
864 .unwrap_or(0);
865 TokenUsage {
866 input: crate::provider::regular_input_tokens(
867 u.prompt_tokens.unwrap_or(0),
868 cache_read,
869 cache_write,
870 ),
871 cached_input: cache_read,
872 output: u.completion_tokens.unwrap_or(0),
873 cache_write,
874 reasoning_tokens: reasoning,
875 }
876 });
877 AssistantMessage {
878 message: Message {
879 role: MessageRole::Assistant,
880 parts,
881 turn_id,
882 origin: MessageOrigin::User,
883 },
884 stop_reason,
885 token_usage: usage.unwrap_or_default(),
886 timing: CallTiming::default(),
887 model: body.model.unwrap_or_default(),
888 response_id: body.id,
889 }
890}
891
892fn parse_stop_reason(s: &str) -> StopReason {
893 match s {
894 "tool_calls" | "function_call" => StopReason::ToolUse,
895 "length" => StopReason::Length,
896 _ => StopReason::End,
897 }
898}
899
900fn next_turn_id_from_req(req: &LlmRequest) -> crate::event::TurnId {
901 req.messages
902 .first()
903 .map(|m| m.turn_id.clone())
904 .unwrap_or_else(crate::event::TurnId::now)
905}
906
907fn net_err(e: reqwest::Error) -> RuntimeError {
908 RuntimeError::ToolFailed(format!("openai net: {e}"))
909}
910
911#[derive(Serialize)]
912struct ChatCompletionsRequest {
913 model: String,
914 stream: bool,
915 #[serde(skip_serializing_if = "Option::is_none")]
916 max_tokens: Option<u32>,
917 #[serde(skip_serializing_if = "Option::is_none")]
918 max_completion_tokens: Option<u32>,
919 messages: Vec<ChatMessage>,
920 #[serde(skip_serializing_if = "Vec::is_empty")]
921 tools: Vec<WireToolSpec>,
922 #[serde(skip_serializing_if = "Option::is_none")]
923 stream_options: Option<StreamOptions>,
924 #[serde(skip_serializing_if = "Option::is_none")]
925 reasoning_effort: Option<String>,
926 #[serde(skip_serializing_if = "Option::is_none")]
927 thinking: Option<ThinkingConfig>,
928 #[serde(skip_serializing_if = "Option::is_none")]
929 prompt_cache_key: Option<String>,
930}
931
932#[derive(Serialize)]
933struct ThinkingConfig {
934 #[serde(rename = "type")]
935 kind: &'static str,
936}
937
938#[derive(Serialize)]
939struct StreamOptions {
940 include_usage: bool,
941}
942
943#[derive(Serialize)]
944struct WireToolSpec {
945 #[serde(rename = "type")]
946 kind: &'static str,
947 function: WireToolFunction,
948}
949
950#[derive(Serialize)]
951struct WireToolFunction {
952 name: String,
953 #[serde(skip_serializing_if = "Option::is_none")]
954 description: Option<String>,
955 parameters: serde_json::Value,
956}
957
958#[derive(Serialize)]
959struct ChatMessage {
960 role: &'static str,
961 #[serde(skip_serializing_if = "Option::is_none")]
962 content: Option<ChatContent>,
963 #[serde(skip_serializing_if = "Option::is_none")]
964 reasoning_content: Option<String>,
965 #[serde(skip_serializing_if = "Option::is_none")]
966 tool_calls: Option<Vec<WireToolCall>>,
967 #[serde(skip_serializing_if = "Option::is_none")]
968 tool_call_id: Option<String>,
969}
970
971#[derive(Serialize)]
972#[serde(untagged)]
973enum ChatContent {
974 Text(String),
975 Parts(Vec<ChatPart>),
976}
977
978#[derive(Serialize)]
979#[serde(tag = "type", rename_all = "snake_case")]
980enum ChatPart {
981 Text { text: String },
982 ImageUrl { image_url: ImageUrl },
983}
984
985#[derive(Serialize)]
986struct ImageUrl {
987 url: String,
988 #[serde(skip_serializing_if = "Option::is_none")]
989 detail: Option<&'static str>,
990}
991
992#[derive(Serialize)]
993struct WireToolCall {
994 id: String,
995 #[serde(rename = "type")]
996 kind: &'static str,
997 function: WireFunctionCall,
998}
999
1000#[derive(Serialize)]
1001struct WireFunctionCall {
1002 name: String,
1003 arguments: String,
1004}
1005
1006#[derive(Deserialize)]
1007struct ChatCompletionsResponse {
1008 choices: Vec<ChatChoice>,
1009 #[serde(default)]
1010 usage: Option<OpenAiUsage>,
1011 #[serde(default)]
1012 model: Option<String>,
1013 #[serde(default)]
1014 id: Option<String>,
1015}
1016
1017#[derive(Deserialize, Default)]
1018struct OpenAiUsage {
1019 #[serde(default)]
1020 prompt_tokens: Option<u64>,
1021 #[serde(default)]
1022 completion_tokens: Option<u64>,
1023 #[serde(default)]
1024 prompt_tokens_details: Option<PromptTokensDetails>,
1025 #[serde(default)]
1026 completion_tokens_details: Option<CompletionTokensDetails>,
1027 #[serde(default)]
1028 prompt_cache_hit_tokens: Option<u64>,
1029 #[serde(default)]
1030 #[allow(dead_code)]
1031 prompt_cache_miss_tokens: Option<u64>,
1032}
1033
1034#[derive(Deserialize, Default)]
1035struct PromptTokensDetails {
1036 #[serde(default)]
1037 cached_tokens: Option<u64>,
1038 #[serde(default)]
1039 cache_write_tokens: Option<u64>,
1040}
1041
1042#[derive(Deserialize, Default)]
1043struct CompletionTokensDetails {
1044 #[serde(default)]
1045 reasoning_tokens: Option<u64>,
1046}
1047
1048#[derive(Deserialize)]
1049struct ChatChoice {
1050 #[serde(default)]
1051 message: Option<ChatChoiceMessage>,
1052 #[serde(default)]
1053 finish_reason: Option<String>,
1054}
1055
1056#[derive(Deserialize)]
1057struct ChatChoiceMessage {
1058 #[serde(default)]
1059 content: Option<String>,
1060 #[serde(default)]
1061 reasoning_content: Option<String>,
1062 #[serde(default)]
1063 tool_calls: Option<Vec<RespToolCall>>,
1064}
1065
1066#[derive(Deserialize)]
1067struct RespToolCall {
1068 id: String,
1069 function: RespFunctionCall,
1070}
1071
1072#[derive(Deserialize)]
1073struct RespFunctionCall {
1074 name: String,
1075 #[serde(default)]
1076 arguments: String,
1077}
1078
1079#[cfg(test)]
1080mod tests {
1081 use super::*;
1082
1083 struct IntentTool;
1084
1085 impl crate::tool::Tool for IntentTool {
1086 fn name(&self) -> &str {
1087 "probe"
1088 }
1089
1090 fn tier(&self) -> crate::tool::Tier {
1091 crate::tool::Tier::Zero
1092 }
1093
1094 fn call<'a>(
1095 &'a self,
1096 _args: crate::tool::ToolArgs,
1097 _ctx: &'a crate::tool::ToolCtx,
1098 ) -> crate::tool::BoxFut<'a, crate::tool::ToolResult> {
1099 Box::pin(async { Ok(crate::Value::Unit) })
1100 }
1101 }
1102
1103 #[test]
1104 fn tool_call_intent_round_trips_through_chat_arguments() {
1105 let tools = vec![crate::tool::tool_spec(&IntentTool)];
1106 let intent = crate::message::ToolCallIntent::new("Inspect provider state");
1107 let (_, calls) = split_assistant_parts(
1108 &[MessagePart::ToolUse {
1109 id: "call-1".into(),
1110 name: "probe".into(),
1111 input: serde_json::json!({"value": 1}),
1112 intent: intent.clone(),
1113 }],
1114 &tools,
1115 );
1116 let arguments: serde_json::Value =
1117 serde_json::from_str(&calls[0].function.arguments).unwrap();
1118 assert_eq!(arguments["_atman_intent"], "Inspect provider state");
1119
1120 let assistant = response_to_assistant(
1121 ChatCompletionsResponse {
1122 choices: vec![ChatChoice {
1123 message: Some(ChatChoiceMessage {
1124 content: None,
1125 reasoning_content: None,
1126 tool_calls: Some(vec![RespToolCall {
1127 id: "call-1".into(),
1128 function: RespFunctionCall {
1129 name: "probe".into(),
1130 arguments: arguments.to_string(),
1131 },
1132 }]),
1133 }),
1134 finish_reason: Some("tool_calls".into()),
1135 }],
1136 usage: None,
1137 model: None,
1138 id: None,
1139 },
1140 crate::event::TurnId::now(),
1141 &tools,
1142 );
1143 assert!(matches!(
1144 assistant.message.parts.as_slice(),
1145 [MessagePart::ToolUse { input, intent: Some(intent), .. }]
1146 if input == &serde_json::json!({"value": 1})
1147 && intent.as_str() == "Inspect provider state"
1148 ));
1149 }
1150
1151 #[test]
1152 fn final_answer_wire_schema_requires_non_empty_intent() {
1153 let provider = OpenAiProvider::new("openai", "test-key");
1154 let request = LlmRequest {
1155 model: "gpt-test".into(),
1156 messages: Vec::new(),
1157 system: None,
1158 input: crate::Value::Unit,
1159 schema: None,
1160 cache_prompt: false,
1161 prompt_cache_key: None,
1162 tools: vec![crate::tool::tool_spec(
1163 &crate::tools::final_answer::FinalAnswer,
1164 )],
1165 reasoning: ReasoningSelection::ProviderDefault,
1166 stall_timeout_secs: 0,
1167 };
1168
1169 let body = serde_json::to_value(provider.build_body(&request, false).unwrap()).unwrap();
1170 let parameters = &body["tools"][0]["function"]["parameters"];
1171 assert_eq!(
1172 parameters["required"],
1173 serde_json::json!(["message", "_atman_intent"])
1174 );
1175 assert_eq!(
1176 parameters["properties"]["_atman_intent"],
1177 serde_json::json!({
1178 "type": "string",
1179 "minLength": 1,
1180 "maxLength": 120,
1181 "pattern": "\\S"
1182 })
1183 );
1184 }
1185
1186 #[test]
1187 fn non_streaming_response_preserves_reasoning_content() {
1188 let assistant = response_to_assistant(
1189 ChatCompletionsResponse {
1190 choices: vec![ChatChoice {
1191 message: Some(ChatChoiceMessage {
1192 content: None,
1193 reasoning_content: Some("completed in reasoning".into()),
1194 tool_calls: None,
1195 }),
1196 finish_reason: Some("stop".into()),
1197 }],
1198 usage: None,
1199 model: None,
1200 id: None,
1201 },
1202 crate::event::TurnId::now(),
1203 &[],
1204 );
1205
1206 assert!(matches!(
1207 assistant.message.parts.as_slice(),
1208 [MessagePart::Thinking { thinking, signature: None }]
1209 if thinking == "completed in reasoning"
1210 ));
1211 }
1212
1213 #[test]
1214 fn context_prefix_uses_chat_projection_and_preserves_appended_messages() {
1215 let provider = OpenAiProvider::new("openai", "test-key");
1216 let mut request = LlmRequest {
1217 model: "gpt-test".into(),
1218 messages: vec![Message::user_text(crate::event::TurnId::now(), "first")],
1219 system: Some("stable".into()),
1220 input: crate::Value::Unit,
1221 schema: None,
1222 cache_prompt: true,
1223 prompt_cache_key: None,
1224 tools: Vec::new(),
1225 reasoning: ReasoningSelection::ProviderDefault,
1226 stall_timeout_secs: 0,
1227 };
1228 let first = provider.context_prefix(&request).unwrap();
1229 let first_bytes = first.initial_observation().wire_prefix_bytes;
1230 request.messages.push(Message::assistant_text(
1231 crate::event::TurnId::now(),
1232 "second",
1233 ));
1234 let second = provider.context_prefix(&request).unwrap();
1235 let observation = second.compare("openai", "openai", "model", "model", &first);
1236
1237 assert_eq!(
1238 observation.profile,
1239 crate::context_plan::ContextPrefixProfile::OpenAiChat
1240 );
1241 assert_eq!(observation.reset_reason, None);
1242 assert_eq!(observation.common_prefix_bytes, first_bytes);
1243 }
1244
1245 #[test]
1246 fn official_chat_request_serializes_prompt_cache_key() {
1247 let provider = OpenAiProvider::new("openai", "test-key");
1248 let request = LlmRequest {
1249 model: "gpt-test".into(),
1250 messages: vec![Message::user_text(crate::event::TurnId::now(), "first")],
1251 system: Some("stable".into()),
1252 input: crate::Value::Unit,
1253 schema: None,
1254 cache_prompt: true,
1255 prompt_cache_key: Some("atman-route".into()),
1256 tools: Vec::new(),
1257 reasoning: ReasoningSelection::ProviderDefault,
1258 stall_timeout_secs: 0,
1259 };
1260
1261 let body = serde_json::to_value(provider.build_body(&request, true).unwrap()).unwrap();
1262 assert_eq!(body["prompt_cache_key"], "atman-route");
1263 }
1264
1265 #[test]
1266 fn internal_context_record_projects_as_mid_conversation_system_message() {
1267 let provider = OpenAiProvider::new("openai", "test-key");
1268 let mut request = LlmRequest {
1269 model: "gpt-test".into(),
1270 messages: vec![Message::user_text(crate::event::TurnId::now(), "before")],
1271 system: Some("stable".into()),
1272 input: crate::Value::Unit,
1273 schema: None,
1274 cache_prompt: true,
1275 prompt_cache_key: None,
1276 tools: Vec::new(),
1277 reasoning: ReasoningSelection::ProviderDefault,
1278 stall_timeout_secs: 0,
1279 };
1280 let before = provider.context_prefix(&request).unwrap();
1281 let before_bytes = before.initial_observation().wire_prefix_bytes;
1282 request.messages.push(Message::context_record(
1283 crate::event::TurnId::now(),
1284 crate::context_plan::ContextRecord::new(
1285 "session.goal",
1286 1,
1287 crate::context_plan::ContextRecordAuthority::User,
1288 crate::context_plan::ContextRecordRetention::Latest,
1289 crate::context_plan::ContextRecordBody::text("finish the task"),
1290 ),
1291 ));
1292
1293 let body = serde_json::to_value(provider.build_body(&request, true).unwrap()).unwrap();
1294 assert_eq!(body["messages"][0]["role"], "system");
1295 assert_eq!(body["messages"][2]["role"], "system");
1296 assert!(
1297 body["messages"][2]["content"]
1298 .as_str()
1299 .is_some_and(|content| content.contains("finish the task"))
1300 );
1301 let after = provider.context_prefix(&request).unwrap();
1302 let observation = after.compare("openai", "openai", "model", "model", &before);
1303 assert_eq!(observation.reset_reason, None);
1304 assert_eq!(observation.common_prefix_bytes, before_bytes);
1305 }
1306}