diff --git a/crates/service/src/gateway/observability/http_bridge/stream_readers/chat_completions.rs b/crates/service/src/gateway/observability/http_bridge/stream_readers/chat_completions.rs
index 498501e8e..f2dd57b5c 100644
--- a/crates/service/src/gateway/observability/http_bridge/stream_readers/chat_completions.rs
+++ b/crates/service/src/gateway/observability/http_bridge/stream_readers/chat_completions.rs
@@ -413,10 +413,124 @@ impl ChatCompletionsFromResponsesSseReader {
         }
     }
 
+    // Never turn a failed or truncated upstream response into a successful chat turn.
+    // HTTP headers may already be sent, so propagate a standard SSE error object.
+    fn error_chunk(&mut self, error: Value, saw_terminal: bool) -> Vec<u8> {
+        if let Ok(mut collector) = self.usage_collector.lock() {
+            collector.saw_terminal |= saw_terminal;
+            if let Some(message) = error.get("message").and_then(Value::as_str) {
+                collector
+                    .terminal_error
+                    .get_or_insert_with(|| message.to_owned());
+            }
+        }
+        self.finished = true;
+        format!("data: {}\n\n", serde_json::json!({ "error": error })).into_bytes()
+    }
+
+    fn transport_error_chunk(&mut self, code: &str, message: String) -> Vec<u8> {
+        self.error_chunk(
+            serde_json::json!({
+                "type": "upstream_error", "code": code, "message": message, "param": null
+            }),
+            false,
+        )
+    }
+
+    fn is_response_failure_event(kind: &str) -> bool {
+        matches!(
+            kind.trim().to_ascii_lowercase().as_str(),
+            "error"
+                | "response.error"
+                | "response.failed"
+                | "response.incomplete"
+                | "response.cancelled"
+                | "response.canceled"
+        )
+    }
+
     fn handle_frame(&mut self, lines: &[String]) -> Option<Vec<u8>> {
-        let value = Self::data_json(lines)?;
+        let value = Self::data_json(lines).unwrap_or(Value::Null);
         self.remember_meta(&value);
         let event_type = Self::event_type(lines, &value);
+        if let Some(kind) = event_type.as_ref() {
+            if let Ok(mut collector) = self.usage_collector.lock() {
+                collector.last_event_type = Some(kind.clone());
+            }
+        }
+        let data_event_type = value.get("type").and_then(Value::as_str);
+        let failure_event = [event_type.as_deref(), data_event_type]
+            .into_iter()
+            .flatten()
+            .find(|kind| Self::is_response_failure_event(kind));
+        // Tool sub-events can carry their own error while the response continues.
+        // Only errors on the response envelope terminate the chat stream.
+        let is_response_sub_event = [event_type.as_deref(), data_event_type]
+            .into_iter()
+            .flatten()
+            .any(|kind| {
+                kind.starts_with("response.")
+                    && !Self::is_response_failure_event(kind)
+                    && !matches!(
+                        kind,
+                        "response.created"
+                            | "response.in_progress"
+                            | "response.completed"
+                            | "response.done"
+                    )
+            });
+        let upstream_error = [
+            value
+                .get("error")
+                .filter(|_| !is_response_sub_event || failure_event.is_some()),
+            value.pointer("/response/error"),
+            value.pointer("/response/status_details/error"),
+        ]
+        .into_iter()
+        .flatten()
+        .find(|error| !error.is_null())
+        .or_else(|| {
+            (event_type.as_deref() == Some("error") || data_event_type == Some("error"))
+                .then_some(&value)
+        });
+        let response_status = value.pointer("/response/status").and_then(Value::as_str);
+        if failure_event.is_some()
+            || upstream_error.is_some()
+            || matches!(
+                response_status,
+                Some("failed" | "incomplete" | "cancelled" | "canceled")
+            )
+        {
+            let incomplete = failure_event == Some("response.incomplete")
+                || response_status == Some("incomplete");
+            let default_code = if incomplete {
+                "upstream_response_incomplete"
+            } else {
+                "upstream_response_failed"
+            };
+            let detail = value
+                .pointer("/response/incomplete_details/reason")
+                .and_then(Value::as_str)
+                .or_else(|| upstream_error.and_then(Value::as_str))
+                .or(failure_event);
+            // Only forward scalar error contract fields, never response output or raw bodies.
+            let field = |name: &str| {
+                upstream_error
+                    .and_then(|error| error.get(name))
+                    .and_then(Value::as_str)
+                    .map(str::to_owned)
+            };
+            let error = serde_json::json!({
+                "type": field("type").filter(|kind| kind != "error").unwrap_or_else(|| "upstream_error".into()),
+                "code": field("code").unwrap_or_else(|| default_code.into()),
+                "message": field("message").unwrap_or_else(|| detail.unwrap_or(default_code).into()),
+                "param": field("param"),
+            });
+            return Some(self.error_chunk(error, true));
+        }
+        if value.is_null() {
+            return None;
+        }
         let is_terminal_event = matches!(
             event_type.as_deref(),
             Some("response.completed") | Some("response.done")
@@ -555,35 +669,31 @@ impl ChatCompletionsFromResponsesSseReader {
                     continue;
                 }
                 Ok(UpstreamSseFramePumpItem::Eof) => {
-                    if let Ok(mut collector) = self.usage_collector.lock() {
-                        if !collector.saw_terminal {
-                            let hint = collector.upstream_error_hint.clone();
-                            collector.terminal_error.get_or_insert_with(|| {
-                                upstream_hint_or_stream_incomplete_message(hint.as_deref())
-                            });
-                        }
-                    }
-                    self.finished = true;
-                    return Ok(Vec::new());
+                    let hint = self
+                        .usage_collector
+                        .lock()
+                        .ok()
+                        .and_then(|collector| collector.upstream_error_hint.clone());
+                    return Ok(self.transport_error_chunk(
+                        "upstream_stream_incomplete",
+                        upstream_hint_or_stream_incomplete_message(hint.as_deref()),
+                    ));
                 }
                 Ok(UpstreamSseFramePumpItem::Error(err)) => {
-                    if let Ok(mut collector) = self.usage_collector.lock() {
-                        collector
-                            .terminal_error
-                            .get_or_insert_with(|| classify_upstream_stream_read_error(&err));
-                    }
-                    self.finished = true;
-                    return Ok(Vec::new());
+                    let classified = classify_upstream_stream_read_error(&err);
+                    let message = if err.trim().is_empty() || classified == err {
+                        classified
+                    } else {
+                        format!("{classified}: {err}")
+                    };
+                    return Ok(self.transport_error_chunk("upstream_stream_read_error", message));
                 }
                 Err(std::sync::mpsc::RecvTimeoutError::Timeout) => {
                     if stream_idle_timed_out(self.last_upstream_activity) {
-                        if let Ok(mut collector) = self.usage_collector.lock() {
-                            collector
-                                .terminal_error
-                                .get_or_insert_with(stream_idle_timeout_message);
-                        }
-                        self.finished = true;
-                        return Ok(Vec::new());
+                        return Ok(self.transport_error_chunk(
+                            "upstream_stream_idle_timeout",
+                            stream_idle_timeout_message(),
+                        ));
                     }
                     if should_emit_keepalive_after_first_frame(self.saw_upstream_frame) {
                         return Ok(SseKeepAliveFrame::Comment.bytes().to_vec());
@@ -591,14 +701,15 @@ impl ChatCompletionsFromResponsesSseReader {
                     continue;
                 }
                 Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => {
-                    if let Ok(mut collector) = self.usage_collector.lock() {
-                        let hint = collector.upstream_error_hint.clone();
-                        collector.terminal_error.get_or_insert_with(|| {
-                            hint.unwrap_or_else(stream_reader_disconnected_message)
-                        });
-                    }
-                    self.finished = true;
-                    return Ok(Vec::new());
+                    let hint = self
+                        .usage_collector
+                        .lock()
+                        .ok()
+                        .and_then(|collector| collector.upstream_error_hint.clone());
+                    return Ok(self.transport_error_chunk(
+                        "upstream_stream_disconnected",
+                        hint.unwrap_or_else(stream_reader_disconnected_message),
+                    ));
                 }
             }
         }
@@ -627,6 +738,10 @@ impl crate::http::gateway_response_body::GatewayResponseBody
     }
 }
 
