From 07faeddab5fec2b68b2876fa563946382a78d985 Mon Sep 17 00:00:00 2001 From: wujunchen Date: Fri, 8 May 2026 22:57:31 +0800 Subject: [PATCH] fix Obsidian plugin review scan Change-Id: I43416145eca8ee04468522c350fdcebeb772ab6e --- eslint.config.mjs | 8 --- manifest.json | 4 +- src/cli.ts | 4 +- src/error-ui.ts | 4 +- src/generation.ts | 9 ++- src/providers.ts | 17 +++-- src/streaming.ts | 117 +++++++++++++++---------------- tests/direct-streaming.test.js | 123 +++++++++++++++++++-------------- versions.json | 3 +- 9 files changed, 150 insertions(+), 139 deletions(-) diff --git a/eslint.config.mjs b/eslint.config.mjs index de3d336..cf570ff 100644 --- a/eslint.config.mjs +++ b/eslint.config.mjs @@ -20,14 +20,6 @@ export default defineConfig([ "@typescript-eslint/no-unsafe-return": "off", }, }, - { - // streaming.ts legitimately uses bare `fetch` because Obsidian's `requestUrl` - // does not support streaming responses (it's a one-shot wrapper). - files: ["src/streaming.ts"], - rules: { - "no-restricted-globals": "off", - }, - }, { ignores: [ "node_modules/", diff --git a/manifest.json b/manifest.json index f7bcf45..ca66634 100644 --- a/manifest.json +++ b/manifest.json @@ -1,8 +1,8 @@ { "id": "parallel-reader", "name": "Parallel Reader", - "version": "1.0.17", - "minAppVersion": "1.7.2", + "version": "1.0.18", + "minAppVersion": "1.8.7", "description": "AI-powered split-view reading: source note on the left, LLM-generated summary cards on the right with scroll-sync highlighting.", "author": "lancivez", "fundingUrl": "https://github.com/fancive", diff --git a/src/cli.ts b/src/cli.ts index 95781dc..beecfa4 100644 --- a/src/cli.ts +++ b/src/cli.ts @@ -163,7 +163,7 @@ export function runCli( settled = true; clearTimers(); if (debug) { - console.info('[parallel-reader] cli ok', { + console.debug('[parallel-reader] cli ok', { cmd, pid: child?.pid, elapsedMs: Date.now() - startedAt, @@ -212,7 +212,7 @@ export function runCli( } if (debug) { - console.info('[parallel-reader] cli spawn', { + console.debug('[parallel-reader] cli spawn', { cmd, argCount: args.length, pid: child?.pid, diff --git a/src/error-ui.ts b/src/error-ui.ts index 16f51fe..035e5ec 100644 --- a/src/error-ui.ts +++ b/src/error-ui.ts @@ -21,7 +21,7 @@ interface ActionableNoticeAction { function showActionableNotice(message: string, actions: ActionableNoticeAction[], durationMs = 12000): Notice { const notice = new Notice('', durationMs); - const root = notice.noticeEl; + const root = notice.messageEl; root.addClass('parallel-reader-error-notice'); root.createDiv({ cls: 'parallel-reader-error-notice-message', text: message }); if (actions.length > 0) { @@ -77,7 +77,7 @@ class TimeoutDiagnosticsModal extends Modal { const reasonKey = this.details.reason === 'idle-timeout' ? 'errorModalReasonIdle' : 'errorModalReasonWall'; contentEl.createEl('p', { text: translate(this.settings, reasonKey) }); - const grid = contentEl.createEl('div', { cls: 'parallel-reader-error-modal-grid' }); + const grid = contentEl.createDiv({ cls: 'parallel-reader-error-modal-grid' }); const addRow = (label: string, value: string) => { const row = grid.createDiv({ cls: 'parallel-reader-error-modal-row' }); row.createSpan({ cls: 'parallel-reader-error-modal-label', text: label }); diff --git a/src/generation.ts b/src/generation.ts index d754a6c..47cd4ea 100644 --- a/src/generation.ts +++ b/src/generation.ts @@ -30,7 +30,14 @@ export async function summarizeDocument( if (useStreaming) { const abortController = new AbortController(); job.onCancel(() => abortController.abort()); - cards = await summarizeViaApiStreaming(system, user, settings, onStreamProgress, abortController.signal); + cards = await summarizeViaApiStreaming( + requestUrl, + system, + user, + settings, + onStreamProgress, + abortController.signal, + ); } else { cards = await summarizeViaApi(requestUrl, system, user, settings); } diff --git a/src/providers.ts b/src/providers.ts index 721b198..d991981 100644 --- a/src/providers.ts +++ b/src/providers.ts @@ -25,7 +25,7 @@ import { getApiPreset, modelForApi, } from './settings'; -import { deltaExtractorForFormat, type StreamProgress, streamingFetch } from './streaming'; +import { deltaExtractorForFormat, type StreamProgress, streamingRequestUrl } from './streaming'; import type { PluginSettings, RawCard } from './types'; export { @@ -193,7 +193,7 @@ async function summarizeViaGoogleGenerativeAi( return parseCardsJson(textFromGoogleGenerativeAiResponse(json), settings); } -// Obsidian's requestUrl is not directly compatible with fetch — we accept it as a typed callback +// Accept Obsidian's requestUrl as a typed callback so tests can inject the transport. export async function summarizeViaApi( requestUrlImpl: RequestUrlFunction, system: string, @@ -220,6 +220,7 @@ export function supportsStreaming(settings: PluginSettings): boolean { } async function streamSummarizeViaOpenAiChat( + requestUrlImpl: RequestUrlFunction, system: string, user: string, settings: PluginSettings, @@ -231,11 +232,12 @@ async function streamSummarizeViaOpenAiChat( const body = buildOpenAiChatBody(system, user, settings, { structured: false }); body.stream = true; const extractor = requiredDeltaExtractor('openai-chat'); - const text = await streamingFetch(url, headers, body, extractor, onProgress, signal, settings); + const text = await streamingRequestUrl(requestUrlImpl, url, headers, body, extractor, onProgress, signal, settings); return parseCardsJson(text.trim(), settings); } async function streamSummarizeViaAnthropicMessages( + requestUrlImpl: RequestUrlFunction, system: string, user: string, settings: PluginSettings, @@ -247,11 +249,12 @@ async function streamSummarizeViaAnthropicMessages( const body = buildAnthropicMessagesBody(system, user, settings, { structured: false }); body.stream = true; const extractor = requiredDeltaExtractor('anthropic-messages'); - const text = await streamingFetch(url, headers, body, extractor, onProgress, signal, settings); + const text = await streamingRequestUrl(requestUrlImpl, url, headers, body, extractor, onProgress, signal, settings); return parseCardsJson(text.trim(), settings); } export async function summarizeViaApiStreaming( + requestUrlImpl: RequestUrlFunction, system: string, user: string, settings: PluginSettings, @@ -261,15 +264,15 @@ export async function summarizeViaApiStreaming( const format = getApiFormat(settings); switch (format) { case 'openai-chat': - return streamSummarizeViaOpenAiChat(system, user, settings, onProgress, signal); + return streamSummarizeViaOpenAiChat(requestUrlImpl, system, user, settings, onProgress, signal); case 'anthropic-messages': - return streamSummarizeViaAnthropicMessages(system, user, settings, onProgress, signal); + return streamSummarizeViaAnthropicMessages(requestUrlImpl, system, user, settings, onProgress, signal); default: throw new Error(`Streaming not supported for format: ${format}`); } } -// Obsidian's requestUrl is not directly compatible with fetch — we accept it as a typed callback +// Accept Obsidian's requestUrl as a typed callback so tests can inject the transport. export async function testApiBackend(requestUrlImpl: RequestUrlFunction, settings: PluginSettings): Promise { await summarizeViaApi(requestUrlImpl, '只输出 JSON：{"cards":[]}', '连通性测试：请原样输出 {"cards":[]}', settings); const format = getApiFormat(settings); diff --git a/src/streaming.ts b/src/streaming.ts index e4c6fc9..7c81bc1 100644 --- a/src/streaming.ts +++ b/src/streaming.ts @@ -1,6 +1,7 @@ 'use strict'; import { translate } from './i18n'; +import type { RequestUrlFunction } from './provider-request'; import type { PluginSettings } from './types'; /** @@ -64,7 +65,7 @@ export function parseSseBuffer(buffer: string, extractDelta: DeltaExtractor): { const json = JSON.parse(data) as Record; const delta = extractDelta(json); if (delta) deltas.push(delta); - } catch (_) { + } catch { // skip non-JSON SSE lines } } @@ -76,68 +77,52 @@ export interface StreamProgress { done: boolean; } -async function doStreamingFetch( +async function doStreamingRequestUrl( + requestUrlImpl: RequestUrlFunction, url: string, headers: Record, body: unknown, extractDelta: DeltaExtractor, onProgress: ((progress: StreamProgress) => void) | undefined, - signal: AbortSignal, settings: PluginSettings | null | undefined, ): Promise { - const response = await fetch(url, { - method: 'POST', - headers, - body: JSON.stringify(body), - signal, - }); - - if (!response.ok) { - const text = await response.text(); + let response: Awaited>; + try { + response = await requestUrlImpl({ + url, + method: 'POST', + headers, + body: JSON.stringify(body), + throw: false, + }); + } catch (e: unknown) { throw new Error( - translate(settings || null, 'errorProviderApiStatus', { + translate(settings || null, 'errorProviderRequestFailed', { label: 'Streaming', - status: response.status, - excerpt: text.slice(0, 500), + error: e instanceof Error ? e.message : String(e), }), ); } - if (!response.body) { - throw new Error('Response has no body for streaming'); + if (response.status >= 400) { + throw new Error( + translate(settings || null, 'errorProviderApiStatus', { + label: 'Streaming', + status: response.status, + excerpt: (response.text || '').slice(0, 500), + }), + ); } - 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(); + const text = response.text || ''; + const buffer = text.endsWith('\n\n') ? text : `${text}\n\n`; + const result = parseSseBuffer(buffer, extractDelta); + for (const delta of result.deltas) { + accumulated += delta; } - - // Flush any unterminated final SSE event (some providers close the stream - // without a trailing \n\n, leaving the last delta in `buffer`). - if (buffer.length > 0) { - const tailBuffer = buffer.endsWith('\n\n') ? buffer : `${buffer}\n\n`; - const tail = parseSseBuffer(tailBuffer, extractDelta); - for (const delta of tail.deltas) accumulated += delta; + if (result.deltas.length > 0) { + onProgress?.({ accumulated, done: false }); } onProgress?.({ accumulated, done: true }); @@ -145,11 +130,12 @@ async function doStreamingFetch( } /** - * Perform a streaming fetch with SSE parsing and configurable timeout. - * Uses the native Fetch API (available in Electron/Obsidian). - * Returns the full accumulated text when done. + * Perform an Obsidian requestUrl call that asks providers for SSE output and parses + * the complete response text. requestUrl is one-shot, so progress arrives after + * the HTTP request completes rather than per network chunk. */ -export async function streamingFetch( +export async function streamingRequestUrl( + requestUrlImpl: RequestUrlFunction, url: string, headers: Record, body: unknown, @@ -158,33 +144,38 @@ export async function streamingFetch( signal?: AbortSignal, settings?: PluginSettings | null, ): Promise { - if (typeof fetch !== 'function') { - throw new Error('Streaming requires fetch API'); - } - const timeoutMs = settings?.streamingTimeoutMs ?? 120000; - const timeoutController = new AbortController(); let abortListener: (() => void) | null = null; + let abortPromise: Promise | null = null; if (signal) { - abortListener = () => timeoutController.abort(); - signal.addEventListener('abort', abortListener, { once: true }); - if (signal.aborted) timeoutController.abort(); + if (signal.aborted) throw new Error('Streaming request aborted'); + abortPromise = new Promise((_, reject) => { + abortListener = () => reject(new Error('Streaming request aborted')); + signal.addEventListener('abort', abortListener, { once: true }); + }); } let timeoutId: number | null = null; const timeoutPromise = new Promise((_, reject) => { timeoutId = activeWindow.setTimeout(() => { - timeoutController.abort(); reject(new Error(`Streaming timed out after ${timeoutMs}ms`)); }, timeoutMs); }); try { - return await Promise.race([ - doStreamingFetch(url, headers, body, extractDelta, onProgress, timeoutController.signal, settings), - timeoutPromise, - ]); + const requestPromise = doStreamingRequestUrl( + requestUrlImpl, + url, + headers, + body, + extractDelta, + onProgress, + settings, + ); + return await Promise.race( + abortPromise ? [requestPromise, timeoutPromise, abortPromise] : [requestPromise, timeoutPromise], + ); } finally { if (timeoutId !== null) activeWindow.clearTimeout(timeoutId); if (signal && abortListener) signal.removeEventListener('abort', abortListener); diff --git a/tests/direct-streaming.test.js b/tests/direct-streaming.test.js index 2357fbe..fb3c2c5 100644 --- a/tests/direct-streaming.test.js +++ b/tests/direct-streaming.test.js @@ -74,7 +74,7 @@ const { assert, requireBundledModule, cleanup } = require('./direct-test-setup') const empty = streaming.parseSseBuffer('', openAiExtractor); assert.deepStrictEqual(empty.deltas, []); - // ── streamingFetch ── + // ── streamingRequestUrl ── function trackedSignal() { const controller = new AbortController(); const signal = controller.signal; @@ -92,41 +92,42 @@ const { assert, requireBundledModule, cleanup } = require('./direct-test-setup') return { controller, signal, activeListeners: () => activeListeners }; } - function streamingBody(text) { - const encoder = new TextEncoder(); - return new ReadableStream({ - start(c) { - c.enqueue(encoder.encode(text)); - c.close(); - }, - }); - } - - const originalFetch = globalThis.fetch; - try { + { const success = trackedSignal(); - globalThis.fetch = async () => ({ - ok: true, - status: 200, - body: streamingBody('data: {"choices":[{"delta":{"content":"ok"}}]}\n\n'), - text: async () => '', - }); - await streaming.streamingFetch( + const progress = []; + const text = await streaming.streamingRequestUrl( + async (params) => { + assert.strictEqual(params.method, 'POST'); + assert.strictEqual(params.url, 'https://example.test'); + assert.strictEqual(params.body, '{"stream":true}'); + return { + status: 200, + json: null, + text: 'data: {"choices":[{"delta":{"content":"ok"}}]}\n\n', + }; + }, 'https://example.test', {}, - {}, + { stream: true }, streaming.deltaExtractorForFormat('openai-chat'), - undefined, + (p) => progress.push(p), success.signal, { streamingTimeoutMs: 1000 }, ); + assert.strictEqual(text, 'ok'); + assert.deepStrictEqual(progress, [ + { accumulated: 'ok', done: false }, + { accumulated: 'ok', done: true }, + ]); assert.strictEqual(success.activeListeners(), 0, 'cleanup after success'); + } + { const httpError = trackedSignal(); - globalThis.fetch = async () => ({ ok: false, status: 500, body: null, text: async () => 'bad' }); await assert.rejects( () => - streaming.streamingFetch( + streaming.streamingRequestUrl( + async () => ({ status: 500, json: null, text: 'bad' }), 'https://example.test', {}, {}, @@ -138,12 +139,14 @@ const { assert, requireBundledModule, cleanup } = require('./direct-test-setup') /HTTP 500|API returned HTTP 500/, ); assert.strictEqual(httpError.activeListeners(), 0, 'cleanup after HTTP error'); + } + { const timeout = trackedSignal(); - globalThis.fetch = async () => new Promise(() => {}); await assert.rejects( () => - streaming.streamingFetch( + streaming.streamingRequestUrl( + async () => new Promise(() => {}), 'https://example.test', {}, {}, @@ -155,16 +158,19 @@ const { assert, requireBundledModule, cleanup } = require('./direct-test-setup') /Streaming timed out/, ); assert.strictEqual(timeout.activeListeners(), 0, 'cleanup after timeout'); + } + { const preAborted = trackedSignal(); preAborted.controller.abort(); - globalThis.fetch = async (_url, opts) => { - if (opts?.signal?.aborted) throw new DOMException('The operation was aborted.', 'AbortError'); - throw new Error('should not reach'); - }; + let called = false; await assert.rejects( () => - streaming.streamingFetch( + streaming.streamingRequestUrl( + async () => { + called = true; + return { status: 200, json: null, text: '' }; + }, 'https://example.test', {}, {}, @@ -175,41 +181,52 @@ const { assert, requireBundledModule, cleanup } = require('./direct-test-setup') ), /abort/i, ); + assert.strictEqual(called, false, 'pre-aborted request is not started'); assert.strictEqual(preAborted.activeListeners(), 0, 'cleanup on pre-aborted'); + } - const abortDuringRead = trackedSignal(); - globalThis.fetch = async (_url, opts) => { - const fetchSignal = opts?.signal; - const stream = new ReadableStream({ - async pull(ctrl) { - ctrl.enqueue(new TextEncoder().encode('data: {"choices":[{"delta":{"content":"first"}}]}\n\n')); - abortDuringRead.controller.abort(); - await new Promise((r) => setTimeout(r, 5)); - if (fetchSignal?.aborted) { - ctrl.error(new DOMException('The operation was aborted.', 'AbortError')); - return; - } - ctrl.close(); - }, - }); - return { ok: true, status: 200, body: stream, text: async () => '' }; - }; + { + const abortDuringRequest = trackedSignal(); await assert.rejects( () => - streaming.streamingFetch( + streaming.streamingRequestUrl( + async () => + new Promise((resolve) => { + setTimeout(() => abortDuringRequest.controller.abort(), 1); + setTimeout(() => resolve({ status: 200, json: null, text: '' }), 20); + }), 'https://example.test', {}, {}, streaming.deltaExtractorForFormat('openai-chat'), undefined, - abortDuringRead.signal, + abortDuringRequest.signal, { streamingTimeoutMs: 5000 }, ), /abort/i, ); - assert.strictEqual(abortDuringRead.activeListeners(), 0, 'cleanup after mid-read abort'); - } finally { - globalThis.fetch = originalFetch; + assert.strictEqual(abortDuringRequest.activeListeners(), 0, 'cleanup after mid-request abort'); + } + + { + const requestFailure = trackedSignal(); + await assert.rejects( + () => + streaming.streamingRequestUrl( + async () => { + throw new Error('network down'); + }, + 'https://example.test', + {}, + {}, + streaming.deltaExtractorForFormat('openai-chat'), + undefined, + requestFailure.signal, + { streamingTimeoutMs: 1000 }, + ), + /network down/, + ); + assert.strictEqual(requestFailure.activeListeners(), 0, 'cleanup after transport failure'); } console.log('direct streaming tests passed'); diff --git a/versions.json b/versions.json index a4182e2..a00ac61 100644 --- a/versions.json +++ b/versions.json @@ -16,5 +16,6 @@ "1.0.14": "1.4.0", "1.0.15": "1.7.2", "1.0.16": "1.7.2", - "1.0.17": "1.7.2" + "1.0.17": "1.7.2", + "1.0.18": "1.8.7" }