fix(stream): report upstream interruption instead of faking success

.github/workflows/kiro-document-candidate.yml

@@ -5,6 +5,7 @@ on:
   push:
     branches:
       - feat/native-document-contract-20260725
+      - feat/stream-interrupt-no-fake-success-20260726
 
 concurrency:
   group: kiro-document-candidate-${{ github.ref }}
@@ -53,6 +54,7 @@ jobs:
           rustfmt --edition 2024 --check \
             src/anthropic/converter.rs \
             src/anthropic/handlers.rs \
+            src/anthropic/stream.rs \
             src/anthropic/types.rs \
             src/anthropic/websearch_loop.rs \
             src/kiro/model/requests/conversation.rs

src/anthropic/handlers.rs

@@ -33,7 +33,9 @@ use uuid::Uuid;
 
 use super::converter::{ConversionError, convert_request_with_mode};
 use super::middleware::{AppState, KeyContext};
-use super::stream::{BufferedStreamContext, SseEvent, StreamContext};
+use super::stream::{
+    BufferedStreamContext, STREAM_INTERRUPTED_CLIENT_MESSAGE, SseEvent, StreamContext,
+};
 use super::types::{
     CountTokensRequest, CountTokensResponse, ErrorResponse, MessagesRequest, Model, ModelsResponse,
     OutputConfig, Thinking,
@@ -1232,8 +1234,11 @@ fn create_sse_stream(
                         }
                         Some(Err(e)) => {
                             tracing::error!("读取响应流失败: {}", e);
-                            // 发送最终事件并结束（记为 error）
-                            let final_events = ctx.generate_final_events();
+                            // 上游断流 ≠ 正常收尾：只关闭未闭合的块并下发 error 事件，
+                            // 绝不补发 message_delta(stop_reason=end_turn) + message_stop，
+                            // 否则客户端会把半截响应当成一次成功完成的回合而不重试。
+                            let final_events =
+                                ctx.generate_interrupted_events(STREAM_INTERRUPTED_CLIENT_MESSAGE);
                             record_stream_usage(&hook, &ctx, credential_id, "error");
                             // 已开始返回内容后上游断流：标记为 interrupted，带已发送字节数
                             tracer.finalize(
@@ -2098,8 +2103,12 @@ fn create_buffered_sse_stream(
                             }
                             Some(Err(e)) => {
                                 tracing::error!("读取响应流失败: {}", e);
-                                // 发生错误，完成处理并返回所有事件
-                                let all_events = ctx.finish_and_get_all_events();
+                                // 上游断流：把已缓冲的事件连同 error 事件一起下发，
+                                // 但不补发 message_delta / message_stop（详见
+                                // BufferedStreamContext::interrupt_and_get_all_events）。
+                                let all_events = ctx.interrupt_and_get_all_events(
+                                    STREAM_INTERRUPTED_CLIENT_MESSAGE,
+                                );
                                 let (i, o, cc, cr, credits) = ctx.final_usage();
                                 hook.record(credential_id, i, o, cc, cr, credits, "error");
                                 // 缓冲模式 chunk 读取失败：上游中途断流

src/anthropic/stream.rs

@@ -22,6 +22,14 @@ use crate::kiro::model::events::Event;
 /// signature，因此该占位字符串只在客户端 ↔ kiro.rs 之间存在，不会影响转发。
 pub(super) const THINKING_SIGNATURE_PLACEHOLDER: &str = "kiro-rs-thinking-signature";
 
+/// 上游断流时下发给客户端的错误文案
+///
+/// 用 `overloaded_error`（Anthropic 官方错误类型之一）而不是伪造正常收尾：客户端
+/// SDK / Claude Code CLI 对该类型走重试路径，而 `stop_reason: end_turn` 会被当成
+/// 本轮已成功完成，半截的工具调用与多步任务就此静默中断。
+pub(super) const STREAM_INTERRUPTED_CLIENT_MESSAGE: &str =
+    "Upstream connection was interrupted before the response finished. Please retry.";
+
 const TOOL_USE_XML_PREFIX: &str = "<tool_use";
 const TOOL_USE_XML_CLOSE: &str = "</tool_use>";
 
@@ -1266,6 +1274,35 @@ impl SseStateManager {
         None
     }
 
+    /// 上游断流收尾：只关闭已打开的块，**绝不**发送 `message_delta` / `message_stop`。
+    ///
+    /// 与 [`Self::generate_final_events`] 的区别是本方法把消息标记为已结束但不发出
+    /// 正常收尾事件——正常收尾会带上 `stop_reason: end_turn`，客户端（Claude Code CLI）
+    /// 据此认为本轮已完整完成，于是把半截回答当成最终答案继续往下走。
+    /// 断流是失败，必须让客户端看到失败。
+    pub fn generate_interrupted_events(&mut self) -> Vec<SseEvent> {
+        let mut events = Vec::new();
+
+        for (index, block) in self.active_blocks.iter_mut() {
+            if block.started && !block.stopped {
+                events.push(SseEvent::new(
+                    "content_block_stop",
+                    json!({
+                        "type": "content_block_stop",
+                        "index": index
+                    }),
+                ));
+                block.stopped = true;
+            }
+        }
+
+        // 占位：后续任何 generate_final_events 调用都不得再补发正常收尾。
+        self.message_delta_sent = true;
+        self.message_ended = true;
+
+        events
+    }
+
     /// 生成最终事件序列
     pub fn generate_final_events(
         &mut self,
@@ -2347,6 +2384,33 @@ impl StreamContext {
         events
     }
 
+    /// 上游断流收尾：关闭未闭合的块，并补发 Anthropic `error` 事件，**不发**
+    /// `message_delta` / `message_stop`。
+    ///
+    /// 背景（Claude Code CLI 工具调用「不严格执行」的一个真因）：原实现在上游
+    /// 断流时也走 `generate_final_events()`，客户端收到 `stop_reason: end_turn`
+    /// + `message_stop`，把截断的半截响应当成一次**成功完成**的回合——工具调用
+    /// 参数写到一半就没了、多步任务在中途"正常"结束。伪造成功比明确报错更糟：
+    /// 客户端不会重试。改为下发 `overloaded_error`，让客户端走重试路径。
+    ///
+    /// 注意：此处**不做**任何 flush（thinking / invoke 嗅探 / XML 过滤器残留、
+    /// 工具 JSON 累积器收尾），因为那些残留本身就是不完整数据，flush 出去只会把
+    /// 半截内容伪装成有效内容。
+    pub fn generate_interrupted_events(&mut self, message: &str) -> Vec<SseEvent> {
+        let mut events = self.state_manager.generate_interrupted_events();
+        events.push(SseEvent::new(
+            "error",
+            json!({
+                "type": "error",
+                "error": {
+                    "type": "overloaded_error",
+                    "message": message
+                }
+            }),
+        ));
+        events
+    }
+
     /// 生成最终事件序列
     pub fn generate_final_events(&mut self) -> Vec<SseEvent> {
         let mut events = Vec::new();
@@ -2589,6 +2653,41 @@ impl BufferedStreamContext {
         std::mem::take(&mut self.event_buffer)
     }
 
+    /// 上游断流收尾（缓冲模式）：下发已缓冲的事件 + `error`，**不补**
+    /// `message_delta` / `message_stop`。
+    ///
+    /// 与 [`Self::finish_and_get_all_events`] 的区别：不做任何残留 flush、不做
+    /// `message_start` 的 usage 更正之外的收尾动作。缓冲模式此时一个字节都还没发给
+    /// 客户端，理论上可以整体丢弃，但保留已缓冲内容对排查更有用；关键是末尾必须是
+    /// `error` 而不是成功收尾，否则客户端把截断响应当成完整回合（详见
+    /// `StreamContext::generate_interrupted_events`）。
+    pub fn interrupt_and_get_all_events(&mut self, message: &str) -> Vec<SseEvent> {
+        if !self.initial_events_generated {
+            let initial_events = self.inner.generate_initial_events();
+            self.event_buffer.extend(initial_events);
+            self.initial_events_generated = true;
+        }
+
+        let (final_input_tokens, cache_creation, cache_read) = self.inner.resolved_usage();
+
+        let interrupted_events = self.inner.generate_interrupted_events(message);
+        self.event_buffer.extend(interrupted_events);
+
+        // 与成功路径一致：把 message_start 里的 usage 更正为真实口径。
+        for event in &mut self.event_buffer {
+            if event.event == "message_start"
+                && let Some(message) = event.data.get_mut("message")
+                && let Some(usage) = message.get_mut("usage")
+            {
+                usage["input_tokens"] = serde_json::json!(final_input_tokens);
+                usage["cache_creation_input_tokens"] = serde_json::json!(cache_creation);
+                usage["cache_read_input_tokens"] = serde_json::json!(cache_read);
+            }
+        }
+
+        std::mem::take(&mut self.event_buffer)
+    }
+
     /// 取出最终用量（在 finish_and_get_all_events 之后调用）
     ///
     /// 返回顺序：(input_tokens, output_tokens, cache_creation_tokens, cache_read_tokens, credits)
@@ -5099,4 +5198,172 @@ mod tests {
                 && e.data["content_block"]["data"] == "encrypted-thinking"
         }));
     }
+
+    // ---- 上游断流：必须报错，绝不伪造成功收尾 ----
+    //
+    // 背景：断流时若走正常收尾（message_delta stop_reason=end_turn + message_stop），
+    // Claude Code CLI 会把截断的半截响应当成一次完整成功的回合，于是不重试、
+    // 继续往下走（工具调用参数写到一半就没了、多步任务中途"正常"结束）。
+
+    fn has_normal_completion(events: &[SseEvent]) -> bool {
+        events
+            .iter()
+            .any(|e| e.event == "message_delta" || e.event == "message_stop")
+    }
+
+    fn overloaded_errors(events: &[SseEvent]) -> Vec<&SseEvent> {
+        events
+            .iter()
+            .filter(|e| e.event == "error" && e.data["error"]["type"] == "overloaded_error")
+            .collect()
+    }
+
+    #[test]
+    fn interrupted_stream_emits_error_and_no_normal_completion() {
+        let mut ctx = StreamContext::new_with_thinking(
+            "test-model",
+            10,
+            false,
+            HashMap::new(),
+            test_known_tools(),
+        );
+        let mut all = ctx.generate_initial_events();
+        all.extend(ctx.process_assistant_response("half an ans"));
+
+        let interrupted = ctx.generate_interrupted_events(STREAM_INTERRUPTED_CLIENT_MESSAGE);
+        all.extend(interrupted.clone());
+
+        assert!(
+            !has_normal_completion(&all),
+            "断流不得下发 message_delta / message_stop，否则客户端把截断当成功: {:?}",
+            all.iter().map(|e| &e.event).collect::<Vec<_>>()
+        );
+        assert_eq!(
+            overloaded_errors(&interrupted).len(),
+            1,
+            "断流必须下发恰好一个 overloaded_error 事件"
+        );
+        assert_eq!(
+            interrupted.last().map(|e| e.event.as_str()),
+            Some("error"),
+            "error 必须是最后一个事件"
+        );
+        // 已打开的 text 块要闭合，客户端才能干净地丢弃这一轮
+        assert!(
+            interrupted.iter().any(|e| e.event == "content_block_stop"),
+            "断流应关闭已打开的内容块"
+        );
+    }
+
+    #[test]
+    fn interrupted_stream_cannot_regress_to_fake_success() {
+        // 防回归：断流收尾之后，任何路径再调 generate_final_events 都不能补出
+        // 正常收尾事件（message_delta / message_stop 的幂等位必须已置位）。
+        let mut ctx = StreamContext::new_with_thinking(
+            "test-model",
+            10,
+            false,
+            HashMap::new(),
+            test_known_tools(),
+        );
+        let _ = ctx.generate_initial_events();
+        let _ = ctx.process_assistant_response("partial");
+        let _ = ctx.generate_interrupted_events(STREAM_INTERRUPTED_CLIENT_MESSAGE);
+
+        let after = ctx.generate_final_events();
+        assert!(
+            !has_normal_completion(&after),
+            "断流后不得再补发正常收尾: {:?}",
+            after.iter().map(|e| &e.event).collect::<Vec<_>>()
+        );
+    }
+
+    #[test]
+    fn interrupted_stream_does_not_flush_truncated_tool_json() {
+        // 上游在工具参数写到一半时断流：不得把半截 JSON 当成完整工具调用发出。
+        let mut ctx = StreamContext::new_with_thinking(
+            "test-model",
+            10,
+            false,
+            HashMap::new(),
+            test_known_tools(),
+        );
+        let mut all = ctx.generate_initial_events();
+        all.extend(ctx.process_tool_use(&tool_evt("t1", "Write", "{\"file_pa", false)));
+
+        let interrupted = ctx.generate_interrupted_events(STREAM_INTERRUPTED_CLIENT_MESSAGE);
+        all.extend(interrupted);
+
+        assert!(
+            !all.iter().any(|e| {
+                e.event == "content_block_start" && e.data["content_block"]["type"] == "tool_use"
+            }),
+            "半截工具调用不得作为 tool_use 块发出"
+        );
+        assert!(!has_normal_completion(&all));
+    }
+
+    #[test]
+    fn buffered_interrupted_stream_emits_error_and_no_normal_completion() {
+        let mut ctx =
+            BufferedStreamContext::new("test-model", 10, false, HashMap::new(), test_known_tools());
+        ctx.process_and_buffer(&Event::AssistantResponse(
+            serde_json::from_value(serde_json::json!({ "content": "half an ans" }))
+                .expect("assistantResponseEvent fixture"),
+        ));
+
+        let all = ctx.interrupt_and_get_all_events(STREAM_INTERRUPTED_CLIENT_MESSAGE);
+
+        assert!(
+            all.iter().any(|e| e.event == "message_start"),
+            "缓冲模式应保留已缓冲事件便于排查"
+        );
+        assert!(
+            !has_normal_completion(&all),
+            "缓冲模式断流同样不得下发 message_delta / message_stop: {:?}",
+            all.iter().map(|e| &e.event).collect::<Vec<_>>()
+        );
+        assert_eq!(overloaded_errors(&all).len(), 1);
+        assert_eq!(all.last().map(|e| e.event.as_str()), Some("error"));
+    }
+
+    #[test]
+    fn buffered_interrupted_stream_cannot_regress_to_fake_success() {
+        let mut ctx =
+            BufferedStreamContext::new("test-model", 10, false, HashMap::new(), test_known_tools());
+        ctx.process_and_buffer(&Event::AssistantResponse(
+            serde_json::from_value(serde_json::json!({ "content": "partial" }))
+                .expect("assistantResponseEvent fixture"),
+        ));
+        let _ = ctx.interrupt_and_get_all_events(STREAM_INTERRUPTED_CLIENT_MESSAGE);
+
+        let after = ctx.finish_and_get_all_events();
+        assert!(
+            !has_normal_completion(&after),
+            "缓冲模式断流后不得再补发正常收尾: {:?}",
+            after.iter().map(|e| &e.event).collect::<Vec<_>>()
+        );
+    }
+
+    #[test]
+    fn normal_completion_still_emits_message_stop() {
+        // 反向对照：正常收尾路径不受影响，仍必须发 message_delta + message_stop。
+        let mut ctx = StreamContext::new_with_thinking(
+            "test-model",
+            10,
+            false,
+            HashMap::new(),
+            test_known_tools(),
+        );
+        let mut all = ctx.generate_initial_events();
+        all.extend(ctx.process_assistant_response("a complete answer"));
+        all.extend(ctx.generate_final_events());
+
+        assert!(
+            all.iter()
+                .any(|e| e.event == "message_delta" && e.data["delta"]["stop_reason"] == "end_turn")
+        );
+        assert!(all.iter().any(|e| e.event == "message_stop"));
+        assert!(overloaded_errors(&all).is_empty());
+    }
 }