diff --git a/crates/anthropic/src/anthropic.rs b/crates/anthropic/src/anthropic.rs
index 3531924f295e1a..3f034bf03c76ae 100644
--- a/crates/anthropic/src/anthropic.rs
+++ b/crates/anthropic/src/anthropic.rs
@@ -644,6 +644,10 @@ pub enum RequestContent {
     #[serde(rename = "compaction")]
     Compaction {
         content: Option<Arc<str>>,
+        /// Opaque metadata from a prior compaction that must be round-tripped
+        /// verbatim for Anthropic to recognize the block.
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        encrypted_content: Option<Arc<str>>,
         #[serde(skip_serializing_if = "Option::is_none")]
         cache_control: Option<CacheControl>,
     },
@@ -679,7 +683,11 @@ pub enum ResponseContent {
         input: serde_json::Value,
     },
     #[serde(rename = "compaction")]
-    Compaction { content: Option<Arc<str>> },
+    Compaction {
+        content: Option<Arc<str>>,
+        #[serde(default)]
+        encrypted_content: Option<Arc<str>>,
+    },
 }
 
 #[derive(Debug, Serialize, Deserialize)]
@@ -932,7 +940,11 @@ pub enum ContentDelta {
     #[serde(rename = "input_json_delta")]
     InputJsonDelta { partial_json: String },
     #[serde(rename = "compaction_delta")]
-    CompactionDelta { content: Option<Arc<str>> },
+    CompactionDelta {
+        content: Option<Arc<str>>,
+        #[serde(default)]
+        encrypted_content: Option<Arc<str>>,
+    },
 }
 
 #[derive(Debug, Serialize, Deserialize)]
diff --git a/crates/anthropic/src/completion.rs b/crates/anthropic/src/completion.rs
index 884ed1a1ddde9b..f02ab0b60487ae 100644
--- a/crates/anthropic/src/completion.rs
+++ b/crates/anthropic/src/completion.rs
@@ -1,15 +1,17 @@
-use anyhow::Result;
+use anyhow::{Result, anyhow};
 use collections::HashMap;
 use futures::{Stream, StreamExt};
 use language_model_core::{
-    CompactionContent, LanguageModelCompletionError, LanguageModelCompletionEvent,
-    LanguageModelProviderName, LanguageModelRequest, LanguageModelRequestToolInput,
-    LanguageModelToolChoice, LanguageModelToolResultContent, LanguageModelToolUse,
-    LanguageModelToolUseInput, MessageContent, Role, StopReason, TokenUsage,
+    CompactedContext, CompactionUpdate, LanguageModelCompletionError, LanguageModelCompletionEvent,
+    LanguageModelProviderId, LanguageModelProviderName, LanguageModelRequest,
+    LanguageModelRequestToolInput, LanguageModelToolChoice, LanguageModelToolResultContent,
+    LanguageModelToolUse, LanguageModelToolUseInput, MessageContent, ProviderCompactionState, Role,
+    SharedString, StopReason, TokenUsage,
     util::{fix_streamed_json, parse_tool_arguments},
 };
 use std::pin::Pin;
 use std::str::FromStr;
