diff --git a/.changeset/docx-asset-references.md b/.changeset/docx-asset-references.md new file mode 100644 index 0000000..0e2a789 --- /dev/null +++ b/.changeset/docx-asset-references.md @@ -0,0 +1,7 @@ +--- +'@ontos-ai/knowhere-sdk': patch +--- + +Preserve complete pinned asset paths when DOCX image names contain nested brackets or Markdown link targets contain parentheses. Keep missing-origin and integrity validation in place before publishing a readable snapshot. + +Persist the archive identity and zero-entry checkpoint before the first resource starts, so a first-entry deadline retains its immutable generation for the next invocation. diff --git a/src/knowledge/__tests__/readable-snapshots.test.ts b/src/knowledge/__tests__/readable-snapshots.test.ts index a9eeb1d..969b35d 100644 --- a/src/knowledge/__tests__/readable-snapshots.test.ts +++ b/src/knowledge/__tests__/readable-snapshots.test.ts @@ -1289,6 +1289,40 @@ describe('readable snapshot storage boundary', (): void => { expect(fixture.requests).toHaveLength(requestsBeforeRepeat + 1); }); + it.each([ + ['images/image-[1] Revenue Chart.png', 'brackets'], + ['images/chart-[1[2]].png', 'brackets'], + ['images/chart (draft).png', 'link'], + ['images/chart (draft (two)).png', 'link'], + ] as const)( + 'prepares pinned assets with balanced delimiters in %s', + async (path, form): Promise => { + const fixture: ReadableFixture = await createFixture(); + updateOriginObject(fixture, path, Buffer.from('pinned-image')); + updateOriginObject( + fixture, + 'chunks.json', + Buffer.from( + JSON.stringify({ + chunks: [ + { + chunk_id: 'image_1', + type: 'image', + path, + content: form === 'brackets' ? `[${path}]` : `[Chart](${path})`, + }, + ], + }), + ), + ); + await expect(fixture.knowledge.prepareReadableSnapshot(reference)).resolves.toMatchObject({ + readable: { snapshotId: 'snapshot_baseline' }, + }); + expect(fixture.objects.has('snapshot_baseline/.knowhere-sdk/readable.json')).toBe(true); + expect(fixture.objects.has(`snapshot_baseline/${path}`)).toBe(false); + }, + ); + it('rejects missing image references and corrupt essential bytes before readability', async (): Promise => { const missing: ReadableFixture = await createFixture(); missing.inventory.splice( diff --git a/src/storage/__tests__/streamed-sync-concurrency.test.ts b/src/storage/__tests__/streamed-sync-concurrency.test.ts index 812926c..8a65e0b 100644 --- a/src/storage/__tests__/streamed-sync-concurrency.test.ts +++ b/src/storage/__tests__/streamed-sync-concurrency.test.ts @@ -272,6 +272,49 @@ describe('bounded parsed archive concurrency', () => { expect(fixture.progress).toHaveLength(0); }); + it('pins the archive before a first-entry timeout and resumes in a new invocation', async () => { + const fixture: Fixture = await createFixture(); + const deadline: AbortController = createSegmentDeadline(); + let hasHeldEntry: boolean = false; + const write = fixture.storage.writeObjectStream?.bind(fixture.storage); + if (!write) throw new Error('Fixture requires streaming storage.'); + fixture.storage.writeObjectStream = async ( + params: ParsedDocumentWriteObjectStreamParams, + ): Promise => { + if (params.path === 'manifest.json') { + hasHeldEntry = true; + await new Promise((_resolve, reject) => { + params.signal?.addEventListener( + 'abort', + () => reject(readCancellationReason(params.signal)), + { + once: true, + }, + ); + }); + } + return write(params); + }; + const pending: Promise = fixture.sync( + { archiveSyncDeadlineMs: 60_000, maxConcurrentFiles: 1 }, + AbortSignal.timeout(2_000), + ); + await vi.waitFor(() => expect(hasHeldEntry).toBe(true)); + const pin: ParsedDocumentSyncProgress | undefined = fixture.progress.at(-1); + deadline.abort(new DOMException('Segment budget expired', 'TimeoutError')); + await expect(pending).resolves.toMatchObject({ completed: false, revisionKey: 'revision' }); + expect(pin).toMatchObject({ + pinnedRevisionKey: 'revision', + nextArchiveEntry: 0, + archiveEntryHashes: [], + status: 'running', + }); + expect(pin?.archiveIdentity).toEqual(expect.any(String)); + fixture.storage.writeObjectStream = write; + await expect(fixture.sync()).resolves.toMatchObject({ completed: true }); + expect(fixture.progress.at(-1)).toMatchObject({ nextArchiveEntry: 10, status: 'completed' }); + }); + it('reports checkpoint failure instead of acknowledging a deadline handoff', async () => { const fixture: Fixture = await createFixture(); const deadline: AbortController = createSegmentDeadline(); @@ -296,7 +339,9 @@ describe('bounded parsed archive concurrency', () => { } return write(params); }; - fixture.storage.writeSyncProgress = (): Promise => Promise.reject(checkpointFailure); + const writeProgress = fixture.storage.writeSyncProgress.bind(fixture.storage); + fixture.storage.writeSyncProgress = (progress): Promise => + progress.nextArchiveEntry === 0 ? writeProgress(progress) : Promise.reject(checkpointFailure); const pending: Promise = fixture.sync( { archiveSyncDeadlineMs: 60_000, @@ -387,7 +432,7 @@ describe('bounded parsed archive concurrency', () => { }); fixture.storage.writeSyncProgress = async (value): Promise => { fixture.progress.push(value); - if (fixture.progress.length === 1) { + if ((value.nextArchiveEntry ?? 0) > 0 && fixture.progress.length === 2) { markCheckpoint?.(); await checkpointGate; } @@ -403,7 +448,7 @@ describe('bounded parsed archive concurrency', () => { releaseCheckpoint?.(); await sync; const running: ParsedDocumentSyncProgress[] = fixture.progress.filter( - (value) => value.status === 'running', + (value) => value.status === 'running' && (value.nextArchiveEntry ?? 0) > 0, ); expect(running.length).toBeLessThan(10); expect(running[0].nextArchiveEntry).toBeGreaterThanOrEqual(4); @@ -916,7 +961,8 @@ describe('bounded parsed archive concurrency', () => { { ...entry, filename: 'chunks.json' }, ]); await expect(fixture.sync({ maxEntriesPerSync: 1 })).rejects.toThrow('Late producer failed'); - expect(fixture.progress).toHaveLength(0); + expect(fixture.progress).toHaveLength(1); + expect(fixture.progress[0].nextArchiveEntry).toBe(0); expect(fixture.objects.has('.knowhere-sdk/commit.json')).toBe(false); }); @@ -962,7 +1008,8 @@ describe('bounded parsed archive concurrency', () => { ]); expect(state).toBe('waiting'); expect(hasCleanedProducer).toBe(false); - expect(fixture.progress).toHaveLength(0); + expect(fixture.progress).toHaveLength(1); + expect(fixture.progress[0].nextArchiveEntry).toBe(0); } finally { releaseCleanup?.(); await sync; @@ -1008,7 +1055,9 @@ describe('bounded parsed archive concurrency', () => { checkpointFileInterval: 50, checkpointIntervalMs: 3, }); - expect(fixture.progress[0].nextArchiveEntry).toBe(2); + expect( + fixture.progress.find((value) => (value.nextArchiveEntry ?? 0) > 0)?.nextArchiveEntry, + ).toBe(2); expect(fixture.progress.filter((value) => value.status === 'running').length).toBeGreaterThan( 1, ); diff --git a/src/storage/readable-parsed-document-snapshots.ts b/src/storage/readable-parsed-document-snapshots.ts index 792018a..0a9ddd6 100644 --- a/src/storage/readable-parsed-document-snapshots.ts +++ b/src/storage/readable-parsed-document-snapshots.ts @@ -515,15 +515,37 @@ function collectReferences(value: unknown, references: Set): void { if (key === 'path' && /^(?:images|tables|page_pdfs)\//.test(child)) references.add(validateObjectPath(child)); if (['ref', 'content', 'summary'].includes(key)) { - for (const match of child.matchAll(/\[((?:images|tables)\/[^\]\r\n]+)\]/g)) - references.add(validateObjectPath(match[1])); - for (const match of child.matchAll(/\]\(((?:images|tables)\/[^)\r\n]+)\)/g)) - references.add(validateObjectPath(match[1])); + for (const path of readInlineResourcePaths(child)) references.add(validateObjectPath(path)); } } else collectReferences(child, references); } } +/** Asset names may contain nested brackets or parentheses inside their outer reference. */ +function readInlineResourcePaths(content: string): readonly string[] { + const paths: string[] = []; + const starts: RegExp = /\[(?=(?:images|tables)\/)|\]\((?=(?:images|tables)\/)/g; + let match: RegExpExecArray | null; + while ((match = starts.exec(content)) !== null) { + const opening: string = match[0].endsWith('(') ? '(' : '['; + const closing: string = opening === '(' ? ')' : ']'; + const start: number = match.index + match[0].length; + let depth: number = 1; + let index: number = start; + for (; index < content.length; index += 1) { + const character: string = content[index]; + if (character === '\r' || character === '\n') break; + if (character === opening) depth += 1; + if (character === closing) depth -= 1; + if (depth !== 0) continue; + paths.push(content.slice(start, index)); + break; + } + starts.lastIndex = index + 1; + } + return paths; +} + function isImagePath(path: string): boolean { return /\.(?:png|jpe?g|gif|webp|svg|avif|bmp|tiff?)$/i.test(path); } diff --git a/src/storage/streamed-parsed-document-sync.ts b/src/storage/streamed-parsed-document-sync.ts index bcbb46d..a09d81c 100644 --- a/src/storage/streamed-parsed-document-sync.ts +++ b/src/storage/streamed-parsed-document-sync.ts @@ -116,6 +116,10 @@ export async function syncRemoteResultArchive( let successfulEntries: number = 0; let checkpointEntry: number = nextEntry; let checkpointAt: number = Date.now(); + let hasPersistedCheckpoint: boolean = + previous?.status === 'running' && + previous.archiveIdentity === source.identity && + previous.pinnedRevisionKey === params.revisionKey; let failure: Error | undefined; let archiveCacheBypassed: boolean | undefined = source.archiveCacheBypassed; @@ -128,7 +132,8 @@ export async function syncRemoteResultArchive( entryHashes[contiguousEntry] = successfulHashes.get(contiguousEntry)!; contiguousEntry += 1; } - if (contiguousEntry === checkpointEntry) return; + if (contiguousEntry === checkpointEntry && (checkpointEntry > 0 || hasPersistedCheckpoint)) + return; await params.storage.writeSyncProgress( { documentId: params.documentId, @@ -151,6 +156,7 @@ export async function syncRemoteResultArchive( checkpointEntry = contiguousEntry; successfulEntries = successfulHashes.size; checkpointAt = Date.now(); + hasPersistedCheckpoint = true; } function startEntry(entry: ArchiveEntry, index: number): void { @@ -214,6 +220,9 @@ export async function syncRemoteResultArchive( }; } } + // Pin identity before the first potentially large entry starts. A timeout + // with zero completed entries must still hand off a durable generation. + await flushProgress(true); for await (const entry of source.entries) { if (failure !== undefined) throw failure; signal.throwIfAborted();