feat: add streaming API responses for OpenAI Chat and Anthropic

Implement SSE streaming for API backends that support it (OpenAI Chat
Completions and Anthropic Messages). During generation the view shows
a live streaming preview with accumulated text. Adds a streaming toggle
in settings (enabled by default). Non-streaming-capable formats (Gemini,
OpenAI Responses) and CLI backends continue using the existing path.

Change-Id: I3361a5ca5aec3e25a7d16e4fb585185f6fc0695e

main.ts

@@ -21,6 +21,8 @@ import {
   buildOpenAiChatBody,
   buildOpenAiResponsesBody,
   summarizeViaApi,
+  summarizeViaApiStreaming,
+  supportsStreaming,
   tokenLimitFieldForOpenAiChat,
 } from './src/providers';
 import { extractJson, normalizeCardsPayload } from './src/schema';
@@ -39,6 +41,7 @@ import {
   pruneCacheEntries,
 } from './src/settings';
 import { ParallelReaderSettingTab } from './src/settings-tab';
+import { deltaExtractorForFormat, parseSseBuffer, type StreamProgress } from './src/streaming';
 import type { CacheEntry, PluginSettings, RawCard, ResolvedCard } from './src/types';
 import { addIconButton, addTextButton, copyToClipboard } from './src/ui-helpers';
 import { folderPathsForTarget } from './src/vault';
@@ -46,18 +49,30 @@ import { ParallelReaderView, VIEW_TYPE_PARALLEL } from './src/view';
 
 /* ---------- Helpers ---------- */
 
