From 9f2af700957af472ac3a8c280d38b527f2155914 Mon Sep 17 00:00:00 2001 From: Sylvester Kaczmarek <16242628+sylvesterkaczmarek@users.noreply.github.com> Date: Thu, 8 Oct 2026 19:01:48 +0100 Subject: [PATCH 1/2] Honor cancellation during PDF knowledge extraction Signed-off-by: Sylvester Kaczmarek <16242628+sylvesterkaczmarek@users.noreply.github.com> --- sdk/typescript/src/knowledge-base.ts | 16 ++- .../knowledge-base-pdf-cancellation.test.ts | 123 ++++++++++++++++++ 2 files changed, 136 insertions(+), 3 deletions(-) create mode 100644 sdk/typescript/tests-ts/knowledge-base-pdf-cancellation.test.ts diff --git a/sdk/typescript/src/knowledge-base.ts b/sdk/typescript/src/knowledge-base.ts index 2df9c138b3..c70cd36f99 100644 --- a/sdk/typescript/src/knowledge-base.ts +++ b/sdk/typescript/src/knowledge-base.ts @@ -97,7 +97,7 @@ export async function readKnowledgeBaseSnapshot( const extension = extname(document).toLowerCase(); const text = extension === ".pdf" - ? await extractPdf(document, bytes) + ? await extractPdf(document, bytes, signal) : extension === ".docx" ? extractDocx(document, bytes) : decodeText(document, bytes); @@ -251,7 +251,11 @@ function decodeText(path: string, bytes: Uint8Array): string { } } -async function extractPdf(path: string, bytes: Uint8Array): Promise { +async function extractPdf( + path: string, + bytes: Uint8Array, + signal?: AbortSignal, +): Promise { try { const { getDocument, VerbosityLevel } = await import("pdfjs-dist/legacy/build/pdf.mjs"); @@ -262,9 +266,14 @@ async function extractPdf(path: string, bytes: Uint8Array): Promise { }); try { const document = await loadingTask.promise; + signal?.throwIfAborted(); const pages: string[] = []; for (let number = 1; number <= document.numPages; number++) { - const content = await (await document.getPage(number)).getTextContent(); + signal?.throwIfAborted(); + const page = await document.getPage(number); + signal?.throwIfAborted(); + const content = await page.getTextContent(); + signal?.throwIfAborted(); pages.push( content.items .map((item) => ("str" in item ? item.str : "")) @@ -276,6 +285,7 @@ async function extractPdf(path: string, bytes: Uint8Array): Promise { await loadingTask.destroy(); } } catch (error) { + if (signal?.aborted) signal.throwIfAborted(); throw new Error(`Cannot extract text from knowledge base PDF: ${path}`, { cause: error, }); diff --git a/sdk/typescript/tests-ts/knowledge-base-pdf-cancellation.test.ts b/sdk/typescript/tests-ts/knowledge-base-pdf-cancellation.test.ts new file mode 100644 index 0000000000..c5c403c4ab --- /dev/null +++ b/sdk/typescript/tests-ts/knowledge-base-pdf-cancellation.test.ts @@ -0,0 +1,123 @@ +import { mkdir, readFile, readdir, writeFile } from "node:fs/promises"; +import { join } from "node:path"; +import { afterEach, expect, mock, test } from "bun:test"; +import { createApiTestFixtures } from "./support/temporary-directories.js"; +import { runTestInSubprocess } from "./support/test-subprocess.js"; + +const { temporaryDirectory, cleanup } = createApiTestFixtures(); +afterEach(cleanup); + +test.each([ + "load", + "page", + "text", + "empty final page", + "cleanup failure", + "not canceled", +] as const)("PDF extraction cancellation boundary: %s", async (boundary) => { + if ( + runTestInSubprocess( + import.meta.path, + `PDF extraction cancellation boundary: ${boundary}`, + ) + ) + return; + + const root = await temporaryDirectory(); + const documentPath = join(root, "architecture.pdf"); + const staging = join(root, "staging"); + await mkdir(staging); + await writeFile(documentPath, "%PDF-synthetic-loader-fixture"); + const reason = new Error("Synthetic PDF extraction cancellation"); + let controller = new AbortController(); + let pages: number[] = []; + let textPages: number[] = []; + let destroyed = 0; + mock.module("pdfjs-dist/legacy/build/pdf.mjs", () => ({ + VerbosityLevel: { ERRORS: 0 }, + getDocument: () => ({ + promise: Promise.resolve().then(() => { + if (boundary === "load") controller.abort(reason); + return { + numPages: boundary === "empty final page" ? 1 : 3, + async getPage(number: number) { + pages.push(number); + if (boundary === "page") controller.abort(reason); + return { + async getTextContent() { + textPages.push(number); + if ( + boundary === "text" || + boundary === "empty final page" || + boundary === "cleanup failure" + ) + controller.abort(reason); + return { + items: + boundary === "empty final page" + ? [] + : [{ str: `Synthetic page ${number}` }], + }; + }, + }; + }, + }; + }), + async destroy() { + destroyed++; + if (boundary === "cleanup failure") + throw new Error("Synthetic PDF cleanup failure"); + }, + }), + })); + const { readKnowledgeBaseSnapshot, prepareKnowledgeBase } = + await import("../src/knowledge-base.js"); + for (const entrypoint of ["snapshot", "prepared"] as const) { + controller = new AbortController(); + pages = []; + textPages = []; + destroyed = 0; + const read = async () => { + if (entrypoint === "snapshot") { + const snapshot = await readKnowledgeBaseSnapshot( + [documentPath], + controller.signal, + ); + return Object.values(snapshot.documents); + } + const prepared = await prepareKnowledgeBase( + [documentPath], + controller.signal, + staging, + ); + try { + return await Promise.all( + (await readdir(prepared.path)).map((name) => + readFile(join(prepared.path, name), "utf8"), + ), + ); + } finally { + await prepared.cleanup(); + } + }; + if (boundary === "not canceled") { + await expect(read()).resolves.toEqual([ + "Synthetic page 1\nSynthetic page 2\nSynthetic page 3", + ]); + } else { + await expect(read()).rejects.toBe(reason); + } + expect(pages).toEqual( + boundary === "not canceled" ? [1, 2, 3] : boundary === "load" ? [] : [1], + ); + expect(textPages).toEqual( + boundary === "not canceled" + ? [1, 2, 3] + : boundary === "load" || boundary === "page" + ? [] + : [1], + ); + expect(destroyed).toBe(1); + expect(await readdir(staging)).toEqual([]); + } +}); From c08f51ba22cd789d092a1a4bf2436bd967b34a89 Mon Sep 17 00:00:00 2001 From: Michael D'Angelo Date: Sun, 25 Oct 2026 20:54:02 +0000 Subject: [PATCH 2/2] fix(knowledge): preserve cancellation through PDF cleanup --- sdk/typescript/src/knowledge-base.ts | 3 +- .../knowledge-base-pdf-cancellation.test.ts | 32 ++++++++++++------- 2 files changed, 23 insertions(+), 12 deletions(-) diff --git a/sdk/typescript/src/knowledge-base.ts b/sdk/typescript/src/knowledge-base.ts index 1dcb4b2596..4b0bebddb6 100644 --- a/sdk/typescript/src/knowledge-base.ts +++ b/sdk/typescript/src/knowledge-base.ts @@ -288,9 +288,10 @@ async function extractPdf( return pages.join("\n"); } finally { await loadingTask.destroy(); + signal?.throwIfAborted(); } } catch (error) { - if (signal?.aborted) signal.throwIfAborted(); + signal?.throwIfAborted(); throw new Error(`Cannot extract text from knowledge base PDF: ${path}`, { cause: error, }); diff --git a/sdk/typescript/tests-ts/knowledge-base-pdf-cancellation.test.ts b/sdk/typescript/tests-ts/knowledge-base-pdf-cancellation.test.ts index c5c403c4ab..3b68f2fff5 100644 --- a/sdk/typescript/tests-ts/knowledge-base-pdf-cancellation.test.ts +++ b/sdk/typescript/tests-ts/knowledge-base-pdf-cancellation.test.ts @@ -10,9 +10,10 @@ afterEach(cleanup); test.each([ "load", "page", - "text", "empty final page", + "during cleanup", "cleanup failure", + "uncanceled cleanup failure", "not canceled", ] as const)("PDF extraction cancellation boundary: %s", async (boundary) => { if ( @@ -29,6 +30,11 @@ test.each([ await mkdir(staging); await writeFile(documentPath, "%PDF-synthetic-loader-fixture"); const reason = new Error("Synthetic PDF extraction cancellation"); + const cleanupError = new Error("Synthetic PDF cleanup failure"); + const emptyPage = + boundary === "empty final page" || boundary === "during cleanup"; + const allPages = + boundary === "not canceled" || boundary === "uncanceled cleanup failure"; let controller = new AbortController(); let pages: number[] = []; let textPages: number[] = []; @@ -39,7 +45,7 @@ test.each([ promise: Promise.resolve().then(() => { if (boundary === "load") controller.abort(reason); return { - numPages: boundary === "empty final page" ? 1 : 3, + numPages: emptyPage ? 1 : 3, async getPage(number: number) { pages.push(number); if (boundary === "page") controller.abort(reason); @@ -47,16 +53,12 @@ test.each([ async getTextContent() { textPages.push(number); if ( - boundary === "text" || boundary === "empty final page" || boundary === "cleanup failure" ) controller.abort(reason); return { - items: - boundary === "empty final page" - ? [] - : [{ str: `Synthetic page ${number}` }], + items: emptyPage ? [] : [{ str: `Synthetic page ${number}` }], }; }, }; @@ -65,8 +67,12 @@ test.each([ }), async destroy() { destroyed++; - if (boundary === "cleanup failure") - throw new Error("Synthetic PDF cleanup failure"); + if (boundary === "during cleanup") controller.abort(reason); + if ( + boundary === "cleanup failure" || + boundary === "uncanceled cleanup failure" + ) + throw cleanupError; }, }), })); @@ -104,14 +110,18 @@ test.each([ await expect(read()).resolves.toEqual([ "Synthetic page 1\nSynthetic page 2\nSynthetic page 3", ]); + } else if (boundary === "uncanceled cleanup failure") { + await expect(read()).rejects.toMatchObject({ + cause: { cause: cleanupError }, + }); } else { await expect(read()).rejects.toBe(reason); } expect(pages).toEqual( - boundary === "not canceled" ? [1, 2, 3] : boundary === "load" ? [] : [1], + allPages ? [1, 2, 3] : boundary === "load" ? [] : [1], ); expect(textPages).toEqual( - boundary === "not canceled" + allPages ? [1, 2, 3] : boundary === "load" || boundary === "page" ? []