Skip to repository content1265 lines · 43.0 KB · rust
tenant.openagents/omega
No repository description is available.
OpenAgents Git authority 2026-07-28T04:40:05.512Z Public web read
NIP-34 coordinate
30617:7649603503856e5148d571eac2766b288a8ff1e9e35d380337a1d2b0015b4f92:omegaMaintainersHidden in public view
References2 branches · 1 tag
Read-only clone
git clone https://openagents.com/git/tenant.openagents/omega.gitBrowse files
responses.rs
1use anyhow::{Context as _, Result, anyhow};
2use futures::{AsyncBufReadExt, AsyncReadExt, StreamExt, io::BufReader, stream::BoxStream};
3use http_client::{
4 AsyncBody, CustomHeaders, HttpClient, Method, Request as HttpRequest, RequestBuilderExt,
5};
6use serde::{Deserialize, Serialize, ser::SerializeSeq as _};
7use serde_json::Value;
8use std::sync::Arc;
9
10use crate::{ReasoningEffort, RequestError, Role, ServiceTier, ToolChoice};
11use language_model_core::{
12 CompactedContext, LanguageModelProviderId, ProviderCompactionState, SharedString,
13};
14
15pub const COMPACTION_STATE_FORMAT: &str = "openai.responses.input-items.v1";
16
17#[derive(Serialize, Debug)]
18pub struct Request {
19 pub model: String,
20 #[serde(skip_serializing_if = "Option::is_none")]
21 pub instructions: Option<String>,
22 #[serde(skip_serializing_if = "ResponseInput::is_empty")]
23 pub input: ResponseInput,
24 #[serde(skip_serializing_if = "Vec::is_empty")]
25 pub include: Vec<ResponseIncludable>,
26 #[serde(default)]
27 pub stream: bool,
28 #[serde(skip_serializing_if = "Option::is_none")]
29 pub temperature: Option<f32>,
30 #[serde(skip_serializing_if = "Option::is_none")]
31 pub top_p: Option<f32>,
32 #[serde(skip_serializing_if = "Option::is_none")]
33 pub max_output_tokens: Option<u64>,
34 #[serde(skip_serializing_if = "Option::is_none")]
35 pub parallel_tool_calls: Option<bool>,
36 #[serde(skip_serializing_if = "Option::is_none")]
37 pub tool_choice: Option<ToolChoice>,
38 #[serde(skip_serializing_if = "Vec::is_empty")]
39 pub tools: Vec<ToolDefinition>,
40 #[serde(skip_serializing_if = "Option::is_none")]
41 pub prompt_cache_key: Option<String>,
42 #[serde(skip_serializing_if = "Option::is_none")]
43 pub reasoning: Option<ReasoningConfig>,
44 #[serde(skip_serializing_if = "Option::is_none")]
45 pub store: Option<bool>,
46 #[serde(skip_serializing_if = "Option::is_none")]
47 pub service_tier: Option<ServiceTier>,
48 #[serde(default, skip_serializing_if = "Option::is_none")]
49 pub context_management: Option<Vec<ContextManagement>>,
50}
51
52impl Request {
53 pub fn into_compact_request(self) -> CompactRequest {
54 CompactRequest {
55 model: self.model,
56 instructions: self.instructions,
57 input: self.input,
58 prompt_cache_key: self.prompt_cache_key,
59 service_tier: self.service_tier,
60 }
61 }
62}
63
64#[derive(Serialize, Debug)]
65pub struct CompactRequest {
66 pub model: String,
67 #[serde(skip_serializing_if = "Option::is_none")]
68 pub instructions: Option<String>,
69 pub input: ResponseInput,
70 #[serde(skip_serializing_if = "Option::is_none")]
71 pub prompt_cache_key: Option<String>,
72 #[serde(skip_serializing_if = "Option::is_none")]
73 pub service_tier: Option<ServiceTier>,
74}
75
76#[derive(Deserialize, Debug)]
77pub struct CompactedResponse {
78 pub id: String,
79 pub created_at: u64,
80 pub object: String,
81 pub output: Vec<Value>,
82 pub usage: ResponseUsage,
83}
84
85impl CompactedResponse {
86 pub fn into_compacted_context(
87 self,
88 owner: LanguageModelProviderId,
89 ) -> Result<CompactedContext> {
90 Ok(CompactedContext::ProviderState(
91 provider_compaction_state_from_items(owner, self.output)?,
92 ))
93 }
94}
95
96/// Packages a canonical replacement window into opaque provider state owned by
97/// `owner`.
98///
99/// Several backends speak the OpenAI Responses protocol (OpenAI itself, Zed
100/// Cloud's OpenAI models, OpenAI-compatible endpoints, and others), but their
101/// encrypted compaction items are not interchangeable: only the backend that
102/// produced an item can decrypt it. The owner recorded here is what
103/// [`provider_compaction_items`] later compares against, so it must identify
104/// the backend whose infrastructure produced the items, not merely the wire
105/// protocol.
106pub fn provider_compaction_state_from_items(
107 owner: LanguageModelProviderId,
108 items: Vec<Value>,
109) -> Result<ProviderCompactionState> {
110 validate_compaction_items(&items)?;
111 Ok(ProviderCompactionState::new(
112 owner,
113 SharedString::new_static(COMPACTION_STATE_FORMAT),
114 serde_json::to_string(&items)?,
115 ))
116}
117
118/// Recovers the canonical replacement window from `state` if it is owned by
119/// `owner`, or `None` when the state belongs to a different backend and the
120/// caller should fall back to replaying the full transcript.
121pub fn provider_compaction_items(
122 state: &ProviderCompactionState,
123 owner: &LanguageModelProviderId,
124) -> Result<Option<Vec<Value>>> {
125 if state.provider_id() != owner {
126 return Ok(None);
127 }
128 if state.format() != COMPACTION_STATE_FORMAT {
129 return Err(anyhow!(
130 "unsupported OpenAI compaction state format: {}",
131 state.format()
132 ));
133 }
134
135 let items = serde_json::from_str::<Vec<Value>>(state.payload())
136 .context("malformed OpenAI compaction state payload")?;
137 validate_compaction_items(&items)?;
138 Ok(Some(items))
139}
140
141fn validate_compaction_items(items: &[Value]) -> Result<()> {
142 if items.is_empty() {
143 return Err(anyhow!("OpenAI returned an empty compaction output"));
144 }
145 if !items.iter().any(|item| {
146 item.get("type")
147 .and_then(Value::as_str)
148 .is_some_and(|item_type| item_type == "compaction")
149 }) {
150 return Err(anyhow!(
151 "OpenAI compaction output did not contain a compaction item"
152 ));
153 }
154 Ok(())
155}
156
157#[derive(Debug, Default)]
158pub struct ResponseInput {
159 provider_items: Vec<Value>,
160 generated_items: Vec<ResponseInputItem>,
161}
162
163impl ResponseInput {
164 pub fn new(provider_items: Vec<Value>, generated_items: Vec<ResponseInputItem>) -> Self {
165 Self {
166 provider_items,
167 generated_items,
168 }
169 }
170
171 pub fn is_empty(&self) -> bool {
172 self.provider_items.is_empty() && self.generated_items.is_empty()
173 }
174
175 /// Filters only the items this crate generated from the request.
176 ///
177 /// Provider items are a canonical replacement window that must be replayed
178 /// verbatim, so they are exempt from filtering. Callers that rewrite the
179 /// input to satisfy backend-specific requirements (and therefore can't
180 /// tolerate arbitrary items inside a replayed window) should not accept
181 /// provider-native compaction state in the first place.
182 pub fn retain(&mut self, predicate: impl FnMut(&ResponseInputItem) -> bool) {
183 self.generated_items.retain(predicate);
184 }
185}
186
187impl Serialize for ResponseInput {
188 fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
189 where
190 S: serde::Serializer,
191 {
192 let mut sequence = serializer
193 .serialize_seq(Some(self.provider_items.len() + self.generated_items.len()))?;
194 for item in &self.provider_items {
195 sequence.serialize_element(item)?;
196 }
197 for item in &self.generated_items {
198 sequence.serialize_element(item)?;
199 }
200 sequence.end()
201 }
202}
203
204/// Server-side context management configuration.
205///
206/// <https://developers.openai.com/api/docs/guides/compaction>
207#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
208#[serde(tag = "type", rename_all = "snake_case")]
209pub enum ContextManagement {
210 Compaction { compact_threshold: u64 },
211}
212
213#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
214#[serde(rename_all = "snake_case")]
215pub enum ResponseIncludable {
216 #[serde(rename = "reasoning.encrypted_content")]
217 ReasoningEncryptedContent,
218}
219
220#[derive(Debug, Serialize, Deserialize)]
221#[serde(tag = "type", rename_all = "snake_case")]
222pub enum ResponseInputItem {
223 Message(ResponseMessageItem),
224 FunctionCall(ResponseFunctionCallItem),
225 FunctionCallOutput(ResponseFunctionCallOutputItem),
226 CustomToolCall(ResponseCustomToolCallItem),
227 CustomToolCallOutput(ResponseCustomToolCallOutputItem),
228 Reasoning(ResponseReasoningInputItem),
229 Compaction(ResponseCompactionItem),
230}
231
232#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
233pub struct ResponseCompactionItem {
234 #[serde(default, skip_serializing_if = "Option::is_none")]
235 pub id: Option<Arc<str>>,
236 pub encrypted_content: Arc<str>,
237}
238
239#[derive(Debug, Serialize, Deserialize)]
240pub struct ResponseMessageItem {
241 pub role: Role,
242 pub content: Vec<ResponseInputContent>,
243 #[serde(default, skip_serializing_if = "Option::is_none")]
244 pub phase: Option<String>,
245}
246
247#[derive(Debug, Serialize, Deserialize)]
248pub struct ResponseFunctionCallItem {
249 pub call_id: String,
250 pub name: String,
251 pub arguments: String,
252}
253
254#[derive(Debug, Serialize, Deserialize)]
255pub struct ResponseFunctionCallOutputItem {
256 pub call_id: String,
257 pub output: ResponseFunctionCallOutputContent,
258}
259
260#[derive(Debug, Serialize, Deserialize)]
261pub struct ResponseCustomToolCallItem {
262 #[serde(default, skip_serializing_if = "Option::is_none")]
263 pub id: Option<String>,
264 pub call_id: String,
265 pub name: String,
266 pub input: String,
267}
268
269#[derive(Debug, Serialize, Deserialize)]
270pub struct ResponseCustomToolCallOutputItem {
271 pub call_id: String,
272 pub output: ResponseFunctionCallOutputContent,
273}
274
275#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
276pub struct ResponseReasoningInputItem {
277 #[serde(default, skip_serializing_if = "Option::is_none")]
278 pub id: Option<String>,
279 #[serde(default)]
280 pub summary: Vec<ResponseReasoningSummaryPart>,
281 #[serde(default, skip_serializing_if = "Vec::is_empty")]
282 pub content: Vec<Value>,
283 #[serde(default, skip_serializing_if = "Option::is_none")]
284 pub encrypted_content: Option<String>,
285 #[serde(default, skip_serializing_if = "Option::is_none")]
286 pub status: Option<String>,
287}
288
289#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
290#[serde(tag = "type", rename_all = "snake_case")]
291pub enum ResponseReasoningSummaryPart {
292 SummaryText { text: String },
293}
294
295#[derive(Debug, Serialize, Deserialize)]
296#[serde(untagged)]
297pub enum ResponseFunctionCallOutputContent {
298 List(Vec<ResponseInputContent>),
299 Text(String),
300}
301
302#[derive(Debug, Clone, Serialize, Deserialize)]
303#[serde(tag = "type")]
304pub enum ResponseInputContent {
305 #[serde(rename = "input_text")]
306 Text { text: String },
307 #[serde(rename = "input_image")]
308 Image { image_url: String },
309 #[serde(rename = "output_text")]
310 OutputText {
311 text: String,
312 #[serde(default)]
313 annotations: Vec<serde_json::Value>,
314 },
315 #[serde(rename = "refusal")]
316 Refusal { refusal: String },
317}
318
319#[derive(Serialize, Debug)]
320pub struct ReasoningConfig {
321 pub effort: ReasoningEffort,
322 #[serde(skip_serializing_if = "Option::is_none")]
323 pub summary: Option<ReasoningSummaryMode>,
324}
325
326#[derive(Serialize, Debug, Clone, Copy, PartialEq, Eq)]
327#[serde(rename_all = "lowercase")]
328pub enum ReasoningSummaryMode {
329 Auto,
330 Concise,
331 Detailed,
332}
333
334#[derive(Serialize, Deserialize, Debug)]
335#[serde(tag = "type", rename_all = "snake_case")]
336pub enum ToolDefinition {
337 Function {
338 name: String,
339 #[serde(skip_serializing_if = "Option::is_none")]
340 description: Option<String>,
341 #[serde(skip_serializing_if = "Option::is_none")]
342 parameters: Option<Value>,
343 #[serde(skip_serializing_if = "Option::is_none")]
344 strict: Option<bool>,
345 },
346 Custom {
347 name: String,
348 #[serde(skip_serializing_if = "Option::is_none")]
349 description: Option<String>,
350 #[serde(skip_serializing_if = "Option::is_none")]
351 format: Option<CustomToolFormat>,
352 },
353}
354
355#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
356#[serde(tag = "type", rename_all = "snake_case")]
357pub enum CustomToolFormat {
358 Text,
359 Grammar {
360 syntax: CustomToolGrammarSyntax,
361 definition: String,
362 },
363}
364
365#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)]
366#[serde(rename_all = "lowercase")]
367pub enum CustomToolGrammarSyntax {
368 Lark,
369 Regex,
370}
371
372#[derive(Serialize, Deserialize, Debug, Clone)]
373pub struct ResponseError {
374 #[serde(default)]
375 pub code: Option<String>,
376 pub message: String,
377 #[serde(default)]
378 pub param: Option<Value>,
379}
380
381/// Payload of the top-level `error` SSE event from the Responses API.
382///
383/// OpenAI's spec documents the error fields as being at the top level of the
384/// event, but in practice the API often nests them under an `error` object.
385#[derive(Deserialize, Debug, Clone, Default)]
386pub struct GenericStreamErrorPayload {
387 #[serde(flatten)]
388 top_level: PartialResponseError,
389 #[serde(default)]
390 error: Option<PartialResponseError>,
391}
392
393#[derive(Deserialize, Debug, Clone, Default)]
394struct PartialResponseError {
395 #[serde(default)]
396 code: Option<String>,
397 #[serde(default)]
398 message: Option<String>,
399 #[serde(default)]
400 param: Option<Value>,
401}
402
403impl GenericStreamErrorPayload {
404 pub fn into_response_error(self) -> ResponseError {
405 let nested = self.error.unwrap_or_default();
406 ResponseError {
407 code: self.top_level.code.or(nested.code),
408 message: self
409 .top_level
410 .message
411 .or(nested.message)
412 .unwrap_or_default(),
413 param: self.top_level.param.or(nested.param),
414 }
415 }
416}
417
418#[derive(Deserialize, Debug)]
419#[serde(tag = "type")]
420pub enum StreamEvent {
421 #[serde(rename = "response.created")]
422 Created { response: ResponseSummary },
423 #[serde(rename = "response.in_progress")]
424 InProgress { response: ResponseSummary },
425 #[serde(rename = "response.output_item.added")]
426 OutputItemAdded {
427 output_index: usize,
428 #[serde(default)]
429 sequence_number: Option<u64>,
430 item: ResponseOutputItem,
431 },
432 #[serde(rename = "response.output_item.done")]
433 OutputItemDone {
434 output_index: usize,
435 #[serde(default)]
436 sequence_number: Option<u64>,
437 item: ResponseOutputItem,
438 },
439 #[serde(rename = "response.content_part.added")]
440 ContentPartAdded {
441 item_id: String,
442 output_index: usize,
443 content_index: usize,
444 part: Value,
445 },
446 #[serde(rename = "response.content_part.done")]
447 ContentPartDone {
448 item_id: String,
449 output_index: usize,
450 content_index: usize,
451 part: Value,
452 },
453 #[serde(rename = "response.output_text.delta")]
454 OutputTextDelta {
455 item_id: String,
456 output_index: usize,
457 #[serde(default)]
458 content_index: Option<usize>,
459 delta: String,
460 },
461 #[serde(rename = "response.output_text.done")]
462 OutputTextDone {
463 item_id: String,
464 output_index: usize,
465 #[serde(default)]
466 content_index: Option<usize>,
467 text: String,
468 },
469 #[serde(rename = "response.refusal.delta")]
470 RefusalDelta {
471 item_id: String,
472 output_index: usize,
473 content_index: usize,
474 delta: String,
475 #[serde(default)]
476 sequence_number: Option<u64>,
477 },
478 #[serde(rename = "response.refusal.done")]
479 RefusalDone {
480 item_id: String,
481 output_index: usize,
482 content_index: usize,
483 refusal: String,
484 #[serde(default)]
485 sequence_number: Option<u64>,
486 },
487 #[serde(rename = "response.reasoning_summary_part.added")]
488 ReasoningSummaryPartAdded {
489 item_id: String,
490 output_index: usize,
491 summary_index: usize,
492 },
493 #[serde(rename = "response.reasoning_summary_text.delta")]
494 ReasoningSummaryTextDelta {
495 item_id: String,
496 output_index: usize,
497 delta: String,
498 },
499 #[serde(rename = "response.reasoning_summary_text.done")]
500 ReasoningSummaryTextDone {
501 item_id: String,
502 output_index: usize,
503 text: String,
504 },
505 #[serde(rename = "response.reasoning_summary_part.done")]
506 ReasoningSummaryPartDone {
507 item_id: String,
508 output_index: usize,
509 summary_index: usize,
510 },
511 #[serde(rename = "response.reasoning.delta")]
512 ReasoningDelta {
513 #[serde(default)]
514 item_id: Option<String>,
515 #[serde(default)]
516 output_index: Option<usize>,
517 delta: String,
518 },
519 #[serde(rename = "response.reasoning.done")]
520 ReasoningDone {
521 #[serde(default)]
522 item_id: Option<String>,
523 #[serde(default)]
524 output_index: Option<usize>,
525 #[serde(default)]
526 text: Option<String>,
527 },
528 #[serde(rename = "response.function_call_arguments.delta")]
529 FunctionCallArgumentsDelta {
530 item_id: String,
531 output_index: usize,
532 delta: String,
533 #[serde(default)]
534 sequence_number: Option<u64>,
535 },
536 #[serde(rename = "response.function_call_arguments.done")]
537 FunctionCallArgumentsDone {
538 item_id: String,
539 output_index: usize,
540 arguments: String,
541 #[serde(default)]
542 sequence_number: Option<u64>,
543 },
544 #[serde(rename = "response.custom_tool_call_input.delta")]
545 CustomToolCallInputDelta {
546 item_id: String,
547 output_index: usize,
548 delta: String,
549 #[serde(default)]
550 sequence_number: Option<u64>,
551 },
552 #[serde(rename = "response.custom_tool_call_input.done")]
553 CustomToolCallInputDone {
554 item_id: String,
555 output_index: usize,
556 input: String,
557 #[serde(default)]
558 sequence_number: Option<u64>,
559 },
560 #[serde(rename = "response.completed")]
561 Completed { response: ResponseSummary },
562 #[serde(rename = "response.incomplete")]
563 Incomplete { response: ResponseSummary },
564 #[serde(rename = "response.failed")]
565 Failed { response: ResponseSummary },
566 #[serde(rename = "response.error")]
567 Error { error: ResponseError },
568 #[serde(rename = "error")]
569 GenericError {
570 #[serde(flatten)]
571 error: GenericStreamErrorPayload,
572 },
573 #[serde(other)]
574 Unknown,
575}
576
577#[derive(Serialize, Deserialize, Debug, Default, Clone)]
578pub struct ResponseSummary {
579 #[serde(default)]
580 pub id: Option<String>,
581 #[serde(default)]
582 pub status: Option<String>,
583 #[serde(default)]
584 pub incomplete_details: Option<ResponseIncompleteDetails>,
585 #[serde(default)]
586 pub error: Option<ResponseError>,
587 #[serde(default)]
588 pub usage: Option<ResponseUsage>,
589 #[serde(default)]
590 pub output: Vec<ResponseOutputItem>,
591 #[serde(default)]
592 pub service_tier: Option<crate::ServiceTier>,
593}
594
595#[derive(Serialize, Deserialize, Debug, Default, Clone)]
596pub struct ResponseIncompleteDetails {
597 #[serde(default)]
598 pub reason: Option<String>,
599}
600
601#[derive(Serialize, Deserialize, Debug, Default, Clone)]
602pub struct ResponseUsage {
603 #[serde(default)]
604 pub input_tokens: Option<u64>,
605 #[serde(default)]
606 pub input_tokens_details: ResponseInputTokensDetails,
607 #[serde(default)]
608 pub output_tokens: Option<u64>,
609 #[serde(default)]
610 pub output_tokens_details: ResponseOutputTokensDetails,
611 #[serde(default)]
612 pub total_tokens: Option<u64>,
613}
614
615#[derive(Serialize, Deserialize, Debug, Default, Clone)]
616pub struct ResponseInputTokensDetails {
617 #[serde(default)]
618 pub cached_tokens: u64,
619}
620
621#[derive(Serialize, Deserialize, Debug, Default, Clone)]
622pub struct ResponseOutputTokensDetails {
623 #[serde(default)]
624 pub reasoning_tokens: u64,
625}
626
627#[derive(Serialize, Deserialize, Debug, Clone)]
628#[serde(tag = "type", rename_all = "snake_case")]
629pub enum ResponseOutputItem {
630 Message(ResponseOutputMessage),
631 FunctionCall(ResponseFunctionToolCall),
632 CustomToolCall(ResponseCustomToolCall),
633 Reasoning(ResponseReasoningItem),
634 Compaction(ResponseCompactionItem),
635 /// Deserialization-only catch-all; must never be serialized back to the API.
636 #[serde(other)]
637 Unknown,
638}
639
640#[derive(Serialize, Deserialize, Debug, Clone)]
641pub struct ResponseReasoningItem {
642 #[serde(default)]
643 pub id: Option<String>,
644 #[serde(default)]
645 pub summary: Vec<ReasoningSummaryPart>,
646 #[serde(default)]
647 pub content: Vec<Value>,
648 #[serde(default)]
649 pub encrypted_content: Option<String>,
650 #[serde(default)]
651 pub status: Option<String>,
652}
653
654#[derive(Serialize, Deserialize, Debug, Clone)]
655#[serde(tag = "type", rename_all = "snake_case")]
656pub enum ReasoningSummaryPart {
657 SummaryText {
658 text: String,
659 },
660 #[serde(other)]
661 Unknown,
662}
663
664#[derive(Serialize, Deserialize, Debug, Clone)]
665pub struct ResponseOutputMessage {
666 #[serde(default)]
667 pub id: Option<String>,
668 #[serde(default)]
669 pub content: Vec<Value>,
670 #[serde(default)]
671 pub role: Option<String>,
672 #[serde(default)]
673 pub status: Option<String>,
674 #[serde(default)]
675 pub phase: Option<String>,
676}
677
678#[derive(Serialize, Deserialize, Debug, Clone)]
679pub struct ResponseFunctionToolCall {
680 #[serde(default)]
681 pub id: Option<String>,
682 #[serde(default)]
683 pub arguments: String,
684 #[serde(default)]
685 pub call_id: Option<String>,
686 #[serde(default)]
687 pub name: Option<String>,
688 #[serde(default)]
689 pub status: Option<String>,
690}
691
692#[derive(Serialize, Deserialize, Debug, Clone)]
693pub struct ResponseCustomToolCall {
694 #[serde(default)]
695 pub id: Option<String>,
696 #[serde(default)]
697 pub status: Option<String>,
698 #[serde(default)]
699 pub call_id: Option<String>,
700 #[serde(default)]
701 pub name: Option<String>,
702 #[serde(default)]
703 pub input: String,
704}
705
706pub async fn compact_response(
707 client: &dyn HttpClient,
708 provider_name: &str,
709 api_url: &str,
710 api_key: &str,
711 request: CompactRequest,
712 extra_headers: &CustomHeaders,
713) -> Result<CompactedResponse, RequestError> {
714 let request = HttpRequest::builder()
715 .method(Method::POST)
716 .uri(format!("{api_url}/responses/compact"))
717 .header("Content-Type", "application/json")
718 .header("Authorization", format!("Bearer {}", api_key.trim()))
719 .extra_headers(extra_headers)
720 .body(AsyncBody::from(
721 serde_json::to_string(&request).map_err(|error| RequestError::Other(error.into()))?,
722 ))
723 .map_err(|error| RequestError::Other(error.into()))?;
724
725 let mut response = client.send(request).await?;
726 let mut body = String::new();
727 response
728 .body_mut()
729 .read_to_string(&mut body)
730 .await
731 .map_err(|error| RequestError::Other(error.into()))?;
732
733 if response.status().is_success() {
734 serde_json::from_str(&body).map_err(|error| RequestError::Other(error.into()))
735 } else {
736 Err(RequestError::HttpResponseError {
737 provider: provider_name.to_owned(),
738 status_code: response.status(),
739 body,
740 headers: response.headers().clone(),
741 })
742 }
743}
744
745pub async fn stream_response(
746 client: &dyn HttpClient,
747 provider_name: &str,
748 api_url: &str,
749 api_key: &str,
750 request: Request,
751 extra_headers: &CustomHeaders,
752) -> Result<BoxStream<'static, Result<StreamEvent>>, RequestError> {
753 let uri = format!("{api_url}/responses");
754 let is_streaming = request.stream;
755 let request = HttpRequest::builder()
756 .method(Method::POST)
757 .uri(uri)
758 .header("Content-Type", "application/json")
759 .header("Authorization", format!("Bearer {}", api_key.trim()))
760 .extra_headers(extra_headers)
761 .body(AsyncBody::from(
762 serde_json::to_string(&request).map_err(|e| RequestError::Other(e.into()))?,
763 ))
764 .map_err(|e| RequestError::Other(e.into()))?;
765
766 let mut response = client.send(request).await?;
767 if response.status().is_success() {
768 if is_streaming {
769 let reader = BufReader::new(response.into_body());
770 Ok(reader
771 .lines()
772 .filter_map(|line| async move {
773 match line {
774 Ok(line) => {
775 let line = line
776 .strip_prefix("data: ")
777 .or_else(|| line.strip_prefix("data:"))?;
778 if line == "[DONE]" || line.is_empty() {
779 None
780 } else {
781 match serde_json::from_str::<StreamEvent>(line) {
782 Ok(event) => Some(Ok(event)),
783 Err(error) => {
784 log::error!(
785 "Failed to parse OpenAI responses stream event: `{}`\nResponse: `{}`",
786 error,
787 line,
788 );
789 Some(Err(anyhow!(error)))
790 }
791 }
792 }
793 }
794 Err(error) => Some(Err(anyhow!(error))),
795 }
796 })
797 .boxed())
798 } else {
799 let mut body = String::new();
800 response
801 .body_mut()
802 .read_to_string(&mut body)
803 .await
804 .map_err(|e| RequestError::Other(e.into()))?;
805
806 match serde_json::from_str::<ResponseSummary>(&body) {
807 Ok(response_summary) => {
808 let events = vec![
809 StreamEvent::Created {
810 response: response_summary.clone(),
811 },
812 StreamEvent::InProgress {
813 response: response_summary.clone(),
814 },
815 ];
816
817 let mut all_events = events;
818 for (output_index, item) in response_summary.output.iter().enumerate() {
819 all_events.push(StreamEvent::OutputItemAdded {
820 output_index,
821 sequence_number: None,
822 item: item.clone(),
823 });
824
825 match item {
826 ResponseOutputItem::Message(message) => {
827 for content_item in &message.content {
828 if let Some(text) = content_item.get("text") {
829 if let Some(text_str) = text.as_str() {
830 if let Some(ref item_id) = message.id {
831 all_events.push(StreamEvent::OutputTextDelta {
832 item_id: item_id.clone(),
833 output_index,
834 content_index: None,
835 delta: text_str.to_string(),
836 });
837 }
838 }
839 }
840 }
841 }
842 ResponseOutputItem::FunctionCall(function_call) => {
843 if let Some(ref item_id) = function_call.id {
844 all_events.push(StreamEvent::FunctionCallArgumentsDone {
845 item_id: item_id.clone(),
846 output_index,
847 arguments: function_call.arguments.clone(),
848 sequence_number: None,
849 });
850 }
851 }
852 ResponseOutputItem::CustomToolCall(custom_tool_call) => {
853 if let Some(ref item_id) = custom_tool_call.id {
854 all_events.push(StreamEvent::CustomToolCallInputDone {
855 item_id: item_id.clone(),
856 output_index,
857 input: custom_tool_call.input.clone(),
858 sequence_number: None,
859 });
860 }
861 }
862 ResponseOutputItem::Reasoning(reasoning) => {
863 if let Some(ref item_id) = reasoning.id {
864 for part in &reasoning.summary {
865 if let ReasoningSummaryPart::SummaryText { text } = part {
866 all_events.push(
867 StreamEvent::ReasoningSummaryTextDelta {
868 item_id: item_id.clone(),
869 output_index,
870 delta: text.clone(),
871 },
872 );
873 }
874 }
875 }
876 }
877 // No synthesized deltas; the `OutputItemDone`
878 // event pushed below carries the full item.
879 ResponseOutputItem::Compaction(_) => {}
880 ResponseOutputItem::Unknown => {}
881 }
882
883 all_events.push(StreamEvent::OutputItemDone {
884 output_index,
885 sequence_number: None,
886 item: item.clone(),
887 });
888 }
889
890 let status = response_summary.status.clone();
891 all_events.push(match status.as_deref() {
892 Some("incomplete") => StreamEvent::Incomplete {
893 response: response_summary,
894 },
895 Some("failed") => StreamEvent::Failed {
896 response: response_summary,
897 },
898 _ => StreamEvent::Completed {
899 response: response_summary,
900 },
901 });
902
903 Ok(futures::stream::iter(all_events.into_iter().map(Ok)).boxed())
904 }
905 Err(error) => {
906 log::error!(
907 "Failed to parse OpenAI non-streaming response: `{}`\nResponse: `{}`",
908 error,
909 body,
910 );
911 Err(RequestError::Other(anyhow!(error)))
912 }
913 }
914 }
915 } else {
916 let mut body = String::new();
917 response
918 .body_mut()
919 .read_to_string(&mut body)
920 .await
921 .map_err(|e| RequestError::Other(e.into()))?;
922
923 Err(RequestError::HttpResponseError {
924 provider: provider_name.to_owned(),
925 status_code: response.status(),
926 body,
927 headers: response.headers().clone(),
928 })
929 }
930}
931
932#[cfg(test)]
933mod tests {
934 use super::*;
935 use futures::executor::block_on;
936 use http_client::FakeHttpClient;
937 use language_model_core::OPEN_AI_PROVIDER_ID;
938 use serde_json::json;
939 use std::sync::{Arc, Mutex};
940
941 #[test]
942 fn compact_response_posts_supported_request_fields() {
943 let captured_request = Arc::new(Mutex::new(None));
944 let captured_request_for_handler = captured_request.clone();
945 let http_client = FakeHttpClient::create(move |request| {
946 let captured_request = captured_request_for_handler.clone();
947 async move {
948 let method = request.method().clone();
949 let uri = request.uri().to_string();
950 let authorization = request
951 .headers()
952 .get("Authorization")
953 .and_then(|value| value.to_str().ok())
954 .map(str::to_string);
955 let mut body = request.into_body();
956 let mut body_text = String::new();
957 body.read_to_string(&mut body_text).await?;
958 *captured_request.lock().unwrap() = Some((method, uri, authorization, body_text));
959
960 Ok(http_client::Response::builder()
961 .status(200)
962 .body(AsyncBody::from(
963 json!({
964 "id": "resp_compact",
965 "created_at": 1_700_000_000,
966 "object": "response.compaction",
967 "output": [{
968 "type": "compaction",
969 "id": "cmp_manual",
970 "encrypted_content": "opaque-state"
971 }],
972 "usage": {
973 "input_tokens": 100,
974 "input_tokens_details": {"cached_tokens": 20},
975 "output_tokens": 10,
976 "output_tokens_details": {"reasoning_tokens": 5},
977 "total_tokens": 110
978 }
979 })
980 .to_string(),
981 ))?)
982 }
983 });
984 let response = block_on(compact_response(
985 http_client.as_ref(),
986 "OpenAI",
987 "https://api.openai.com/v1",
988 "secret",
989 compact_test_request(),
990 &CustomHeaders::default(),
991 ))
992 .unwrap();
993
994 assert_eq!(
995 provider_compaction_items(
996 &match response
997 .into_compacted_context(OPEN_AI_PROVIDER_ID)
998 .unwrap()
999 {
1000 CompactedContext::ProviderState(state) => state,
1001 CompactedContext::Summary { .. } => panic!("expected provider state"),
1002 },
1003 &OPEN_AI_PROVIDER_ID
1004 )
1005 .unwrap(),
1006 Some(vec![json!({
1007 "type": "compaction",
1008 "id": "cmp_manual",
1009 "encrypted_content": "opaque-state"
1010 })])
1011 );
1012 let (method, uri, authorization, body) = captured_request.lock().unwrap().take().unwrap();
1013 assert_eq!(method, Method::POST);
1014 assert_eq!(uri, "https://api.openai.com/v1/responses/compact");
1015 assert_eq!(authorization.as_deref(), Some("Bearer secret"));
1016 assert_eq!(
1017 serde_json::from_str::<Value>(&body).unwrap(),
1018 json!({
1019 "model": "gpt-5.4",
1020 "input": [{
1021 "type": "message",
1022 "role": "user",
1023 "content": [{
1024 "type": "input_text",
1025 "text": "Retain this context."
1026 }]
1027 }],
1028 "prompt_cache_key": "thread-123",
1029 "service_tier": "priority"
1030 })
1031 );
1032 }
1033
1034 #[test]
1035 fn compact_response_reports_http_and_deserialization_errors() {
1036 let http_client = FakeHttpClient::create(|_| async move {
1037 Ok(http_client::Response::builder()
1038 .status(429)
1039 .header("retry-after", "5")
1040 .body(AsyncBody::from("rate limited"))?)
1041 });
1042
1043 let error = block_on(compact_response(
1044 http_client.as_ref(),
1045 "OpenAI",
1046 "https://api.openai.com/v1",
1047 "secret",
1048 compact_test_request(),
1049 &CustomHeaders::default(),
1050 ))
1051 .unwrap_err();
1052
1053 match error {
1054 RequestError::HttpResponseError {
1055 provider,
1056 status_code,
1057 body,
1058 headers,
1059 } => {
1060 assert_eq!(provider, "OpenAI");
1061 assert_eq!(status_code, 429);
1062 assert_eq!(body, "rate limited");
1063 assert_eq!(headers["retry-after"], "5");
1064 }
1065 error => panic!("expected an HTTP response error, got {error:?}"),
1066 }
1067
1068 let http_client = FakeHttpClient::create(|_| async move {
1069 Ok(http_client::Response::builder()
1070 .status(200)
1071 .body(AsyncBody::from("not valid JSON"))?)
1072 });
1073
1074 let error = block_on(compact_response(
1075 http_client.as_ref(),
1076 "OpenAI",
1077 "https://api.openai.com/v1",
1078 "secret",
1079 compact_test_request(),
1080 &CustomHeaders::default(),
1081 ))
1082 .unwrap_err();
1083
1084 assert!(
1085 matches!(error, RequestError::Other(_)),
1086 "expected malformed JSON to produce a request error, got {error:?}"
1087 );
1088 }
1089
1090 #[test]
1091 fn compacted_response_preserves_canonical_output_items() {
1092 let output = vec![
1093 json!({
1094 "type": "message",
1095 "role": "user",
1096 "content": "Retained user context.",
1097 "provider_extension": {"preserve": true}
1098 }),
1099 json!({
1100 "type": "compaction",
1101 "id": "cmp_manual",
1102 "encrypted_content": "opaque-state"
1103 }),
1104 ];
1105 let response: CompactedResponse = serde_json::from_value(json!({
1106 "id": "resp_compact",
1107 "created_at": 1_700_000_000,
1108 "object": "response.compaction",
1109 "output": &output,
1110 "usage": {
1111 "input_tokens": 100,
1112 "input_tokens_details": {"cached_tokens": 20},
1113 "output_tokens": 10,
1114 "output_tokens_details": {"reasoning_tokens": 5},
1115 "total_tokens": 110
1116 }
1117 }))
1118 .unwrap();
1119
1120 let CompactedContext::ProviderState(state) = response
1121 .into_compacted_context(OPEN_AI_PROVIDER_ID)
1122 .unwrap()
1123 else {
1124 panic!("expected provider state");
1125 };
1126 assert_eq!(
1127 provider_compaction_items(&state, &OPEN_AI_PROVIDER_ID).unwrap(),
1128 Some(output)
1129 );
1130 }
1131
1132 #[test]
1133 fn compacted_response_rejects_output_without_compaction_item() {
1134 let response: CompactedResponse = serde_json::from_value(json!({
1135 "id": "resp_compact",
1136 "created_at": 1_700_000_000,
1137 "object": "response.compaction",
1138 "output": [{
1139 "type": "message",
1140 "role": "user",
1141 "content": "Retained user context."
1142 }],
1143 "usage": {
1144 "input_tokens": 100,
1145 "input_tokens_details": {"cached_tokens": 20},
1146 "output_tokens": 10,
1147 "output_tokens_details": {"reasoning_tokens": 5},
1148 "total_tokens": 110
1149 }
1150 }))
1151 .unwrap();
1152
1153 assert!(
1154 response
1155 .into_compacted_context(OPEN_AI_PROVIDER_ID)
1156 .unwrap_err()
1157 .to_string()
1158 .contains("compaction item")
1159 );
1160 }
1161
1162 #[test]
1163 fn compacted_response_rejects_empty_output() {
1164 let response: CompactedResponse = serde_json::from_value(json!({
1165 "id": "resp_compact",
1166 "created_at": 1_700_000_000,
1167 "object": "response.compaction",
1168 "output": [],
1169 "usage": {
1170 "input_tokens": 100,
1171 "input_tokens_details": {"cached_tokens": 20},
1172 "output_tokens": 10,
1173 "output_tokens_details": {"reasoning_tokens": 5},
1174 "total_tokens": 110
1175 }
1176 }))
1177 .unwrap();
1178
1179 assert!(
1180 response
1181 .into_compacted_context(OPEN_AI_PROVIDER_ID)
1182 .unwrap_err()
1183 .to_string()
1184 .contains("empty")
1185 );
1186 }
1187
1188 #[test]
1189 fn provider_compaction_items_ignores_state_owned_by_another_provider() {
1190 let items = vec![json!({
1191 "type": "compaction",
1192 "id": "cmp_manual",
1193 "encrypted_content": "opaque-state"
1194 })];
1195 let state =
1196 provider_compaction_state_from_items(OPEN_AI_PROVIDER_ID, items.clone()).unwrap();
1197
1198 assert_eq!(
1199 provider_compaction_items(&state, &LanguageModelProviderId::new("anthropic")).unwrap(),
1200 None
1201 );
1202
1203 // The same window stamped for a different OpenAI-protocol backend is
1204 // opaque to OpenAI proper: encrypted compaction items are only
1205 // decryptable by the infrastructure that produced them.
1206 let compatible_backend = LanguageModelProviderId::new("my-compatible-endpoint");
1207 let state =
1208 provider_compaction_state_from_items(compatible_backend.clone(), items).unwrap();
1209 assert_eq!(
1210 provider_compaction_items(&state, &OPEN_AI_PROVIDER_ID).unwrap(),
1211 None
1212 );
1213 assert!(
1214 provider_compaction_items(&state, &compatible_backend)
1215 .unwrap()
1216 .is_some()
1217 );
1218 }
1219
1220 #[test]
1221 fn provider_compaction_items_rejects_unknown_open_ai_format() {
1222 let state = ProviderCompactionState::new(
1223 OPEN_AI_PROVIDER_ID,
1224 "openai.responses.input-items.v2",
1225 "[]",
1226 );
1227 assert!(
1228 provider_compaction_items(&state, &OPEN_AI_PROVIDER_ID)
1229 .unwrap_err()
1230 .to_string()
1231 .contains("unsupported OpenAI compaction state format")
1232 );
1233 }
1234
1235 #[test]
1236 fn provider_compaction_items_rejects_malformed_open_ai_state() {
1237 let state = ProviderCompactionState::new(
1238 OPEN_AI_PROVIDER_ID,
1239 COMPACTION_STATE_FORMAT,
1240 "not valid JSON",
1241 );
1242
1243 assert!(provider_compaction_items(&state, &OPEN_AI_PROVIDER_ID).is_err());
1244 }
1245
1246 fn compact_test_request() -> CompactRequest {
1247 CompactRequest {
1248 model: "gpt-5.4".to_string(),
1249 instructions: None,
1250 input: ResponseInput::new(
1251 Vec::new(),
1252 vec![ResponseInputItem::Message(ResponseMessageItem {
1253 role: Role::User,
1254 content: vec![ResponseInputContent::Text {
1255 text: "Retain this context.".to_string(),
1256 }],
1257 phase: None,
1258 })],
1259 ),
1260 prompt_cache_key: Some("thread-123".to_string()),
1261 service_tier: Some(ServiceTier::Priority),
1262 }
1263 }
1264}
1265