+#[cfg(test)]
+#[path = "chat_completions_repair_tests.rs"]
+mod repair_tests;
+
 #[cfg(test)]
 impl Read for ChatCompletionsFromResponsesSseReader {
     fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
diff --git a/crates/service/src/gateway/observability/http_bridge/stream_readers/chat_completions_repair_tests.rs b/crates/service/src/gateway/observability/http_bridge/stream_readers/chat_completions_repair_tests.rs
new file mode 100644
index 000000000..ec9a274bc
--- /dev/null
+++ b/crates/service/src/gateway/observability/http_bridge/stream_readers/chat_completions_repair_tests.rs
@@ -0,0 +1,234 @@
+use super::*;
+use std::io::{self, Cursor, Read};
+
+fn run_fixture(sse: &str) -> (String, PassthroughSseCollector) {
+    let collector = Arc::new(Mutex::new(PassthroughSseCollector::default()));
+    let mut reader = ChatCompletionsFromResponsesSseReader::from_pump(
+        UpstreamSseFramePump::from_reader(Cursor::new(sse.as_bytes().to_vec())),
+        Arc::clone(&collector),
+        Instant::now(),
+    );
+    let mut body = String::new();
+    reader.read_to_string(&mut body).unwrap();
+    let state = collector.lock().unwrap().clone();
+    (body, state)
+}
+
+fn error_from_body(body: &str) -> Value {
+    body.lines()
+        .filter_map(|line| line.strip_prefix("data: "))
+        .filter_map(|line| serde_json::from_str::<Value>(line).ok())
+        .find_map(|v| v.get("error").cloned())
+        .expect("downstream must receive an explicit error")
+}
+
+fn no_false_success(body: &str) {
+    assert!(!body.contains("\"finish_reason\":\"stop\""));
+    assert!(!body.contains("\"finish_reason\":\"tool_calls\""));
+    assert!(!body.contains("data: [DONE]"));
+}
+
+#[test]
+fn repair_eof_emits_error_without_fabricating_success() {
+    let (body, state) =
+        run_fixture("data: {\"type\":\"response.output_text.delta\",\"delta\":\"partial\"}\n\n");
+    assert_eq!(error_from_body(&body)["code"], "upstream_stream_incomplete");
+    assert!(state.terminal_error.is_some());
+    assert!(!state.saw_terminal);
+    assert_eq!(
+        state.last_event_type.as_deref(),
+        Some("response.output_text.delta")
+    );
+    no_false_success(&body);
+}
+
+#[test]
+fn repair_explicit_failure_preserves_code_param_and_message() {
+    let (body, state) = run_fixture("data: {\"type\":\"response.failed\",\"response\":{\"id\":\"resp_synthetic\",\"status\":\"failed\",\"error\":{\"code\":\"usage_limit_reached\",\"type\":\"server_error\",\"param\":\"input\",\"message\":\"Synthetic quota failure\"}}}\n\n");
+    let error = error_from_body(&body);
+    assert_eq!(error["code"], "usage_limit_reached");
+    assert_eq!(error["message"], "Synthetic quota failure");
+    assert_eq!(error["param"], "input");
+    assert!(state.saw_terminal);
+    assert_eq!(state.last_event_type.as_deref(), Some("response.failed"));
+    no_false_success(&body);
+}
+
+#[test]
+fn repair_incomplete_is_not_a_success() {
+    let (body, state) = run_fixture("data: {\"type\":\"response.incomplete\",\"response\":{\"status\":\"incomplete\",\"incomplete_details\":{\"reason\":\"max_output_tokens\"}}}\n\n");
+    assert_eq!(
+        error_from_body(&body)["code"],
+        "upstream_response_incomplete"
+    );
+    assert!(state.saw_terminal);
+    no_false_success(&body);
+}
+
+#[test]
+fn repair_valid_completion_keeps_stop_and_usage() {
+    let (body, state) = run_fixture("data: {\"type\":\"response.output_text.delta\",\"delta\":\"OK\"}\n\ndata: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\",\"usage\":{\"input_tokens\":1,\"output_tokens\":1,\"total_tokens\":2}}}\n\n");
+    assert!(body.contains("\"finish_reason\":\"stop\""));
+    assert!(body.contains("data: [DONE]"));
+    assert!(state.saw_terminal);
+    assert!(state.terminal_error.is_none());
+}
+
+fn assert_tool_failure_can_complete(event: &str) {
+    for error in [
+        Value::Null,
+        serde_json::json!({ "message": "Tool unavailable" }),
+    ] {
+        let tool_failure = serde_json::json!({
+            "type": event,
+            "item_id": "mcp_review",
+            "output_index": 0,
+            "sequence_number": 1,
+            "error": error,
+        });
+        for prefix in [String::new(), format!("event: {event}\n")] {
+            let sse = format!(
+                "{prefix}data: {tool_failure}\n\n\
+                 data: {{\"type\":\"response.output_text.delta\",\"delta\":\"The tool is unavailable; here is the fallback answer.\"}}\n\n\
+                 data: {{\"type\":\"response.completed\",\"response\":{{\"id\":\"resp_review\",\"status\":\"completed\",\"usage\":{{\"input_tokens\":1,\"output_tokens\":1,\"total_tokens\":2}}}}}}\n\n"
+            );
+            let (body, state) = run_fixture(&sse);
+            assert!(body.contains("The tool is unavailable; here is the fallback answer."));
+            assert!(body.contains("\"finish_reason\":\"stop\""));
+            assert!(body.contains("data: [DONE]"));
+            assert!(state.saw_terminal);
+            assert_eq!(state.last_event_type.as_deref(), Some("response.completed"));
+            assert_eq!(state.usage.total_tokens, Some(2));
+            assert!(state.terminal_error.is_none());
+        }
+    }
+}
+
+#[test]
+fn review_mcp_tool_failure_does_not_abort_completed_response() {
+    assert_tool_failure_can_complete("response.mcp_call.failed");
+}
+
+#[test]
+fn review_mcp_list_tools_failure_does_not_abort_completed_response() {
+    assert_tool_failure_can_complete("response.mcp_list_tools.failed");
+}
+
+struct BrokenReader;
+impl Read for BrokenReader {
+    fn read(&mut self, _: &mut [u8]) -> io::Result<usize> {
+        Err(io::Error::new(
+            io::ErrorKind::ConnectionReset,
+            "synthetic connection reset",
+        ))
+    }
+}
+
+#[test]
+fn repair_read_error_reaches_downstream() {
+    let collector = Arc::new(Mutex::new(PassthroughSseCollector::default()));
+    let mut reader = ChatCompletionsFromResponsesSseReader::from_pump(
+        UpstreamSseFramePump::from_reader(BrokenReader),
+        collector,
+        Instant::now(),
+    );
+    let mut body = String::new();
+    reader.read_to_string(&mut body).unwrap();
+    let error = error_from_body(&body);
+    assert_eq!(error["code"], "upstream_stream_read_error");
+    assert!(error["message"]
+        .as_str()
+        .unwrap()
+        .contains("synthetic connection reset"));
+    no_false_success(&body);
+}
+
+#[test]
+fn repair_top_level_error_fields_are_preserved() {
+    let (body, _) = run_fixture("event: error\ndata: {\"type\":\"error\",\"code\":\"upstream_503\",\"message\":\"Synthetic unavailable\",\"param\":null}\n\n");
+    let error = error_from_body(&body);
+    assert_eq!(error["code"], "upstream_503");
+    assert_eq!(error["message"], "Synthetic unavailable");
+    no_false_success(&body);
+}
+
+#[test]
+fn repair_done_marker_without_completed_is_not_success() {
+    let (body, state) = run_fixture("data: [DONE]\n\n");
+    assert_eq!(error_from_body(&body)["code"], "upstream_stream_incomplete");
+    assert!(!state.saw_terminal);
+    no_false_success(&body);
+}
+
+#[test]
+fn repair_terminal_error_variants_fail_closed() {
+    for sse in [
+        "event: error\ndata: not-json\n\n",
+        "event: response.cancelled\n\n",
+        "event: response.canceled\n\n",
+        "event: response.incomplete\n\n",
+        "event: response.error\ndata: not-json\n\n",
+        "event: response.failed\ndata: not-json\n\n",
+        "event: message\ndata: {\"type\":\"response.failed\"}\n\n",
+        "data: {\"type\":\"response.completed\",\"response\":{\"status\":\"failed\",\"error\":{\"message\":\"Synthetic failed despite event name\"}}}\n\n",
+        "data: {\"response\":{\"status\":\"incomplete\",\"status_details\":{\"error\":{\"code\":\"timeout\",\"message\":\"Synthetic upstream timeout\"}}}}\n\n",
+    ] {
+        let (body, state) = run_fixture(sse);
+        error_from_body(&body);
+        assert!(state.terminal_error.is_some());
+        no_false_success(&body);
+    }
+}
+
+#[test]
+fn repair_idle_timeout_reaches_downstream() {
+    let _guard = crate::test_env_guard();
+    struct Restore(u64);
+    impl Drop for Restore {
+        fn drop(&mut self) {
+            crate::gateway::set_upstream_stream_timeout_ms(self.0);
+        }
+    }
+    let _restore = Restore(crate::gateway::current_upstream_stream_timeout_ms());
+    crate::gateway::set_upstream_stream_timeout_ms(10);
+    let (_sender, receiver) = tokio::sync::mpsc::channel(1);
+    let collector = Arc::new(Mutex::new(PassthroughSseCollector::default()));
+    let mut reader = ChatCompletionsFromResponsesSseReader::from_pump(
+        UpstreamSseFramePump::from_stream(
+            crate::gateway::upstream::GatewayByteStream::from_receiver(receiver),
+        ),
+        collector,
+        Instant::now(),
+    );
+    let mut body = String::new();
+    reader.read_to_string(&mut body).unwrap();
+    assert_eq!(
+        error_from_body(&body)["code"],
+        "upstream_stream_idle_timeout"
+    );
+    no_false_success(&body);
+}
+
+// Optional local synthetic artefacts for exercising the real Pi parser. No credentials.
+#[test]
+fn repair_export_synthetic_wires() {
+    let Ok(dir) = std::env::var("CM_REPAIR_FIXTURE_DIR") else {
+        return;
+    };
+    let root = std::path::Path::new(&dir);
+    assert!(root.is_absolute());
+    std::fs::create_dir_all(root).unwrap();
+    let cases = [
+        ("eof", "data: {\"type\":\"response.output_text.delta\",\"delta\":\"partial\"}\n\n"),
+        ("failed", "data: {\"type\":\"response.failed\",\"response\":{\"status\":\"failed\",\"error\":{\"code\":\"usage_limit_reached\",\"type\":\"server_error\",\"message\":\"Synthetic quota failure\"}}}\n\n"),
+        ("complete", "data: {\"type\":\"response.output_text.delta\",\"delta\":\"SYNTHETIC_OK\"}\n\ndata: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\"}}\n\n"),
+        ("tool", "data: {\"type\":\"response.completed\",\"response\":{\"status\":\"completed\",\"output\":[{\"type\":\"function_call\",\"call_id\":\"call_fixture\",\"name\":\"read_fixture\",\"arguments\":\"{}\"}]}}\n\n"),
+    ];
+    for (name, sse) in cases {
+        let (body, state) = run_fixture(sse);
+        std::fs::write(root.join(format!("{name}.sse")), body).unwrap();
+        std::fs::write(root.join(format!("{name}.json")), serde_json::to_vec_pretty(&serde_json::json!({
+            "saw_terminal":state.saw_terminal,"terminal_error":state.terminal_error,"last_event_type":state.last_event_type
+        })).unwrap()).unwrap();
+    }
+}