-async function summarizeDocument(content: string, settings: PluginSettings, job: GenerationJob) {
+async function summarizeDocument(
+  content: string,
+  settings: PluginSettings,
+  job: GenerationJob,
+  onStreamProgress?: (progress: StreamProgress) => void,
+) {
   const { system, user } = buildPrompts(content, settings);
   let cards: RawCard[];
+  const useStreaming = isApiBackend(settings.backend) && supportsStreaming(settings);
+  const abortController = useStreaming ? new AbortController() : null;
+  if (abortController) {
+    job.onCancel(() => abortController.abort());
+  }
   switch (settings.backend) {
     case 'codex':
       cards = await summarizeViaCodex(system, user, settings, job);
       break;
     case 'api':
-      cards = await summarizeViaApi(requestUrl, system, user, settings);
-      break;
     case 'anthropic-api':
-      cards = await summarizeViaApi(requestUrl, system, user, settings);
+      if (useStreaming) {
+        cards = await summarizeViaApiStreaming(system, user, settings, onStreamProgress, abortController!.signal);
+      } else {
+        cards = await summarizeViaApi(requestUrl, system, user, settings);
+      }
       break;
     default:
       cards = await summarizeViaClaudeCode(system, user, settings, job);
@@ -600,7 +615,14 @@ class ParallelReaderPlugin extends Plugin {
         new Notice(this.t('generatingNotice'));
 
         job.setPhase('generating');
-        const sections = await summarizeDocument(content, this.settings, job);
+        const streamProgress = view
+          ? (progress: StreamProgress) => {
+              if (!progress.done && this.viewIsShowingFile(view, file)) {
+                view!.renderStreamingPreview(file, progress.accumulated);
+              }
+            }
+          : undefined;
+        const sections = await summarizeDocument(content, this.settings, job, streamProgress);
         job.throwIfCancelled();
         if (sections.length === 0) {
           new Notice(this.t('noCardsReturned'));
@@ -797,4 +819,7 @@ export const __test = {
   tokenLimitFieldForOpenAiChat,
   updateCardAt,
   visibleTopProbeY,
+  supportsStreaming,
+  deltaExtractorForFormat,
+  parseSseBuffer,
 };

src/i18n.ts

@@ -143,6 +143,8 @@ export const STRINGS: Record<string, Record<string, string>> = {
     cachedNotesName: '已缓存笔记：{count} 篇',
     cachedNotesDesc: '缓存以源笔记 SHA1 + 生成配置指纹作为失效键，源笔记或模型配置修改后会显示 stale 提示',
     clearAllCacheButton: '清除所有缓存',
+    settingStreamingName: '流式输出',
+    settingStreamingDesc: '启用后生成时实时显示 LLM 输出进度（仅 OpenAI Chat 和 Anthropic 格式支持）',
   },
   en: {
     appTitle: 'Parallel Reader',
@@ -290,6 +292,9 @@ export const STRINGS: Record<string, Record<string, string>> = {
     cachedNotesName: 'Cached notes: {count}',
     cachedNotesDesc: 'Cache is invalidated by source SHA1 and generation settings fingerprint.',
     clearAllCacheButton: 'Clear all caches',
+    settingStreamingName: 'Streaming output',
+    settingStreamingDesc:
+      'Show real-time LLM output progress during generation (OpenAI Chat and Anthropic formats only)',
   },
 };
 

src/providers.ts

@@ -19,6 +19,7 @@ import {
   getApiPreset,
   modelForApi,
 } from './settings';
+import { deltaExtractorForFormat, type StreamProgress, streamingFetch } from './streaming';
 import type { PluginSettings, RawCard } from './types';
 
 function endpointUrl(baseUrl: string, suffixes: string[]) {
@@ -416,6 +417,62 @@ export async function summarizeViaApi(
   }
 }
 
+export function supportsStreaming(settings: PluginSettings): boolean {
+  if (!settings.streaming) return false;
+  const format = getApiFormat(settings);
+  return !!deltaExtractorForFormat(format);
+}
+
+async function streamSummarizeViaOpenAiChat(
+  system: string,
+  user: string,
+  settings: PluginSettings,
+  onProgress?: (progress: StreamProgress) => void,
+  signal?: AbortSignal,
+) {
+  const url = endpointUrl(getApiBaseUrl(settings), ['/chat/completions']);
+  const headers = buildApiHeaders(settings);
+  const body = buildOpenAiChatBody(system, user, settings, { structured: false });
+  body.stream = true;
+  const extractor = deltaExtractorForFormat('openai-chat')!;
+  const text = await streamingFetch(url, headers, body, extractor, onProgress, signal, settings);
+  return parseCardsJson(text.trim(), settings);
+}
+
+async function streamSummarizeViaAnthropicMessages(
+  system: string,
+  user: string,
+  settings: PluginSettings,
+  onProgress?: (progress: StreamProgress) => void,
+  signal?: AbortSignal,
+) {
+  const url = endpointUrl(getApiBaseUrl(settings), ['/messages']);
+  const headers = buildApiHeaders(settings, { 'anthropic-version': '2023-06-01' });
+  const body = buildAnthropicMessagesBody(system, user, settings, { structured: false });
+  body.stream = true;
+  const extractor = deltaExtractorForFormat('anthropic-messages')!;
+  const text = await streamingFetch(url, headers, body, extractor, onProgress, signal, settings);
+  return parseCardsJson(text.trim(), settings);
+}
+
+export async function summarizeViaApiStreaming(
+  system: string,
+  user: string,
+  settings: PluginSettings,
+  onProgress?: (progress: StreamProgress) => void,
+  signal?: AbortSignal,
+): Promise<RawCard[]> {
+  const format = getApiFormat(settings);
+  switch (format) {
+    case 'openai-chat':
+      return streamSummarizeViaOpenAiChat(system, user, settings, onProgress, signal);
+    case 'anthropic-messages':
+      return streamSummarizeViaAnthropicMessages(system, user, settings, onProgress, signal);
+    default:
+      throw new Error(`Streaming not supported for format: ${format}`);
+  }
+}
+
 // eslint-disable-next-line @typescript-eslint/no-explicit-any -- Obsidian's requestUrl is not compatible with fetch
 export async function testApiBackend(requestUrlImpl: any, settings: PluginSettings): Promise<string> {
   await summarizeViaApi(requestUrlImpl, '只输出 JSON：{"cards":[]}', '连通性测试：请原样输出 {"cards":[]}', settings);

src/settings-tab.ts

@@ -219,6 +219,16 @@ export class ParallelReaderSettingTab extends PluginSettingTab {
           }
         }),
       );
+
+      new Setting(containerEl)
+        .setName(tr('settingStreamingName'))
+        .setDesc(tr('settingStreamingDesc'))
+        .addToggle((toggle) =>
+          toggle.setValue(this.plugin.settings.streaming ?? true).onChange(async (v) => {
+            this.plugin.settings.streaming = v;
+            await this.plugin.saveSettings();
+          }),
+        );
     }
 
     new Setting(containerEl)

src/settings.ts

@@ -40,6 +40,7 @@ export const DEFAULT_SETTINGS: PluginSettings = {
   model: 'claude-sonnet-4-6',
   exportFolder: 'Reading/Articles',
   cliTimeoutMs: 120000,
+  streaming: true,
 };
 
 export const API_FORMATS: Record<string, ApiFormat> = {

src/streaming.ts

@@ -0,0 +1,140 @@
+'use strict';
+
+import { translate } from './i18n';
+import type { PluginSettings } from './types';
+
+/**
+ * Extract delta text from an OpenAI Chat SSE chunk.
+ * Shape: { choices: [{ delta: { content: "..." } }] }
+ */
+function openAiChatDelta(json: Record<string, unknown>): string {
+  const choices = json.choices as Array<{ delta?: { content?: string } }> | undefined;
+  return choices?.[0]?.delta?.content || '';
+}
+
+/**
+ * Extract delta text from an Anthropic Messages SSE chunk.
+ * Event types:
+ *   content_block_delta → { delta: { text: "..." } }
+ */
+function anthropicDelta(json: Record<string, unknown>): string {
+  if (json.type === 'content_block_delta') {
+    const delta = json.delta as { text?: string } | undefined;
+    return delta?.text || '';
+  }
+  return '';
+}
+
+export type DeltaExtractor = (json: Record<string, unknown>) => string;
+
+export function deltaExtractorForFormat(format: string): DeltaExtractor | null {
+  switch (format) {
+    case 'openai-chat':
+      return openAiChatDelta;
+    case 'anthropic-messages':
+      return anthropicDelta;
+    default:
+      return null;
+  }
+}
+
+/**
+ * Parse a buffer of SSE text into individual data payloads.
+ * Returns { events: parsed JSON objects, rest: unconsumed buffer }.
+ */
+export function parseSseBuffer(buffer: string, extractDelta: DeltaExtractor): { deltas: string[]; rest: string } {
+  const deltas: string[] = [];
+  const lines = buffer.split('\n');
+  const rest = lines.pop()!; // keep incomplete line
+
+  for (const line of lines) {
+    const trimmed = line.trim();
+    if (!trimmed || trimmed === 'data: [DONE]') continue;
+    if (trimmed.startsWith('event:')) continue;
+    if (!trimmed.startsWith('data:')) continue;
+
+    const data = trimmed.slice(trimmed.startsWith('data: ') ? 6 : 5);
+    try {
+      const json = JSON.parse(data) as Record<string, unknown>;
+      const delta = extractDelta(json);
+      if (delta) deltas.push(delta);
+    } catch (_) {
+      // skip non-JSON SSE lines
+    }
+  }
+  return { deltas, rest };
+}
+
+export interface StreamProgress {
+  accumulated: string;
+  done: boolean;
+}
+
+/**
+ * Perform a streaming fetch with SSE parsing.
+ * Uses the native Fetch API (available in Electron/Obsidian).
+ * Returns the full accumulated text when done.
+ */
+export async function streamingFetch(
+  url: string,
+  headers: Record<string, string>,
+  body: unknown,
+  extractDelta: DeltaExtractor,
+  onProgress?: (progress: StreamProgress) => void,
+  signal?: AbortSignal,
+  settings?: PluginSettings | null,
+): Promise<string> {
+  if (typeof globalThis.fetch !== 'function') {
+    throw new Error('Streaming requires fetch API');
+  }
+
+  const response = await globalThis.fetch(url, {
+    method: 'POST',
+    headers,
+    body: JSON.stringify(body),
+    signal,
+  });
+
+  if (!response.ok) {
+    const text = await response.text();
+    throw new Error(
+      translate(settings || null, 'errorProviderApiStatus', {
+        label: 'Streaming',
+        status: response.status,
+        excerpt: text.slice(0, 500),
+      }),
+    );
+  }
+
+  if (!response.body) {
+    throw new Error('Response has no body for streaming');
+  }
+
+  const reader = response.body.getReader();
+  const decoder = new TextDecoder();
+  let accumulated = '';
+  let buffer = '';
+
+  try {
+    while (true) {
+      const { done, value } = await reader.read();
+      if (done) break;
+
+      buffer += decoder.decode(value, { stream: true });
+      const result = parseSseBuffer(buffer, extractDelta);
+      buffer = result.rest;
+
+      for (const delta of result.deltas) {
+        accumulated += delta;
+      }
+      if (result.deltas.length > 0) {
+        onProgress?.({ accumulated, done: false });
+      }
+    }
+  } finally {
+    reader.releaseLock();
+  }
+
+  onProgress?.({ accumulated, done: true });
+  return accumulated;
+}

src/types.ts

@@ -65,6 +65,7 @@ export interface PluginSettings {
   model: string;
   exportFolder: string;
   cliTimeoutMs: number;
+  streaming: boolean;
 }
 
 /* ---------- Provider types ---------- */

src/view.ts

@@ -94,6 +94,39 @@ export class ParallelReaderView extends ItemView {
     this.render();
   }
 
+  renderStreamingPreview(file: TFile, text: string) {
+    if (this.sourceFile?.path !== file.path) {
+      this.sourceFile = file;
+    }
+    const container = this.containerEl.children[1];
+    const existing = container.querySelector('.parallel-reader-streaming-preview');
+    if (existing) {
+      const pre = existing.querySelector('pre');
+      if (pre) pre.textContent = text.slice(-2000);
+      const counter = existing.querySelector('.parallel-reader-stream-counter');
+      if (counter) counter.textContent = `${text.length} chars`;
+      return;
+    }
+    container.empty();
+    const header = container.createDiv({ cls: 'parallel-reader-header' });
+    const headerRow = header.createDiv({ cls: 'parallel-reader-header-row' });
+    headerRow.createEl('div', { text: file.basename, cls: 'parallel-reader-title' });
+    const actions = headerRow.createDiv({ cls: 'parallel-reader-actions' });
+    addIconButton(actions, 'square', this.plugin.t('actionCancel'), () => {
+      this.plugin.cancelGenerationForFile(file);
+    });
+
+    const state = container.createDiv({
+      cls: 'parallel-reader-state parallel-reader-loading parallel-reader-streaming-preview',
+    });
+    state.createDiv({ cls: 'parallel-reader-spinner' });
+    const titleEl = state.createEl('div', { cls: 'parallel-reader-state-title' });
+    titleEl.createSpan({ text: this.plugin.t('loadingGenerating') + ' ' });
+    titleEl.createSpan({ cls: 'parallel-reader-stream-counter', text: `${text.length} chars` });
+    const pre = state.createEl('pre', { cls: 'parallel-reader-stream-text' });
+    pre.textContent = text.slice(-2000);
+  }
+
   async renderError(file: TFile, message: string) {
     this.sourceFile = file;
     this.sections = [];

styles.css

@@ -349,3 +349,23 @@
   border-radius: 3px;
   font-size: 11px;
 }
+
+.parallel-reader-stream-text {
+  margin-top: 10px;
+  max-height: 200px;
+  overflow-y: auto;
+  font-size: 11px;
+  line-height: 1.5;
+  color: var(--text-faint);
+  white-space: pre-wrap;
+  word-break: break-all;
+  background: var(--background-secondary);
+  border-radius: 4px;
+  padding: 8px;
+}
+
+.parallel-reader-stream-counter {
+  font-size: 11px;
+  color: var(--text-faint);
+  font-weight: normal;
+}

tests/modules.test.js

@@ -390,4 +390,66 @@ assert.strictEqual(t.classifyGenerationError(new Error('random error')), 'unknow
 assert.strictEqual(t.classifyGenerationError(null), 'unknown', 'null error');
 assert.strictEqual(t.classifyGenerationError('string error'), 'unknown', 'string error');
 
+// ── streaming.ts ──
+
+// supportsStreaming
+assert.strictEqual(
+  t.supportsStreaming({ ...baseSettings, streaming: true, apiFormat: 'openai-chat' }),
+  true, 'OpenAI Chat supports streaming'
+);
+assert.strictEqual(
+  t.supportsStreaming({ ...baseSettings, streaming: true, apiFormat: 'anthropic-messages', apiProvider: 'anthropic' }),
+  true, 'Anthropic Messages supports streaming'
+);
+assert.strictEqual(
+  t.supportsStreaming({ ...baseSettings, streaming: true, apiFormat: 'google-generative-ai', apiProvider: 'google' }),
+  false, 'Gemini does not support streaming'
+);
+assert.strictEqual(
+  t.supportsStreaming({ ...baseSettings, streaming: false, apiFormat: 'openai-chat' }),
+  false, 'streaming disabled returns false'
+);
+
+// deltaExtractorForFormat
+const openaiExtract = t.deltaExtractorForFormat('openai-chat');
+assert.ok(openaiExtract, 'openai-chat extractor exists');
+assert.strictEqual(
+  openaiExtract({ choices: [{ delta: { content: 'hello' } }] }),
+  'hello', 'extracts OpenAI delta'
+);
+assert.strictEqual(openaiExtract({ choices: [{ delta: {} }] }), '', 'empty delta');
+assert.strictEqual(openaiExtract({}), '', 'missing choices');
+
+const anthropicExtract = t.deltaExtractorForFormat('anthropic-messages');
+assert.ok(anthropicExtract, 'anthropic extractor exists');
+assert.strictEqual(
+  anthropicExtract({ type: 'content_block_delta', delta: { text: 'world' } }),
+  'world', 'extracts Anthropic delta'
+);
+assert.strictEqual(
+  anthropicExtract({ type: 'message_start' }),
+  '', 'non-delta event returns empty'
+);
+
+assert.strictEqual(t.deltaExtractorForFormat('google-generative-ai'), null, 'no extractor for gemini');
+
+// parseSseBuffer
+const sseResult = t.parseSseBuffer(
+  'data: {"choices":[{"delta":{"content":"hi"}}]}\n\ndata: {"choices":[{"delta":{"content":" there"}}]}\ndata: [DONE]\npartial',
+  openaiExtract,
+);
+assert.deepStrictEqual(sseResult.deltas, ['hi', ' there'], 'SSE parser extracts deltas');
+assert.strictEqual(sseResult.rest, 'partial', 'SSE parser keeps partial line');
+
+const sseEmpty = t.parseSseBuffer('', openaiExtract);
+assert.deepStrictEqual(sseEmpty.deltas, []);
+assert.strictEqual(sseEmpty.rest, '');
+
+// Anthropic SSE format
+const anthSse = t.parseSseBuffer(
+  'event: content_block_delta\ndata: {"type":"content_block_delta","delta":{"text":"token"}}\n\nevent: message_stop\ndata: {"type":"message_stop"}\n\n',
+  anthropicExtract,
+);
+assert.deepStrictEqual(anthSse.deltas, ['token'], 'Anthropic SSE extracts text delta');
+
 console.log('modules tests passed');