obsidian-plugin/fancive_obsidian-parallel-reader
1
0
Fork 0
mirror of https://github.com/fancive/obsidian-parallel-reader.git synced 2026-07-22 06:53:43 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix Obsidian plugin review scan

Browse source 
Change-Id: I43416145eca8ee04468522c350fdcebeb772ab6e
This commit is contained in:
wujunchen 2026-05-08 22:57:31 +08:00
parent a57743e7e2
commit 07faeddab5
9 changed files with 150 additions and 139 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
eslint.config.mjs
View file

@ -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/",

4
manifest.json
View file

@ -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",

4
src/cli.ts
View file

@ -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,

4
src/error-ui.ts
View file

@ -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 });

9
src/generation.ts
View file

@ -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);
}

17
src/providers.ts
View file

@ -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<string> {
await summarizeViaApi(requestUrlImpl, '只输出 JSON：{"cards":[]}', '连通性测试：请原样输出 {"cards":[]}', settings);
const format = getApiFormat(settings);

117
src/streaming.ts
View file

@ -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<string, unknown>;
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<string, string>,
body: unknown,
extractDelta: DeltaExtractor,
onProgress: ((progress: StreamProgress) => void) | undefined,
signal: AbortSignal,
settings: PluginSettings | null | undefined,
): Promise<string> {
const response = await fetch(url, {
method: 'POST',
headers,
body: JSON.stringify(body),
signal,
});
if (!response.ok) {
const text = await response.text();
let response: Awaited<ReturnType<RequestUrlFunction>>;
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<string, string>,
body: unknown,
@ -158,33 +144,38 @@ export async function streamingFetch(
signal?: AbortSignal,
settings?: PluginSettings | null,
): Promise<string> {
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<never> | 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<never>((_, reject) => {
abortListener = () => reject(new Error('Streaming request aborted'));
signal.addEventListener('abort', abortListener, { once: true });
});
}
let timeoutId: number | null = null;
const timeoutPromise = new Promise<never>((_, 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);

123
tests/direct-streaming.test.js
View file

@ -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');

3
versions.json
View file

@ -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"
}