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/converter.rs

@@ -637,6 +637,32 @@ impl std::fmt::Display for ConversionError {
 
 impl std::error::Error for ConversionError {}
 
+/// 由会话锚点派生稳定的 `agentContinuationId`
+///
+/// 原实现每个请求都 `Uuid::new_v4()`，于是同一个客户端会话的每一轮在上游看来都是
+/// 一条全新的 agent 任务线：多步任务（工具调用链、长任务续写）失去连续性。
+/// `conversationId` 已经从 `metadata.user_id` 的 session UUID 派生，这里用同一个
+/// 锚点派生 continuation，让「同一会话 = 同一条任务线」在上游成立。
+///
+/// 取 SHA-256 而不直接复用 conversationId：两个字段在上游是不同维度，直接相等会
+/// 让上游把「会话」和「任务线」当成同一个键；派生值保证一一对应但互不相等。
+/// domain 前缀防止与其它用途的摘要撞用途。
+fn derive_agent_continuation_id(conversation_anchor: &str) -> String {
+    let mut hasher = Sha256::new();
+    hasher.update(b"kiro-rs/agent-continuation/v1\0");
+    hasher.update(conversation_anchor.as_bytes());
+    let hex = format!("{:x}", hasher.finalize());
+    // 沿用 UUID 形状：上游只要求稳定字符串，UUID 形状便于日志辨识。
+    format!(
+        "{}-{}-{}-{}-{}",
+        &hex[..8],
+        &hex[8..12],
+        &hex[12..16],
+        &hex[16..20],
+        &hex[20..32]
+    )
+}
+
 /// 从 metadata.user_id 中提取 session UUID
 ///
 /// 支持两种格式:
@@ -747,13 +773,20 @@ pub fn convert_request_with_mode(
 
     // 3. 生成会话 ID 和代理 ID
     // 优先从 metadata.user_id 中提取 session UUID 作为 conversationId
-    let conversation_id = req
+    let session_anchor = req
         .metadata
         .as_ref()
         .and_then(|m| m.user_id.as_ref())
-        .and_then(|user_id| extract_session_id(user_id))
+        .and_then(|user_id| extract_session_id(user_id));
+    let conversation_id = session_anchor
+        .clone()
         .unwrap_or_else(|| Uuid::new_v4().to_string());
-    let agent_continuation_id = Uuid::new_v4().to_string();
+    // 有会话锚点时派生稳定的 continuation（同一会话 = 同一条 agent 任务线）；
+    // 拿不到锚点则退回随机值，保持旧行为——宁可丢连续性，不可乱绑任务线。
+    let agent_continuation_id = match session_anchor.as_deref() {
+        Some(anchor) => derive_agent_continuation_id(anchor),
+        None => Uuid::new_v4().to_string(),
+    };
 
     // 4. 确定触发类型
     let chat_trigger_type = determine_chat_trigger_type(req);
@@ -4018,4 +4051,131 @@ mod tests {
             matches!(err, ConversionError::UnsupportedDocument(message) if message.contains("URL source"))
         );
     }
+
+    // ---- 会话连续性：agentContinuationId 必须由会话锚点确定性派生 ----
+    //
+    // 原实现每请求一个随机 UUID，上游因此把同一客户端会话的每一轮当成新的 agent
+    // 任务线，多步任务（工具调用链）失去连续性。
+
+    #[test]
+    fn agent_continuation_id_is_stable_for_the_same_anchor() {
+        let anchor = "0b4445e1-f5be-49e1-87ce-62bbc28ad705";
+        assert_eq!(
+            derive_agent_continuation_id(anchor),
+            derive_agent_continuation_id(anchor),
+            "同一会话锚点必须派生出同一个 continuation"
+        );
+    }
+
+    #[test]
+    fn agent_continuation_id_differs_across_anchors() {
+        let a = derive_agent_continuation_id("0b4445e1-f5be-49e1-87ce-62bbc28ad705");
+        let b = derive_agent_continuation_id("0b4445e1-f5be-49e1-87ce-62bbc28ad706");
+        assert_ne!(a, b, "不同会话不得共用 agent 任务线");
+    }
+
+    #[test]
+    fn agent_continuation_id_is_not_equal_to_the_anchor() {
+        // conversationId 直接用锚点；continuation 必须是另一个值，否则上游会把
+        // 「会话」与「任务线」当成同一个键。
+        let anchor = "0b4445e1-f5be-49e1-87ce-62bbc28ad705";
+        assert_ne!(derive_agent_continuation_id(anchor), anchor);
+    }
+
+    #[test]
+    fn agent_continuation_id_keeps_uuid_shape() {
+        let id = derive_agent_continuation_id("0b4445e1-f5be-49e1-87ce-62bbc28ad705");
+        assert_eq!(id.len(), 36, "{}", id);
+        assert_eq!(id.chars().filter(|c| *c == '-').count(), 4, "{}", id);
+        assert!(
+            id.chars().all(|c| c.is_ascii_hexdigit() || c == '-'),
+            "{}",
+            id
+        );
+    }
+
+    #[test]
+    fn distinct_anchors_do_not_collide_in_bulk() {
+        let ids: std::collections::HashSet<String> = (0..2000)
+            .map(|i| derive_agent_continuation_id(&format!("session-{i}")))
+            .collect();
+        assert_eq!(ids.len(), 2000, "派生值发生碰撞");
+    }
+
+    /// 构造一个带 metadata.user_id 的最小请求，用于端到端验证转换结果。
+    fn request_with_user_id(user_id: Option<&str>) -> MessagesRequest {
+        MessagesRequest {
+            force_web_search_loop: false,
+            model: "claude-sonnet-4-5-20250929".to_string(),
+            max_tokens: 1024,
+            messages: vec![super::super::types::Message {
+                role: "user".to_string(),
+                content: serde_json::json!("hi"),
+            }],
+            stream: false,
+            system: None,
+            tools: None,
+            tool_choice: None,
+            thinking: None,
+            output_config: None,
+            metadata: Some(super::super::types::Metadata {
+                user_id: user_id.map(|s| s.to_string()),
+            }),
+        }
+    }
+
+    #[test]
+    fn same_client_session_reuses_conversation_and_continuation() {
+        let user_id = "user_abc_account__session_0b4445e1-f5be-49e1-87ce-62bbc28ad705";
+        let first = convert_request(&request_with_user_id(Some(user_id))).unwrap();
+        let second = convert_request(&request_with_user_id(Some(user_id))).unwrap();
+
+        let first_state = &first.conversation_state;
+        let second_state = &second.conversation_state;
+
+        assert_eq!(
+            first_state.conversation_id, second_state.conversation_id,
+            "同一会话的两轮必须复用 conversationId"
+        );
+        assert_eq!(
+            first_state.agent_continuation_id, second_state.agent_continuation_id,
+            "同一会话的两轮必须复用 agentContinuationId"
+        );
+    }
+
+    #[test]
+    fn distinct_client_sessions_do_not_share_continuation() {
+        let a = convert_request(&request_with_user_id(Some(
+            "user_abc_account__session_0b4445e1-f5be-49e1-87ce-62bbc28ad705",
+        )))
+        .unwrap();
+        let b = convert_request(&request_with_user_id(Some(
+            "user_abc_account__session_11111111-2222-3333-4444-555555555555",
+        )))
+        .unwrap();
+
+        assert_ne!(
+            a.conversation_state.conversation_id,
+            b.conversation_state.conversation_id
+        );
+        assert_ne!(
+            a.conversation_state.agent_continuation_id,
+            b.conversation_state.agent_continuation_id
+        );
+    }
+
+    #[test]
+    fn missing_session_anchor_falls_back_to_random_per_request() {
+        // 拿不到锚点时保持旧行为：每请求独立随机，宁可丢连续性也不乱绑任务线。
+        let first = convert_request(&request_with_user_id(None)).unwrap();
+        let second = convert_request(&request_with_user_id(None)).unwrap();
+        assert_ne!(
+            first.conversation_state.conversation_id,
+            second.conversation_state.conversation_id
+        );
+        assert_ne!(
+            first.conversation_state.agent_continuation_id,
+            second.conversation_state.agent_continuation_id
+        );
+    }
 }

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 读取失败：上游中途断流