+use std::sync::Arc;
 
 use crate::{
     AdaptiveThinkingDisplay, AnthropicError, AnthropicModelMode, CacheControl, CacheControlType,
@@ -19,6 +21,46 @@ use crate::{
     completion_error_from_anthropic_api,
 };
 
+pub const COMPACTION_STATE_FORMAT: &str = "anthropic.messages.encrypted-content.v1";
+
+/// Packages a compaction block's opaque `encrypted_content` into provider
+/// state owned by `owner`.
+///
+/// Anthropic requires the metadata to be round-tripped verbatim, and only the
+/// backend whose infrastructure produced it can make sense of it. The owner
+/// recorded here is what [`provider_compaction_encrypted_content`] later
+/// compares against, so it must identify that backend, not merely the wire
+/// protocol.
+pub fn provider_compaction_state_from_encrypted_content(
+    owner: LanguageModelProviderId,
+    encrypted_content: impl Into<Arc<str>>,
+) -> ProviderCompactionState {
+    ProviderCompactionState::new(
+        owner,
+        SharedString::new_static(COMPACTION_STATE_FORMAT),
+        encrypted_content,
+    )
+}
+
+/// Recovers the `encrypted_content` to round-trip from `state` if it is owned
+/// by `owner`, or `None` when the state belongs to a different backend and the
+/// summary should be replayed without it.
+pub fn provider_compaction_encrypted_content(
+    state: &ProviderCompactionState,
+    owner: &LanguageModelProviderId,
+) -> Result<Option<Arc<str>>> {
+    if state.provider_id() != owner {
+        return Ok(None);
+    }
+    if state.format() != COMPACTION_STATE_FORMAT {
+        return Err(anyhow!(
+            "unsupported Anthropic compaction state format: {}",
+            state.format()
+        ));
+    }
+    Ok(Some(state.payload().into()))
+}
+
 #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
 pub enum AnthropicPromptCacheMode {
     Disabled,
@@ -68,7 +110,10 @@ fn mark_last_cacheable_content(content: &mut [RequestContent], cache_control: Ca
     }
 }
 
-fn to_anthropic_content(content: MessageContent) -> Result<Option<RequestContent>> {
+fn to_anthropic_content(
+    content: MessageContent,
+    compaction_state_owner: &LanguageModelProviderId,
+) -> Result<Option<RequestContent>> {
     match content {
         MessageContent::Text(text) => {
             let text = if text.chars().last().is_some_and(|c| c.is_whitespace()) {
@@ -159,17 +204,23 @@ fn to_anthropic_content(content: MessageContent) -> Result<Option<RequestContent
                 cache_control: None,
             }))
         }
-        MessageContent::Compaction(CompactionContent::Summary { content }) => {
+        MessageContent::Compaction(CompactedContext::Summary {
+            content,
+            provider_state,
+        }) => {
+            let encrypted_content = match &provider_state {
+                Some(state) => {
+                    provider_compaction_encrypted_content(state, compaction_state_owner)?
+                }
+                None => None,
+            };
             Ok(Some(RequestContent::Compaction {
-                content,
+                content: Some(content),
+                encrypted_content,
                 cache_control: None,
             }))
         }
-        // Encrypted compaction blocks come from other providers, and a
-        // Pending block is a streaming-only UI signal; neither is replayed.
-        MessageContent::Compaction(
-            CompactionContent::Encrypted { .. } | CompactionContent::Pending,
-        ) => Ok(None),
+        MessageContent::Compaction(CompactedContext::ProviderState(_)) => Ok(None),
     }
 }
 
@@ -180,6 +231,7 @@ pub fn into_anthropic(
     max_output_tokens: u64,
     mode: AnthropicModelMode,
     cache_mode: AnthropicPromptCacheMode,
+    compaction_state_owner: &LanguageModelProviderId,
 ) -> Result<crate::Request> {
     let mut new_messages: Vec<Message> = Vec::new();
     let mut system_message = String::new();
@@ -196,7 +248,7 @@ pub fn into_anthropic(
             Role::User | Role::Assistant => {
                 let mut anthropic_message_content = Vec::new();
                 for content in message.content {
-                    if let Some(content) = to_anthropic_content(content)? {
+                    if let Some(content) = to_anthropic_content(content, compaction_state_owner)? {
                         anthropic_message_content.push(content);
                     }
                 }
@@ -358,18 +410,28 @@ pub fn into_anthropic(
 
 pub struct AnthropicEventMapper {
     tool_uses_by_index: HashMap<usize, RawToolUse>,
+    compactions_by_index: HashMap<usize, RawCompaction>,
     usage: Usage,
     stop_reason: StopReason,
     provider_name: LanguageModelProviderName,
+    compaction_state_owner: LanguageModelProviderId,
 }
 
 impl AnthropicEventMapper {
-    pub fn new(provider_name: LanguageModelProviderName) -> Self {
+    /// `compaction_state_owner` identifies the backend whose infrastructure
+    /// produced this stream, so that any `encrypted_content` it emits is only
+    /// ever round-tripped back to that same backend.
+    pub fn new(
+        provider_name: LanguageModelProviderName,
+        compaction_state_owner: LanguageModelProviderId,
+    ) -> Self {
         Self {
             tool_uses_by_index: HashMap::default(),
+            compactions_by_index: HashMap::default(),
             usage: Usage::default(),
             stop_reason: StopReason::EndTurn,
             provider_name,
+            compaction_state_owner,
         }
     }
 
@@ -421,10 +483,28 @@ impl AnthropicEventMapper {
                     );
                     Vec::new()
                 }
-                ResponseContent::Compaction { content } => {
-                    vec![Ok(LanguageModelCompletionEvent::Compaction(
-                        CompactionContent::Summary { content },
-                    ))]
+                ResponseContent::Compaction {
+                    content,
+                    encrypted_content,
+                } => {
+                    let mut events = vec![Ok(LanguageModelCompletionEvent::Compaction(
+                        CompactionUpdate::Started,
+                    ))];
+                    let compaction = self.compactions_by_index.entry(index).or_default();
+                    if let Some(encrypted_content) =
+                        encrypted_content.filter(|encrypted| !encrypted.is_empty())
+                    {
+                        compaction.encrypted_content = Some(encrypted_content);
+                    }
+                    if let Some(content) = content
+                        && !content.is_empty()
+                    {
+                        compaction.summary.push_str(&content);
+                        events.push(Ok(LanguageModelCompletionEvent::Compaction(
+                            CompactionUpdate::SummaryDelta(content),
+                        )));
+                    }
+                    events
                 }
             },
             Event::ContentBlockDelta { index, delta } => match delta {
@@ -443,9 +523,30 @@ impl AnthropicEventMapper {
                         signature: Some(signature),
                     })]
                 }
-                ContentDelta::CompactionDelta { content } => {
+                ContentDelta::CompactionDelta {
+                    content,
+                    encrypted_content,
+                } => {
+                    let Some(compaction) = self.compactions_by_index.get_mut(&index) else {
+                        return vec![Err(LanguageModelCompletionError::Other(anyhow::anyhow!(
+                            "Anthropic streamed a compaction delta before starting its content block"
+                        )))];
+                    };
+                    // Unlike summary text, `encrypted_content` arrives whole:
+                    // a later delta carries a complete replacement value, not
+                    // a chunk to append (Anthropic's own SDKs assign it, the
+                    // way they do thinking signatures).
+                    if let Some(encrypted_content) =
+                        encrypted_content.filter(|encrypted| !encrypted.is_empty())
+                    {
+                        compaction.encrypted_content = Some(encrypted_content);
+                    }
+                    let Some(content) = content.filter(|content| !content.is_empty()) else {
+                        return Vec::new();
+                    };
+                    compaction.summary.push_str(&content);
                     vec![Ok(LanguageModelCompletionEvent::Compaction(
-                        CompactionContent::Summary { content },
+                        CompactionUpdate::SummaryDelta(content),
                     ))]
                 }
                 ContentDelta::InputJsonDelta { partial_json } => {
@@ -475,7 +576,29 @@ impl AnthropicEventMapper {
                 }
             },
             Event::ContentBlockStop { index } => {
-                if let Some(tool_use) = self.tool_uses_by_index.remove(&index) {
+                if let Some(compaction) = self.compactions_by_index.remove(&index) {
+                    // A compaction block that closes without content is a
+                    // documented failed compaction, which the server treats
+                    // as a no-op: there is nothing to persist, and the
+                    // conversation continues on the uncompacted transcript.
+                    if compaction.summary.is_empty() {
+                        return vec![Ok(LanguageModelCompletionEvent::Compaction(
+                            CompactionUpdate::Failed,
+                        ))];
+                    }
+                    let provider_state = compaction.encrypted_content.map(|encrypted_content| {
+                        provider_compaction_state_from_encrypted_content(
+                            self.compaction_state_owner.clone(),
+                            encrypted_content,
+                        )
+                    });
+                    vec![Ok(LanguageModelCompletionEvent::Compaction(
+                        CompactionUpdate::Finished(CompactedContext::Summary {
+                            content: compaction.summary.into(),
+                            provider_state,
+                        }),
+                    ))]
+                } else if let Some(tool_use) = self.tool_uses_by_index.remove(&index) {
                     let input_json = tool_use.input_json.trim();
                     let event_result = match parse_tool_arguments(input_json) {
                         Ok(input) => Ok(LanguageModelCompletionEvent::ToolUse(
@@ -533,6 +656,17 @@ impl AnthropicEventMapper {
                 ))]
             }
             Event::MessageStop => {
+                // Anthropic closes every content block before ending the
+                // message, so an unclosed compaction block means the stream
+                // was malformed and its finalized summary never arrived.
+                // Consumers would otherwise see `Started` with no terminal
+                // event and treat the compaction as still in progress.
+                if !self.compactions_by_index.is_empty() {
+                    self.compactions_by_index.clear();
+                    return vec![Err(LanguageModelCompletionError::Other(anyhow::anyhow!(
+                        "Anthropic ended the stream without finishing its compaction summary"
+                    )))];
+                }
                 vec![Ok(LanguageModelCompletionEvent::Stop(self.stop_reason))]
             }
             Event::Error { error } => {
@@ -552,6 +686,12 @@ struct RawToolUse {
     input_json: String,
 }
 
+#[derive(Default)]
+struct RawCompaction {
+    summary: String,
+    encrypted_content: Option<Arc<str>>,
+}
+
 /// Updates usage data by preferring counts from `new`.
 fn update_usage(usage: &mut Usage, new: &Usage) {
     if let Some(input_tokens) = new.input_tokens {
@@ -582,7 +722,8 @@ mod tests {
     use super::*;
     use crate::{AnthropicModelMode, UsageIteration, UsageIterationType};
     use language_model_core::{
-        ANTHROPIC_PROVIDER_NAME, LanguageModelImage, LanguageModelRequestMessage, MessageContent,
+        ANTHROPIC_PROVIDER_ID, ANTHROPIC_PROVIDER_NAME, LanguageModelImage,
+        LanguageModelRequestMessage, MessageContent,
     };
 
     #[test]
@@ -631,6 +772,7 @@ mod tests {
             4096,
             AnthropicModelMode::Default,
             AnthropicPromptCacheMode::Automatic,
+            &ANTHROPIC_PROVIDER_ID,
         )
         .unwrap();
 
@@ -738,6 +880,7 @@ mod tests {
             4096,
             AnthropicModelMode::Default,
             AnthropicPromptCacheMode::Legacy,
+            &ANTHROPIC_PROVIDER_ID,
         )
         .unwrap();
 
@@ -797,6 +940,7 @@ mod tests {
             128_000,
             AnthropicModelMode::AdaptiveThinking,
             AnthropicPromptCacheMode::Automatic,
+            &ANTHROPIC_PROVIDER_ID,
         )
         .unwrap();
 
@@ -850,6 +994,7 @@ mod tests {
             4096,
             AnthropicModelMode::Default,
             AnthropicPromptCacheMode::Automatic,
+            &ANTHROPIC_PROVIDER_ID,
         )
         .unwrap();
 
@@ -896,6 +1041,7 @@ mod tests {
                 budget_tokens: Some(10000),
             },
             AnthropicPromptCacheMode::Automatic,
+            &ANTHROPIC_PROVIDER_ID,
         )
         .unwrap()
     }
@@ -996,6 +1142,7 @@ mod tests {
             4096,
             AnthropicModelMode::Default,
             AnthropicPromptCacheMode::Disabled,
+            &ANTHROPIC_PROVIDER_ID,
         )
         .unwrap();
 
@@ -1021,8 +1168,9 @@ mod tests {
     #[test]
     fn test_compaction_content_replayed_as_compaction_block() {
         let result = request_with_assistant_content(vec![
-            MessageContent::Compaction(CompactionContent::Summary {
-                content: Some("Summary of the conversation so far.".into()),
+            MessageContent::Compaction(CompactedContext::Summary {
+                content: "Summary of the conversation so far.".into(),
+                provider_state: None,
             }),
             MessageContent::Text("Response".to_string()),
         ]);
@@ -1042,26 +1190,84 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_compaction_encrypted_content_replayed_only_for_owning_backend() {
+        let summary_owned_by = |owner: LanguageModelProviderId| {
+            MessageContent::Compaction(CompactedContext::Summary {
+                content: "Summary of the conversation so far.".into(),
+                provider_state: Some(provider_compaction_state_from_encrypted_content(
+                    owner,
+                    "opaque-compaction-payload",
+                )),
+            })
+        };
+
+        let owned = to_anthropic_content(
+            summary_owned_by(ANTHROPIC_PROVIDER_ID),
+            &ANTHROPIC_PROVIDER_ID,
+        )
+        .unwrap()
+        .expect("compaction block should be produced");
+        assert_eq!(
+            serde_json::to_value(&owned).unwrap(),
+            serde_json::json!({
+                "type": "compaction",
+                "content": "Summary of the conversation so far.",
+                "encrypted_content": "opaque-compaction-payload"
+            })
+        );
+
+        // State produced by a different Anthropic-protocol backend must not
+        // be round-tripped: the summary is still replayed, but without the
+        // foreign encrypted payload.
+        let foreign = to_anthropic_content(
+            summary_owned_by(LanguageModelProviderId::new("other-anthropic-backend")),
+            &ANTHROPIC_PROVIDER_ID,
+        )
+        .unwrap()
+        .expect("compaction block should be produced");
+        assert_eq!(
+            serde_json::to_value(&foreign).unwrap(),
+            serde_json::json!({
+                "type": "compaction",
+                "content": "Summary of the conversation so far."
+            })
+        );
+    }
+
     #[test]
     fn test_event_mapper_maps_compaction_block_and_deltas() {
-        let mut mapper = AnthropicEventMapper::new(ANTHROPIC_PROVIDER_NAME);
+        let mut mapper = AnthropicEventMapper::new(ANTHROPIC_PROVIDER_NAME, ANTHROPIC_PROVIDER_ID);
 
         let start_event: Event = serde_json::from_value(serde_json::json!({
             "type": "content_block_start",
             "index": 0,
-            "content_block": { "type": "compaction", "content": null }
+            "content_block": { "type": "compaction", "content": "Summary " }
         }))
         .unwrap();
         let delta_event: Event = serde_json::from_value(serde_json::json!({
             "type": "content_block_delta",
             "index": 0,
-            "delta": { "type": "compaction_delta", "content": "Summary chunk" }
+            "delta": { "type": "compaction_delta", "content": "in " }
+        }))
+        .unwrap();
+        let second_delta_event: Event = serde_json::from_value(serde_json::json!({
+            "type": "content_block_delta",
+            "index": 0,
+            "delta": { "type": "compaction_delta", "content": "chunks" }
+        }))
+        .unwrap();
+        let stop_event: Event = serde_json::from_value(serde_json::json!({
+            "type": "content_block_stop",
+            "index": 0
         }))
         .unwrap();
 
         let mut events = Vec::new();
         events.extend(mapper.map_event(start_event));
         events.extend(mapper.map_event(delta_event));
+        events.extend(mapper.map_event(second_delta_event));
+        events.extend(mapper.map_event(stop_event));
         let events = events
             .into_iter()
             .collect::<Result<Vec<_>, _>>()
@@ -1070,14 +1276,186 @@ mod tests {
         assert_eq!(
             events,
             vec![
-                LanguageModelCompletionEvent::Compaction(CompactionContent::Summary {
-                    content: None
-                }),
-                LanguageModelCompletionEvent::Compaction(CompactionContent::Summary {
-                    content: Some("Summary chunk".into())
-                }),
+                LanguageModelCompletionEvent::Compaction(CompactionUpdate::Started),
+                LanguageModelCompletionEvent::Compaction(CompactionUpdate::SummaryDelta(
+                    "Summary ".into()
+                )),
+                LanguageModelCompletionEvent::Compaction(CompactionUpdate::SummaryDelta(
+                    "in ".into()
+                )),
+                LanguageModelCompletionEvent::Compaction(CompactionUpdate::SummaryDelta(
+                    "chunks".into()
+                )),
+                LanguageModelCompletionEvent::Compaction(CompactionUpdate::Finished(
+                    CompactedContext::Summary {
+                        content: "Summary in chunks".into(),
+                        provider_state: None,
+                    }
+                )),
+            ]
+        );
+    }
+
+    /// Mirrors the stream shape in Anthropic's SDK fixtures: the block starts
+    /// with both fields null, then a single delta carries the summary text
+    /// alongside the opaque `encrypted_content` that must be round-tripped.
+    #[test]
+    fn test_event_mapper_captures_encrypted_content_as_provider_state() {
+        let mut mapper = AnthropicEventMapper::new(ANTHROPIC_PROVIDER_NAME, ANTHROPIC_PROVIDER_ID);
+
+        let start_event: Event = serde_json::from_value(serde_json::json!({
+            "type": "content_block_start",
+            "index": 0,
+            "content_block": { "type": "compaction", "content": null, "encrypted_content": null }
+        }))
+        .unwrap();
+        let delta_event: Event = serde_json::from_value(serde_json::json!({
+            "type": "content_block_delta",
+            "index": 0,
+            "delta": {
+                "type": "compaction_delta",
+                "content": "Earlier conversation summarized.",
+                "encrypted_content": "opaque-compaction-payload"
+            }
+        }))
+        .unwrap();
+        let stop_event: Event = serde_json::from_value(serde_json::json!({
+            "type": "content_block_stop",
+            "index": 0
+        }))
+        .unwrap();
+
+        let mut events = Vec::new();
+        events.extend(mapper.map_event(start_event));
+        events.extend(mapper.map_event(delta_event));
+        events.extend(mapper.map_event(stop_event));
+        let mut events = events
+            .into_iter()
+            .collect::<Result<Vec<_>, _>>()
+            .expect("all events should map successfully");
+
+        let Some(LanguageModelCompletionEvent::Compaction(CompactionUpdate::Finished(
+            CompactedContext::Summary {
+                content,
+                provider_state: Some(state),
+            },
+        ))) = events.pop()
+        else {
+            panic!("expected a finished summary carrying provider state");
+        };
+        assert_eq!(content.as_ref(), "Earlier conversation summarized.");
+        assert_eq!(
+            provider_compaction_encrypted_content(&state, &ANTHROPIC_PROVIDER_ID)
+                .unwrap()
+                .as_deref(),
+            Some("opaque-compaction-payload")
+        );
+        assert_eq!(
+            provider_compaction_encrypted_content(
+                &state,
+                &LanguageModelProviderId::new("other-anthropic-backend")
+            )
+            .unwrap(),
+            None
+        );
+    }
+
+    /// A compaction block that closes without any content is Anthropic's
+    /// documented representation of a failed compaction, which the server
+    /// treats as a no-op. It must surface as `Failed` -- not as an error that
+    /// would kill the rest of the response.
+    #[test]
+    fn test_event_mapper_maps_null_content_compaction_to_failed() {
+        let mut mapper = AnthropicEventMapper::new(ANTHROPIC_PROVIDER_NAME, ANTHROPIC_PROVIDER_ID);
+        let start_event: Event = serde_json::from_value(serde_json::json!({
+            "type": "content_block_start",
+            "index": 0,
+            "content_block": { "type": "compaction", "content": null, "encrypted_content": null }
+        }))
+        .unwrap();
+        let stop_event: Event = serde_json::from_value(serde_json::json!({
+            "type": "content_block_stop",
+            "index": 0
+        }))
+        .unwrap();
+
+        assert_eq!(
+            mapper
+                .map_event(start_event)
+                .into_iter()
+                .collect::<Result<Vec<_>, _>>()
+                .unwrap(),
+            vec![LanguageModelCompletionEvent::Compaction(
+                CompactionUpdate::Started
+            )]
+        );
+        assert_eq!(
+            mapper
+                .map_event(stop_event)
+                .into_iter()
+                .collect::<Result<Vec<_>, _>>()
+                .unwrap(),
+            vec![LanguageModelCompletionEvent::Compaction(
+                CompactionUpdate::Failed
+            )]
+        );
+    }
+
+    #[test]
+    fn test_event_mapper_rejects_compaction_delta_before_start() {
+        let mut mapper = AnthropicEventMapper::new(ANTHROPIC_PROVIDER_NAME, ANTHROPIC_PROVIDER_ID);
+        let delta_event: Event = serde_json::from_value(serde_json::json!({
+            "type": "content_block_delta",
+            "index": 0,
+            "delta": { "type": "compaction_delta", "content": "Summary chunk" }
+        }))
+        .unwrap();
+
+        let error = mapper.map_event(delta_event).pop().unwrap().unwrap_err();
+
+        assert!(
+            error
+                .to_string()
+                .contains("compaction delta before starting")
+        );
+    }
+
+    #[test]
+    fn test_event_mapper_rejects_stream_end_with_unfinished_compaction() {
+        let mut mapper = AnthropicEventMapper::new(ANTHROPIC_PROVIDER_NAME, ANTHROPIC_PROVIDER_ID);
+        let start_event: Event = serde_json::from_value(serde_json::json!({
+            "type": "content_block_start",
+            "index": 0,
+            "content_block": { "type": "compaction", "content": "Summary " }
+        }))
+        .unwrap();
+        let stop_event: Event = serde_json::from_value(serde_json::json!({
+            "type": "message_stop"
+        }))
+        .unwrap();
+
+        let started = mapper
+            .map_event(start_event)
+            .into_iter()
+            .collect::<Result<Vec<_>, _>>()
+            .unwrap();
+        assert_eq!(
+            started,
+            vec![
+                LanguageModelCompletionEvent::Compaction(CompactionUpdate::Started),
+                LanguageModelCompletionEvent::Compaction(CompactionUpdate::SummaryDelta(
+                    "Summary ".into()
+                )),
             ]
         );
+
+        let error = mapper.map_event(stop_event).pop().unwrap().unwrap_err();
+
+        assert!(
+            error
+                .to_string()
+                .contains("without finishing its compaction summary")
+        );
     }
 
     #[test]
diff --git a/crates/language_model/src/language_model.rs b/crates/language_model/src/language_model.rs
index 2e060049d0b056..5760d071d02ea4 100644
--- a/crates/language_model/src/language_model.rs
+++ b/crates/language_model/src/language_model.rs
@@ -35,6 +35,16 @@ impl DisabledReason {
     }
 }
 
+/// The outcome of an explicit [`LanguageModel::compact`] request.
+#[derive(Debug, Clone, PartialEq)]
+pub struct CompactionResult {
+    /// The replacement context to persist and use in subsequent requests.
+    pub context: CompactedContext,
+    /// Token usage of the compaction request itself, as reported by the
+    /// provider.
+    pub usage: TokenUsage,
+}
+
 pub struct LanguageModelTextStream {
     pub message_id: Option<String>,
     pub stream: BoxStream<'static, Result<String, LanguageModelCompletionError>>,
@@ -131,6 +141,24 @@ pub trait LanguageModel: Send + Sync {
         false
     }
 
+    fn supports_explicit_compaction(&self) -> bool {
+        false
+    }
+
+    fn compact(
+        &self,
+        _request: LanguageModelRequest,
+        _cx: &AsyncApp,
+    ) -> BoxFuture<'static, Result<CompactionResult, LanguageModelCompletionError>> {
+        let provider = self.provider_name();
+        async move {
+            Err(LanguageModelCompletionError::Other(anyhow::anyhow!(
+                "{provider} does not support explicit compaction"
+            )))
+        }
+        .boxed()
+    }
+
     /// Whether this model supports images
     fn supports_images(&self) -> bool;
 
diff --git a/crates/language_model_core/src/language_model_core.rs b/crates/language_model_core/src/language_model_core.rs
index e514a0155a80f0..c5dfcc60c2a8ec 100644
--- a/crates/language_model_core/src/language_model_core.rs
+++ b/crates/language_model_core/src/language_model_core.rs
@@ -59,7 +59,21 @@ pub enum LanguageModelCompletionEvent {
     },
     ReasoningDetails(serde_json::Value),
     UsageUpdate(TokenUsage),
-    Compaction(CompactionContent),
+    Compaction(CompactionUpdate),
+}
+
+#[derive(Debug, PartialEq, Clone, Serialize, Deserialize)]
+pub enum CompactionUpdate {
+    /// A streamed response has started producing replacement context.
+    Started,
+    /// A chunk of a natural-language summary, suitable for incremental display.
+    SummaryDelta(Arc<str>),
+    /// The complete context to persist and use in subsequent requests.
+    Finished(CompactedContext),
+    /// The provider abandoned the compaction without producing replacement
+    /// context. This is a documented outcome, not a protocol error: the
+    /// conversation simply continues on the uncompacted transcript.
+    Failed,
 }
 
 impl LanguageModelCompletionEvent {
diff --git a/crates/language_model_core/src/request.rs b/crates/language_model_core/src/request.rs
index a661874b0eb329..68222eff967ed0 100644
--- a/crates/language_model_core/src/request.rs
+++ b/crates/language_model_core/src/request.rs
@@ -4,7 +4,8 @@ use serde::{Deserialize, Serialize};
 
 use crate::role::Role;
 use crate::{
-    LanguageModelToolUse, LanguageModelToolUseId, LanguageModelToolUseInput, SharedString,
+    LanguageModelProviderId, LanguageModelToolUse, LanguageModelToolUseId,
+    LanguageModelToolUseInput, SharedString,
 };
 
 /// Dimensions of a `LanguageModelImage`
@@ -262,19 +263,58 @@ pub enum MessageContent {
     Image(LanguageModelImage),
     ToolUse(LanguageModelToolUse),
     ToolResult(LanguageModelToolResult),
-    Compaction(CompactionContent),
+    Compaction(CompactedContext),
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, Eq, PartialEq, Hash)]
-pub enum CompactionContent {
-    Pending,
+pub enum CompactedContext {
     Summary {
-        content: Option<Arc<str>>,
-    },
-    Encrypted {
-        id: Option<Arc<str>>,
-        encrypted_content: Arc<str>,
+        content: Arc<str>,
+        /// Opaque state the producing backend needs round-tripped alongside
+        /// the summary (e.g. Anthropic's `encrypted_content`). `None` when the
+        /// summary stands alone.
+        #[serde(default)]
+        provider_state: Option<ProviderCompactionState>,
     },
+    ProviderState(ProviderCompactionState),
+}
+
+/// Opaque context produced by a provider's native compaction mechanism.
+///
+/// Only the provider identified by `provider_id` may interpret `payload`.
+/// `format` lets that provider evolve its representation without exposing it
+/// through the shared language model API.
+#[derive(Debug, Clone, Serialize, Deserialize, Eq, PartialEq, Hash)]
+pub struct ProviderCompactionState {
+    provider_id: LanguageModelProviderId,
+    format: SharedString,
+    payload: Arc<str>,
+}
+
+impl ProviderCompactionState {
+    pub fn new(
+        provider_id: LanguageModelProviderId,
+        format: impl Into<SharedString>,
+        payload: impl Into<Arc<str>>,
+    ) -> Self {
+        Self {
+            provider_id,
+            format: format.into(),
+            payload: payload.into(),
+        }
+    }
+
+    pub fn provider_id(&self) -> &LanguageModelProviderId {
+        &self.provider_id
+    }
+
+    pub fn format(&self) -> &str {
+        &self.format
+    }
+
+    pub fn payload(&self) -> &str {
+        &self.payload
+    }
 }
 
 impl MessageContent {
diff --git a/crates/language_models/src/provider/anthropic.rs b/crates/language_models/src/provider/anthropic.rs
index a3df6cfe88624c..8edb4998ffdd3e 100644
--- a/crates/language_models/src/provider/anthropic.rs
+++ b/crates/language_models/src/provider/anthropic.rs
@@ -616,6 +616,7 @@ impl LanguageModel for AnthropicModel {
             self.model.max_output_tokens,
             self.model.mode.clone(),
             AnthropicPromptCacheMode::Automatic,
+            &PROVIDER_ID,
         ) {
             Ok(request) => request,
             Err(error) => return async move { Err(error.into()) }.boxed(),
@@ -626,7 +627,7 @@ impl LanguageModel for AnthropicModel {
         let request = self.stream_completion(request, cx);
         let future = self.request_limiter.stream(async move {
             let response = request.await?;
-            Ok(AnthropicEventMapper::new(PROVIDER_NAME).map_stream(response))
+            Ok(AnthropicEventMapper::new(PROVIDER_NAME, PROVIDER_ID).map_stream(response))
         });
         async move { Ok(future.await?.boxed()) }.boxed()
     }
diff --git a/crates/language_models/src/provider/anthropic_compatible.rs b/crates/language_models/src/provider/anthropic_compatible.rs
index 4d31e596b2197a..c4970e699c1df9 100644
--- a/crates/language_models/src/provider/anthropic_compatible.rs
+++ b/crates/language_models/src/provider/anthropic_compatible.rs
@@ -446,6 +446,7 @@ impl LanguageModel for AnthropicCompatibleLanguageModel {
             self.model.max_output_tokens,
             self.model.mode.clone(),
             self.cache_mode,
+            &self.provider_id,
         ) {
             Ok(request) => request,
             Err(error) => return async move { Err(error.into()) }.boxed(),
@@ -455,9 +456,10 @@ impl LanguageModel for AnthropicCompatibleLanguageModel {
         }
         let completion_request = self.stream_completion(request, cx);
         let provider_name = self.provider_name.clone();
+        let provider_id = self.provider_id.clone();
         let future = self.request_limiter.stream(async move {
             let response = completion_request.await?;
-            Ok(AnthropicEventMapper::new(provider_name).map_stream(response))
+            Ok(AnthropicEventMapper::new(provider_name, provider_id).map_stream(response))
         });
         async move { Ok(future.await?.boxed()) }.boxed()
     }
diff --git a/crates/language_models/src/provider/bedrock.rs b/crates/language_models/src/provider/bedrock.rs
index 15a3d22fa9436e..f27b379baf77bf 100644
--- a/crates/language_models/src/provider/bedrock.rs
+++ b/crates/language_models/src/provider/bedrock.rs
@@ -1325,7 +1325,7 @@ struct MantleResponseEventMapper {
 impl MantleResponseEventMapper {
     fn new() -> Self {
         Self {
-            open_ai_mapper: OpenAiResponseEventMapper::new(),
+            open_ai_mapper: OpenAiResponseEventMapper::new(PROVIDER_ID),
             current_message: None,
             previous_message: None,
             pending_message_events: Vec::new(),
@@ -1898,7 +1898,7 @@ impl LanguageModel for BedrockMantleModel {
 
         match self.model.protocol() {
             MantleProtocol::Responses => {
-                let request = into_open_ai_response(
+                let request = match into_open_ai_response(
                     request,
                     &model_id,
                     self.model.supports_tools(),
@@ -1906,7 +1906,11 @@ impl LanguageModel for BedrockMantleModel {
                     max_output_tokens,
                     mantle_default_reasoning_effort(&self.model),
                     self.model.supports_thinking(),
-                );
+                    &PROVIDER_ID,
+                ) {
+                    Ok(request) => request,
+                    Err(error) => return async move { Err(error.into()) }.boxed(),
+                };
                 let completions = self.stream_response(request, cx);
                 async move {
                     let mapper = MantleResponseEventMapper::new();
@@ -3653,7 +3657,9 @@ mod tests {
             Some(MantleModel::Grok4_3.max_output_tokens()),
             mantle_default_reasoning_effort(&MantleModel::Grok4_3),
             MantleModel::Grok4_3.supports_thinking(),
-        );
+            &PROVIDER_ID,
+        )
+        .unwrap();
 
         assert_eq!(
             serde_json::to_value(&request).unwrap()["reasoning"],
@@ -3704,7 +3710,9 @@ mod tests {
             Some(128_000),
             Some(ReasoningEffort::Medium),
             false,
-        );
+            &PROVIDER_ID,
+        )
+        .unwrap();
 
         assert!(request.context_management.is_some());
         strip_unsupported_mantle_response_fields(&mut request);
diff --git a/crates/language_models/src/provider/copilot_chat.rs b/crates/language_models/src/provider/copilot_chat.rs
index 4074fab35c2453..f5d16161ba3ca4 100644
--- a/crates/language_models/src/provider/copilot_chat.rs
+++ b/crates/language_models/src/provider/copilot_chat.rs
@@ -368,6 +368,7 @@ impl LanguageModel for CopilotChatLanguageModel {
                         AnthropicModelMode::Default
                     },
                     AnthropicPromptCacheMode::Legacy,
+                    &PROVIDER_ID,
                 )?;
 
                 anthropic_request.temperature = None;
@@ -413,7 +414,7 @@ impl LanguageModel for CopilotChatLanguageModel {
                 request_limiter
                     .stream(async move {
                         let events = stream.await?;
-                        let mapper = AnthropicEventMapper::new(PROVIDER_NAME);
+                        let mapper = AnthropicEventMapper::new(PROVIDER_NAME, PROVIDER_ID);
                         Ok(mapper.map_stream(events).boxed())
                     })
                     .await
diff --git a/crates/language_models/src/provider/open_ai.rs b/crates/language_models/src/provider/open_ai.rs
index eabb0ed5032958..9b59c56577c25c 100644
--- a/crates/language_models/src/provider/open_ai.rs
+++ b/crates/language_models/src/provider/open_ai.rs
@@ -5,16 +5,19 @@ use futures::{FutureExt, StreamExt, future::BoxFuture};
 use gpui::{App, AppContext, AsyncApp, Context, Entity, SharedString, Task};
 use http_client::{CustomHeaders, HttpClient};
 use language_model::{
-    ApiKeyConfiguration, ApiKeyState, AuthenticateError, EnvVar, FastModeConfirmation, IconOrSvg,
-    LanguageModel, LanguageModelCompletionError, LanguageModelCompletionEvent,
-    LanguageModelEffortLevel, LanguageModelId, LanguageModelName, LanguageModelProvider,
-    LanguageModelProviderId, LanguageModelProviderName, LanguageModelProviderState,
-    LanguageModelRequest, LanguageModelToolChoice, OPEN_AI_PROVIDER_ID, OPEN_AI_PROVIDER_NAME,
-    ProviderSettingsView, RateLimiter, env_var,
+    ApiKeyConfiguration, ApiKeyState, AuthenticateError, CompactionResult, EnvVar,
+    FastModeConfirmation, IconOrSvg, LanguageModel, LanguageModelCompletionError,
+    LanguageModelCompletionEvent, LanguageModelEffortLevel, LanguageModelId, LanguageModelName,
+    LanguageModelProvider, LanguageModelProviderId, LanguageModelProviderName,
+    LanguageModelProviderState, LanguageModelRequest, LanguageModelToolChoice, OPEN_AI_PROVIDER_ID,
+    OPEN_AI_PROVIDER_NAME, ProviderSettingsView, RateLimiter, env_var,
 };
 use open_ai::{
     ResponseStreamEvent,
-    responses::{Request as ResponseRequest, StreamEvent as ResponsesStreamEvent, stream_response},
+    responses::{
+        CompactRequest, CompactedResponse, Request as ResponseRequest,
+        StreamEvent as ResponsesStreamEvent, compact_response, stream_response,
+    },
     stream_completion,
 };
 use settings::{OpenAiAvailableModel as AvailableModel, Settings, SettingsStore};
@@ -22,6 +25,7 @@ use std::sync::{Arc, LazyLock};
 use strum::IntoEnumIterator;
 use ui::IconName;
 
+use open_ai::completion::token_usage_from_response_usage;
 pub use open_ai::completion::{
     ChatCompletionMaxTokensParameter, OpenAiEventMapper, OpenAiResponseEventMapper, into_open_ai,
     into_open_ai_response,
@@ -425,6 +429,40 @@ impl OpenAiLanguageModel {
 
         async move { Ok(future.await?.boxed()) }.boxed()
     }
+
+    fn compact_response(
+        &self,
+        request: CompactRequest,
+        cx: &AsyncApp,
+    ) -> BoxFuture<'static, Result<CompactedResponse, LanguageModelCompletionError>> {
+        let http_client = self.http_client.clone();
+
+        let (api_key, api_url, extra_headers) = self.state.read_with(cx, |state, cx| {
+            let api_url = OpenAiLanguageModelProvider::api_url(cx);
+            let extra_headers = OpenAiLanguageModelProvider::settings(cx)
+                .custom_headers
+                .clone();
+            (state.api_key_state.key(&api_url), api_url, extra_headers)
+        });
+
+        let provider = PROVIDER_NAME;
+        let future = self.request_limiter.run(async move {
+            let Some(api_key) = api_key else {
+                return Err(LanguageModelCompletionError::NoApiKey { provider });
+            };
+            Ok(compact_response(
+                http_client.as_ref(),
+                provider.0.as_str(),
+                &api_url,
+                &api_key,
+                request,
+                &extra_headers,
+            )
+            .await?)
+        });
+
+        future.boxed()
+    }
 }
 
 impl LanguageModel for OpenAiLanguageModel {
@@ -499,6 +537,53 @@ impl LanguageModel for OpenAiLanguageModel {
         self.model.supports_compaction()
     }
 
+    fn supports_explicit_compaction(&self) -> bool {
+        self.model.supports_compaction()
+    }
+
+    fn compact(
+        &self,
+        mut request: LanguageModelRequest,
+        cx: &AsyncApp,
+    ) -> BoxFuture<'static, Result<CompactionResult, LanguageModelCompletionError>> {
+        if !self.supports_explicit_compaction() {
+            return async {
+                Err(LanguageModelCompletionError::Other(anyhow::anyhow!(
+                    "this OpenAI model does not support explicit compaction"
+                )))
+            }
+            .boxed();
+        }
+
+        normalize_open_ai_response_thinking_effort(&mut request, &self.model);
+        let request = match into_open_ai_response(
+            request,
+            self.model.id(),
+            self.model.supports_parallel_tool_calls(),
+            self.model.supports_prompt_cache_key(),
+            self.max_output_tokens(),
+            default_thinking_reasoning_effort(&self.model),
+            self.model
+                .supported_reasoning_efforts()
+                .contains(&open_ai::ReasoningEffort::None),
+            &OPEN_AI_PROVIDER_ID,
+        ) {
+            Ok(request) => request,
+            Err(error) => return async move { Err(error.into()) }.boxed(),
+        };
+        let request = request.into_compact_request();
+        let response = self.compact_response(request, cx);
+        async move {
+            let response = response.await?;
+            let usage = token_usage_from_response_usage(&response.usage);
+            let context = response
+                .into_compacted_context(OPEN_AI_PROVIDER_ID)
+                .map_err(LanguageModelCompletionError::Other)?;
+            Ok(CompactionResult { context, usage })
+        }
+        .boxed()
+    }
+
     fn supported_effort_levels(&self) -> Vec<LanguageModelEffortLevel> {
         supported_thinking_effort_levels(&self.model)
     }
@@ -538,7 +623,7 @@ impl LanguageModel for OpenAiLanguageModel {
         }
         if self.model.uses_responses_api() {
             normalize_open_ai_response_thinking_effort(&mut request, &self.model);
-            let request = into_open_ai_response(
+            let request = match into_open_ai_response(
                 request,
                 self.model.id(),
                 self.model.supports_parallel_tool_calls(),
@@ -548,10 +633,14 @@ impl LanguageModel for OpenAiLanguageModel {
                 self.model
                     .supported_reasoning_efforts()
                     .contains(&open_ai::ReasoningEffort::None),
-            );
+                &OPEN_AI_PROVIDER_ID,
+            ) {
+                Ok(request) => request,
+                Err(error) => return async move { Err(error.into()) }.boxed(),
+            };
             let completions = self.stream_response(request, cx);
             async move {
-                let mapper = OpenAiResponseEventMapper::new();
+                let mapper = OpenAiResponseEventMapper::new(OPEN_AI_PROVIDER_ID);
                 Ok(mapper.map_stream(completions.await?).boxed())
             }
             .boxed()
diff --git a/crates/language_models/src/provider/open_ai_compatible.rs b/crates/language_models/src/provider/open_ai_compatible.rs
index ff67344ac5aba6..6e9a30e91b7f13 100644
--- a/crates/language_models/src/provider/open_ai_compatible.rs
+++ b/crates/language_models/src/provider/open_ai_compatible.rs
@@ -445,7 +445,7 @@ impl LanguageModel for OpenAiCompatibleLanguageModel {
             .boxed()
         } else {
             disable_response_thinking_for_none_effort(&mut request, &self.model);
-            let request = into_open_ai_response(
+            let request = match into_open_ai_response(
                 request,
                 &self.model.name,
                 self.model.capabilities.parallel_tool_calls,
@@ -453,10 +453,15 @@ impl LanguageModel for OpenAiCompatibleLanguageModel {
                 self.max_output_tokens(),
                 default_thinking_reasoning_effort(&self.model),
                 supports_none_reasoning_effort(&self.model),
-            );
+                &self.provider_id,
+            ) {
+                Ok(request) => request,
+                Err(error) => return async move { Err(error.into()) }.boxed(),
+            };
             let completions = self.stream_response(request, cx);
+            let compaction_state_owner = self.provider_id.clone();
             async move {
-                let mapper = OpenAiResponseEventMapper::new();
+                let mapper = OpenAiResponseEventMapper::new(compaction_state_owner);
                 Ok(mapper.map_stream(completions.await?).boxed())
             }
             .boxed()
@@ -635,7 +640,9 @@ mod tests {
             model.max_output_tokens,
             default_thinking_reasoning_effort(&model),
             supports_none_reasoning_effort(&model),
-        );
+            &LanguageModelProviderId::new("test-compatible-provider"),
+        )
+        .unwrap();
         let serialized = serde_json::to_value(request).unwrap();
 
         assert_eq!(
@@ -664,7 +671,9 @@ mod tests {
             model.max_output_tokens,
             default_thinking_reasoning_effort(&model),
             supports_none_reasoning_effort(&model),
-        );
+            &LanguageModelProviderId::new("test-compatible-provider"),
+        )
+        .unwrap();
         let serialized = serde_json::to_value(request).unwrap();
 
         assert_eq!(serialized.get("reasoning"), None);
diff --git a/crates/language_models/src/provider/openai_subscribed.rs b/crates/language_models/src/provider/openai_subscribed.rs
index 4e2dee2075cac9..e67c9f3b65ba79 100644
--- a/crates/language_models/src/provider/openai_subscribed.rs
+++ b/crates/language_models/src/provider/openai_subscribed.rs
@@ -519,7 +519,7 @@ impl LanguageModel for OpenAiSubscribedLanguageModel {
         // The Codex backend rejects `max_output_tokens` (`Unsupported parameter`),
         // unlike the public OpenAI Responses API. Pass `None` so the field is
         // omitted from the serialized request body entirely.
-        let mut responses_request = into_open_ai_response(
+        let mut responses_request = match into_open_ai_response(
             request,
             self.model.id(),
             self.model.supports_parallel_tool_calls(),
@@ -529,26 +529,18 @@ impl LanguageModel for OpenAiSubscribedLanguageModel {
             self.model
                 .supported_reasoning_efforts()
                 .contains(&ReasoningEffort::None),
-        );
+            &PROVIDER_ID,
+        ) {
+            Ok(request) => request,
+            Err(error) => return async move { Err(error.into()) }.boxed(),
+        };
         responses_request.store = Some(false);
 
-        // The Codex backend requires system messages to be in the top-level
-        // `instructions` field rather than as input items.
-        let mut instructions = Vec::new();
-        responses_request.input.retain(|item| {
-            if let open_ai::responses::ResponseInputItem::Message(msg) = item {
-                if msg.role == open_ai::Role::System {
-                    for part in &msg.content {
-                        if let open_ai::responses::ResponseInputContent::Text { text } = part {
-                            instructions.push(text.clone());
-                        }
-                    }
-                    return false;
-                }
-            }
-            true
-        });
-        responses_request.instructions = Some(instructions.join("\n\n"));
+        // `into_open_ai_response` already hoists system messages into
+        // `instructions`, which is the only form the Codex backend accepts.
+        // Codex has only ever been sent requests with the field present
+        // (possibly empty), so keep sending it even without system messages.
+        responses_request.instructions.get_or_insert_default();
 
         let state = self.state.downgrade();
         let http_client = self.http_client.clone();
@@ -594,7 +586,7 @@ impl LanguageModel for OpenAiSubscribedLanguageModel {
         });
 
         async move {
-            let mapper = OpenAiResponseEventMapper::new();
+            let mapper = OpenAiResponseEventMapper::new(PROVIDER_ID);
             Ok(mapper.map_stream(future.await?.boxed()).boxed())
         }
         .boxed()
diff --git a/crates/language_models/src/provider/opencode.rs b/crates/language_models/src/provider/opencode.rs
index 6604995ffee14a..2a4e27c3c7ce78 100644
--- a/crates/language_models/src/provider/opencode.rs
+++ b/crates/language_models/src/provider/opencode.rs
@@ -678,6 +678,7 @@ impl LanguageModel for OpenCodeLanguageModel {
                         .unwrap_or(8192),
                     mode,
                     anthropic::completion::AnthropicPromptCacheMode::Automatic,
+                    &PROVIDER_ID,
                 ) {
                     Ok(request) => request,
                     Err(error) => return async move { Err(error.into()) }.boxed(),
@@ -685,7 +686,7 @@ impl LanguageModel for OpenCodeLanguageModel {
                 let stream =
                     self.stream_anthropic(anthropic_request, http_client, extra_headers, cx);
                 async move {
-                    let mapper = AnthropicEventMapper::new(PROVIDER_NAME);
+                    let mapper = AnthropicEventMapper::new(PROVIDER_NAME, PROVIDER_ID);
                     Ok(mapper.map_stream(stream.await?).boxed())
                 }
                 .boxed()
@@ -725,7 +726,7 @@ impl LanguageModel for OpenCodeLanguageModel {
                     .model
                     .supported_reasoning_effort_levels()
                     .is_some_and(|levels| levels.contains(&ReasoningEffort::None));
-                let response_request = into_open_ai_response(
+                let response_request = match into_open_ai_response(
                     request,
                     self.model.id(),
                     true,
@@ -733,11 +734,15 @@ impl LanguageModel for OpenCodeLanguageModel {
                     self.model.max_output_tokens(self.subscription),
                     None,
                     supports_none_reasoning_effort,
-                );
+                    &PROVIDER_ID,
+                ) {
+                    Ok(request) => request,
+                    Err(error) => return async move { Err(error.into()) }.boxed(),
+                };
                 let stream =
                     self.stream_openai_response(response_request, http_client, extra_headers, cx);
                 async move {
-                    let mapper = OpenAiResponseEventMapper::new();
+                    let mapper = OpenAiResponseEventMapper::new(PROVIDER_ID);
                     Ok(mapper.map_stream(stream.await?).boxed())
                 }
                 .boxed()
diff --git a/crates/language_models_cloud/Cargo.toml b/crates/language_models_cloud/Cargo.toml
index 2e408e66c7165d..571d8e86ebc6db 100644
--- a/crates/language_models_cloud/Cargo.toml
+++ b/crates/language_models_cloud/Cargo.toml
@@ -29,4 +29,6 @@ serde_json.workspace = true
 thiserror.workspace = true
 
 [dev-dependencies]
+gpui = { workspace = true, features = ["test-support"] }
+http_client = { workspace = true, features = ["test-support"] }
 language_model = { workspace = true, features = ["test-support"] }
diff --git a/crates/language_models_cloud/src/language_models_cloud.rs b/crates/language_models_cloud/src/language_models_cloud.rs
index 46060e92c7b21e..695d4324890d49 100644
--- a/crates/language_models_cloud/src/language_models_cloud.rs
+++ b/crates/language_models_cloud/src/language_models_cloud.rs
@@ -19,8 +19,8 @@ use http_client::{
     AsyncBody, HttpClient, HttpClientWithUrl, HttpRequestExt, Method, Response, StatusCode,
 };
 use language_model::{
-    ANTHROPIC_PROVIDER_ID, ANTHROPIC_PROVIDER_NAME, DisabledReason, GOOGLE_PROVIDER_ID,
-    GOOGLE_PROVIDER_NAME, LanguageModel, LanguageModelCompletionError,
+    ANTHROPIC_PROVIDER_ID, ANTHROPIC_PROVIDER_NAME, CompactionResult, DisabledReason,
+    GOOGLE_PROVIDER_ID, GOOGLE_PROVIDER_NAME, LanguageModel, LanguageModelCompletionError,
     LanguageModelCompletionEvent, LanguageModelEffortLevel, LanguageModelId, LanguageModelName,
     LanguageModelProviderId, LanguageModelProviderName, LanguageModelRequest,
     LanguageModelToolChoice, LanguageModelToolSchemaFormat, OPEN_AI_PROVIDER_ID,
@@ -43,7 +43,7 @@ use anthropic::completion::{AnthropicEventMapper, AnthropicPromptCacheMode, into
 use google_ai::completion::{GoogleEventMapper, into_google};
 use open_ai::completion::{
     ChatCompletionMaxTokensParameter, OpenAiEventMapper, OpenAiResponseEventMapper, into_open_ai,
-    into_open_ai_response,
+    into_open_ai_response, token_usage_from_response_usage,
 };
 
 const PROVIDER_ID: LanguageModelProviderId = ZED_CLOUD_PROVIDER_ID;
@@ -124,9 +124,49 @@ impl<TP: CloudLlmTokenProvider> CloudLanguageModel<TP> {
         auth_context: TP::AuthContext,
         app_version: Option<Version>,
         body: CompletionBody,
+    ) -> Result<PerformLlmCompletionResponse, LanguageModelCompletionError> {
+        Self::perform_llm_request(
+            "/completions",
+            true,
+            http_client,
+            token_provider,
+            auth_context,
+            app_version,
+            body,
+        )
+        .await
+    }
+
+    async fn perform_llm_compaction(
+        http_client: &HttpClientWithUrl,
+        token_provider: &TP,
+        auth_context: TP::AuthContext,
+        app_version: Option<Version>,
+        body: CompletionBody,
+    ) -> Result<PerformLlmCompletionResponse, LanguageModelCompletionError> {
+        Self::perform_llm_request(
+            "/completions/compact",
+            false,
+            http_client,
+            token_provider,
+            auth_context,
+            app_version,
+            body,
+        )
+        .await
+    }
+
+    async fn perform_llm_request(
+        path: &str,
+        request_status_messages: bool,
+        http_client: &HttpClientWithUrl,
+        token_provider: &TP,
+        auth_context: TP::AuthContext,
+        app_version: Option<Version>,
+        body: CompletionBody,
     ) -> Result<PerformLlmCompletionResponse, LanguageModelCompletionError> {
         let url = http_client
-            .build_zed_llm_url("/completions", &[])
+            .build_zed_llm_url(path, &[])
             .map_err(LanguageModelCompletionError::Other)?;
         let body = serde_json::to_string(&body).map_err(|error| {
             LanguageModelCompletionError::SerializeRequest {
@@ -136,17 +176,20 @@ impl<TP: CloudLlmTokenProvider> CloudLanguageModel<TP> {
         })?;
         let mut response =
             authenticated_llm_request(http_client, token_provider, auth_context, |token| {
-                Ok(http_client::Request::builder()
+                let mut request = http_client::Request::builder()
                     .method(Method::POST)
                     .uri(url.as_ref())
                     .when_some(app_version.as_ref(), |builder, app_version| {
                         builder.header(ZED_VERSION_HEADER_NAME, app_version.to_string())
                     })
                     .header("Content-Type", "application/json")
-                    .header("Authorization", format!("Bearer {token}"))
-                    .header(CLIENT_SUPPORTS_STATUS_MESSAGES_HEADER_NAME, "true")
-                    .header(CLIENT_SUPPORTS_STATUS_STREAM_ENDED_HEADER_NAME, "true")
-                    .body(body.clone().into())?)
+                    .header("Authorization", format!("Bearer {token}"));
+                if request_status_messages {
+                    request = request
+                        .header(CLIENT_SUPPORTS_STATUS_MESSAGES_HEADER_NAME, "true")
+                        .header(CLIENT_SUPPORTS_STATUS_STREAM_ENDED_HEADER_NAME, "true");
+                }
+                Ok(request.body(body.clone().into())?)
             })
             .await
             .map_err(|error| LanguageModelCompletionError::HttpSend {
@@ -156,10 +199,11 @@ impl<TP: CloudLlmTokenProvider> CloudLanguageModel<TP> {
 
         let status = response.status();
         if status.is_success() {
-            let includes_status_messages = response
-                .headers()
-                .get(SERVER_SUPPORTS_STATUS_MESSAGES_HEADER_NAME)
-                .is_some();
+            let includes_status_messages = request_status_messages
+                && response
+                    .headers()
+                    .get(SERVER_SUPPORTS_STATUS_MESSAGES_HEADER_NAME)
+                    .is_some();
 
             return Ok(PerformLlmCompletionResponse {
                 response,
@@ -376,6 +420,104 @@ impl<TP: CloudLlmTokenProvider + 'static> LanguageModel for CloudLanguageModel<T
         self.model.supports_server_side_compaction
     }
 
+    fn supports_explicit_compaction(&self) -> bool {
+        self.model.provider == cloud_llm_client::LanguageModelProvider::OpenAi
+            && self.model.supports_server_side_compaction
+    }
+
+    fn compact(
+        &self,
+        request: LanguageModelRequest,
+        cx: &AsyncApp,
+    ) -> BoxFuture<'static, Result<CompactionResult, LanguageModelCompletionError>> {
+        if !self.supports_explicit_compaction() {
+            return async {
+                Err(LanguageModelCompletionError::Other(anyhow::anyhow!(
+                    "this cloud model does not support explicit compaction"
+                )))
+            }
+            .boxed();
+        }
+
+        let thread_id = request.thread_id.clone();
+        let prompt_id = request.prompt_id.clone();
+        let app_version = self.app_version.clone();
+        let model_provider = self.model.provider;
+        let provider_name = provider_name(&self.model.provider);
+        let supports_none_reasoning_effort =
+            self.model.supported_effort_levels.iter().any(|effort| {
+                open_ai::ReasoningEffort::from_str(&effort.value)
+                    .is_ok_and(|effort| effort == open_ai::ReasoningEffort::None)
+            });
+        // Cloud proxies to OpenAI's own infrastructure, so the resulting
+        // compaction state is owned by (and interchangeable with) OpenAI
+        // proper, not by the cloud transport.
+        let request = match into_open_ai_response(
+            request,
+            &self.model.id.0,
+            self.model.supports_parallel_tool_calls,
+            true,
+            None,
+            None,
+            supports_none_reasoning_effort,
+            &OPEN_AI_PROVIDER_ID,
+        ) {
+            Ok(request) => request,
+            Err(error) => return async move { Err(error.into()) }.boxed(),
+        };
+        let compact_request = request.into_compact_request();
+        let http_client = self.http_client.clone();
+        let token_provider = self.token_provider.clone();
+        let auth_context = token_provider.auth_context(cx);
+        let future = self.request_limiter.run(async move {
+            let PerformLlmCompletionResponse {
+                response,
+                includes_status_messages,
+            } = Self::perform_llm_compaction(
+                &http_client,
+                &*token_provider,
+                auth_context,
+                app_version,
+                CompletionBody {
+                    thread_id,
+                    prompt_id,
+                    provider: model_provider,
+                    model: compact_request.model.clone(),
+                    provider_request: serde_json::to_value(compact_request).map_err(|error| {
+                        LanguageModelCompletionError::SerializeRequest {
+                            provider: provider_name.clone(),
+                            error,
+                        }
+                    })?,
+                },
+            )
+            .await?;
+
+            let events = response_lines::<open_ai::responses::CompactedResponse>(
+                response,
+                includes_status_messages,
+            );
+            futures::pin_mut!(events);
+            while let Some(event) = events.next().await {
+                match event.map_err(|error| error.into_completion_error(provider_name.clone()))? {
+                    CompletionEvent::Event(response) => {
+                        let usage = token_usage_from_response_usage(&response.usage);
+                        let context = response
+                            .into_compacted_context(OPEN_AI_PROVIDER_ID)
+                            .map_err(LanguageModelCompletionError::Other)?;
+                        return Ok(CompactionResult { context, usage });
+                    }
+                    CompletionEvent::Status(_) => {}
+                }
+            }
+
+            Err(LanguageModelCompletionError::StreamEndedUnexpectedly {
+                provider: provider_name,
+            })
+        });
+        future.boxed()
+    }
+
     fn supported_effort_levels(&self) -> Vec<LanguageModelEffortLevel> {
         self.model
             .supported_effort_levels
@@ -475,6 +617,10 @@ impl<TP: CloudLlmTokenProvider + 'static> LanguageModel for CloudLanguageModel<T
                         AnthropicModelMode::Default
                     },
                     AnthropicPromptCacheMode::Automatic,
+                    // Cloud proxies to Anthropic's own infrastructure, so
+                    // compaction state is owned by (and interchangeable with)
+                    // Anthropic proper, not by the cloud transport.
+                    &ANTHROPIC_PROVIDER_ID,
                 ) {
                     Ok(request) => request,
                     Err(error) => return async move { Err(error.into()) }.boxed(),
@@ -518,7 +664,8 @@ impl<TP: CloudLlmTokenProvider + 'static> LanguageModel for CloudLanguageModel<T
                     )
                     .await?;
 
-                    let mut mapper = AnthropicEventMapper::new(provider_name.clone());
+                    let mut mapper =
+                        AnthropicEventMapper::new(provider_name.clone(), ANTHROPIC_PROVIDER_ID);
                     Ok(map_cloud_completion_events(
                         Box::pin(response_lines(response, includes_status_messages)),
                         &provider_name,
@@ -541,7 +688,7 @@ impl<TP: CloudLlmTokenProvider + 'static> LanguageModel for CloudLanguageModel<T
                             .is_ok_and(|effort| effort == open_ai::ReasoningEffort::None)
                     });
 
-                let mut request = into_open_ai_response(
+                let mut request = match into_open_ai_response(
                     request,
                     &self.model.id.0,
                     self.model.supports_parallel_tool_calls,
@@ -549,7 +696,11 @@ impl<TP: CloudLlmTokenProvider + 'static> LanguageModel for CloudLanguageModel<T
                     None,
                     None,
                     supports_none_reasoning_effort,
-                );
+                    &OPEN_AI_PROVIDER_ID,
+                ) {
+                    Ok(request) => request,
+                    Err(error) => return async move { Err(error.into()) }.boxed(),
+                };
 
                 if enable_thinking && let Some(effort) = effort {
                     request.reasoning = Some(open_ai::responses::ReasoningConfig {
@@ -583,7 +734,7 @@ impl<TP: CloudLlmTokenProvider + 'static> LanguageModel for CloudLanguageModel<T
                     )
                     .await?;
 
-                    let mut mapper = OpenAiResponseEventMapper::new();
+                    let mut mapper = OpenAiResponseEventMapper::new(OPEN_AI_PROVIDER_ID);
                     Ok(map_cloud_completion_events(
                         Box::pin(response_lines(response, includes_status_messages)),
                         &provider_name,
@@ -969,8 +1120,167 @@ pub fn response_lines<T: DeserializeOwned>(
 #[cfg(test)]
 mod tests {
     use super::*;
+    use http_client::FakeHttpClient;
     use http_client::http::{HeaderMap, StatusCode};
-    use language_model::LanguageModelCompletionError;
+    use language_model::{
+        LanguageModelCompletionError, LanguageModelRequestMessage, MessageContent, Role, Speed,
+    };
+    use serde_json::json;
+    use std::sync::Mutex;
+
+    #[gpui::test]
+    async fn cloud_explicit_compaction_forwards_supported_request_fields(
+        cx: &mut gpui::TestAppContext,
+    ) {
+        let captured_request = Arc::new(Mutex::new(None));
+        let captured_request_for_handler = captured_request.clone();
+        let http_client = FakeHttpClient::create(move |request| {
+            let captured_request = captured_request_for_handler.clone();
+            async move {
+                let method = request.method().clone();
+                let uri = request.uri().to_string();
+                let authorization = request
+                    .headers()
+                    .get("Authorization")
+                    .and_then(|value| value.to_str().ok())
+                    .map(str::to_string);
+                let requested_status_messages = request
+                    .headers()
+                    .contains_key(CLIENT_SUPPORTS_STATUS_MESSAGES_HEADER_NAME);
+                let requested_stream_end = request
+                    .headers()
+                    .contains_key(CLIENT_SUPPORTS_STATUS_STREAM_ENDED_HEADER_NAME);
+                let mut body = request.into_body();
+                let mut body_text = String::new();
+                body.read_to_string(&mut body_text).await?;
+                *captured_request.lock().unwrap() = Some((
+                    method,
+                    uri,
+                    authorization,
+                    requested_status_messages,
+                    requested_stream_end,
+                    body_text,
+                ));
+
+                Ok(http_client::Response::builder()
+                    .status(200)
+                    .body(AsyncBody::from(format!(
+                        "{}\n",
+                        json!({
+                            "id": "resp_compact",
+                            "created_at": 1_700_000_000,
+                            "object": "response.compaction",
+                            "output": [{
+                                "type": "compaction",
+                                "id": "cmp_manual",
+                                "encrypted_content": "opaque-state"
+                            }],
+                            "usage": {
+                                "input_tokens": 100,
+                                "input_tokens_details": {"cached_tokens": 20},
+                                "output_tokens": 10,
+                                "output_tokens_details": {"reasoning_tokens": 5},
+                                "total_tokens": 110
+                            }
+                        })
+                    )))?)
+            }
+        });
+        let model = cloud_test_model(http_client);
+        let request = compact_test_request();
+
+        let result = model.compact(request, &cx.to_async()).await.unwrap();
+
+        assert_eq!(
+            result.usage,
+            language_model::TokenUsage {
+                input_tokens: 80,
+                output_tokens: 10,
+                cache_creation_input_tokens: 0,
+                cache_read_input_tokens: 20,
+            }
+        );
+        let language_model::CompactedContext::ProviderState(state) = result.context else {
+            panic!("expected provider compaction state");
+        };
+        assert_eq!(
+            open_ai::responses::provider_compaction_items(&state, &OPEN_AI_PROVIDER_ID).unwrap(),
+            Some(vec![json!({
+                "type": "compaction",
+                "id": "cmp_manual",
+                "encrypted_content": "opaque-state"
+            })])
+        );
+        let (method, uri, authorization, requested_status_messages, requested_stream_end, body) =
+            captured_request.lock().unwrap().take().unwrap();
+        assert_eq!(method, Method::POST);
+        assert_eq!(uri, "http://test.example/completions/compact?");
+        assert_eq!(authorization.as_deref(), Some("Bearer test-token"));
+        assert!(!requested_status_messages);
+        assert!(!requested_stream_end);
+        let body = serde_json::from_str::<serde_json::Value>(&body).unwrap();
+        assert_eq!(body["thread_id"], "thread-123");
+        assert_eq!(body["provider"], "open_ai");
+        assert_eq!(body["model"], "gpt-5.4");
+        assert_eq!(
+            body["provider_request"],
+            json!({
+                "model": "gpt-5.4",
+                "input": [{
+                    "type": "message",
+                    "role": "user",
+                    "content": [{
+                        "type": "input_text",
+                        "text": "Retain this context."
+                    }]
+                }],
+                "prompt_cache_key": "thread-123",
+                "service_tier": "priority"
+            })
+        );
+    }
+
+    #[gpui::test]
+    async fn cloud_explicit_compaction_rejects_output_without_compaction_item(
+        cx: &mut gpui::TestAppContext,
+    ) {
+        let http_client = FakeHttpClient::create(|_| async move {
+            Ok(http_client::Response::builder()
+                .status(200)
+                .body(AsyncBody::from(format!(
+                    "{}\n",
+                    json!({
+                        "id": "resp_compact",
+                        "created_at": 1_700_000_000,
+                        "object": "response.compaction",
+                        "output": [{
+                            "type": "message",
+                            "role": "assistant",
+                            "content": "This is not an opaque compaction item."
+                        }],
+                        "usage": {
+                            "input_tokens": 100,
+                            "input_tokens_details": {"cached_tokens": 20},
+                            "output_tokens": 10,
+                            "output_tokens_details": {"reasoning_tokens": 5},
+                            "total_tokens": 110
+                        }
+                    })
+                )))?)
+        });
+        let model = cloud_test_model(http_client);
+
+        let error = model
+            .compact(compact_test_request(), &cx.to_async())
+            .await
+            .unwrap_err();
+
+        assert!(
+            matches!(&error, LanguageModelCompletionError::Other(_)),
+            "expected invalid canonical output to be rejected, got {error:?}"
+        );
+        assert!(error.to_string().contains("compaction item"));
+    }
 
     #[test]
     fn test_api_error_conversion_with_upstream_http_error() {
@@ -1144,4 +1454,76 @@ mod tests {
             "Expected DeserializeResponse, got: {deserialize:?}"
         );
     }
+
+    fn compact_test_request() -> LanguageModelRequest {
+        LanguageModelRequest {
+            thread_id: Some("thread-123".to_string()),
+            messages: vec![LanguageModelRequestMessage {
+                role: Role::User,
+                content: vec![MessageContent::Text("Retain this context.".to_string())],
+                cache: false,
+                reasoning_details: None,
+            }],
+            speed: Some(Speed::Fast),
+            ..Default::default()
+        }
+    }
+
+    fn cloud_test_model(
+        http_client: Arc<HttpClientWithUrl>,
+    ) -> CloudLanguageModel<TestTokenProvider> {
+        CloudLanguageModel {
+            id: LanguageModelId::from("gpt-5.4".to_string()),
+            model: Arc::new(cloud_llm_client::LanguageModel {
+                provider: cloud_llm_client::LanguageModelProvider::OpenAi,
+                id: cloud_llm_client::LanguageModelId(Arc::from("gpt-5.4")),
+                display_name: "GPT-5.4".to_string(),
+                is_latest: true,
+                max_token_count: 1_000_000,
+                max_token_count_in_max_mode: None,
+                max_output_tokens: 128_000,
+                supports_tools: true,
+                supports_images: true,
+                supports_thinking: true,
+                supports_disabling_thinking: true,
+                supports_fast_mode: true,
+                supports_server_side_compaction: true,
+                supported_effort_levels: Vec::new(),
+                supports_streaming_tools: true,
+                supports_parallel_tool_calls: true,
+                is_disabled: false,
+                disabled_reason: None,
+            }),
+            token_provider: Arc::new(TestTokenProvider),
+            http_client,
+            app_version: None,
+            request_limiter: RateLimiter::new(4),
+        }
+    }
+
+    struct TestTokenProvider;
+
+    impl CloudLlmTokenProvider for TestTokenProvider {
+        type AuthContext = ();
+
+        fn auth_context(&self, _cx: &impl AppContext) -> Self::AuthContext {}
+
+        fn cached_token(
+            &self,
+            _auth_context: Self::AuthContext,
+        ) -> BoxFuture<'static, Result<String>> {
+            async { Ok("test-token".to_string()) }.boxed()
+        }
+
+        fn refresh_token(
+            &self,
+            _auth_context: Self::AuthContext,
+        ) -> BoxFuture<'static, Result<String>> {
+            async { Ok("refreshed-test-token".to_string()) }.boxed()
+        }
+
+        fn has_data_retention_consent(&self, _cx: &impl AppContext) -> bool {
+            false
+        }
+    }
 }
diff --git a/crates/open_ai/Cargo.toml b/crates/open_ai/Cargo.toml
index 5083e97c560147..8b89be9bada9b0 100644
--- a/crates/open_ai/Cargo.toml
+++ b/crates/open_ai/Cargo.toml
@@ -30,4 +30,5 @@ strum.workspace = true
 thiserror.workspace = true
 
 [dev-dependencies]
+http_client = { workspace = true, features = ["test-support"] }
 pretty_assertions.workspace = true
diff --git a/crates/open_ai/src/completion.rs b/crates/open_ai/src/completion.rs
index 18798a54b17ad7..8b8da6e657edaf 100644
--- a/crates/open_ai/src/completion.rs
+++ b/crates/open_ai/src/completion.rs
@@ -2,25 +2,26 @@ use anyhow::{Result, anyhow};
 use collections::HashMap;
 use futures::{Stream, StreamExt};
 use language_model_core::{
-    CompactionContent, LanguageModelCompletionError, LanguageModelCompletionEvent,
+    CompactedContext, CompactionUpdate, LanguageModelCompletionError, LanguageModelCompletionEvent,
     LanguageModelCustomToolFormat, LanguageModelCustomToolGrammarSyntax, LanguageModelImage,
-    LanguageModelRequest, LanguageModelRequestMessage, LanguageModelRequestToolInput,
-    LanguageModelToolChoice, LanguageModelToolResultContent, LanguageModelToolUse,
-    LanguageModelToolUseId, LanguageModelToolUseInput, MessageContent, Role, StopReason,
-    TokenUsage,
+    LanguageModelProviderId, LanguageModelRequest, LanguageModelRequestMessage,
+    LanguageModelRequestToolInput, LanguageModelToolChoice, LanguageModelToolResultContent,
+    LanguageModelToolUse, LanguageModelToolUseId, LanguageModelToolUseInput, MessageContent, Role,
+    StopReason, TokenUsage,
     util::{fix_streamed_json, is_context_window_exceeded_message, parse_tool_arguments},
 };
 use std::pin::Pin;
 use std::sync::Arc;
 
 use crate::responses::{
-    ContextManagement, Request as ResponseRequest, ResponseCompactionItem,
-    ResponseCustomToolCallItem, ResponseCustomToolCallOutputItem, ResponseError,
-    ResponseFunctionCallItem, ResponseFunctionCallOutputContent, ResponseFunctionCallOutputItem,
-    ResponseIncludable, ResponseInputContent, ResponseInputItem, ResponseMessageItem,
-    ResponseOutputItem, ResponseOutputMessage, ResponseReasoningInputItem, ResponseReasoningItem,
+    ContextManagement, Request as ResponseRequest, ResponseCustomToolCallItem,
+    ResponseCustomToolCallOutputItem, ResponseError, ResponseFunctionCallItem,
+    ResponseFunctionCallOutputContent, ResponseFunctionCallOutputItem, ResponseIncludable,
+    ResponseInputContent, ResponseInputItem, ResponseMessageItem, ResponseOutputItem,
+    ResponseOutputMessage, ResponseReasoningInputItem, ResponseReasoningItem,
     ResponseReasoningSummaryPart, ResponseSummary as ResponsesSummary,
     ResponseUsage as ResponsesUsage, StreamEvent as ResponsesStreamEvent,
+    provider_compaction_items, provider_compaction_state_from_items,
 };
 use crate::{
     FunctionContent, FunctionDefinition, ImageUrl, MessagePart, ReasoningEffort,
@@ -227,6 +228,11 @@ pub fn into_open_ai(
     })
 }
 
+/// `compaction_state_owner` identifies the backend this request will be sent
+/// to for the purpose of replaying provider-native compaction state. Multiple
+/// backends share this conversion, but their encrypted compaction items are
+/// not interchangeable, so only state owned by this id is replayed; state
+/// owned by any other backend falls back to replaying the full transcript.
 pub fn into_open_ai_response(
     request: LanguageModelRequest,
     model_id: &str,
@@ -235,7 +241,8 @@ pub fn into_open_ai_response(
     max_output_tokens: Option<u64>,
     default_reasoning_effort: Option<ReasoningEffort>,
     supports_none_reasoning_effort: bool,
-) -> ResponseRequest {
+    compaction_state_owner: &LanguageModelProviderId,
+) -> Result<ResponseRequest> {
     let stream = !model_id.starts_with("o1-");
 
     let LanguageModelRequest {
@@ -255,17 +262,38 @@ pub fn into_open_ai_response(
 
     let service_tier = service_tier_for(speed);
 
+    let mut provider_items = Vec::new();
     let mut input_items = Vec::new();
     let mut replayed_reasoning_item_indexes = HashMap::default();
     let mut tool_use_kinds_by_id = HashMap::default();
+    let mut system_instructions = Vec::new();
     for (index, message) in messages.into_iter().enumerate() {
+        // System messages go to the top-level `instructions` field rather
+        // than the input item list. `instructions` is applied per request
+        // (the Responses API documents it as a system message inserted into
+        // the model's context), so the current system prompt survives when
+        // replaying provider compaction state replaces the accumulated input
+        // items below. This also matches the Codex backend, which rejects
+        // system-role input items outright.
+        if message.role == Role::System {
+            for content in message.content {
+                if let MessageContent::Text(text) = content
+                    && !text.trim().is_empty()
+                {
+                    system_instructions.push(text);
+                }
+            }
+            continue;
+        }
         append_message_to_response_items(
             message,
             index,
+            compaction_state_owner,
             &mut replayed_reasoning_item_indexes,
             &mut tool_use_kinds_by_id,
+            &mut provider_items,
             &mut input_items,
-        );
+        )?;
     }
 
     let tools: Vec<_> = tools
@@ -328,10 +356,14 @@ pub fn into_open_ai_response(
         Vec::new()
     };
 
-    ResponseRequest {
+    Ok(ResponseRequest {
         model: model_id.into(),
-        instructions: None,
-        input: input_items,
+        instructions: if system_instructions.is_empty() {
+            None
+        } else {
+            Some(system_instructions.join("\n\n"))
+        },
+        input: crate::responses::ResponseInput::new(provider_items, input_items),
         store: Some(false),
         include,
         stream,
@@ -358,16 +390,18 @@ pub fn into_open_ai_response(
         service_tier,
         context_management: compact_at_tokens
             .map(|compact_threshold| vec![ContextManagement::Compaction { compact_threshold }]),
-    }
+    })
 }
 
 fn append_message_to_response_items(
     message: LanguageModelRequestMessage,
     index: usize,
+    compaction_state_owner: &LanguageModelProviderId,
     replayed_reasoning_item_indexes: &mut HashMap<String, usize>,
     tool_use_kinds_by_id: &mut HashMap<LanguageModelToolUseId, ReplayToolKind>,
+    provider_items: &mut Vec<serde_json::Value>,
     input_items: &mut Vec<ResponseInputItem>,
-) {
+) -> Result<()> {
     let mut content_parts: Vec<ResponseInputContent> = Vec::new();
 
     let LanguageModelRequestMessage {
@@ -396,27 +430,26 @@ fn append_message_to_response_items(
                 push_response_text_part(&role, text, &mut content_parts);
             }
             MessageContent::Thinking { .. } | MessageContent::RedactedThinking(_) => {}
-            MessageContent::Compaction(CompactionContent::Encrypted {
-                id,
-                encrypted_content,
-            }) => {
-                flush_response_parts(
-                    &role,
-                    index,
-                    phase.as_deref(),
-                    &mut content_parts,
-                    input_items,
-                );
-                input_items.push(ResponseInputItem::Compaction(ResponseCompactionItem {
-                    id,
-                    encrypted_content,
-                }));
+            MessageContent::Compaction(CompactedContext::ProviderState(state)) => {
+                // The canonical replacement window already encodes all
+                // context retained at the compaction point, so everything
+                // accumulated before it -- earlier input items, earlier parts
+                // of this same message, and replay bookkeeping -- is
+                // superseded and must not be resent. When a transcript
+                // contains multiple compactions, each later window supersedes
+                // the previous one, so the last compaction wins. State owned
+                // by another backend yields `None`, in which case we fall
+                // back to replaying the full transcript.
+                if let Some(items) = provider_compaction_items(&state, compaction_state_owner)? {
+                    content_parts.clear();
+                    input_items.clear();
+                    replayed_reasoning_item_indexes.clear();
+                    tool_use_kinds_by_id.clear();
+                    provider_items.clear();
+                    provider_items.extend(items);
+                }
             }
-            // Summary compaction blocks come from other providers, and a
-            // Pending block is a streaming-only UI signal; neither is replayed.
-            MessageContent::Compaction(
-                CompactionContent::Summary { .. } | CompactionContent::Pending,
-            ) => {}
+            MessageContent::Compaction(CompactedContext::Summary { .. }) => {}
             MessageContent::Image(image) => {
                 push_response_image_part(&role, image, &mut content_parts);
             }
@@ -514,6 +547,7 @@ fn append_message_to_response_items(
         &mut content_parts,
         input_items,
     );
+    Ok(())
 }
 
 #[derive(Clone, Copy)]
@@ -835,11 +869,15 @@ struct RawToolCall {
 }
 
 pub struct OpenAiResponseEventMapper {
+    /// The backend whose infrastructure produced this stream; stamped on any
+    /// compaction state it emits so replay is limited to the same backend.
+    compaction_state_owner: LanguageModelProviderId,
     function_calls_by_item: HashMap<String, PendingResponseFunctionCall>,
     custom_tool_calls_by_item: HashMap<String, PendingResponseCustomToolCall>,
     reasoning_items: Vec<ResponseReasoningInputItem>,
     current_message_phase: Option<String>,
     pending_stop_reason: Option<StopReason>,
+    pending_compaction_items: usize,
 }
 
 #[derive(Default)]
@@ -856,13 +894,15 @@ struct PendingResponseCustomToolCall {
 }
 
 impl OpenAiResponseEventMapper {
-    pub fn new() -> Self {
+    pub fn new(compaction_state_owner: LanguageModelProviderId) -> Self {
         Self {
+            compaction_state_owner,
             function_calls_by_item: HashMap::default(),
             custom_tool_calls_by_item: HashMap::default(),
             reasoning_items: Vec::new(),
             current_message_phase: None,
             pending_stop_reason: None,
+            pending_compaction_items: 0,
         }
     }
 
@@ -931,8 +971,9 @@ impl OpenAiResponseEventMapper {
                         }
                     }
                     ResponseOutputItem::Compaction(_) => {
+                        self.pending_compaction_items += 1;
                         events.push(Ok(LanguageModelCompletionEvent::Compaction(
-                            CompactionContent::Pending,
+                            CompactionUpdate::Started,
                         )));
                     }
                     ResponseOutputItem::Reasoning(_) | ResponseOutputItem::Unknown => {}
@@ -1114,12 +1155,20 @@ impl OpenAiResponseEventMapper {
                     }
                 }
                 ResponseOutputItem::Compaction(compaction) => {
-                    vec![Ok(LanguageModelCompletionEvent::Compaction(
-                        CompactionContent::Encrypted {
-                            id: compaction.id,
-                            encrypted_content: compaction.encrypted_content,
-                        },
-                    ))]
+                    self.pending_compaction_items = self.pending_compaction_items.saturating_sub(1);
+                    match serde_json::to_value(ResponseInputItem::Compaction(compaction))
+                        .map_err(anyhow::Error::from)
+                        .and_then(|item| {
+                            provider_compaction_state_from_items(
+                                self.compaction_state_owner.clone(),
+                                vec![item],
+                            )
+                        }) {
+                        Ok(state) => vec![Ok(LanguageModelCompletionEvent::Compaction(
+                            CompactionUpdate::Finished(CompactedContext::ProviderState(state)),
+                        ))],
+                        Err(error) => vec![Err(LanguageModelCompletionError::Other(error))],
+                    }
                 }
                 ResponseOutputItem::FunctionCall(_) | ResponseOutputItem::Unknown => Vec::new(),
             },
@@ -1140,6 +1189,16 @@ impl OpenAiResponseEventMapper {
         response: ResponsesSummary,
         default_reason: StopReason,
     ) -> Vec<Result<LanguageModelCompletionEvent, LanguageModelCompletionError>> {
+        // A compaction item that was added but never done means the server
+        // already pruned its context, but we never received the canonical
+        // replacement window. Continuing as if the turn succeeded would
+        // leave the conversation unable to continue coherently.
+        if self.pending_compaction_items > 0 {
+            return vec![Err(LanguageModelCompletionError::Other(anyhow!(
+                "response completed with an unfinished compaction item"
+            )))];
+        }
+
         let mut events = Vec::new();
 
         events.extend(self.capture_reasoning_items_from_output(&response.output));
@@ -1427,7 +1486,7 @@ fn response_content_is_refusal(content: &serde_json::Value) -> bool {
     content_type == Some("refusal") || !refusal.is_empty()
 }
 
-fn token_usage_from_response_usage(usage: &ResponsesUsage) -> TokenUsage {
+pub fn token_usage_from_response_usage(usage: &ResponsesUsage) -> TokenUsage {
     let cache_read_input_tokens = usage.input_tokens_details.cached_tokens;
 
     TokenUsage {
@@ -1479,7 +1538,8 @@ mod tests {
         LanguageModelCustomToolFormat, LanguageModelCustomToolGrammarSyntax, LanguageModelImage,
         LanguageModelRequestMessage, LanguageModelRequestTool, LanguageModelRequestToolInput,
         LanguageModelToolResult, LanguageModelToolResultContent, LanguageModelToolUse,
-        LanguageModelToolUseId, LanguageModelToolUseInput, SharedString, Speed,
+        LanguageModelToolUseId, LanguageModelToolUseInput, OPEN_AI_PROVIDER_ID, SharedString,
+        Speed,
     };
     use pretty_assertions::assert_eq;
     use serde_json::json;
@@ -1491,7 +1551,7 @@ mod tests {
 
     fn map_response_events(events: Vec<ResponsesStreamEvent>) -> Vec<LanguageModelCompletionEvent> {
         block_on(async {
-            OpenAiResponseEventMapper::new()
+            OpenAiResponseEventMapper::new(OPEN_AI_PROVIDER_ID)
                 .map_stream(Box::pin(futures::stream::iter(events.into_iter().map(Ok))))
                 .collect::<Vec<_>>()
                 .await
@@ -1828,19 +1888,15 @@ mod tests {
             Some(2048),
             Some(ReasoningEffort::Low),
             false,
-        );
+            &OPEN_AI_PROVIDER_ID,
+        )
+        .unwrap();
 
         let serialized = serde_json::to_value(&response).unwrap();
         let expected = json!({
             "model": "custom-model",
+            "instructions": "System context",
             "input": [
-                {
-                    "type": "message",
-                    "role": "system",
-                    "content": [
-                        { "type": "input_text", "text": "System context" }
-                    ]
-                },
                 {
                     "type": "message",
                     "role": "user",
@@ -2010,8 +2066,17 @@ mod tests {
             compact_at_tokens: None,
         };
 
-        let response =
-            into_open_ai_response(request, "custom-model", false, false, None, None, false);
+        let response = into_open_ai_response(
+            request,
+            "custom-model",
+            false,
+            false,
+            None,
+            None,
+            false,
+            &OPEN_AI_PROVIDER_ID,
+        )
+        .unwrap();
         let serialized = serde_json::to_value(response).unwrap();
         assert_eq!(
             serialized,
@@ -2110,7 +2175,9 @@ mod tests {
             None,
             Some(ReasoningEffort::Low),
             false,
-        );
+            &OPEN_AI_PROVIDER_ID,
+        )
+        .unwrap();
 
         let serialized = serde_json::to_value(&response).unwrap();
         assert_eq!(
@@ -2185,8 +2252,17 @@ mod tests {
             compact_at_tokens: None,
         };
 
-        let response =
-            into_open_ai_response(request, "custom-model", false, false, None, None, false);
+        let response = into_open_ai_response(
+            request,
+            "custom-model",
+            false,
+            false,
+            None,
+            None,
+            false,
+            &OPEN_AI_PROVIDER_ID,
+        )
+        .unwrap();
         let serialized = serde_json::to_value(&response).unwrap();
 
         assert_eq!(
@@ -2250,7 +2326,9 @@ mod tests {
             None,
             Some(ReasoningEffort::Medium),
             false,
-        );
+            &OPEN_AI_PROVIDER_ID,
+        )
+        .unwrap();
 
         let serialized = serde_json::to_value(&response).unwrap();
         assert_eq!(serialized.get("reasoning"), None);
@@ -2286,7 +2364,17 @@ mod tests {
                 compact_at_tokens: None,
             };
 
-            let response = into_open_ai_response(request, "gpt-5.4", true, true, None, None, true);
+            let response = into_open_ai_response(
+                request,
+                "gpt-5.4",
+                true,
+                true,
+                None,
+                None,
+                true,
+                &OPEN_AI_PROVIDER_ID,
+            )
+            .unwrap();
 
             let serialized = serde_json::to_value(&response)?;
             assert_eq!(
@@ -2420,7 +2508,9 @@ mod tests {
             None,
             Some(ReasoningEffort::Medium),
             true,
-        );
+            &OPEN_AI_PROVIDER_ID,
+        )
+        .unwrap();
 
         let serialized = serde_json::to_value(&response)?;
         assert_eq!(serialized["reasoning"], json!({ "effort": "none" }));
@@ -2459,7 +2549,9 @@ mod tests {
             None,
             Some(ReasoningEffort::Medium),
             true,
-        );
+            &OPEN_AI_PROVIDER_ID,
+        )
+        .unwrap();
 
         let serialized = serde_json::to_value(&response)?;
         assert_eq!(
@@ -2510,7 +2602,9 @@ mod tests {
             None,
             Some(ReasoningEffort::Medium),
             false,
-        );
+            &OPEN_AI_PROVIDER_ID,
+        )
+        .unwrap();
 
         let serialized = serde_json::to_value(&response).unwrap();
         assert_eq!(
@@ -2600,7 +2694,9 @@ mod tests {
             None,
             Some(ReasoningEffort::Medium),
             false,
-        );
+            &OPEN_AI_PROVIDER_ID,
+        )
+        .unwrap();
 
         let serialized = serde_json::to_value(&response).unwrap();
         assert_eq!(
@@ -2680,8 +2776,17 @@ mod tests {
             compact_at_tokens: None,
         };
 
-        let response =
-            into_open_ai_response(request, "custom-model", false, false, None, None, false);
+        let response = into_open_ai_response(
+            request,
+            "custom-model",
+            false,
+            false,
+            None,
+            None,
+            false,
+            &OPEN_AI_PROVIDER_ID,
+        )
+        .unwrap();
         let serialized = serde_json::to_value(&response).unwrap();
 
         assert_eq!(
@@ -2804,7 +2909,7 @@ mod tests {
 
     #[test]
     fn responses_stream_failed_uses_response_error_message() {
-        let mut mapper = OpenAiResponseEventMapper::new();
+        let mut mapper = OpenAiResponseEventMapper::new(OPEN_AI_PROVIDER_ID);
         let mapped = mapper.map_event(ResponsesStreamEvent::Failed {
             response: ResponseSummary {
                 status: Some("failed".into()),
@@ -2836,7 +2941,7 @@ mod tests {
         }))
         .expect("documented error event");
 
-        let mut mapper = OpenAiResponseEventMapper::new();
+        let mut mapper = OpenAiResponseEventMapper::new(OPEN_AI_PROVIDER_ID);
         let mapped = mapper.map_event(event);
 
         assert_eq!(mapped.len(), 1);
@@ -2861,7 +2966,7 @@ mod tests {
         }))
         .expect("nested error event");
 
-        let mut mapper = OpenAiResponseEventMapper::new();
+        let mut mapper = OpenAiResponseEventMapper::new(OPEN_AI_PROVIDER_ID);
         let mapped = mapper.map_event(event);
 
         assert_eq!(mapped.len(), 1);
@@ -2886,7 +2991,7 @@ mod tests {
         }))
         .expect("nested error event");
 
-        let mut mapper = OpenAiResponseEventMapper::new();
+        let mut mapper = OpenAiResponseEventMapper::new(OPEN_AI_PROVIDER_ID);
         let mapped = mapper.map_event(event);
 
         assert_eq!(mapped.len(), 1);
@@ -2899,7 +3004,7 @@ mod tests {
 
     #[test]
     fn responses_stream_maps_failed_context_length_exceeded_to_prompt_too_large() {
-        let mut mapper = OpenAiResponseEventMapper::new();
+        let mut mapper = OpenAiResponseEventMapper::new(OPEN_AI_PROVIDER_ID);
         let mapped = mapper.map_event(ResponsesStreamEvent::Failed {
             response: ResponseSummary {
                 status: Some("failed".into()),
@@ -2931,7 +3036,7 @@ mod tests {
         }))
         .expect("response error event");
 
-        let mut mapper = OpenAiResponseEventMapper::new();
+        let mut mapper = OpenAiResponseEventMapper::new(OPEN_AI_PROVIDER_ID);
         let mapped = mapper.map_event(event);
 
         assert_eq!(mapped.len(), 1);
@@ -3994,6 +4099,128 @@ mod tests {
         }));
     }
 
+    #[test]
+    fn into_open_ai_response_prepends_provider_input_unchanged() {
+        let provider_input = json!([
+            {
+                "type": "message",
+                "role": "user",
+                "content": "Retained user context.",
+                "provider_extension": {"preserve": true}
+            },
+            {
+                "type": "compaction",
+                "id": "cmp_manual",
+                "encrypted_content": "opaque-state"
+            }
+        ]);
+        let request = LanguageModelRequest {
+            messages: vec![
+                LanguageModelRequestMessage {
+                    role: Role::Assistant,
+                    content: vec![MessageContent::Compaction(CompactedContext::ProviderState(
+                        provider_compaction_state_from_items(
+                            OPEN_AI_PROVIDER_ID,
+                            provider_input.as_array().unwrap().clone(),
+                        )
+                        .unwrap(),
+                    ))],
+                    cache: false,
+                    reasoning_details: None,
+                },
+                LanguageModelRequestMessage {
+                    role: Role::User,
+                    content: vec![MessageContent::Text("Continue.".into())],
+                    cache: false,
+                    reasoning_details: None,
+                },
+            ],
+            ..Default::default()
+        };
+
+        let response = into_open_ai_response(
+            request,
+            "gpt-5.4",
+            true,
+            true,
+            None,
+            None,
+            false,
+            &OPEN_AI_PROVIDER_ID,
+        )
+        .unwrap();
+
+        assert_eq!(
+            serde_json::to_value(&response).unwrap()["input"],
+            json!([
+                {
+                    "type": "message",
+                    "role": "user",
+                    "content": "Retained user context.",
+                    "provider_extension": {"preserve": true}
+                },
+                {
+                    "type": "compaction",
+                    "id": "cmp_manual",
+                    "encrypted_content": "opaque-state"
+                },
+                {
+                    "type": "message",
+                    "role": "user",
+                    "content": [{"type": "input_text", "text": "Continue."}]
+                }
+            ])
+        );
+    }
+
+    #[test]
+    fn open_ai_response_converts_to_compact_request_with_supported_controls() {
+        let request = LanguageModelRequest {
+            thread_id: Some("thread-123".to_string()),
+            messages: vec![LanguageModelRequestMessage {
+                role: Role::User,
+                content: vec![MessageContent::Text("Retain this context.".into())],
+                cache: false,
+                reasoning_details: None,
+            }],
+            speed: Some(Speed::Fast),
+            compact_at_tokens: Some(100_000),
+            ..Default::default()
+        };
+
+        let mut response_request = into_open_ai_response(
+            request,
+            "gpt-5.4",
+            true,
+            true,
+            None,
+            None,
+            false,
+            &OPEN_AI_PROVIDER_ID,
+        )
+        .unwrap();
+        response_request.instructions = Some("Preserve implementation details.".to_string());
+        let compact_request = response_request.into_compact_request();
+
+        assert_eq!(
+            serde_json::to_value(compact_request).unwrap(),
+            json!({
+                "model": "gpt-5.4",
+                "instructions": "Preserve implementation details.",
+                "input": [{
+                    "type": "message",
+                    "role": "user",
+                    "content": [{
+                        "type": "input_text",
+                        "text": "Retain this context."
+                    }]
+                }],
+                "prompt_cache_key": "thread-123",
+                "service_tier": "priority"
+            })
+        );
+    }
+
     #[test]
     fn into_open_ai_response_maps_compact_at_tokens_to_context_management() {
         let request = LanguageModelRequest {
@@ -4007,7 +4234,17 @@ mod tests {
             ..Default::default()
         };
 
-        let response = into_open_ai_response(request, "gpt-5.1", true, true, None, None, false);
+        let response = into_open_ai_response(
+            request,
+            "gpt-5.1",
+            true,
+            true,
+            None,
+            None,
+            false,
+            &OPEN_AI_PROVIDER_ID,
+        )
+        .unwrap();
 
         assert_eq!(
             serde_json::to_value(&response).unwrap()["context_management"],
@@ -4027,7 +4264,17 @@ mod tests {
             ..Default::default()
         };
 
-        let response = into_open_ai_response(request, "gpt-5.1", true, true, None, None, false);
+        let response = into_open_ai_response(
+            request,
+            "gpt-5.1",
+            true,
+            true,
+            None,
+            None,
+            false,
+            &OPEN_AI_PROVIDER_ID,
+        )
+        .unwrap();
 
         assert!(
             serde_json::to_value(&response)
@@ -4038,15 +4285,21 @@ mod tests {
     }
 
     #[test]
-    fn into_open_ai_response_replays_encrypted_compaction_block() {
+    fn into_open_ai_response_replays_provider_compaction_block() {
+        let state = provider_compaction_state_from_items(
+            OPEN_AI_PROVIDER_ID,
+            vec![json!({
+                "type": "compaction",
+                "id": "cmp_1",
+                "encrypted_content": "encrypted-blob"
+            })],
+        )
+        .unwrap();
         let request = LanguageModelRequest {
             messages: vec![LanguageModelRequestMessage {
                 role: Role::Assistant,
                 content: vec![
-                    MessageContent::Compaction(CompactionContent::Encrypted {
-                        id: Some("cmp_1".into()),
-                        encrypted_content: "encrypted-blob".into(),
-                    }),
+                    MessageContent::Compaction(CompactedContext::ProviderState(state)),
                     MessageContent::Text("Done.".into()),
                 ],
                 cache: false,
@@ -4055,7 +4308,17 @@ mod tests {
             ..Default::default()
         };
 
-        let response = into_open_ai_response(request, "gpt-5.1", true, true, None, None, false);
+        let response = into_open_ai_response(
+            request,
+            "gpt-5.1",
+            true,
+            true,
+            None,
+            None,
+            false,
+            &OPEN_AI_PROVIDER_ID,
+        )
+        .unwrap();
 
         assert_eq!(
             serde_json::to_value(&response).unwrap()["input"],
@@ -4076,6 +4339,329 @@ mod tests {
         );
     }
 
+    #[test]
+    fn into_open_ai_response_hoists_system_messages_into_instructions() {
+        let request = LanguageModelRequest {
+            messages: vec![
+                LanguageModelRequestMessage {
+                    role: Role::System,
+                    content: vec![MessageContent::Text("You are a coding assistant.".into())],
+                    cache: false,
+                    reasoning_details: None,
+                },
+                LanguageModelRequestMessage {
+                    role: Role::User,
+                    content: vec![MessageContent::Text("Hello.".into())],
+                    cache: false,
+                    reasoning_details: None,
+                },
+                LanguageModelRequestMessage {
+                    role: Role::System,
+                    content: vec![MessageContent::Text("Prefer terse answers.".into())],
+                    cache: false,
+                    reasoning_details: None,
+                },
+            ],
+            ..Default::default()
+        };
+
+        let response = into_open_ai_response(
+            request,
+            "gpt-5.1",
+            true,
+            true,
+            None,
+            None,
+            false,
+            &OPEN_AI_PROVIDER_ID,
+        )
+        .unwrap();
+
+        let serialized = serde_json::to_value(&response).unwrap();
+        assert_eq!(
+            serialized["instructions"],
+            json!("You are a coding assistant.\n\nPrefer terse answers.")
+        );
+        assert_eq!(
+            serialized["input"],
+            json!([
+                {
+                    "type": "message",
+                    "role": "user",
+                    "content": [
+                        { "type": "input_text", "text": "Hello." }
+                    ]
+                }
+            ])
+        );
+    }
+
+    /// Replaying provider compaction state discards all input items
+    /// accumulated before it, but the system prompt must not be lost with
+    /// them: it lives in the per-request `instructions` field, outside the
+    /// compacted window, so the model keeps running on the current prompt
+    /// rather than whatever was frozen into the window at compaction time.
+    #[test]
+    fn into_open_ai_response_preserves_system_prompt_across_compaction_replay() {
+        let state = provider_compaction_state_from_items(
+            OPEN_AI_PROVIDER_ID,
+            vec![json!({
+                "type": "compaction",
+                "id": "cmp_1",
+                "encrypted_content": "encrypted-blob"
+            })],
+        )
+        .unwrap();
+        let request = LanguageModelRequest {
+            messages: vec![
+                LanguageModelRequestMessage {
+                    role: Role::System,
+                    content: vec![MessageContent::Text("Current system prompt.".into())],
+                    cache: false,
+                    reasoning_details: None,
+                },
+                LanguageModelRequestMessage {
+                    role: Role::User,
+                    content: vec![MessageContent::Text("Old context.".into())],
+                    cache: false,
+                    reasoning_details: None,
+                },
+                LanguageModelRequestMessage {
+                    role: Role::Assistant,
+                    content: vec![MessageContent::Compaction(CompactedContext::ProviderState(
+                        state,
+                    ))],
+                    cache: false,
+                    reasoning_details: None,
+                },
+                LanguageModelRequestMessage {
+                    role: Role::User,
+                    content: vec![MessageContent::Text("Continue.".into())],
+                    cache: false,
+                    reasoning_details: None,
+                },
+            ],
+            ..Default::default()
+        };
+
+        let response = into_open_ai_response(
+            request,
+            "gpt-5.1",
+            true,
+            true,
+            None,
+            None,
+            false,
+            &OPEN_AI_PROVIDER_ID,
+        )
+        .unwrap();
+
+        let serialized = serde_json::to_value(&response).unwrap();
+        assert_eq!(serialized["instructions"], json!("Current system prompt."));
+        assert_eq!(
+            serialized["input"],
+            json!([
+                {
+                    "type": "compaction",
+                    "id": "cmp_1",
+                    "encrypted_content": "encrypted-blob"
+                },
+                {
+                    "type": "message",
+                    "role": "user",
+                    "content": [
+                        { "type": "input_text", "text": "Continue." }
+                    ]
+                }
+            ])
+        );
+    }
+
+    #[test]
+    fn into_open_ai_response_rejects_malformed_provider_compaction_state() {
+        let request = LanguageModelRequest {
+            messages: vec![LanguageModelRequestMessage {
+                role: Role::Assistant,
+                content: vec![MessageContent::Compaction(CompactedContext::ProviderState(
+                    language_model_core::ProviderCompactionState::new(
+                        language_model_core::OPEN_AI_PROVIDER_ID,
+                        crate::responses::COMPACTION_STATE_FORMAT,
+                        "not valid JSON",
+                    ),
+                ))],
+                cache: false,
+                reasoning_details: None,
+            }],
+            ..Default::default()
+        };
+
+        let error = into_open_ai_response(
+            request,
+            "gpt-5.1",
+            true,
+            true,
+            None,
+            None,
+            false,
+            &OPEN_AI_PROVIDER_ID,
+        )
+        .unwrap_err();
+
+        assert!(
+            error
+                .to_string()
+                .contains("malformed OpenAI compaction state payload")
+        );
+    }
+
+    #[test]
+    fn into_open_ai_response_ignores_compaction_state_owned_by_another_backend() {
+        // Several backends share this request conversion (OpenAI itself,
+        // OpenAI-compatible endpoints, Codex, Mantle), but an encrypted
+        // compaction item is only decryptable by the backend that produced
+        // it. Replaying OpenAI-owned state through a different backend would
+        // discard the entire transcript in exchange for an opaque blob that
+        // backend cannot read, so the state must be ignored and the full
+        // transcript replayed instead.
+        let state = provider_compaction_state_from_items(
+            OPEN_AI_PROVIDER_ID,
+            vec![json!({
+                "type": "compaction",
+                "id": "cmp_1",
+                "encrypted_content": "encrypted-blob"
+            })],
+        )
+        .unwrap();
+        let request = LanguageModelRequest {
+            messages: vec![
+                LanguageModelRequestMessage {
+                    role: Role::User,
+                    content: vec![MessageContent::Text("Set up the project.".into())],
+                    cache: false,
+                    reasoning_details: None,
+                },
+                LanguageModelRequestMessage {
+                    role: Role::Assistant,
+                    content: vec![
+                        MessageContent::Compaction(CompactedContext::ProviderState(state)),
+                        MessageContent::Text("Done.".into()),
+                    ],
+                    cache: false,
+                    reasoning_details: None,
+                },
+                LanguageModelRequestMessage {
+                    role: Role::User,
+                    content: vec![MessageContent::Text("Continue.".into())],
+                    cache: false,
+                    reasoning_details: None,
+                },
+            ],
+            ..Default::default()
+        };
+
+        let response = into_open_ai_response(
+            request,
+            "compatible-model",
+            true,
+            true,
+            None,
+            None,
+            false,
+            &LanguageModelProviderId::new("my-compatible-endpoint"),
+        )
+        .unwrap();
+
+        assert_eq!(
+            serde_json::to_value(&response).unwrap()["input"],
+            json!([
+                {
+                    "type": "message",
+                    "role": "user",
+                    "content": [{ "type": "input_text", "text": "Set up the project." }]
+                },
+                {
+                    "type": "message",
+                    "role": "assistant",
+                    "content": [
+                        { "type": "output_text", "text": "Done.", "annotations": [] }
+                    ]
+                },
+                {
+                    "type": "message",
+                    "role": "user",
+                    "content": [{ "type": "input_text", "text": "Continue." }]
+                }
+            ])
+        );
+    }
+
+    #[test]
+    fn responses_stream_stamps_compaction_state_with_owning_backend() {
+        let owner = LanguageModelProviderId::new("my-compatible-endpoint");
+        let mut mapper = OpenAiResponseEventMapper::new(owner.clone());
+        let item: ResponseOutputItem = serde_json::from_value(json!({
+            "type": "compaction",
+            "id": "cmp_1",
+            "encrypted_content": "encrypted-blob"
+        }))
+        .unwrap();
+
+        let mut events = mapper.map_event(ResponsesStreamEvent::OutputItemDone {
+            output_index: 0,
+            sequence_number: None,
+            item,
+        });
+
+        let Some(Ok(LanguageModelCompletionEvent::Compaction(CompactionUpdate::Finished(
+            CompactedContext::ProviderState(state),
+        )))) = events.pop()
+        else {
+            panic!("expected finished provider compaction state");
+        };
+        assert_eq!(state.provider_id(), &owner);
+        assert!(
+            crate::responses::provider_compaction_items(&state, &owner)
+                .unwrap()
+                .is_some()
+        );
+        // OpenAI proper must not attempt to replay a window produced by a
+        // different backend's infrastructure.
+        assert_eq!(
+            crate::responses::provider_compaction_items(&state, &OPEN_AI_PROVIDER_ID).unwrap(),
+            None
+        );
+    }
+
+    #[test]
+    fn responses_stream_rejects_completion_with_unfinished_compaction() {
+        let mut mapper = OpenAiResponseEventMapper::new(OPEN_AI_PROVIDER_ID);
+        let item: ResponseOutputItem = serde_json::from_value(json!({
+            "type": "compaction",
+            "id": "cmp_1",
+            "encrypted_content": "encrypted-blob"
+        }))
+        .unwrap();
+
+        let started = mapper.map_event(ResponsesStreamEvent::OutputItemAdded {
+            output_index: 0,
+            sequence_number: None,
+            item,
+        });
+        assert!(matches!(
+            started.as_slice(),
+            [Ok(LanguageModelCompletionEvent::Compaction(
+                CompactionUpdate::Started
+            ))]
+        ));
+
+        let mut completed = mapper.map_event(ResponsesStreamEvent::Completed {
+            response: ResponseSummary::default(),
+        });
+        let error = completed.pop().unwrap().unwrap_err();
+
+        assert!(error.to_string().contains("unfinished compaction"));
+    }
+
     #[test]
     fn responses_stream_maps_compaction_output_item() {
         let item: ResponseOutputItem = serde_json::from_value(json!({
@@ -4102,11 +4688,20 @@ mod tests {
         assert_eq!(
             mapped,
             vec![
-                LanguageModelCompletionEvent::Compaction(CompactionContent::Pending),
-                LanguageModelCompletionEvent::Compaction(CompactionContent::Encrypted {
-                    id: Some("cmp_1".into()),
-                    encrypted_content: "encrypted-blob".into(),
-                }),
+                LanguageModelCompletionEvent::Compaction(CompactionUpdate::Started),
+                LanguageModelCompletionEvent::Compaction(CompactionUpdate::Finished(
+                    CompactedContext::ProviderState(
+                        provider_compaction_state_from_items(
+                            OPEN_AI_PROVIDER_ID,
+                            vec![json!({
+                                "type": "compaction",
+                                "id": "cmp_1",
+                                "encrypted_content": "encrypted-blob"
+                            })],
+                        )
+                        .unwrap()
+                    )
+                )),
             ]
         );
     }
diff --git a/crates/open_ai/src/responses.rs b/crates/open_ai/src/responses.rs
index f508a2746057da..dc2d6e8decb523 100644
--- a/crates/open_ai/src/responses.rs
+++ b/crates/open_ai/src/responses.rs
@@ -1,21 +1,26 @@
-use anyhow::{Result, anyhow};
+use anyhow::{Context as _, Result, anyhow};
 use futures::{AsyncBufReadExt, AsyncReadExt, StreamExt, io::BufReader, stream::BoxStream};
 use http_client::{
     AsyncBody, CustomHeaders, HttpClient, Method, Request as HttpRequest, RequestBuilderExt,
 };
-use serde::{Deserialize, Serialize};
+use serde::{Deserialize, Serialize, ser::SerializeSeq as _};
 use serde_json::Value;
 use std::sync::Arc;
 
 use crate::{ReasoningEffort, RequestError, Role, ServiceTier, ToolChoice};
+use language_model_core::{
+    CompactedContext, LanguageModelProviderId, ProviderCompactionState, SharedString,
+};
+
+pub const COMPACTION_STATE_FORMAT: &str = "openai.responses.input-items.v1";
 
 #[derive(Serialize, Debug)]
 pub struct Request {
     pub model: String,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub instructions: Option<String>,
-    #[serde(skip_serializing_if = "Vec::is_empty")]
-    pub input: Vec<ResponseInputItem>,
+    #[serde(skip_serializing_if = "ResponseInput::is_empty")]
+    pub input: ResponseInput,
     #[serde(skip_serializing_if = "Vec::is_empty")]
     pub include: Vec<ResponseIncludable>,
     #[serde(default)]
@@ -44,6 +49,158 @@ pub struct Request {
     pub context_management: Option<Vec<ContextManagement>>,
 }
 
+impl Request {
+    pub fn into_compact_request(self) -> CompactRequest {
+        CompactRequest {
+            model: self.model,
+            instructions: self.instructions,
+            input: self.input,
+            prompt_cache_key: self.prompt_cache_key,
+            service_tier: self.service_tier,
+        }
+    }
+}
+
+#[derive(Serialize, Debug)]
+pub struct CompactRequest {
+    pub model: String,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub instructions: Option<String>,
+    pub input: ResponseInput,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub prompt_cache_key: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub service_tier: Option<ServiceTier>,
+}
+
+#[derive(Deserialize, Debug)]
+pub struct CompactedResponse {
+    pub id: String,
+    pub created_at: u64,
+    pub object: String,
+    pub output: Vec<Value>,
+    pub usage: ResponseUsage,
+}
+
+impl CompactedResponse {
+    pub fn into_compacted_context(
+        self,
+        owner: LanguageModelProviderId,
+    ) -> Result<CompactedContext> {
+        Ok(CompactedContext::ProviderState(
+            provider_compaction_state_from_items(owner, self.output)?,
+        ))
+    }
+}
+
+/// Packages a canonical replacement window into opaque provider state owned by
+/// `owner`.
+///
+/// Several backends speak the OpenAI Responses protocol (OpenAI itself, Zed
+/// Cloud's OpenAI models, OpenAI-compatible endpoints, and others), but their
+/// encrypted compaction items are not interchangeable: only the backend that
+/// produced an item can decrypt it. The owner recorded here is what
+/// [`provider_compaction_items`] later compares against, so it must identify
+/// the backend whose infrastructure produced the items, not merely the wire
+/// protocol.
+pub fn provider_compaction_state_from_items(
+    owner: LanguageModelProviderId,
+    items: Vec<Value>,
+) -> Result<ProviderCompactionState> {
+    validate_compaction_items(&items)?;
+    Ok(ProviderCompactionState::new(
+        owner,
+        SharedString::new_static(COMPACTION_STATE_FORMAT),
+        serde_json::to_string(&items)?,
+    ))
+}
+
+/// Recovers the canonical replacement window from `state` if it is owned by
+/// `owner`, or `None` when the state belongs to a different backend and the
+/// caller should fall back to replaying the full transcript.
+pub fn provider_compaction_items(
+    state: &ProviderCompactionState,
+    owner: &LanguageModelProviderId,
+) -> Result<Option<Vec<Value>>> {
+    if state.provider_id() != owner {
+        return Ok(None);
+    }
+    if state.format() != COMPACTION_STATE_FORMAT {
+        return Err(anyhow!(
+            "unsupported OpenAI compaction state format: {}",
+            state.format()
+        ));
+    }
+
+    let items = serde_json::from_str::<Vec<Value>>(state.payload())
+        .context("malformed OpenAI compaction state payload")?;
+    validate_compaction_items(&items)?;
+    Ok(Some(items))
+}
+
+fn validate_compaction_items(items: &[Value]) -> Result<()> {
+    if items.is_empty() {
+        return Err(anyhow!("OpenAI returned an empty compaction output"));
+    }
+    if !items.iter().any(|item| {
+        item.get("type")
+            .and_then(Value::as_str)
+            .is_some_and(|item_type| item_type == "compaction")
+    }) {
+        return Err(anyhow!(
+            "OpenAI compaction output did not contain a compaction item"
+        ));
+    }
+    Ok(())
+}
+
+#[derive(Debug, Default)]
+pub struct ResponseInput {
+    provider_items: Vec<Value>,
+    generated_items: Vec<ResponseInputItem>,
+}
+
+impl ResponseInput {
+    pub fn new(provider_items: Vec<Value>, generated_items: Vec<ResponseInputItem>) -> Self {
+        Self {
+            provider_items,
+            generated_items,
+        }
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.provider_items.is_empty() && self.generated_items.is_empty()
+    }
+
+    /// Filters only the items this crate generated from the request.
+    ///
+    /// Provider items are a canonical replacement window that must be replayed
+    /// verbatim, so they are exempt from filtering. Callers that rewrite the
+    /// input to satisfy backend-specific requirements (and therefore can't
+    /// tolerate arbitrary items inside a replayed window) should not accept
+    /// provider-native compaction state in the first place.
+    pub fn retain(&mut self, predicate: impl FnMut(&ResponseInputItem) -> bool) {
+        self.generated_items.retain(predicate);
+    }
+}
+
+impl Serialize for ResponseInput {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        let mut sequence = serializer
+            .serialize_seq(Some(self.provider_items.len() + self.generated_items.len()))?;
+        for item in &self.provider_items {
+            sequence.serialize_element(item)?;
+        }
+        for item in &self.generated_items {
+            sequence.serialize_element(item)?;
+        }
+        sequence.end()
+    }
+}
+
 /// Server-side context management configuration.
 ///
 /// <https://developers.openai.com/api/docs/guides/compaction>
@@ -546,6 +703,45 @@ pub struct ResponseCustomToolCall {
     pub input: String,
 }
 
+pub async fn compact_response(
+    client: &dyn HttpClient,
+    provider_name: &str,
+    api_url: &str,
+    api_key: &str,
+    request: CompactRequest,
+    extra_headers: &CustomHeaders,
+) -> Result<CompactedResponse, RequestError> {
+    let request = HttpRequest::builder()
+        .method(Method::POST)
+        .uri(format!("{api_url}/responses/compact"))
+        .header("Content-Type", "application/json")
+        .header("Authorization", format!("Bearer {}", api_key.trim()))
+        .extra_headers(extra_headers)
+        .body(AsyncBody::from(
+            serde_json::to_string(&request).map_err(|error| RequestError::Other(error.into()))?,
+        ))
+        .map_err(|error| RequestError::Other(error.into()))?;
+
+    let mut response = client.send(request).await?;
+    let mut body = String::new();
+    response
+        .body_mut()
+        .read_to_string(&mut body)
+        .await
+        .map_err(|error| RequestError::Other(error.into()))?;
+
+    if response.status().is_success() {
+        serde_json::from_str(&body).map_err(|error| RequestError::Other(error.into()))
+    } else {
+        Err(RequestError::HttpResponseError {
+            provider: provider_name.to_owned(),
+            status_code: response.status(),
+            body,
+            headers: response.headers().clone(),
+        })
+    }
+}
+
 pub async fn stream_response(
     client: &dyn HttpClient,
     provider_name: &str,
@@ -732,3 +928,337 @@ pub async fn stream_response(
         })
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use futures::executor::block_on;
+    use http_client::FakeHttpClient;
+    use language_model_core::OPEN_AI_PROVIDER_ID;
+    use serde_json::json;
+    use std::sync::{Arc, Mutex};
+
+    #[test]
+    fn compact_response_posts_supported_request_fields() {
+        let captured_request = Arc::new(Mutex::new(None));
+        let captured_request_for_handler = captured_request.clone();
+        let http_client = FakeHttpClient::create(move |request| {
+            let captured_request = captured_request_for_handler.clone();
+            async move {
+                let method = request.method().clone();
+                let uri = request.uri().to_string();
+                let authorization = request
+                    .headers()
+                    .get("Authorization")
+                    .and_then(|value| value.to_str().ok())
+                    .map(str::to_string);
+                let mut body = request.into_body();
+                let mut body_text = String::new();
+                body.read_to_string(&mut body_text).await?;
+                *captured_request.lock().unwrap() = Some((method, uri, authorization, body_text));
+
+                Ok(http_client::Response::builder()
+                    .status(200)
+                    .body(AsyncBody::from(
+                        json!({
+                            "id": "resp_compact",
+                            "created_at": 1_700_000_000,
+                            "object": "response.compaction",
+                            "output": [{
+                                "type": "compaction",
+                                "id": "cmp_manual",
+                                "encrypted_content": "opaque-state"
+                            }],
+                            "usage": {
+                                "input_tokens": 100,
+                                "input_tokens_details": {"cached_tokens": 20},
+                                "output_tokens": 10,
+                                "output_tokens_details": {"reasoning_tokens": 5},
+                                "total_tokens": 110
+                            }
+                        })
+                        .to_string(),
+                    ))?)
+            }
+        });
+        let response = block_on(compact_response(
+            http_client.as_ref(),
+            "OpenAI",
+            "https://api.openai.com/v1",
+            "secret",
+            compact_test_request(),
+            &CustomHeaders::default(),
+        ))
+        .unwrap();
+
+        assert_eq!(
+            provider_compaction_items(
+                &match response
+                    .into_compacted_context(OPEN_AI_PROVIDER_ID)
+                    .unwrap()
+                {
+                    CompactedContext::ProviderState(state) => state,
+                    CompactedContext::Summary { .. } => panic!("expected provider state"),
+                },
+                &OPEN_AI_PROVIDER_ID
+            )
+            .unwrap(),
+            Some(vec![json!({
+                "type": "compaction",
+                "id": "cmp_manual",
+                "encrypted_content": "opaque-state"
+            })])
+        );
+        let (method, uri, authorization, body) = captured_request.lock().unwrap().take().unwrap();
+        assert_eq!(method, Method::POST);
+        assert_eq!(uri, "https://api.openai.com/v1/responses/compact");
+        assert_eq!(authorization.as_deref(), Some("Bearer secret"));
+        assert_eq!(
+            serde_json::from_str::<Value>(&body).unwrap(),
+            json!({
+                "model": "gpt-5.4",
+                "input": [{
+                    "type": "message",
+                    "role": "user",
+                    "content": [{
+                        "type": "input_text",
+                        "text": "Retain this context."
+                    }]
+                }],
+                "prompt_cache_key": "thread-123",
+                "service_tier": "priority"
+            })
+        );
+    }
+
+    #[test]
+    fn compact_response_reports_http_and_deserialization_errors() {
+        let http_client = FakeHttpClient::create(|_| async move {
+            Ok(http_client::Response::builder()
+                .status(429)
+                .header("retry-after", "5")
+                .body(AsyncBody::from("rate limited"))?)
+        });
+
+        let error = block_on(compact_response(
+            http_client.as_ref(),
+            "OpenAI",
+            "https://api.openai.com/v1",
+            "secret",
+            compact_test_request(),
+            &CustomHeaders::default(),
+        ))
+        .unwrap_err();
+
+        match error {
+            RequestError::HttpResponseError {
+                provider,
+                status_code,
+                body,
+                headers,
+            } => {
+                assert_eq!(provider, "OpenAI");
+                assert_eq!(status_code, 429);
+                assert_eq!(body, "rate limited");
+                assert_eq!(headers["retry-after"], "5");
+            }
+            error => panic!("expected an HTTP response error, got {error:?}"),
+        }
+
+        let http_client = FakeHttpClient::create(|_| async move {
+            Ok(http_client::Response::builder()
+                .status(200)
+                .body(AsyncBody::from("not valid JSON"))?)
+        });
+
+        let error = block_on(compact_response(
+            http_client.as_ref(),
+            "OpenAI",
+            "https://api.openai.com/v1",
+            "secret",
+            compact_test_request(),
+            &CustomHeaders::default(),
+        ))
+        .unwrap_err();
+
+        assert!(
+            matches!(error, RequestError::Other(_)),
+            "expected malformed JSON to produce a request error, got {error:?}"
+        );
+    }
+
+    #[test]
+    fn compacted_response_preserves_canonical_output_items() {
+        let output = vec![
+            json!({
+                "type": "message",
+                "role": "user",
+                "content": "Retained user context.",
+                "provider_extension": {"preserve": true}
+            }),
+            json!({
+                "type": "compaction",
+                "id": "cmp_manual",
+                "encrypted_content": "opaque-state"
+            }),
+        ];
+        let response: CompactedResponse = serde_json::from_value(json!({
+            "id": "resp_compact",
+            "created_at": 1_700_000_000,
+            "object": "response.compaction",
+            "output": &output,
+            "usage": {
+                "input_tokens": 100,
+                "input_tokens_details": {"cached_tokens": 20},
+                "output_tokens": 10,
+                "output_tokens_details": {"reasoning_tokens": 5},
+                "total_tokens": 110
+            }
+        }))
+        .unwrap();
+
+        let CompactedContext::ProviderState(state) = response
+            .into_compacted_context(OPEN_AI_PROVIDER_ID)
+            .unwrap()
+        else {
+            panic!("expected provider state");
+        };
+        assert_eq!(
+            provider_compaction_items(&state, &OPEN_AI_PROVIDER_ID).unwrap(),
+            Some(output)
+        );
+    }
+
+    #[test]
+    fn compacted_response_rejects_output_without_compaction_item() {
+        let response: CompactedResponse = serde_json::from_value(json!({
+            "id": "resp_compact",
+            "created_at": 1_700_000_000,
+            "object": "response.compaction",
+            "output": [{
+                "type": "message",
+                "role": "user",
+                "content": "Retained user context."
+            }],
+            "usage": {
+                "input_tokens": 100,
+                "input_tokens_details": {"cached_tokens": 20},
+                "output_tokens": 10,
+                "output_tokens_details": {"reasoning_tokens": 5},
+                "total_tokens": 110
+            }
+        }))
+        .unwrap();
+
+        assert!(
+            response
+                .into_compacted_context(OPEN_AI_PROVIDER_ID)
+                .unwrap_err()
+                .to_string()
+                .contains("compaction item")
+        );
+    }
+
+    #[test]
+    fn compacted_response_rejects_empty_output() {
+        let response: CompactedResponse = serde_json::from_value(json!({
+            "id": "resp_compact",
+            "created_at": 1_700_000_000,
+            "object": "response.compaction",
+            "output": [],
+            "usage": {
+                "input_tokens": 100,
+                "input_tokens_details": {"cached_tokens": 20},
+                "output_tokens": 10,
+                "output_tokens_details": {"reasoning_tokens": 5},
+                "total_tokens": 110
+            }
+        }))
+        .unwrap();
+
+        assert!(
+            response
+                .into_compacted_context(OPEN_AI_PROVIDER_ID)
+                .unwrap_err()
+                .to_string()
+                .contains("empty")
+        );
+    }
+
+    #[test]
+    fn provider_compaction_items_ignores_state_owned_by_another_provider() {
+        let items = vec![json!({
+            "type": "compaction",
+            "id": "cmp_manual",
+            "encrypted_content": "opaque-state"
+        })];
+        let state =
+            provider_compaction_state_from_items(OPEN_AI_PROVIDER_ID, items.clone()).unwrap();
+
+        assert_eq!(
+            provider_compaction_items(&state, &LanguageModelProviderId::new("anthropic")).unwrap(),
+            None
+        );
+
+        // The same window stamped for a different OpenAI-protocol backend is
+        // opaque to OpenAI proper: encrypted compaction items are only
+        // decryptable by the infrastructure that produced them.
+        let compatible_backend = LanguageModelProviderId::new("my-compatible-endpoint");
+        let state =
+            provider_compaction_state_from_items(compatible_backend.clone(), items).unwrap();
+        assert_eq!(
+            provider_compaction_items(&state, &OPEN_AI_PROVIDER_ID).unwrap(),
+            None
+        );
+        assert!(
+            provider_compaction_items(&state, &compatible_backend)
+                .unwrap()
+                .is_some()
+        );
+    }
+
+    #[test]
+    fn provider_compaction_items_rejects_unknown_open_ai_format() {
+        let state = ProviderCompactionState::new(
+            OPEN_AI_PROVIDER_ID,
+            "openai.responses.input-items.v2",
+            "[]",
+        );
+        assert!(
+            provider_compaction_items(&state, &OPEN_AI_PROVIDER_ID)
+                .unwrap_err()
+                .to_string()
+                .contains("unsupported OpenAI compaction state format")
+        );
+    }
+
+    #[test]
+    fn provider_compaction_items_rejects_malformed_open_ai_state() {
+        let state = ProviderCompactionState::new(
+            OPEN_AI_PROVIDER_ID,
+            COMPACTION_STATE_FORMAT,
+            "not valid JSON",
+        );
+
+        assert!(provider_compaction_items(&state, &OPEN_AI_PROVIDER_ID).is_err());
+    }
+
+    fn compact_test_request() -> CompactRequest {
+        CompactRequest {
+            model: "gpt-5.4".to_string(),
+            instructions: None,
+            input: ResponseInput::new(
+                Vec::new(),
+                vec![ResponseInputItem::Message(ResponseMessageItem {
+                    role: Role::User,
+                    content: vec![ResponseInputContent::Text {
+                        text: "Retain this context.".to_string(),
+                    }],
+                    phase: None,
+                })],
+            ),
+            prompt_cache_key: Some("thread-123".to_string()),
+            service_tier: Some(ServiceTier::Priority),
+        }
+    }
+}