Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions .changeset/docx-asset-references.md
Original file line number Diff line number Diff line change
@@ -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.
34 changes: 34 additions & 0 deletions src/knowledge/__tests__/readable-snapshots.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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<void> => {
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<void> => {
const missing: ReadableFixture = await createFixture();
missing.inventory.splice(
Expand Down
61 changes: 55 additions & 6 deletions src/storage/__tests__/streamed-sync-concurrency.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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<ParsedDocumentWriteObjectResult> => {
if (params.path === 'manifest.json') {
hasHeldEntry = true;
await new Promise<void>((_resolve, reject) => {
params.signal?.addEventListener(
'abort',
() => reject(readCancellationReason(params.signal)),
{
once: true,
},
);
});
}
return write(params);
};
const pending: Promise<KnowledgeSyncParsedDocumentResponse> = 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();
Expand All @@ -296,7 +339,9 @@ describe('bounded parsed archive concurrency', () => {
}
return write(params);
};
fixture.storage.writeSyncProgress = (): Promise<void> => Promise.reject(checkpointFailure);
const writeProgress = fixture.storage.writeSyncProgress.bind(fixture.storage);
fixture.storage.writeSyncProgress = (progress): Promise<void> =>
progress.nextArchiveEntry === 0 ? writeProgress(progress) : Promise.reject(checkpointFailure);
const pending: Promise<KnowledgeSyncParsedDocumentResponse> = fixture.sync(
{
archiveSyncDeadlineMs: 60_000,
Expand Down Expand Up @@ -387,7 +432,7 @@ describe('bounded parsed archive concurrency', () => {
});
fixture.storage.writeSyncProgress = async (value): Promise<void> => {
fixture.progress.push(value);
if (fixture.progress.length === 1) {
if ((value.nextArchiveEntry ?? 0) > 0 && fixture.progress.length === 2) {
markCheckpoint?.();
await checkpointGate;
}
Expand All @@ -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);
Expand Down Expand Up @@ -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);
});

Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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,
);
Expand Down
30 changes: 26 additions & 4 deletions src/storage/readable-parsed-document-snapshots.ts
Original file line number Diff line number Diff line change
Expand Up @@ -515,15 +515,37 @@ function collectReferences(value: unknown, references: Set<string>): 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);
}
Expand Down
11 changes: 10 additions & 1 deletion src/storage/streamed-parsed-document-sync.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand All @@ -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,
Expand All @@ -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 {
Expand Down Expand Up @@ -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();
Expand Down
Loading