refactor(ai): SSE 读取内核收敛与 NDJSON 喂入纯函数+单测（观察 2026-09-05-2）

Author: Aparencia

app/src-tauri/src/ai_chat_stream.rs

@@ -174,19 +174,35 @@ pub fn stream_sse_content(
         .set("Authorization", &format!("Bearer {}", client.config.api_key.trim()))
         .send_string(&payload.to_string())
         .map_err(map_status)?;
-    let reader = BufReader::new(resp.into_reader());
+    let (content, usage_json, cancelled, completed) =
+        read_sse_lines(BufReader::new(resp.into_reader()), None, |d| emit(d));
+    Ok(StreamOutcome { content, usage_json, cancelled, completed })
+}
+
+/// 通用 SSE 读取内核（观察 2026-09-05-2：收敛与 stream_chat 高度同构的循环——
+/// 断流行 completed=false，调用方不得当成功；cancel 可选供聊天路径复用）。
+fn read_sse_lines(
+    reader: impl std::io::BufRead,
+    cancel: Option<&CancelFlag>,
+    mut on_delta: impl FnMut(&str),
+) -> (String, Option<String>, bool, bool) {
     let mut content = String::new();
     let mut usage_json: Option<String> = None;
+    let mut cancelled = false;
     let mut completed = false;
     for line in reader.lines() {
+        if cancel.is_some_and(|c| c.is_cancelled()) {
+            cancelled = true;
+            break;
+        }
         let line = match line {
             Ok(l) => l,
             Err(_) => break, // 传输中途断流：completed=false——调用方不得当成功
         };
         match parse_sse_line(&line) {
             SseEvent::Delta(d) => {
                 content.push_str(&d);
-                emit(&d);
+                on_delta(&d);
             }
             SseEvent::Done => {
                 completed = true;
@@ -199,7 +215,7 @@ pub fn stream_sse_content(
             }
         }
     }
-    Ok(StreamOutcome { content, usage_json, cancelled: false, completed })
+    (content, usage_json, cancelled, completed)
 }
 
 /// HTTP 状态 → AiClientError（与 post_completions 同归一口径——四下一致）。

app/src-tauri/src/ai_note_refine.rs

@@ -216,8 +216,6 @@ impl AiNoteRefineAdapter {
         dims: Option<&ResolvedDims>,
         mut on_section: impl FnMut(crate::ai_refine_protocol::AiRefineSection),
     ) -> Result<AiRefineResponse, AiClientError> {
-        use crate::ai_refine_protocol::parse_section_ndjson_line;
-
         let mut system = self.prompt.build_system(&request.profile, dims);
         system.push_str("\n\n");
         system.push_str(crate::ai_refine_protocol::NDJSON_SYSTEM_SUFFIX);
@@ -250,20 +248,21 @@ impl AiNoteRefineAdapter {
             &self.client,
             payload,
             |delta| {
-                pending.push_str(delta);
-                while let Some(pos) = pending.find('\n') {
-                    let line: String = pending.drain(..=pos).collect();
-                    if let Some(sec) = parse_section_ndjson_line(&line) {
-                        sections.push(sec.clone());
-                        on_section(sec);
-                    }
+                // 观察 2026-09-05-2：行缓冲收敛至 ndjson_feed 纯函数（可单测）
+                let before = sections.len();
+                crate::ndjson_feed::feed_ndjson(&mut pending, delta, &mut sections);
+                for sec in sections[before..].iter().cloned() {
+                    on_section(sec);
                 }
             },
         )?;
         // 末行无换行（SSE 结束前 flush）
-        if let Some(sec) = parse_section_ndjson_line(&pending) {
-            sections.push(sec.clone());
-            on_section(sec);
+        {
+            let before = sections.len();
+            crate::ndjson_feed::flush_ndjson(&mut pending, &mut sections);
+            for sec in sections[before..].iter().cloned() {
+                on_section(sec);
+            }
         }
         // B2（审查）：未收到 [DONE] 即断流——已累积节不可信（尾节可能丢失），
         // 整体视作失败走同拍非流式回退（禁止静默截断当成功）

app/src-tauri/src/lib.rs

@@ -348,6 +348,8 @@ mod media_state;
 mod note_filter;
 mod note_filter_ai;
 mod note_filter_discourse;
+// 观察 2026-09-05-2：NDJSON 流式喂入缓冲纯函数（审查收口）
+mod ndjson_feed;
 // v0.12.0 M1（ADR-021）：正文源多态——BodySource 判定 + OCR 精简过滤链
 mod note_body_source;
 mod note_filter_ocr;

app/src-tauri/src/ndjson_feed.rs

@@ -0,0 +1,81 @@
+//! NDJSON 流式喂入缓冲（观察 2026-09-05-2，v0.19.7 审查收口）。
+//!
+//! @ai-context: 精修逐节流式的行缓冲逻辑从 adapter 闭包抽为纯函数（可单测）：
+//!              chunk 可能切在行中/行间/CRLF 边界；`feed_ndjson` 只解析完整行，
+//!              残留留 pending；`flush_ndjson` 处理末行无换行与尾随垃圾。
+
+use crate::ai_refine_protocol::{AiRefineSection, parse_section_ndjson_line};
+
+/// 喂入增量文本：解析全部完整行并入 sink（未换行残留在 pending）。
+pub fn feed_ndjson(pending: &mut String, chunk: &str, sink: &mut Vec<AiRefineSection>) {
+    pending.push_str(chunk);
+    while let Some(pos) = pending.find('\n') {
+        let line: String = pending.drain(..=pos).collect();
+        if let Some(sec) = parse_section_ndjson_line(&line) {
+            sink.push(sec);
+        }
+    }
+}
+
+/// 流结束 flush：解析剩余（末行无换行/模型尾部垃圾行忽略）。
+pub fn flush_ndjson(pending: &mut String, sink: &mut Vec<AiRefineSection>) {
+    if pending.is_empty() {
+        return;
+    }
+    let tail = std::mem::take(pending);
+    if let Some(sec) = parse_section_ndjson_line(&tail) {
+        sink.push(sec);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn sec(heading: &str) -> AiRefineSection {
+        // 解析侧只关心 heading/blocks 形状——空 blocks 元素同构
+        let line = format!(r#"{{"heading":"{}","blocks":[]}}"#, heading);
+        parse_section_ndjson_line(&line).expect("构造节失败")
+    }
+
+    #[test]
+    fn chunk_split_across_lines_accumulates() {
+        let mut pending = String::new();
+        let mut sink = Vec::new();
+        // 第 1 个对象被切成两半 + 第 2 个对象完整
+        feed_ndjson(&mut pending, r#"{"heading":"节A","blocks":[]}"#, &mut sink);
+        feed_ndjson(&mut pending, "\n", &mut sink);
+        feed_ndjson(&mut pending, r#"{"heading":"节B","blocks":[]}"#, &mut sink);
+        assert_eq!(sink.len(), 0, "首行未换行前不解析");
+        feed_ndjson(&mut pending, "\n", &mut sink);
+        assert_eq!(sink.len(), 2);
+        assert_eq!(sink[0], sec("节A"));
+        assert_eq!(sink[1], sec("节B"));
+    }
+
+    #[test]
+    fn crlf_and_tail_garbage_handled() {
+        let mut pending = String::new();
+        let mut sink = Vec::new();
+        feed_ndjson(&mut pending, "{\"heading\":\"A\",\"blocks\":[]}\r\n{\"heading\":\"B\",\"blocks\":[]}\n", &mut sink);
+        assert_eq!(sink.len(), 2, "CRLF 尾部 \\r 应被 trim 忽略");
+        // 末行无换行 + 尾随解释文本 → flush 只取可解析对象
+        feed_ndjson(&mut pending, "{\"heading\":\"C\",\"blocks\":[]}", &mut sink);
+        feed_ndjson(&mut pending, "（以上为整理结果）", &mut sink);
+        flush_ndjson(&mut pending, &mut sink);
+        assert_eq!(sink.len(), 3);
+        assert_eq!(sink[2], sec("C"));
+        assert!(pending.is_empty());
+    }
+
+    #[test]
+    fn compact_array_lines_rejected_per_line() {
+        // 模型输出完整数组（违反逐节约定）→ 行级全部拒绝；整包回退由
+        // adapter 层按全文解析（见 refine_stream_ndjson）
+        let mut pending = String::new();
+        let mut sink = Vec::new();
+        feed_ndjson(&mut pending, "[{\"heading\":\"A\",\"blocks\":[]}]", &mut sink);
+        flush_ndjson(&mut pending, &mut sink);
+        assert!(sink.is_empty());
+    }
+}