mirror of
https://github.com/fancive/obsidian-parallel-reader.git
synced 2026-07-22 06:53:43 +00:00
fix: address 10 P2 behavioral and concurrency findings
Findings from the original review (P2): - streaming: flush unterminated final SSE event at EOF (some providers close the stream without a trailing blank line, dropping the last delta) - cache-manager: validate entry shape on load — drop entries where cards is not an array, anchor is not a string, or bullets is not an array. Tolerates missing optional fields (treated as cache miss). - view: card edit/delete now check cacheReplaceCards return and surface failures via a localized Notice instead of pretending success - main: clear-current / clear-all / file-menu-clear refresh open view via renderEmpty so stale UI does not display deleted data - prompt + settings-tab: card count is normalized via a single helper used by buildPrompts and onChange. Prompt and fingerprint stay in sync. UI value is written back to the textbox after clamping. - generation-job-manager: global concurrency limit (default 3) with a cancellable wait queue. Race-safe slot accounting via a reserved counter so resolved-but-not-yet-set waiters are visible to fast-path start() callers. - runForFile: now returns RunForFileResult; accepts preloadedContent + silentView + skipEditConfirm options used by batch (and reflected on the PluginHost interface). - batch: avoid double file read, do not steal UI focus, classify results correctly (generated / cached / already-running / empty / ...) Follow-up from the 1.0.11 review: - generationFingerprint: codex backend excludes settings.model from the hash. Codex ignores --model; previously editing model spuriously invalidated all codex cache. Codex review pass: - generation-job-manager: race fix — releaseSlot's resolve microtask and a synchronous start() fast-path could briefly overshoot maxConcurrent. Reserve the slot synchronously inside the wrapped resolve to close the window. - cache-manager: anchor type validation in addition to bullets. NOTE: Codex backend users will see a one-time "stale cache" banner on existing notes due to the fingerprint change; regenerate to refresh. Change-Id: I7721c7dfe51dea3f51b0215764f721523c2f6806
This commit is contained in:
fancivez
wujunchen
parent
d1f36d52a0
commit
76d2e8050a
18 changed files with 486 additions and 41 deletions
77
main.ts
77
main.ts
|
|
@ -40,6 +40,7 @@ import type {
|
|||
PluginSettings,
|
||||
ResolvedCard,
|
||||
RunForFileOptions,
|
||||
RunForFileResult,
|
||||
} from './src/types';
|
||||
import { copyToClipboard } from './src/ui-helpers';
|
||||
import { ParallelReaderView, VIEW_TYPE_PARALLEL } from './src/view';
|
||||
|
|
@ -120,6 +121,7 @@ class ParallelReaderPlugin extends Plugin {
|
|||
return;
|
||||
}
|
||||
await this.cacheManager.delete(active.file.path);
|
||||
this.refreshViewAfterCacheDelete(active.file.path);
|
||||
new Notice(this.t('cacheClearedFile', { name: active.file.basename }));
|
||||
},
|
||||
});
|
||||
|
|
@ -129,6 +131,7 @@ class ParallelReaderPlugin extends Plugin {
|
|||
callback: async () => {
|
||||
const n = Object.keys(this.cacheManager.cache).length;
|
||||
await this.cacheManager.clear();
|
||||
this.refreshViewAfterCacheClear();
|
||||
new Notice(this.t('cacheClearedAll', { count: n }));
|
||||
},
|
||||
});
|
||||
|
|
@ -178,6 +181,7 @@ class ParallelReaderPlugin extends Plugin {
|
|||
}
|
||||
|
||||
onunload() {
|
||||
if (this.jobs) this.jobs.cancelAllWaiters();
|
||||
this.flushSettingsSave().catch((e: unknown) => console.error('[parallel-reader] flush settings on unload', e));
|
||||
this.flushCacheSave().catch((e: unknown) => console.error('[parallel-reader] flush cache on unload', e));
|
||||
}
|
||||
|
|
@ -359,12 +363,23 @@ class ParallelReaderPlugin extends Plugin {
|
|||
.setIcon('trash')
|
||||
.onClick(async () => {
|
||||
await this.cacheManager.delete(file.path);
|
||||
this.refreshViewAfterCacheDelete(file.path);
|
||||
new Notice(this.t('cacheClearedFile', { name: file.basename }));
|
||||
}),
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
private refreshViewAfterCacheDelete(filePath: string) {
|
||||
const view = this.getParallelView();
|
||||
if (view?.sourceFile?.path === filePath) view.renderEmpty();
|
||||
}
|
||||
|
||||
private refreshViewAfterCacheClear() {
|
||||
const view = this.getParallelView();
|
||||
if (view) view.renderEmpty();
|
||||
}
|
||||
|
||||
async handleFileRename(file: TFile, oldPath: string) {
|
||||
if (!(file instanceof TFile) || !oldPath) return;
|
||||
const wasMarkdown = oldPath.endsWith('.md');
|
||||
|
|
@ -434,36 +449,52 @@ class ParallelReaderPlugin extends Plugin {
|
|||
return this.runForFile(mdView.file, force);
|
||||
}
|
||||
|
||||
async runForFile(file: TFile | null, force: boolean, options: RunForFileOptions = {}) {
|
||||
async runForFile(
|
||||
file: TFile | null,
|
||||
force: boolean,
|
||||
options: RunForFileOptions = {},
|
||||
preloadedContent?: string,
|
||||
): Promise<RunForFileResult> {
|
||||
if (!file) {
|
||||
new Notice(this.t('openNoteFirst'));
|
||||
return;
|
||||
return 'error';
|
||||
}
|
||||
if (this.jobs.isRunning(file.path)) {
|
||||
if (this.jobs.isPending(file.path)) {
|
||||
new Notice(this.t('alreadyGenerating'));
|
||||
return;
|
||||
return 'already-running';
|
||||
}
|
||||
if (
|
||||
!options.skipEditConfirm &&
|
||||
shouldConfirmRegenerate(this.cacheManager.get(file.path), force) &&
|
||||
!(await this.confirmRegenerateEditedCards())
|
||||
) {
|
||||
new Notice(this.t('regenerateCancelled'));
|
||||
return;
|
||||
return 'cancelled';
|
||||
}
|
||||
|
||||
let view: ParallelReaderView | null = null;
|
||||
return this.jobs
|
||||
let outcome: RunForFileResult = 'error';
|
||||
|
||||
await this.jobs
|
||||
.start(file.path, async (job) => {
|
||||
job.setPhase('reading');
|
||||
const content = await this.app.vault.read(file);
|
||||
const content = preloadedContent ?? (await this.app.vault.read(file));
|
||||
job.throwIfCancelled();
|
||||
if (!content.trim()) {
|
||||
new Notice(this.t('emptyNote'));
|
||||
outcome = 'empty';
|
||||
return;
|
||||
}
|
||||
|
||||
view = await this.ensureView();
|
||||
if (!view) return;
|
||||
if (options.silentView) {
|
||||
view = this.getParallelView() ?? null;
|
||||
} else {
|
||||
view = await this.ensureView();
|
||||
if (!view) {
|
||||
outcome = 'no-view';
|
||||
return;
|
||||
}
|
||||
}
|
||||
job.throwIfCancelled();
|
||||
|
||||
job.setPhase('cache-check');
|
||||
|
|
@ -471,12 +502,15 @@ class ParallelReaderPlugin extends Plugin {
|
|||
const entry = this.cacheManager.get(file.path);
|
||||
if (entry && cacheEntryMatches(entry, content, this.settings)) {
|
||||
this.cacheTouch(file.path);
|
||||
if (this.activeFileStillMatches(file)) view.loadFor(file, resolveCardAnchors(content, entry.cards), false);
|
||||
if (view && this.activeFileStillMatches(file)) {
|
||||
view.loadFor(file, resolveCardAnchors(content, entry.cards), false);
|
||||
}
|
||||
outcome = 'cached';
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
view.renderLoading(file, this.t('loadingGenerating'));
|
||||
if (view && this.viewIsShowingFile(view, file)) view.renderLoading(file, this.t('loadingGenerating'));
|
||||
const maxDocChars = Number(this.settings.maxDocChars) || DEFAULT_SETTINGS.maxDocChars;
|
||||
if (content.length > maxDocChars) new Notice(this.t('longNoteTruncated', { count: maxDocChars }));
|
||||
new Notice(this.t('generatingNotice'));
|
||||
|
|
@ -487,13 +521,14 @@ class ParallelReaderPlugin extends Plugin {
|
|||
if (sections.length === 0) {
|
||||
if (view && this.viewIsShowingFile(view, file)) view.renderError(file, this.t('noCardsReturned'));
|
||||
new Notice(this.t('noCardsReturned'));
|
||||
outcome = 'empty';
|
||||
return;
|
||||
}
|
||||
const rawCards = sections.map((s) => ({ title: s.title, anchor: s.anchor, gist: s.gist, bullets: s.bullets }));
|
||||
job.setPhase('saving');
|
||||
await this.cacheManager.put(file.path, content, rawCards, this.settings);
|
||||
job.throwIfCancelled();
|
||||
if (this.viewIsShowingFile(view, file)) view.loadFor(file, sections, false);
|
||||
if (view && this.viewIsShowingFile(view, file)) view.loadFor(file, sections, false);
|
||||
const unanchored = sections.filter((s) => s.startLine < 0).length;
|
||||
new Notice(
|
||||
this.t('generationDone', {
|
||||
|
|
@ -501,11 +536,17 @@ class ParallelReaderPlugin extends Plugin {
|
|||
suffix: unanchored ? this.t('unanchoredSuffix', { count: unanchored }) : '',
|
||||
}),
|
||||
);
|
||||
outcome = 'generated';
|
||||
})
|
||||
.catch((e: unknown) => {
|
||||
this.handleGenerationError(e, file, view);
|
||||
if (e instanceof GenerationJobAlreadyRunningError) outcome = 'already-running';
|
||||
else if (e instanceof GenerationJobCancelledError) outcome = 'cancelled';
|
||||
else outcome = 'error';
|
||||
if (options.rethrowErrors) throw e;
|
||||
});
|
||||
|
||||
return outcome;
|
||||
}
|
||||
|
||||
private streamProgressFor(
|
||||
|
|
@ -581,9 +622,16 @@ class ParallelReaderPlugin extends Plugin {
|
|||
continue;
|
||||
}
|
||||
try {
|
||||
await this.runForFile(file, false, { rethrowErrors: true });
|
||||
const result = await this.runForFile(
|
||||
file,
|
||||
false,
|
||||
{ rethrowErrors: true, silentView: true, skipEditConfirm: true },
|
||||
content,
|
||||
);
|
||||
if (batch.cancelled) break;
|
||||
stats = recordBatchProcessed(stats);
|
||||
if (result === 'generated') stats = recordBatchProcessed(stats);
|
||||
else if (result === 'cached' || result === 'already-running') stats = recordBatchSkip(stats);
|
||||
else stats = recordBatchError(stats);
|
||||
} catch (e: unknown) {
|
||||
if (batch.cancelled && e instanceof GenerationJobCancelledError) break;
|
||||
stats = recordBatchError(stats);
|
||||
|
|
@ -591,6 +639,7 @@ class ParallelReaderPlugin extends Plugin {
|
|||
}
|
||||
}
|
||||
} finally {
|
||||
if (batch.cancelled) this.jobs.cancelAllWaiters();
|
||||
if (this.activeBatch === batch) this.activeBatch = null;
|
||||
}
|
||||
const batchVars: Record<string, string | number> = {
|
||||
|
|
|
|||
|
|
@ -11,6 +11,27 @@ import {
|
|||
} from './settings';
|
||||
import type { CacheEntry, PluginSettings, RawCard, ResolvedCard } from './types';
|
||||
|
||||
/**
|
||||
* Reject cache entries whose shape would crash downstream consumers (e.g. cards.map).
|
||||
* Missing optional fields like contentHash/settingsHash are tolerated — they just
|
||||
* cause a normal cache miss instead of returning stale data.
|
||||
*/
|
||||
function isValidCacheEntry(entry: unknown): entry is CacheEntry {
|
||||
if (!entry || typeof entry !== 'object') return false;
|
||||
const cards = (entry as { cards?: unknown }).cards;
|
||||
if (!Array.isArray(cards)) return false;
|
||||
for (const c of cards) {
|
||||
if (!c || typeof c !== 'object') return false;
|
||||
const card = c as { bullets?: unknown; anchor?: unknown };
|
||||
// bullets is allowed to be missing (defaults to []), but if present must be an array
|
||||
if (card.bullets !== undefined && !Array.isArray(card.bullets)) return false;
|
||||
// anchor is allowed to be missing, but if present must be a string
|
||||
// (downstream resolveCardAnchors / findLineForAnchor expects string)
|
||||
if (card.anchor !== undefined && typeof card.anchor !== 'string') return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
export class CacheManager {
|
||||
cache: Record<string, CacheEntry> = {};
|
||||
private _timer: ReturnType<typeof setTimeout> | null = null;
|
||||
|
|
@ -41,8 +62,16 @@ export class CacheManager {
|
|||
try {
|
||||
const raw = await this.adapter.read(this.filePath());
|
||||
const parsed = JSON.parse(raw);
|
||||
if (parsed && typeof parsed === 'object' && parsed.entries && typeof parsed.entries === 'object')
|
||||
return parsed.entries;
|
||||
if (parsed && typeof parsed === 'object' && parsed.entries && typeof parsed.entries === 'object') {
|
||||
const validated: Record<string, CacheEntry> = {};
|
||||
let dropped = 0;
|
||||
for (const [path, entry] of Object.entries(parsed.entries as Record<string, unknown>)) {
|
||||
if (isValidCacheEntry(entry)) validated[path] = entry;
|
||||
else dropped++;
|
||||
}
|
||||
if (dropped > 0) console.warn('[parallel-reader] dropped', dropped, 'malformed cache entries');
|
||||
return validated;
|
||||
}
|
||||
} catch (e: unknown) {
|
||||
const message = e instanceof Error ? e.message : String(e);
|
||||
if (!/not found|does not exist|ENOENT/i.test(message))
|
||||
|
|
|
|||
|
|
@ -76,10 +76,20 @@ export class GenerationJob {
|
|||
}
|
||||
}
|
||||
|
||||
type Waiter = { key: string; resolve: () => void; reject: (err: Error) => void };
|
||||
|
||||
export class GenerationJobManager {
|
||||
private jobs: Map<string, GenerationJob>;
|
||||
private waiters: Waiter[] = [];
|
||||
/**
|
||||
* Live count of slots in use OR resolved-but-not-yet-set. This is the
|
||||
* authoritative concurrency gauge — using `jobs.size` alone has a race
|
||||
* window between `releaseSlot.resolve()` and the awaiter's `jobs.set()`
|
||||
* during which a fresh `start()` could see a free slot and double-book.
|
||||
*/
|
||||
private reserved = 0;
|
||||
|
||||
constructor() {
|
||||
constructor(private maxConcurrent: number = 3) {
|
||||
this.jobs = new Map();
|
||||
}
|
||||
|
||||
|
|
@ -91,8 +101,75 @@ export class GenerationJobManager {
|
|||
return this.jobs.has(key);
|
||||
}
|
||||
|
||||
/** Returns true if a key is either running or queued. */
|
||||
isPending(key: string): boolean {
|
||||
return this.jobs.has(key) || this.waiters.some((w) => w.key === key);
|
||||
}
|
||||
|
||||
waitingCount(): number {
|
||||
return this.waiters.length;
|
||||
}
|
||||
|
||||
/**
|
||||
* Reject all queued waiters with cancellation. Used on plugin unload or batch cancel
|
||||
* so queued promises don't leak. Does not affect `reserved` — queued waiters were
|
||||
* not yet counted toward `reserved` (they get incremented on resolve).
|
||||
*/
|
||||
cancelAllWaiters(): number {
|
||||
const drained = this.waiters.splice(0);
|
||||
for (const w of drained) w.reject(new GenerationJobCancelledError(w.key));
|
||||
return drained.length;
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns null if a slot is immediately available (and reserves it synchronously,
|
||||
* so the caller can take the fast path). Returns a Promise when queueing is required.
|
||||
*/
|
||||
private waitSlot(key: string): Promise<void> | null {
|
||||
if (this.reserved < this.maxConcurrent) {
|
||||
this.reserved++;
|
||||
return null;
|
||||
}
|
||||
return new Promise<void>((resolve, reject) => {
|
||||
// Wrap resolve so the slot is reserved synchronously inside releaseSlot()
|
||||
// before any other start() can read `reserved`.
|
||||
this.waiters.push({
|
||||
key,
|
||||
resolve: () => {
|
||||
this.reserved++;
|
||||
resolve();
|
||||
},
|
||||
reject,
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
private releaseSlot(): void {
|
||||
this.reserved--;
|
||||
const next = this.waiters.shift();
|
||||
if (next) next.resolve(); // wrapped resolve increments `reserved` synchronously
|
||||
}
|
||||
|
||||
async start<T>(key: string, runner: (job: GenerationJob) => Promise<T>): Promise<T> {
|
||||
if (this.jobs.has(key)) throw new GenerationJobAlreadyRunningError(key);
|
||||
// Reject same-key dedup at entry (running OR queued).
|
||||
if (this.isPending(key)) throw new GenerationJobAlreadyRunningError(key);
|
||||
const wait = this.waitSlot(key);
|
||||
if (wait) {
|
||||
try {
|
||||
await wait;
|
||||
} catch (err) {
|
||||
// Cancelled while queued (cancelAllWaiters) — slot was never reserved
|
||||
// for this caller, so no releaseSlot is needed.
|
||||
throw err;
|
||||
}
|
||||
if (this.jobs.has(key)) {
|
||||
// Same-key racily inserted while we waited; release the slot we got.
|
||||
this.reserved--;
|
||||
const next = this.waiters.shift();
|
||||
if (next) next.resolve();
|
||||
throw new GenerationJobAlreadyRunningError(key);
|
||||
}
|
||||
}
|
||||
const job = new GenerationJob(key);
|
||||
this.jobs.set(key, job);
|
||||
try {
|
||||
|
|
@ -107,6 +184,7 @@ export class GenerationJobManager {
|
|||
throw err;
|
||||
} finally {
|
||||
this.jobs.delete(key);
|
||||
this.releaseSlot();
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -32,6 +32,7 @@ export const STRINGS: Record<string, Record<string, string>> = {
|
|||
actionCopyError: '复制错误信息',
|
||||
emptyCard: '（未生成）',
|
||||
cardUntitled: '(无标题)',
|
||||
cardPersistFailed: '卡片改动未保存（缓存失效）',
|
||||
generationAlreadyRunning: '该笔记正在生成对照笔记',
|
||||
anchorMismatch: 'anchor 匹配失败，无法滚动联动',
|
||||
menuCopyMarkdown: '复制 Markdown',
|
||||
|
|
@ -206,6 +207,7 @@ export const STRINGS: Record<string, Record<string, string>> = {
|
|||
actionCopyError: 'Copy error',
|
||||
emptyCard: '(not generated)',
|
||||
cardUntitled: '(Untitled)',
|
||||
cardPersistFailed: 'Card change not persisted (cache missing).',
|
||||
generationAlreadyRunning: 'This note is already being summarized.',
|
||||
anchorMismatch: 'Anchor did not match; scroll sync is unavailable',
|
||||
menuCopyMarkdown: 'Copy Markdown',
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
'use strict';
|
||||
|
||||
import { DEFAULT_SETTINGS, MAX_DOC_CHARS, PROMPT_LANGUAGES } from './settings';
|
||||
import { DEFAULT_SETTINGS, MAX_DOC_CHARS, normalizeCardCount, PROMPT_LANGUAGES } from './settings';
|
||||
import type { PluginSettings, PromptPair } from './types';
|
||||
|
||||
export function promptLanguageInstruction(language: string): string {
|
||||
|
|
@ -119,8 +119,8 @@ export function buildPrompts(content: string, settings: PluginSettings): PromptP
|
|||
const promptLanguage = (PROMPT_LANGUAGES as Record<string, string>)[settings.promptLanguage]
|
||||
? settings.promptLanguage
|
||||
: DEFAULT_SETTINGS.promptLanguage;
|
||||
const minCards = Math.max(1, Number(settings.minCards) || DEFAULT_SETTINGS.minCards);
|
||||
const maxCards = Math.max(minCards, Number(settings.maxCards) || DEFAULT_SETTINGS.maxCards);
|
||||
const minCards = normalizeCardCount(settings.minCards, DEFAULT_SETTINGS.minCards);
|
||||
const maxCards = Math.max(minCards, normalizeCardCount(settings.maxCards, DEFAULT_SETTINGS.maxCards));
|
||||
const languageInstruction = promptLanguageInstruction(promptLanguage);
|
||||
const doc =
|
||||
content.length > maxDocChars
|
||||
|
|
|
|||
|
|
@ -12,6 +12,7 @@ import {
|
|||
DEFAULT_SETTINGS,
|
||||
getApiFormat,
|
||||
getApiPreset,
|
||||
normalizeCardCount,
|
||||
normalizeCliTimeoutMs,
|
||||
normalizeStreamingTimeoutMs,
|
||||
PROMPT_LANGUAGES,
|
||||
|
|
@ -310,29 +311,36 @@ export class ParallelReaderSettingTab extends PluginSettingTab {
|
|||
new Setting(containerEl)
|
||||
.setName(this.tr('settingCardRangeName'))
|
||||
.setDesc(this.tr('settingCardRangeDesc'))
|
||||
.addText((t) =>
|
||||
t
|
||||
.addText((textComponent) =>
|
||||
textComponent
|
||||
.setPlaceholder('Min')
|
||||
.setValue(String(this.plugin.settings.minCards || DEFAULT_SETTINGS.minCards))
|
||||
.onChange((v) => {
|
||||
const n = parseInt(v, 10);
|
||||
if (!Number.isNaN(n) && n > 0) {
|
||||
this.plugin.settings.minCards = n;
|
||||
if (this.plugin.settings.maxCards < n) this.plugin.settings.maxCards = n;
|
||||
this.plugin.saveSettingsDebounced();
|
||||
}
|
||||
const trimmed = v.trim();
|
||||
if (trimmed === '') return;
|
||||
const normalized = normalizeCardCount(trimmed, DEFAULT_SETTINGS.minCards);
|
||||
this.plugin.settings.minCards = normalized;
|
||||
if (this.plugin.settings.maxCards < normalized) this.plugin.settings.maxCards = normalized;
|
||||
if (String(normalized) !== trimmed) textComponent.setValue(String(normalized));
|
||||
this.plugin.saveSettingsDebounced();
|
||||
}),
|
||||
)
|
||||
.addText((t) =>
|
||||
t
|
||||
.addText((textComponent) =>
|
||||
textComponent
|
||||
.setPlaceholder('Max')
|
||||
.setValue(String(this.plugin.settings.maxCards || DEFAULT_SETTINGS.maxCards))
|
||||
.onChange((v) => {
|
||||
const n = parseInt(v, 10);
|
||||
if (!Number.isNaN(n) && n > 0) {
|
||||
this.plugin.settings.maxCards = Math.max(n, this.plugin.settings.minCards || DEFAULT_SETTINGS.minCards);
|
||||
this.plugin.saveSettingsDebounced();
|
||||
const trimmed = v.trim();
|
||||
if (trimmed === '') return;
|
||||
const normalized = normalizeCardCount(trimmed, DEFAULT_SETTINGS.maxCards);
|
||||
this.plugin.settings.maxCards = Math.max(
|
||||
normalized,
|
||||
this.plugin.settings.minCards || DEFAULT_SETTINGS.minCards,
|
||||
);
|
||||
if (String(this.plugin.settings.maxCards) !== trimmed) {
|
||||
textComponent.setValue(String(this.plugin.settings.maxCards));
|
||||
}
|
||||
this.plugin.saveSettingsDebounced();
|
||||
}),
|
||||
);
|
||||
|
||||
|
|
|
|||
|
|
@ -285,7 +285,9 @@ export function generationFingerprint(settings: PluginSettings): string {
|
|||
maxCards: normalized.maxCards,
|
||||
customSystemPromptHash: hashContent(normalized.customSystemPrompt || ''),
|
||||
backend: normalized.backend,
|
||||
model: normalized.model,
|
||||
// Codex backend ignores settings.model (uses its own config); excluding it from
|
||||
// the fingerprint avoids spurious cache invalidation when the user edits model.
|
||||
model: normalized.backend === 'codex' ? '' : normalized.model,
|
||||
apiProvider: apiBackend ? normalized.apiProvider : '',
|
||||
apiFormat: apiBackend ? format : '',
|
||||
apiBaseUrl,
|
||||
|
|
|
|||
|
|
@ -132,6 +132,14 @@ async function doStreamingFetch(
|
|||
reader.releaseLock();
|
||||
}
|
||||
|
||||
// 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;
|
||||
}
|
||||
|
||||
onProgress?.({ accumulated, done: true });
|
||||
return accumulated;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -51,6 +51,7 @@ export {
|
|||
generationFingerprint,
|
||||
getApiBaseUrl,
|
||||
modelForApi,
|
||||
normalizeCardCount,
|
||||
normalizeCliTimeoutMs,
|
||||
normalizeSettings,
|
||||
normalizeStreamingTimeoutMs,
|
||||
|
|
|
|||
13
src/types.ts
13
src/types.ts
|
|
@ -105,8 +105,14 @@ export type ErrorKind = 'auth' | 'timeout' | 'rate-limit' | 'schema' | 'config'
|
|||
|
||||
export interface RunForFileOptions {
|
||||
rethrowErrors?: boolean;
|
||||
/** Skip ensureView+revealLeaf; only update view if it's already showing this file (used by batch). */
|
||||
silentView?: boolean;
|
||||
/** Skip the "you have edited cards" confirm dialog (used by unattended batch). */
|
||||
skipEditConfirm?: boolean;
|
||||
}
|
||||
|
||||
export type RunForFileResult = 'generated' | 'cached' | 'cancelled' | 'already-running' | 'empty' | 'error' | 'no-view';
|
||||
|
||||
/* ---------- Prompt types ---------- */
|
||||
|
||||
export interface PromptPair {
|
||||
|
|
@ -160,7 +166,12 @@ export interface PluginHost {
|
|||
t(key: string, vars?: Record<string, string | number>): string;
|
||||
isGeneratingFile(file: TFile | null): boolean;
|
||||
cancelGenerationForFile(file: TFile | null): boolean;
|
||||
runForFile(file: TFile | null, force: boolean, options?: RunForFileOptions): Promise<void>;
|
||||
runForFile(
|
||||
file: TFile | null,
|
||||
force: boolean,
|
||||
options?: RunForFileOptions,
|
||||
preloadedContent?: string,
|
||||
): Promise<RunForFileResult | void>;
|
||||
copyCurrentViewMarkdown(): Promise<void>;
|
||||
scrollEditorToLine(line: number, file: TFile | null): Promise<void>;
|
||||
cacheReplaceCards(filePath: string, cards: ResolvedCard[]): Promise<boolean>;
|
||||
|
|
|
|||
12
src/view.ts
12
src/view.ts
|
|
@ -391,8 +391,12 @@ export class ParallelReaderView extends ItemView {
|
|||
const previousLength = this.sections.length;
|
||||
this.sections = nextSections;
|
||||
this.activeIdx = activeIndexAfterCardDelete(index, previousLength, this.activeIdx);
|
||||
await this.plugin.cacheReplaceCards(this.sourceFile.path, nextSections);
|
||||
const ok = await this.plugin.cacheReplaceCards(this.sourceFile.path, nextSections);
|
||||
this.render();
|
||||
if (!ok) {
|
||||
new Notice(this.plugin.t('cardPersistFailed'));
|
||||
return false;
|
||||
}
|
||||
new Notice(this.plugin.t('cardDeleted'));
|
||||
return true;
|
||||
}
|
||||
|
|
@ -410,8 +414,12 @@ export class ParallelReaderView extends ItemView {
|
|||
const nextSections = updateCardAt(this.sections, index, patch);
|
||||
if (nextSections.length !== this.sections.length) return false;
|
||||
this.sections = nextSections;
|
||||
await this.plugin.cacheReplaceCards(this.sourceFile.path, nextSections);
|
||||
const ok = await this.plugin.cacheReplaceCards(this.sourceFile.path, nextSections);
|
||||
this.render();
|
||||
if (!ok) {
|
||||
new Notice(this.plugin.t('cardPersistFailed'));
|
||||
return false;
|
||||
}
|
||||
new Notice(this.plugin.t('cardSaved'));
|
||||
return true;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -274,6 +274,69 @@ function createFakeAdapter() {
|
|||
await scheduleManager.flush();
|
||||
assert.strictEqual(scheduleAdapter.files.has(scheduleManager.filePath()), false, 'flush is no-op when not dirty');
|
||||
|
||||
// ── readFile drops malformed entries (cards not array / bullets not array) ──
|
||||
{
|
||||
const validateAdapter = createFakeAdapter();
|
||||
const validateManager = new cacheManagerModule.CacheManager(
|
||||
validateAdapter,
|
||||
'.obsidian',
|
||||
'parallel-reader',
|
||||
() => ({ ...settings.DEFAULT_SETTINGS, maxCacheEntries: 100 }),
|
||||
);
|
||||
validateAdapter.files.set(
|
||||
validateManager.filePath(),
|
||||
JSON.stringify({
|
||||
version: 1,
|
||||
entries: {
|
||||
'good.md': { generatedAt: '2024-01-01T00:00:00.000Z', cards: [{ title: 't', bullets: ['x'] }] },
|
||||
'cards-null.md': { cards: null },
|
||||
'cards-string.md': { cards: 'oops' },
|
||||
'bullets-string.md': { cards: [{ title: 't', bullets: 'not-array' }] },
|
||||
'bullets-missing.md': { cards: [{ title: 't' }] }, // tolerated
|
||||
'card-null.md': { cards: [null] }, // dangerous: c.anchor would crash
|
||||
'anchor-number.md': { cards: [{ anchor: 42, bullets: [] }] }, // dangerous: anchor.trim() would crash
|
||||
'not-an-object.md': 42,
|
||||
'null-entry.md': null,
|
||||
},
|
||||
}),
|
||||
);
|
||||
const origWarn = console.warn;
|
||||
let droppedMessage = '';
|
||||
console.warn = (...args) => {
|
||||
droppedMessage = args.join(' ');
|
||||
};
|
||||
try {
|
||||
const loaded = await validateManager.readFile();
|
||||
assert.ok(loaded['good.md'], 'good entry kept');
|
||||
assert.ok(loaded['bullets-missing.md'], 'entry with missing bullets tolerated (defaults to [])');
|
||||
assert.strictEqual(loaded['cards-null.md'], undefined, 'cards=null dropped');
|
||||
assert.strictEqual(loaded['cards-string.md'], undefined, 'cards=string dropped');
|
||||
assert.strictEqual(loaded['bullets-string.md'], undefined, 'bullets=string dropped');
|
||||
assert.strictEqual(loaded['card-null.md'], undefined, 'cards=[null] dropped');
|
||||
assert.strictEqual(loaded['anchor-number.md'], undefined, 'anchor=number dropped');
|
||||
assert.strictEqual(loaded['not-an-object.md'], undefined, 'non-object entry dropped');
|
||||
assert.strictEqual(loaded['null-entry.md'], undefined, 'null entry dropped');
|
||||
assert.ok(/dropped 7 malformed/.test(droppedMessage), 'console.warn reports drop count');
|
||||
} finally {
|
||||
console.warn = origWarn;
|
||||
}
|
||||
}
|
||||
|
||||
// ── readFile tolerates parsed=null / non-object entries field ──
|
||||
{
|
||||
const edgeAdapter = createFakeAdapter();
|
||||
const edgeManager = new cacheManagerModule.CacheManager(
|
||||
edgeAdapter,
|
||||
'.obsidian',
|
||||
'parallel-reader',
|
||||
() => settings.DEFAULT_SETTINGS,
|
||||
);
|
||||
edgeAdapter.files.set(edgeManager.filePath(), JSON.stringify(null));
|
||||
assert.deepStrictEqual(await edgeManager.readFile(), {}, 'parsed=null returns empty cache');
|
||||
edgeAdapter.files.set(edgeManager.filePath(), JSON.stringify({ entries: 'oops' }));
|
||||
assert.deepStrictEqual(await edgeManager.readFile(), {}, 'entries=non-object returns empty cache');
|
||||
}
|
||||
|
||||
console.log('direct cache tests passed');
|
||||
} finally {
|
||||
cleanup();
|
||||
|
|
|
|||
|
|
@ -88,7 +88,7 @@ const { assert, requireBundledModule, cleanup } = require('./direct-test-setup')
|
|||
// ── buildPrompts: card count normalization ──
|
||||
// 0 is falsy so || picks DEFAULT (5), then Math.max(1, 5) = 5
|
||||
const zeroCards = prompt.buildPrompts('doc', { ...base, minCards: 0, maxCards: 0 });
|
||||
assert.ok(zeroCards.system.includes('5-'), 'zero minCards falls back to default');
|
||||
assert.ok(zeroCards.system.includes('1-1'), 'zero minCards/maxCards clamps to 1 (normalizeCardCount floor)');
|
||||
|
||||
const bigCards = prompt.buildPrompts('doc', { ...base, minCards: 5, maxCards: 3 });
|
||||
assert.ok(bigCards.system.includes('5-5'), 'maxCards raised to match minCards');
|
||||
|
|
@ -97,6 +97,12 @@ const { assert, requireBundledModule, cleanup } = require('./direct-test-setup')
|
|||
const defaultMax = prompt.buildPrompts('doc', { ...base, maxDocChars: undefined });
|
||||
assert.ok(defaultMax.user.includes('doc'), 'undefined maxDocChars uses default without truncation');
|
||||
|
||||
// ── buildPrompts: card count is normalized (capped at 30) — keeps prompt ↔ fingerprint in sync ──
|
||||
const oversized = prompt.buildPrompts('doc', { ...base, minCards: 100, maxCards: 200 });
|
||||
assert.ok(oversized.system.includes('30-30'), 'over-cap card count is clamped to 30 in prompt');
|
||||
assert.ok(!oversized.system.includes('100'), 'raw 100 must not appear in prompt');
|
||||
assert.ok(!oversized.system.includes('200'), 'raw 200 must not appear in prompt');
|
||||
|
||||
console.log('direct prompt tests passed');
|
||||
} finally {
|
||||
cleanup();
|
||||
|
|
|
|||
|
|
@ -27,6 +27,145 @@ async function testSingleFlightAndCleanup() {
|
|||
assert.strictEqual(manager.get('note.md'), null);
|
||||
}
|
||||
|
||||
async function testGlobalConcurrencyLimit() {
|
||||
const manager = new GenerationJobManager(2); // max 2 concurrent
|
||||
const releases = [null, null, null, null];
|
||||
const blockers = releases.map(
|
||||
(_, i) =>
|
||||
new Promise((r) => {
|
||||
releases[i] = r;
|
||||
}),
|
||||
);
|
||||
const startedAt = [];
|
||||
|
||||
const promises = ['a.md', 'b.md', 'c.md', 'd.md'].map((path, i) =>
|
||||
manager.start(path, async () => {
|
||||
startedAt.push(i);
|
||||
await blockers[i];
|
||||
return path;
|
||||
}),
|
||||
);
|
||||
|
||||
// Let microtasks flush
|
||||
await new Promise((r) => setTimeout(r, 5));
|
||||
|
||||
assert.strictEqual(manager.isRunning('a.md'), true, 'a is running');
|
||||
assert.strictEqual(manager.isRunning('b.md'), true, 'b is running');
|
||||
assert.strictEqual(manager.isRunning('c.md'), false, 'c is queued, not running yet');
|
||||
assert.strictEqual(manager.isRunning('d.md'), false, 'd is queued');
|
||||
assert.strictEqual(manager.isPending('c.md'), true, 'c is pending (queued)');
|
||||
assert.strictEqual(manager.waitingCount(), 2, 'two waiters queued');
|
||||
assert.deepStrictEqual(startedAt, [0, 1], 'only first 2 entered runner');
|
||||
|
||||
// Release first -> c gets slot
|
||||
releases[0]();
|
||||
await promises[0];
|
||||
await new Promise((r) => setTimeout(r, 5));
|
||||
assert.strictEqual(manager.isRunning('c.md'), true, 'c promoted after a finishes');
|
||||
assert.strictEqual(manager.waitingCount(), 1, 'one waiter left');
|
||||
|
||||
// Same-key dedup while queued: starting c again throws (without waiting)
|
||||
await assert.rejects(
|
||||
() => manager.start('d.md', async () => 'dup'),
|
||||
GenerationJobAlreadyRunningError,
|
||||
'same-key (queued) dedup throws',
|
||||
);
|
||||
|
||||
releases[1]();
|
||||
releases[2]();
|
||||
releases[3]();
|
||||
await Promise.all(promises);
|
||||
assert.strictEqual(manager.waitingCount(), 0);
|
||||
assert.strictEqual(manager.isRunning('d.md'), false, 'all jobs settled');
|
||||
}
|
||||
|
||||
async function testNoOverbookingDuringRelease() {
|
||||
// Regression for the microtask race: when a queued waiter is resolved by
|
||||
// releaseSlot(), a synchronous start() call must NOT slip past with the now-stale
|
||||
// jobs.size and bring concurrent runners above maxConcurrent.
|
||||
const manager = new GenerationJobManager(2);
|
||||
const releases = [null, null, null];
|
||||
const runnerEntered = [];
|
||||
const blockers = releases.map(
|
||||
(_, i) =>
|
||||
new Promise((r) => {
|
||||
releases[i] = r;
|
||||
}),
|
||||
);
|
||||
|
||||
const a = manager.start('a.md', async () => {
|
||||
runnerEntered.push('a');
|
||||
await blockers[0];
|
||||
return 'a';
|
||||
});
|
||||
const b = manager.start('b.md', async () => {
|
||||
runnerEntered.push('b');
|
||||
await blockers[1];
|
||||
return 'b';
|
||||
});
|
||||
// c is queued (slot full); c will block too so we can observe state during transitions.
|
||||
const c = manager.start('c.md', async () => {
|
||||
runnerEntered.push('c');
|
||||
await blockers[2];
|
||||
return 'c';
|
||||
});
|
||||
|
||||
await new Promise((r) => setTimeout(r, 5));
|
||||
assert.strictEqual(manager.waitingCount(), 1, 'c is queued');
|
||||
assert.deepStrictEqual(runnerEntered, ['a', 'b'], 'only 2 runners entered');
|
||||
|
||||
// Release a → c's waiter resolves and continuation queued. c will run and block.
|
||||
releases[0]();
|
||||
await a;
|
||||
await new Promise((r) => setTimeout(r, 5));
|
||||
// After a settles + microtasks: c is now in jobs (b + c running), runnerEntered = a,b,c
|
||||
assert.deepStrictEqual(runnerEntered, ['a', 'b', 'c'], 'c promoted after a finished');
|
||||
|
||||
// Now slots are full again (b + c). New start('d.md') MUST queue, not jump in.
|
||||
const d = manager.start('d.md', async () => {
|
||||
runnerEntered.push('d');
|
||||
return 'd';
|
||||
});
|
||||
await new Promise((r) => setTimeout(r, 5));
|
||||
assert.strictEqual(manager.waitingCount(), 1, 'd queued because b+c are running');
|
||||
assert.deepStrictEqual(runnerEntered, ['a', 'b', 'c'], 'd not entered while b+c running');
|
||||
|
||||
releases[1]();
|
||||
releases[2]();
|
||||
await Promise.all([b, c, d]);
|
||||
assert.ok(runnerEntered.includes('d'), 'd entered eventually');
|
||||
assert.strictEqual(manager.waitingCount(), 0);
|
||||
}
|
||||
|
||||
async function testCancelAllWaiters() {
|
||||
const manager = new GenerationJobManager(1);
|
||||
let release;
|
||||
const blocker = new Promise((r) => {
|
||||
release = r;
|
||||
});
|
||||
|
||||
const running = manager.start('first.md', async () => {
|
||||
await blocker;
|
||||
return 'first';
|
||||
});
|
||||
|
||||
// queued: should be cancellable
|
||||
const queued1 = manager.start('q1.md', async () => 'q1');
|
||||
const queued2 = manager.start('q2.md', async () => 'q2');
|
||||
|
||||
await new Promise((r) => setTimeout(r, 5));
|
||||
assert.strictEqual(manager.waitingCount(), 2, 'two queued');
|
||||
|
||||
const drained = manager.cancelAllWaiters();
|
||||
assert.strictEqual(drained, 2, 'cancelAllWaiters returns drained count');
|
||||
await assert.rejects(queued1, GenerationJobCancelledError, 'q1 rejects with cancelled');
|
||||
await assert.rejects(queued2, GenerationJobCancelledError, 'q2 rejects with cancelled');
|
||||
|
||||
release();
|
||||
assert.strictEqual(await running, 'first', 'running job still completes');
|
||||
assert.strictEqual(manager.waitingCount(), 0);
|
||||
}
|
||||
|
||||
async function testCancelSignalsRunnerAndCleansUp() {
|
||||
const manager = new GenerationJobManager();
|
||||
let cancelHookCalled = false;
|
||||
|
|
@ -58,6 +197,9 @@ function testErrorClassification() {
|
|||
|
||||
(async () => {
|
||||
await testSingleFlightAndCleanup();
|
||||
await testGlobalConcurrencyLimit();
|
||||
await testNoOverbookingDuringRelease();
|
||||
await testCancelAllWaiters();
|
||||
await testCancelSignalsRunnerAndCleansUp();
|
||||
testErrorClassification();
|
||||
console.log('generation job manager tests passed');
|
||||
|
|
|
|||
|
|
@ -7,9 +7,11 @@ const originalConsoleError = console.error;
|
|||
plugin.cacheManager = { get: () => null };
|
||||
plugin.jobs = {
|
||||
isRunning: () => false,
|
||||
isPending: () => false,
|
||||
start: async () => {
|
||||
throw new Error('backend down');
|
||||
},
|
||||
cancelAllWaiters: () => 0,
|
||||
};
|
||||
plugin.t = (key) => key;
|
||||
|
||||
|
|
|
|||
|
|
@ -24,8 +24,21 @@ assert.strictEqual(fp1, fp2, 'same settings = same fingerprint');
|
|||
assert.notStrictEqual(
|
||||
fp1,
|
||||
t.generationFingerprint({ ...baseSettings, model: 'other' }),
|
||||
'different model = different fp',
|
||||
'non-codex backend: different model = different fp',
|
||||
);
|
||||
|
||||
// Codex backend: model is ignored by fingerprint (codex uses its own config)
|
||||
{
|
||||
const codexA = t.generationFingerprint({ ...baseSettings, backend: 'codex', model: 'gpt-5-codex' });
|
||||
const codexB = t.generationFingerprint({ ...baseSettings, backend: 'codex', model: 'claude-sonnet-4-6' });
|
||||
assert.strictEqual(codexA, codexB, 'codex backend: model change does NOT affect fingerprint');
|
||||
// But other fields still affect codex fingerprint
|
||||
assert.notStrictEqual(
|
||||
codexA,
|
||||
t.generationFingerprint({ ...baseSettings, backend: 'codex', model: 'gpt-5-codex', minCards: 999 }),
|
||||
'codex backend: other settings still affect fingerprint',
|
||||
);
|
||||
}
|
||||
assert.notStrictEqual(
|
||||
fp1,
|
||||
t.generationFingerprint({ ...baseSettings, apiMaxTokens: 8192 }),
|
||||
|
|
@ -81,6 +94,13 @@ assert.strictEqual(
|
|||
);
|
||||
assert.strictEqual(t.shouldSkipBatchFile(null, 'test', baseSettings), false, 'missing cache entry is not skipped');
|
||||
|
||||
// normalizeCardCount: cap to 30, fall back on invalid
|
||||
assert.strictEqual(t.normalizeCardCount(50, 5), 30, 'card count over 30 is capped');
|
||||
assert.strictEqual(t.normalizeCardCount(0, 5), 1, 'card count 0 floors to 1');
|
||||
assert.strictEqual(t.normalizeCardCount(-1, 5), 1, 'negative card count floors to 1');
|
||||
assert.strictEqual(t.normalizeCardCount('bad', 5), 5, 'non-numeric card count falls back to default');
|
||||
assert.strictEqual(t.normalizeCardCount('15', 5), 15, 'numeric string accepted');
|
||||
|
||||
// normalizeCliTimeoutMs (mirrors normalizeStreamingTimeoutMs)
|
||||
assert.strictEqual(t.normalizeCliTimeoutMs('60000'), 60000, 'cli timeout accepts numeric strings');
|
||||
assert.strictEqual(t.normalizeCliTimeoutMs(999), 120000, 'cli timeout below minimum falls back to default');
|
||||
|
|
|
|||
|
|
@ -91,4 +91,19 @@ assert.deepStrictEqual(sseMixed.deltas, ['x'], 'comment and event lines ignored'
|
|||
const sseBad = t.parseSseBuffer('data: not_json\n\ndata: {"choices":[{"delta":{"content":"y"}}]}\n\n', openaiExtract);
|
||||
assert.deepStrictEqual(sseBad.deltas, ['y'], 'bad JSON line skipped, good line extracted');
|
||||
|
||||
// ── parseSseBuffer flush behavior: appending \n\n to unterminated event ──
|
||||
// (used by doStreamingFetch on EOF when provider closes without trailing \n\n)
|
||||
{
|
||||
const unterminated = 'data: {"choices":[{"delta":{"content":"final"}}]}';
|
||||
const flushed = t.parseSseBuffer(`${unterminated}\n\n`, openaiExtract);
|
||||
assert.deepStrictEqual(flushed.deltas, ['final'], 'flush of unterminated event extracts last delta');
|
||||
assert.strictEqual(flushed.rest, '', 'flush leaves no remainder');
|
||||
}
|
||||
{
|
||||
// Buffer ending with single \n (often a partial line) — flush by appending one more \n
|
||||
const partial = 'data: {"choices":[{"delta":{"content":"tail"}}]}\n';
|
||||
const flushed = t.parseSseBuffer(`${partial}\n`, openaiExtract);
|
||||
assert.deepStrictEqual(flushed.deltas, ['tail'], 'single-newline buffer flushed correctly');
|
||||
}
|
||||
|
||||
console.log('streaming tests passed');
|
||||
|
|
|
|||
|
|
@ -37,6 +37,7 @@ const expectedExports = [
|
|||
'normalizeSettings',
|
||||
'normalizeStreamingTimeoutMs',
|
||||
'normalizeCliTimeoutMs',
|
||||
'normalizeCardCount',
|
||||
'applyApiProviderPreset',
|
||||
];
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue