From e42beed0813091eb91cde403e87384ad60597f74 Mon Sep 17 00:00:00 2001 From: Maximo Guk <62088388+Maximo-Guk@users.noreply.github.com> Date: Mon, 9 Nov 2026 16:17:31 -0600 Subject: [PATCH 1/4] Share the streamed base64-in-JSON request body builder convertToMarkdown() built its request body, a JSON document holding the file as one base64 string encoded chunk by chunk, in a private helper tied to the conversion's own JSON. The helper now takes the JSON around the string, so another streamed Workers AI request can use it. No behaviour change. Co-Authored-By: Claude Fable 5.2 --- .../src/workers-ai-markdown.ts | 40 ++++++++++++------- 1 file changed, 25 insertions(+), 15 deletions(-) diff --git a/packages/gatekeeper-google/src/workers-ai-markdown.ts b/packages/gatekeeper-google/src/workers-ai-markdown.ts index d04b41518a..95aecd0b92 100644 --- a/packages/gatekeeper-google/src/workers-ai-markdown.ts +++ b/packages/gatekeeper-google/src/workers-ai-markdown.ts @@ -63,7 +63,10 @@ export type MarkdownConversion = { export async function convertToMarkdown( ai: AiFetcher, source: MarkdownSource, options: object, ): Promise { - let { body, sent, cancelSource } = requestBody(source, options); + // {"files":[{name, mimeType, data}],"options":…} + let file = `"name":${JSON.stringify(source.name)},"mimeType":${JSON.stringify(source.mimeType)}`; + let { body, sent, cancelSource } = base64JsonBody( + `{"files":[{${file},"data":"`, source.body, `"}],"options":${JSON.stringify(options)}}`); let fail = async (failure: ConversionFailure): Promise => { // A fetch holds the body locked, so the source is released through its reader instead. await cancelSource(); @@ -103,27 +106,34 @@ export async function convertToMarkdown( }; } -/** - * The conversion request's JSON body, `{"files":[{name, mimeType, data}],"options":…}`, with `data` - * base64-encoded from the source one chunk at a time as the request is sent. `sent` counts the - * source's bytes read and keeps the error the source itself failed with, if any. Cancelling the - * body cancels the source; `cancelSource` does so too while the body is locked to a request, and - * does nothing once the source has been read to its end. - */ -function requestBody(source: MarkdownSource, options: object): { +/** A request body that streams its source, and what has become of that source. */ +export type StreamedBody = { body: ReadableStream; + /** The source's bytes read so far, and the error the source itself failed with, if any. */ sent: { bytes: number; error?: { error: unknown } }; + /** + * Cancels the source while the body is locked to a request; cancelling the body does so too. + * Does nothing once the source has been read to its end. + */ cancelSource(): Promise; -} { +}; + +/** + * A JSON request body holding `source` as one base64 string: `prefix`, which ends inside the + * string's opening quote, then the source base64-encoded one chunk at a time as the request is + * sent, then `suffix`, which begins with the closing quote. + */ +export function base64JsonBody( + prefix: string, source: ReadableStream, suffix: string, +): StreamedBody { let encoder = new TextEncoder(); - let reader = source.body.getReader(); - let sent: { bytes: number; error?: { error: unknown } } = { bytes: 0 }; + let reader = source.getReader(); + let sent: StreamedBody["sent"] = { bytes: 0 }; // The 0-2 bytes left over from the previous chunk, since base64 encodes whole 3-byte groups. let carry = new Uint8Array(0); let body = new ReadableStream({ start(controller) { - let file = `"name":${JSON.stringify(source.name)},"mimeType":${JSON.stringify(source.mimeType)}`; - controller.enqueue(encoder.encode(`{"files":[{${file},"data":"`)); + controller.enqueue(encoder.encode(prefix)); }, async pull(controller) { // A pull that enqueues nothing is not called again, so read until there is something to @@ -137,7 +147,7 @@ function requestBody(source: MarkdownSource, options: object): { throw error; } if (chunk.done) { - controller.enqueue(encoder.encode(`${carry.toBase64()}"}],"options":${JSON.stringify(options)}}`)); + controller.enqueue(encoder.encode(`${carry.toBase64()}${suffix}`)); controller.close(); return; } From 85f4fff3daa67371bc342450dfc4bc1f04b63ef8 Mon Sep 17 00:00:00 2001 From: Maximo Guk <62088388+Maximo-Guk@users.noreply.github.com> Date: Mon, 9 Nov 2026 16:33:56 -0600 Subject: [PATCH 2/4] Transcribe an image with a Workers AI vision model, streamed both ways transcribeImage() posts what Ai.run() would, through the binding's fetch, so the image is base64-encoded into the request as it arrives instead of being held as one string, and reads the model's answer as server-sent events only as the text is asked for. It mirrors convertToMarkdown(): the same failure shape, logged by status and error code only, and the same release of both streams on failure or cancel. The model is @cf/google/gemma-4-26b-a4b-it at temperature 0 with thinking off. Measured on scanned pages, thinking tripled a page's tokens and time and held the text back until it was over, without reading a poor scan any better. Nothing calls it yet. Co-Authored-By: Claude Fable 5.2 --- .../__tests__/workers-ai-fixture.ts | 47 ++++ .../__tests__/workers-ai-vision.test.ts | 200 ++++++++++++++++ .../src/workers-ai-vision.ts | 217 ++++++++++++++++++ 3 files changed, 464 insertions(+) create mode 100644 packages/gatekeeper-google/__tests__/workers-ai-vision.test.ts create mode 100644 packages/gatekeeper-google/src/workers-ai-vision.ts diff --git a/packages/gatekeeper-google/__tests__/workers-ai-fixture.ts b/packages/gatekeeper-google/__tests__/workers-ai-fixture.ts index 18c8330c97..d26439da5a 100644 --- a/packages/gatekeeper-google/__tests__/workers-ai-fixture.ts +++ b/packages/gatekeeper-google/__tests__/workers-ai-fixture.ts @@ -83,6 +83,53 @@ export function fakeAi(answer: Response | Error | ((sent: Sent) => Response)) { return { ai, sent }; } +/** The body of a vision request, as far as the tests read it. */ +export type VisionJson = { + inputs: { + messages: { role: string; content: { type: string; text?: string; image_url?: { url: string } }[] }[]; + [setting: string]: unknown; + }; + options: unknown; +}; + +/** The image a vision request carried: its data URI's MIME type, and its bytes. */ +export function sentImage(sent: Sent): { mimeType: string; bytes: Uint8Array } { + let { inputs } = sent.json as unknown as VisionJson; + let url = inputs.messages[0]!.content.find(part => part.type === "image_url")!.image_url!.url; + let [, mimeType, data] = /^data:([^;]+);base64,(.*)$/.exec(url)!; + return { mimeType: mimeType!, bytes: decodeBase64(data!) }; +} + +/** One event of a model's streamed answer. */ +function event(delta: object, finishReason: string | null = null): string { + return `data: ${JSON.stringify({ + id: "a", object: "chat.completion.chunk", model: "@cf/google/gemma-4-26b-a4b-it", + choices: [{ index: 0, delta, logprobs: null, finish_reason: finishReason }], + usage: { prompt_tokens: 0, completion_tokens: 1 }, p: "abc", + })}\n\n`; +} + +/** + * A model's streamed answer as Workers AI sends it: an opening event, one per piece of text, a + * closing one carrying `finish`, the usage summary, and `[DONE]`. + */ +export function tokenEvents(pieces: string[], finish = "stop"): string { + return event({ reasoning_content: null, role: "assistant", content: "" }) + + pieces.map(content => event({ content })).join("") + + event({ content: "" }, finish) + + `data: ${JSON.stringify({ id: "a", choices: [], usage: {} })}\n\n` + + `data: ${JSON.stringify({ response: "", usage: { prompt_tokens: 318, completion_tokens: 340 } })}\n\n` + + "data: [DONE]\n\n"; +} + +/** A response streaming server-sent `events` out `size` bytes at a time. */ +export function eventResponse(events: string, size = 11, status = 200) { + let bytes = new TextEncoder().encode(events); + return new Response(byteStream(bytes, [size]).stream, { + status, headers: { "Content-Type": "text/event-stream" }, + }); +} + export function decodeBase64(data: string): Uint8Array { return Uint8Array.from(atob(data), ch => ch.charCodeAt(0)); } diff --git a/packages/gatekeeper-google/__tests__/workers-ai-vision.test.ts b/packages/gatekeeper-google/__tests__/workers-ai-vision.test.ts new file mode 100644 index 0000000000..006fe23e44 --- /dev/null +++ b/packages/gatekeeper-google/__tests__/workers-ai-vision.test.ts @@ -0,0 +1,200 @@ +import { describe, expect, it } from "vitest"; +import type { AiFetcher } from "../src/workers-ai-markdown"; +import { transcribeImage, TranscriptionError, type Transcription } from "../src/workers-ai-vision"; +import { + BYTES, byteStream, eventResponse, fakeAi, sentImage, tokenEvents, type VisionJson, +} from "./workers-ai-fixture"; + +const RUN_URL = "https://workers-binding.ai/run?version=3"; +const GATEWAY_RUN_URL = "https://workers-binding.ai/ai-gateway/run?version=3"; +const PAGE = ["Harbour Ledger", " Review\n\n", "Lot 55-0193-G, ", "18,442.07 florins."]; + +function jpeg(body: ReadableStream = byteStream(BYTES).stream) { + return { mimeType: "image/jpeg" as const, body }; +} + +/** Every piece of a transcription's text, joined. */ +async function drain(transcription: Transcription): Promise { + let text = ""; + for (let piece = await transcription.next(); piece !== undefined; piece = await transcription.next()) { + expect(piece).not.toBe(""); + text += piece; + } + return text; +} + +/** The text a transcription answered with `events` gives. */ +async function textOf(events: string, size?: number): Promise { + return drain(await transcribeImage(fakeAi(eventResponse(events, size)).ai, undefined, jpeg())); +} + +async function failure(run: Promise): Promise { + let error = await run.catch((caught: unknown) => caught); + expect(error).toBeInstanceOf(TranscriptionError); + return error as TranscriptionError; +} + +describe("transcribeImage", () => { + it("posts the image as a data URI to the model, with thinking off", async () => { + let { ai, sent } = fakeAi(eventResponse(tokenEvents(PAGE))); + + let transcription = await transcribeImage(ai, undefined, jpeg(byteStream(BYTES, [1, 2, 300]).stream)); + expect(await drain(transcription)).toBe(PAGE.join("")); + expect(transcription).toMatchObject({ sentBytes: BYTES.byteLength, truncated: false }); + + expect(sent).toHaveLength(1); + expect(sent[0]!.url).toBe(RUN_URL); + expect(sent[0]!.init.method).toBe("POST"); + expect(sent[0]!.init.headers).toEqual({ + "Content-Type": "application/json", + "cf-consn-sdk-version": "2.0.0", + "cf-consn-model-id": "@cf/google/gemma-4-26b-a4b-it", + }); + expect(sentImage(sent[0]!)).toEqual({ mimeType: "image/jpeg", bytes: BYTES }); + let { inputs, options } = sent[0]!.json as unknown as VisionJson; + expect(options).toEqual({}); + let { messages, ...settings } = inputs; + expect(settings).toEqual({ + temperature: 0, max_tokens: 8192, stream: true, + chat_template_kwargs: { enable_thinking: false }, + }); + expect(messages).toHaveLength(1); + expect(messages[0]!.role).toBe("user"); + expect(messages[0]!.content.map(part => part.type)).toEqual(["text", "image_url"]); + expect(messages[0]!.content[0]!.text).toContain("Transcribe"); + }); + + it("goes through the named AI Gateway's endpoint", async () => { + let { ai, sent } = fakeAi(eventResponse(tokenEvents(PAGE))); + await drain(await transcribeImage(ai, "team-gateway", { mimeType: "image/png", body: byteStream(BYTES).stream })); + expect(sent[0]!.url).toBe(GATEWAY_RUN_URL); + expect((sent[0]!.json as unknown as VisionJson).options).toEqual({ gateway: { id: "team-gateway" } }); + expect(sentImage(sent[0]!).mimeType).toBe("image/png"); + }); + + it("reads the text whatever the size of the response's chunks", async () => { + let pieces = ["Tab\tle | \"quoted\" \\ back", "slash\n", "é ü 你好 ", "😀", " end"]; + for (let size of [1, 2, 3, 5, 64, 100_000]) { + expect(await textOf(tokenEvents(pieces), size)).toBe(pieces.join("")); + } + }); + + it("keeps the halves of a surrogate pair sent in two events together", async () => { + let transcription = await transcribeImage( + fakeAi(eventResponse(tokenEvents(["a\ud83d", "\ude00b"]), 100_000)).ai, undefined, jpeg()); + let pieces: string[] = []; + for (let piece = await transcription.next(); piece !== undefined; piece = await transcription.next()) { + pieces.push(piece); + } + expect(pieces.join("")).toBe("a😀b"); + for (let piece of pieces) expect(new TextDecoder().decode(new TextEncoder().encode(piece))).toBe(piece); + }); + + it("passes over what the model thinks, and takes text sent as bare responses", async () => { + let thinking = `data: ${JSON.stringify({ choices: [{ delta: { reasoning_content: "I need", reasoning: "to" } }] })}\n\n`; + expect(await textOf(`${thinking}${tokenEvents(["text"])}`)).toBe("text"); + expect(await textOf(`: comment\nevent: x\ndata: {"response":"plain "}\r\ndata:{"response":"events"}\n\ndata: [DONE]\n`)) + .toBe("plain events"); + }); + + it("gives no text for a page the model finds none on", async () => { + expect(await textOf(tokenEvents([]))).toBe(""); + }); + + it("says when the model was stopped at its output limit", async () => { + let transcription = await transcribeImage( + fakeAi(eventResponse(tokenEvents(["cut"], "length"))).ai, undefined, jpeg()); + expect(await drain(transcription)).toBe("cut"); + expect(transcription.truncated).toBe(true); + }); + + it("stops reading the response once the model says it is done", async () => { + let state = { cancelled: false, pulls: 0 }; + let events = new TextEncoder().encode(tokenEvents(["all"])); + let body = new ReadableStream({ + pull(controller) { + // Sends the whole answer, then never ends. + if (state.pulls++ === 0) controller.enqueue(events); + }, + cancel() { state.cancelled = true; }, + }, { highWaterMark: 0 }); + expect(await drain(await transcribeImage(fakeAi(new Response(body)).ai, undefined, jpeg()))).toBe("all"); + expect(state.cancelled).toBe(true); + }); + + it("reports a refused request by its status and error code only", async () => { + let image = byteStream(BYTES); + let body = JSON.stringify({ + name: "AiError", internalCode: 8006, httpCode: 400, + message: "AiError: Invalid data for body - secret page text", + }); + let error = await failure(transcribeImage( + fakeAi(new Response(body, { status: 400 })).ai, undefined, jpeg(image.stream))); + expect(error.failure).toEqual({ httpStatus: 400, providerCode: 8006 }); + expect(error.message).toBe("Workers AI could not transcribe the image."); + expect(JSON.stringify(error.failure)).not.toContain("secret"); + expect(error.sentBytes).toBe(BYTES.byteLength); + }); + + it("reports a request the binding throws on, and releases the image", async () => { + let image = byteStream(BYTES, [10]); + let thrown = new Error("Network connection lost."); + let ai = { fetch: async () => { throw thrown; } } as unknown as AiFetcher; + let error = await failure(transcribeImage(ai, undefined, jpeg(image.stream))); + expect(error.failure).toEqual({ error: thrown }); + expect(image.state.cancelled).toBe(true); + }); + + it.each([ + ["ends before saying it is done", tokenEvents(["partial"]).replace("data: [DONE]\n\n", "")], + ["reports an error partway", `${tokenEvents(["partial"]).split("data: [DONE]")[0]}data: {"error":"secret upstream text"}\n\n`], + ["holds an event that is not JSON", `data: {"response":"partial"}\n\ndata: {secret page text\n\n`], + ])("fails, without the service's words, on a response that %s", async (_label, events) => { + let transcription = await transcribeImage(fakeAi(eventResponse(events)).ai, undefined, jpeg()); + let error = await failure(drain(transcription)); + expect(String((error.failure as { error: Error }).error.message)).not.toContain("secret"); + expect(error.sentBytes).toBe(BYTES.byteLength); + }); + + it("fails on a response that is not a stream of events", async () => { + let transcription = await transcribeImage( + fakeAi(eventResponse("x".repeat(300_000), 100_000)).ai, undefined, jpeg()); + await failure(drain(transcription)); + }); + + it("throws the image stream's own error as it is", async () => { + let brokeOff = new Error("The Google Drive download failed."); + let sentChunk = false; + let body = new ReadableStream({ + pull(controller) { + if (sentChunk) throw brokeOff; + sentChunk = true; + controller.enqueue(BYTES); + }, + }); + await expect(transcribeImage(fakeAi(eventResponse(tokenEvents(PAGE))).ai, undefined, jpeg(body))) + .rejects.toBe(brokeOff); + }); + + it("releases the response and the image when cancelled", async () => { + let response = { cancelled: false }; + let body = new ReadableStream({ + pull(controller) { controller.enqueue(new TextEncoder().encode(tokenEvents(["more "]).split("data: [DONE]")[0])); }, + cancel() { response.cancelled = true; }, + }, { highWaterMark: 0 }); + // The service answers before it has read the whole image. + let image = byteStream(BYTES, [10]); + let ai = { + fetch: async (_url: string, init: RequestInit) => { + await (init.body as ReadableStream).getReader().read(); + return new Response(body); + }, + } as unknown as AiFetcher; + + let transcription = await transcribeImage(ai, undefined, jpeg(image.stream)); + expect(await transcription.next()).toBe("more "); + await transcription.cancel(); + expect(response.cancelled).toBe(true); + expect(image.state.cancelled).toBe(true); + }); +}); diff --git a/packages/gatekeeper-google/src/workers-ai-vision.ts b/packages/gatekeeper-google/src/workers-ai-vision.ts new file mode 100644 index 0000000000..e9bb7d0339 --- /dev/null +++ b/packages/gatekeeper-google/src/workers-ai-vision.ts @@ -0,0 +1,217 @@ +// A page image transcribed by a Workers AI vision model, streamed in both directions: the image is +// encoded into the request as it arrives, and the text is read from the response as it is wanted, +// so neither is ever held whole. + +import { readTextPrefix } from "@gadgets/gatekeeper-kit/response-body"; +import { base64JsonBody, type AiFetcher, type ConversionFailure } from "./workers-ai-markdown"; + +const MODEL = "@cf/google/gemma-4-26b-a4b-it"; +/** + * Where the binding's `run()` posts, directly and through an AI Gateway. That method takes its + * input as one object, image included; posting the same JSON here lets the image stream through. + */ +const RUN_URL = "https://workers-binding.ai/run?version=3"; +const GATEWAY_RUN_URL = "https://workers-binding.ai/ai-gateway/run?version=3"; + +const PROMPT = "Transcribe the text of this scanned page exactly as it is printed, in reading " + + "order. Write each paragraph on one line, headings and lists as Markdown, and tables as " + + "Markdown tables. Do not describe the page, comment on it, or correct what it says. If the " + + "page has no text, reply with nothing at all."; +/** Enough for the densest page; what a model stuck repeating itself is stopped at. */ +const MAX_TOKENS = 8192; +/** + * The model's settings. `temperature: 0` because a transcription is not to be varied. Thinking is + * off because, measured on scanned pages, it tripled the tokens and the time a page takes (about + * 1,100 tokens against 340) and held the text back until it was over, without making a poor scan + * any better read: it saved a word or two on clean pages only. + */ +const SETTINGS = { + temperature: 0, max_tokens: MAX_TOKENS, stream: true, + chat_template_kwargs: { enable_thinking: false }, +}; + +/** Bytes of a failed request's response read for its error code. */ +const MAX_ERROR_BYTES = 1024; +/** Characters of one line of the response held; a line carries one token. */ +const MAX_LINE_CHARS = 256 * 1024; + +/** One image to transcribe. */ +export type ImageSource = { mimeType: "image/jpeg" | "image/png"; body: ReadableStream }; + +/** A failed transcription, with why it failed and how much of the image had been sent. */ +export class TranscriptionError extends Error { + constructor(readonly failure: ConversionFailure, readonly sentBytes: number) { + super("Workers AI could not transcribe the image."); + } +} + +/** A transcription's text, read from the model's response only as it is asked for. */ +export type Transcription = { + /** Bytes of the image sent so far. */ + readonly sentBytes: number; + /** Whether the model was stopped at its output limit, so the text ends early. Known once `next` has returned `undefined`. */ + readonly truncated: boolean; + /** + * The next piece of the text, never empty, or `undefined` once the response has ended whole. + * Throws a {@link TranscriptionError} once the response breaks or ends early, having released + * the response and the image. + */ + next(): Promise; + /** Stops reading the response and the image. */ + cancel(reason?: unknown): Promise; +}; + +/** + * Has the vision model transcribe `image` with the Workers AI binding, through AI Gateway + * `gatewayId` when one is named. Resolves once the service answers, before any of the text is + * read. A failed request rejects with a {@link TranscriptionError}, and a failure of the image + * stream itself with that failure as it is; either way the image is released. + */ +export async function transcribeImage( + ai: AiFetcher, gatewayId: string | undefined, image: ImageSource, +): Promise { + // {"inputs":{"messages":[{role, content:[{text}, {image_url:{url}}]}], …settings},"options":…} + let text = JSON.stringify({ type: "text", text: PROMPT }); + let options = gatewayId ? { gateway: { id: gatewayId } } : {}; + let { body, sent, cancelSource } = base64JsonBody( + `{"inputs":{"messages":[{"role":"user","content":[${text},{"type":"image_url","image_url":` + + `{"url":"data:${image.mimeType};base64,`, + image.body, + `"}}]}],${JSON.stringify(SETTINGS).slice(1, -1)}},"options":${JSON.stringify(options)}}`); + let fail = async (failure: ConversionFailure): Promise => { + // A fetch holds the body locked, so the image is released through its reader instead. + await cancelSource(); + if (sent.error) throw sent.error.error; + throw new TranscriptionError(failure, sent.bytes); + }; + + let response: Response; + try { + response = await ai.fetch(gatewayId ? GATEWAY_RUN_URL : RUN_URL, { + method: "POST", + headers: { + "Content-Type": "application/json", + "cf-consn-sdk-version": "2.0.0", + "cf-consn-model-id": MODEL, + }, + body, + }); + } catch (error) { + return fail({ error }); + } + if (!response.ok || !response.body) { + // Only the service's numeric error code is kept from its answer, never its text. + let prefix = await readTextPrefix(response, MAX_ERROR_BYTES).catch(() => ""); + let code = /"(?:internalCode|code)"\s*:\s*(\d+)/.exec(prefix)?.[1]; + return fail({ httpStatus: response.status, ...(code ? { providerCode: Number(code) } : {}) }); + } + let reader = new TokenReader(response.body); + return { + get sentBytes() { return sent.bytes; }, + get truncated() { return reader.truncated; }, + async next() { + try { + return await reader.next(); + } catch (error) { + await reader.cancel(); + return fail({ error }); + } + }, + async cancel(reason) { + await reader.cancel(reason); + await cancelSource(); + }, + }; +} + +/** What is read of one event of the response; a model's answer comes in either shape. */ +type TokenEvent = { + choices?: { delta?: { content?: unknown }; finish_reason?: unknown }[]; + response?: unknown; + error?: unknown; + errors?: unknown; +}; + +/** + * The text of a streamed model response: server-sent events, each `data: ` line a JSON object + * carrying the next piece of text, ended by `data: [DONE]`. Only the answer's text is kept; what + * a model thinks before answering comes in a field of its own and is passed over. + */ +class TokenReader { + truncated = false; + #reader: ReadableStreamDefaultReader; + #decoder = new TextDecoder(); + /** The line being read. */ + #line = ""; + /** Text read but not yet returned by {@link next}. */ + #text = ""; + #done = false; + #ended = false; + + constructor(body: ReadableStream) { + this.#reader = body.getReader(); + } + + /** + * The next piece of the text, never empty, or `undefined` once the response has said it is done. + * Throws once the response breaks, reports an error, or ends without saying so. + */ + async next(): Promise { + for (;;) { + // The halves of a surrogate pair can arrive in two events and are kept together so that + // neither is encoded alone. + let end = this.#text.length; + let last = this.#text.charCodeAt(end - 1); + if (!this.#done && !this.#ended && last >= 0xd800 && last <= 0xdbff) end -= 1; + if (end > 0) { + let text = this.#text.slice(0, end); + this.#text = this.#text.slice(end); + return text; + } + if (this.#done) { + await this.cancel(); + return undefined; + } + if (this.#ended) throw new Error("The model's response ended early."); + let chunk = await this.#reader.read(); + if (chunk.done) this.#ended = true; + let lines = (this.#line + (chunk.done ? `${this.#decoder.decode()}\n` + : this.#decoder.decode(chunk.value, { stream: true }))).split("\n"); + this.#line = lines.pop()!; + if (this.#line.length > MAX_LINE_CHARS) throw new Error("The model's response is not a stream of events."); + for (let line of lines) this.#event(line); + } + } + + /** Stops reading the response. */ + async cancel(reason?: unknown): Promise { + await this.#reader.cancel(reason).catch(() => undefined); + } + + #event(line: string): void { + if (this.#done || !line.startsWith("data:")) return; + let data = line.slice("data:".length).trim(); + if (data === "[DONE]") { + this.#done = true; + return; + } + let event: TokenEvent; + try { + event = JSON.parse(data) as TokenEvent; + } catch { + // The parser's own message would quote the text. + throw new Error("The model's response held an event that is not JSON."); + } + if (typeof event !== "object" || event === null) return; + if (event.error !== undefined || event.errors !== undefined) { + throw new Error("The model reported an error partway through."); + } + if (!Array.isArray(event.choices)) { + if (typeof event.response === "string") this.#text += event.response; + return; + } + let choice = event.choices[0]; + if (typeof choice?.delta?.content === "string") this.#text += choice.delta.content; + if (choice?.finish_reason === "length") this.truncated = true; + } +} From 19f7dc61c669547d71f276aeba938a1c15c4919e Mon Sep 17 00:00:00 2001 From: Maximo Guk <62088388+Maximo-Guk@users.noreply.github.com> Date: Mon, 9 Nov 2026 16:33:57 -0600 Subject: [PATCH 3/4] Locate a PDF's page images as its bytes stream past A scanner writes each page of a PDF as one image. PdfImageScan reads a PDF fed to it in chunks cut anywhere and reports where each page-sized image's data sits, how long it is and how it is encoded, holding only the object it is reading: stream data is counted past, never kept. It walks objects from the front, since the cross-reference table is at the end, takes lengths given directly, by a reference before or after the stream, or not at all, passes over masks and small images, and notes an encrypted file. A JPEG page is a JPEG file as it is stored. A losslessly compressed grey or RGB page is deflated rows of samples, which is what a PNG's image data is, so pngFromDeflated() wraps it as a PNG while it streams: the PDF's own bytes inside PNG's framing when the rows carry PNG filter bytes, re-deflated with one added per row when they do not. No pixel is decoded. Nothing calls either yet. Co-Authored-By: Claude Fable 5.2 --- .../__tests__/pdf-fixture.ts | 103 ++++ .../__tests__/pdf-images.test.ts | 189 +++++++ .../__tests__/pdf-png.test.ts | 87 +++ packages/gatekeeper-google/src/pdf-images.ts | 531 ++++++++++++++++++ packages/gatekeeper-google/src/pdf-png.ts | 92 +++ 5 files changed, 1002 insertions(+) create mode 100644 packages/gatekeeper-google/__tests__/pdf-fixture.ts create mode 100644 packages/gatekeeper-google/__tests__/pdf-images.test.ts create mode 100644 packages/gatekeeper-google/__tests__/pdf-png.test.ts create mode 100644 packages/gatekeeper-google/src/pdf-images.ts create mode 100644 packages/gatekeeper-google/src/pdf-png.ts diff --git a/packages/gatekeeper-google/__tests__/pdf-fixture.ts b/packages/gatekeeper-google/__tests__/pdf-fixture.ts new file mode 100644 index 0000000000..a808fdb70f --- /dev/null +++ b/packages/gatekeeper-google/__tests__/pdf-fixture.ts @@ -0,0 +1,103 @@ +// Builds small PDFs whose pages are each one image, the way a scanner writes them, for the tests +// of what readPdf() does with a PDF that has no text of its own. + +/** One page's image, as it is stored in the file. */ +export type PdfImage = { + /** The image stream's bytes, already encoded as `filter` says. */ + data: Uint8Array; + width: number; + height: number; + /** The `/Filter` value as written, `/DCTDecode` (a JPEG) unless given; `""` for none. */ + filter?: string; + /** The `/ColorSpace` value as written, `/DeviceRGB` unless given. */ + colorSpace?: string; + bitsPerComponent?: number; + /** Further dictionary entries, as written, such as `/ImageMask true`. */ + extra?: string; + /** Writes `/Length` as a reference to an object that follows the stream. */ + indirectLength?: boolean; + /** Writes this `/Length` instead of the true one. */ + declaredLength?: number; +}; + +/** A built PDF, and where each image's stream data sits in it, in page order. */ +export type BuiltPdf = { bytes: Uint8Array; images: { offset: number; length: number }[] }; + +/** + * The number of page `i`'s first object. Objects 1 and 2 are the catalog and page tree; each page + * then takes four: the page, its content, its image, and the image's length (used only when that + * is indirect). + */ +function pageId(i: number): number { + return 3 + i * 4; +} + +/** + * A PDF of one US Letter page per image, each drawn across its whole page. Page `i`'s image is + * object `5 + 4i`. `objects` are written after the pages, as given, numbered on from the last + * page's (`3 + 4 * pages.length` is the first), and `trailer` adds entries to the trailer + * dictionary as written, such as an `/Encrypt` reference. + */ +export function imagePdf( + pages: PdfImage[], options: { objects?: string[]; trailer?: string } = {}, +): BuiltPdf { + let encoder = new TextEncoder(); + let parts: Uint8Array[] = []; + let size = 0; + let push = (part: string | Uint8Array) => { + let bytes = typeof part === "string" ? encoder.encode(part) : part; + parts.push(bytes); + size += bytes.byteLength; + }; + let offsets: number[] = []; + let object = (id: number, write: () => void) => { + offsets[id] = size; + push(`${id} 0 obj\n`); + write(); + push("\nendobj\n"); + }; + let images: BuiltPdf["images"] = []; + + push("%PDF-1.7\n%\xe2\xe3\xcf\xd3\n"); + object(1, () => push("<< /Type /Catalog /Pages 2 0 R >>")); + object(2, () => push(`<< /Type /Pages /Count ${pages.length} /Kids [${ + pages.map((_, i) => `${pageId(i)} 0 R`).join(" ")}] >>`)); + pages.forEach((image, i) => { + let id = pageId(i); + object(id, () => push(`<< /Type /Page /Parent 2 0 R /MediaBox [0 0 612 792] /Contents ${ + id + 1} 0 R /Resources << /XObject << /Im0 ${id + 2} 0 R >> >> >>`)); + let content = "q 612 0 0 792 0 0 cm /Im0 Do Q"; + object(id + 1, () => push(`<< /Length ${content.length} >>\nstream\n${content}\nendstream`)); + object(id + 2, () => { + let filter = image.filter ?? "/DCTDecode"; + let length = image.indirectLength ? `${id + 3} 0 R` + : String(image.declaredLength ?? image.data.byteLength); + push(`<< /Type /XObject /Subtype /Image /Width ${image.width} /Height ${image.height + } /ColorSpace ${image.colorSpace ?? "/DeviceRGB"} /BitsPerComponent ${ + image.bitsPerComponent ?? 8}${filter ? ` /Filter ${filter}` : ""}${ + image.extra ? ` ${image.extra}` : ""} /Length ${length} >>\nstream\n`); + images.push({ offset: size, length: image.data.byteLength }); + push(image.data); + push("\nendstream"); + }); + object(id + 3, () => push(String(image.data.byteLength))); + }); + + let objects = options.objects ?? []; + objects.forEach((body, i) => object(pageId(pages.length) + i, () => push(body))); + + let count = pageId(pages.length) + objects.length; + let xref = size; + push(`xref\n0 ${count}\n0000000000 65535 f \n`); + for (let id = 1; id < count; id++) push(`${String(offsets[id]).padStart(10, "0")} 00000 n \n`); + push(`trailer\n<< /Size ${count} /Root 1 0 R${options.trailer ? ` ${options.trailer}` : "" + } >>\nstartxref\n${xref}\n%%EOF\n`); + + let bytes = new Uint8Array(size); + let at = 0; + for (let part of parts) { + bytes.set(part, at); + at += part.byteLength; + } + return { bytes, images }; +} diff --git a/packages/gatekeeper-google/__tests__/pdf-images.test.ts b/packages/gatekeeper-google/__tests__/pdf-images.test.ts new file mode 100644 index 0000000000..dfb831f3b7 --- /dev/null +++ b/packages/gatekeeper-google/__tests__/pdf-images.test.ts @@ -0,0 +1,189 @@ +import { describe, expect, it } from "vitest"; +import { MAX_PAGE_IMAGES, PdfImageScan, type PdfImages } from "../src/pdf-images"; +import { imagePdf, type PdfImage } from "./pdf-fixture"; + +/** Bytes standing for an image's data: every byte value, and the keywords a careless scan trips on. */ +function data(length: number, seed = 1): Uint8Array { + let bytes = Uint8Array.from({ length }, (_, i) => (i * 7 + seed * 13) % 256); + let keywords = new TextEncoder().encode(" 9 0 obj << /Length 3 >> stream\nendobj "); + if (length >= 100) bytes.set(keywords, 20); + return bytes; +} + +function page(overrides: Partial = {}): PdfImage { + return { data: data(900), width: 1275, height: 1650, ...overrides }; +} + +/** A page stored as deflated samples, 1,000 pixels wide. */ +function lossless(extra: string, colorSpace: string, bitsPerComponent = 8): PdfImage { + return page({ filter: "/FlateDecode", colorSpace, bitsPerComponent, extra, width: 1000 }); +} + +/** Scans `bytes` fed in chunks of the given sizes, cycling. */ +function scan(bytes: Uint8Array, sizes: number[] = [bytes.byteLength]): PdfImages { + let scanner = new PdfImageScan(); + for (let at = 0, turn = 0; at < bytes.byteLength; turn++) { + let size = sizes[turn % sizes.length]!; + scanner.feed(bytes.subarray(at, at + size)); + at += size; + } + return scanner.end(); +} + +function pdf(text: string): Uint8Array { + return Uint8Array.from(text, ch => ch.charCodeAt(0)); +} + +const IMAGE = "/Type /XObject /Subtype /Image /Width 600 /Height 800 /BitsPerComponent 8"; + +describe("PdfImageScan", () => { + it("records where each page image sits, however the file is cut into chunks", () => { + let built = imagePdf([page(), page({ data: data(1300, 2) }), page({ data: data(40, 3) })]); + for (let sizes of [[built.bytes.byteLength], [1], [2], [3], [7], [64], [5, 1, 300], [1024]]) { + let found = scan(built.bytes, sizes); + expect(found.images.map(({ offset, length }) => ({ offset, length }))).toEqual(built.images); + expect(found).toMatchObject({ encrypted: false, more: false, bytes: built.bytes.byteLength }); + expect(found.images[0]).toEqual({ + ...built.images[0], width: 1275, height: 1650, encoding: "jpeg", damaged: false, + }); + } + let [first] = built.images; + expect(built.bytes.slice(first!.offset, first!.offset + first!.length)).toEqual(data(900)); + }); + + it("reads a length kept in an object after the stream, at every chunk size", () => { + let built = imagePdf([page({ indirectLength: true }), page({ indirectLength: true, data: data(77, 4) })]); + for (let sizes of [[built.bytes.byteLength], [1], [3], [10], [11], [64]]) { + let found = scan(built.bytes, sizes); + expect(found.images.map(({ offset, length, damaged }) => ({ offset, length, damaged }))) + .toEqual(built.images.map(image => ({ ...image, damaged: false }))); + } + }); + + it("reads a length kept in an object before the stream", () => { + let bytes = pdf(`%PDF-1.4\n7 0 obj\n5\nendobj\n8 0 obj\n<< ${IMAGE} /Filter /DCTDecode ` + + `/ColorSpace /DeviceRGB /Length 7 0 R >>\nstream\nab\ncd\nendstream\nendobj\n`); + let [image] = scan(bytes).images; + expect(image).toMatchObject({ length: 5, damaged: false }); + expect(String.fromCharCode(...bytes.slice(image!.offset, image!.offset + 5))).toBe("ab\ncd"); + }); + + it("finds the end of data whose length object it never sees, less the line end before it", () => { + for (let [lineEnd, length] of [["\n", 5], ["\r\n", 5], ["", 5], ["\n\n", 6]] as const) { + let bytes = pdf(`8 0 obj\n<< ${IMAGE} /Filter /DCTDecode /Length 99 0 R >>\n` + + `stream\r\nab\ncd${lineEnd}endstream\nendobj\n`); + for (let sizes of [[bytes.byteLength], [1], [4]]) { + expect(scan(bytes, sizes).images[0]).toMatchObject({ length, damaged: false }); + } + } + }); + + it("marks an image whose data is not as long as the file says", () => { + for (let declaredLength of [899, 902, 5, 100_000]) { + let built = imagePdf([page({ declaredLength }), page({ data: data(300, 5) })]); + let found = scan(built.bytes, [50]); + expect(found.images[0]!.damaged).toBe(true); + // A length that runs past the next object swallows it; one that falls short does not. + if (declaredLength < 2000) expect(found.images[1]).toMatchObject({ ...built.images[1], damaged: false }); + } + }); + + it("names each encoding in our own words", () => { + let encodings = [ + ["/DCTDecode", "jpeg"], ["[/DCTDecode]", "jpeg"], ["[ /DCTDecode ]", "jpeg"], + ["/CCITTFaxDecode", "fax"], ["/JBIG2Decode", "jbig2"], ["/JPXDecode", "jpeg2000"], + ["/LZWDecode", "other"], ["[/FlateDecode /DCTDecode]", "other"], + ["[/ASCII85Decode /DCTDecode]", "other"], ["", "other"], ["12 0 R", "other"], + ] as const; + let built = imagePdf(encodings.map(([filter]) => page({ filter }))); + expect(scan(built.bytes, [100]).images.map(image => image.encoding)) + .toEqual(encodings.map(([, encoding]) => encoding)); + }); + + it("describes deflated grey and RGB rows as a PNG's", () => { + let built = imagePdf([ + lossless("", "/DeviceRGB"), + lossless("", "/DeviceGray", 1), + lossless("/DecodeParms << /Predictor 15 /Colors 3 /Columns 1000 >>", "/DeviceRGB"), + lossless("/DecodeParms [ << /Predictor 12 /Columns 1000 /BitsPerComponent 1 >> ]", "[/CalGray << /WhitePoint [1 1 1] >>]", 1), + // A colour space behind a reference, and a profile that follows the image using it. + lossless("", "27 0 R"), + lossless("", "[/ICCBased 28 0 R]", 16), + ], { objects: ["[/ICCBased 28 0 R]", "<< /N 3 /Alternate /DeviceRGB /Length 2 >>\nstream\nxx\nendstream"] }); + expect(scan(built.bytes, [33]).images.map(({ encoding, png }) => ({ encoding, png }))).toEqual([ + { encoding: "lossless", png: { colorType: 2, bitDepth: 8, rowBytes: 3000, filtered: false } }, + { encoding: "lossless", png: { colorType: 0, bitDepth: 1, rowBytes: 125, filtered: false } }, + { encoding: "lossless", png: { colorType: 2, bitDepth: 8, rowBytes: 3000, filtered: true } }, + { encoding: "lossless", png: { colorType: 0, bitDepth: 1, rowBytes: 125, filtered: true } }, + { encoding: "lossless", png: { colorType: 2, bitDepth: 8, rowBytes: 3000, filtered: false } }, + { encoding: "lossless", png: { colorType: 2, bitDepth: 16, rowBytes: 6000, filtered: false } }, + ]); + }); + + it("leaves unread the deflated images a PNG cannot hold", () => { + let built = imagePdf([ + lossless("", "/DeviceCMYK"), + lossless("", "[/Indexed /DeviceRGB 255 <00ff00>]"), + lossless("", "/DeviceRGB", 4), + lossless("", "30 0 R"), + lossless("/DecodeParms << /Predictor 2 /Colors 3 /Columns 1000 >>", "/DeviceRGB"), + lossless("/DecodeParms << /Predictor 15 /Colors 3 /Columns 999 >>", "/DeviceRGB"), + lossless("/DecodeParms << /Predictor 15 /Columns 1000 >>", "/DeviceRGB"), + ]); + expect(scan(built.bytes).images.map(image => image.encoding)).toEqual(Array(7).fill("other")); + }); + + it("passes over masks, the images used as masks, and images too small to be a page", () => { + let built = imagePdf([ + page({ extra: "/SMask 7 0 R" }), + page({ colorSpace: "/DeviceGray" }), // Object 9: the first page's soft mask. + page({ extra: "/ImageMask true" }), + page({ width: 199 }), + page({ height: 120, width: 120 }), + page({ width: 200, height: 200 }), + ]); + // Page i's image is object 5 + 4i, so the mask is named by the number of the second. + let bytes = pdf(String.fromCharCode(...built.bytes).replace("/SMask 7 0 R", "/SMask 9 0 R")); + expect(scan(bytes).images.map(image => image.width)).toEqual([1275, 200]); + }); + + it("is not misled by keywords inside strings, comments and nested dictionaries", () => { + let bytes = pdf(`1 0 obj\n<< /Title (a stream\n 2 0 obj endobj \\) endstream) /K <> >>\nendobj\n` + + `2 0 obj\n% endobj stream\n<< ${IMAGE} /Filter /DCTDecode /Length 4 /Note (stream\n) >>\r\nstream\r\nDATA\r\nendstream endobj\n`); + for (let sizes of [[bytes.byteLength], [1], [9]]) { + expect(scan(bytes, sizes).images).toMatchObject([{ length: 4, encoding: "jpeg", damaged: false }]); + } + }); + + it("recovers after an object too large to be read", () => { + let filler = `1 0 obj\n[ ${"1234567 ".repeat(20_000)}]\nendobj\n`; + let built = imagePdf([page()]); + let bytes = pdf(filler + String.fromCharCode(...built.bytes)); + expect(scan(bytes, [4096]).images).toMatchObject([{ offset: built.images[0]!.offset + filler.length, length: 900 }]); + }); + + it("notes an encrypted file, from a trailer or a cross-reference stream", () => { + expect(scan(imagePdf([page()], { trailer: "/Encrypt 99 0 R" }).bytes, [13]).encrypted).toBe(true); + let bytes = pdf("5 0 obj\n<< /Type /XRef /Size 6 /Encrypt 4 0 R /Length 3 >>\nstream\nabc\nendstream\nendobj\n"); + expect(scan(bytes, [5]).encrypted).toBe(true); + // The word in an object of its own, or in stream data, says nothing. + let plain = pdf("5 0 obj\n<< /Length 9 >>\nstream\n/Encrypt \nendstream\nendobj\n6 0 obj\n(/Encrypt)\nendobj\n"); + expect(scan(plain, [5]).encrypted).toBe(false); + }); + + it("keeps a bounded number of page images and says when there were more", () => { + let one = `<< ${IMAGE} /Filter /DCTDecode /Length 1 >>\nstream\nx\nendstream\nendobj\n`; + let text = ""; + for (let i = 1; i <= MAX_PAGE_IMAGES + 3; i++) text += `${i} 0 obj\n${one}`; + let found = scan(pdf(text), [65_536]); + expect(found.images).toHaveLength(MAX_PAGE_IMAGES); + expect(found.more).toBe(true); + }); + + it("finds nothing in a file with no images, or one that is not a PDF", () => { + expect(scan(pdf("%PDF-1.4\n1 0 obj\n<< /Type /Catalog >>\nendobj\ntrailer\n<< /Root 1 0 R >>\n%%EOF"))) + .toEqual({ images: [], more: false, encrypted: false, bytes: 76 }); + expect(scan(data(5000), [17]).images).toEqual([]); + expect(scan(new Uint8Array(0))).toEqual({ images: [], more: false, encrypted: false, bytes: 0 }); + }); +}); diff --git a/packages/gatekeeper-google/__tests__/pdf-png.test.ts b/packages/gatekeeper-google/__tests__/pdf-png.test.ts new file mode 100644 index 0000000000..9a21f4d11e --- /dev/null +++ b/packages/gatekeeper-google/__tests__/pdf-png.test.ts @@ -0,0 +1,87 @@ +import { deflateSync, inflateSync } from "node:zlib"; +import { describe, expect, it } from "vitest"; +import type { PngLayout } from "../src/pdf-images"; +import { pngFromDeflated } from "../src/pdf-png"; +import { byteStream } from "./workers-ai-fixture"; + +/** A PNG's header fields, its image data inflated, and whether every chunk's checksum holds. */ +async function readPng(png: ReadableStream) { + let bytes = new Uint8Array(await new Response(png).arrayBuffer()); + let view = new DataView(bytes.buffer); + expect([...bytes.subarray(0, 8)]).toEqual([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a]); + let types: string[] = [], data: Uint8Array[] = [], header: number[] = []; + for (let at = 8; at < bytes.byteLength;) { + let length = view.getUint32(at); + let type = String.fromCharCode(...bytes.subarray(at + 4, at + 8)); + let body = bytes.subarray(at + 8, at + 8 + length); + expect(view.getUint32(at + 8 + length)).toBe(crc32(bytes.subarray(at + 4, at + 8 + length))); + types.push(type); + if (type === "IDAT") data.push(body); + if (type === "IHDR") header = [view.getUint32(at + 8), view.getUint32(at + 12), ...body.subarray(8)]; + at += 12 + length; + } + return { types, header, rows: new Uint8Array(inflateSync(Buffer.concat(data))) }; +} + +/** PNG's checksum, bit by bit. */ +function crc32(bytes: Uint8Array): number { + let crc = 0xffffffff; + for (let byte of bytes) { + crc ^= byte; + for (let bit = 0; bit < 8; bit++) crc = crc & 1 ? 0xedb88320 ^ (crc >>> 1) : crc >>> 1; + } + return ~crc >>> 0; +} + +const SIZE = { width: 5, height: 4 }; +/** Four rows of five RGB pixels. */ +const SAMPLES = Uint8Array.from({ length: 60 }, (_, i) => (i * 37) % 256); +const RGB: PngLayout = { colorType: 2, bitDepth: 8, rowBytes: 15, filtered: false }; + +describe("pngFromDeflated", () => { + it("adds a filter byte to each row of plain samples, however they are cut into chunks", async () => { + let deflated = new Uint8Array(deflateSync(SAMPLES)); + for (let sizes of [[deflated.byteLength], [1], [2, 7], [16]]) { + let png = await readPng(pngFromDeflated(byteStream(deflated, sizes).stream, SIZE, RGB)); + expect(png.header).toEqual([5, 4, 8, 2, 0, 0, 0]); + expect(png.types[0]).toBe("IHDR"); + expect(png.types.at(-1)).toBe("IEND"); + expect(png.types.slice(1, -1).every(type => type === "IDAT")).toBe(true); + expect(png.rows.byteLength).toBe(4 * 16); + for (let row = 0; row < 4; row++) { + expect(png.rows[row * 16]).toBe(0); + expect(png.rows.subarray(row * 16 + 1, row * 16 + 16)).toEqual(SAMPLES.subarray(row * 15, row * 15 + 15)); + } + } + }); + + it("frames rows that already carry filter bytes without touching them", async () => { + let filtered = Uint8Array.from({ length: 4 * 6 }, (_, i) => (i % 6 === 0 ? 2 : i * 11 % 256)); + let deflated = new Uint8Array(deflateSync(filtered)); + let layout: PngLayout = { colorType: 0, bitDepth: 8, rowBytes: 5, filtered: true }; + let png = await readPng(pngFromDeflated(byteStream(deflated, [9]).stream, SIZE, layout)); + expect(png.header).toEqual([5, 4, 8, 0, 0, 0, 0]); + expect(png.rows).toEqual(filtered); + // The data is the PDF's own bytes: nothing was inflated and deflated again. + let again = new Uint8Array(await new Response( + pngFromDeflated(byteStream(deflated, [9]).stream, SIZE, layout)).arrayBuffer()); + expect(Buffer.from(again).includes(Buffer.from(deflated.subarray(0, 9)))).toBe(true); + }); + + it("fails when the data does not inflate, and releases it", async () => { + let source = byteStream(Uint8Array.from({ length: 400 }, (_, i) => i % 251), [50]); + await expect(readPng(pngFromDeflated(source.stream, SIZE, RGB))).rejects.toThrow(); + }); + + it("cancels the data when it is cancelled", async () => { + let endless = { cancelled: false }; + let source = new ReadableStream({ + pull(controller) { controller.enqueue(Uint8Array.of(1, 2, 3)); }, + cancel() { endless.cancelled = true; }, + }, { highWaterMark: 0 }); + let reader = pngFromDeflated(source, SIZE, { ...RGB, filtered: true }).getReader(); + await reader.read(); + await reader.cancel(); + expect(endless.cancelled).toBe(true); + }); +}); diff --git a/packages/gatekeeper-google/src/pdf-images.ts b/packages/gatekeeper-google/src/pdf-images.ts new file mode 100644 index 0000000000..a81ebb15db --- /dev/null +++ b/packages/gatekeeper-google/src/pdf-images.ts @@ -0,0 +1,531 @@ +// Finds where a PDF keeps its page images, from the file's bytes as they stream past. A scanner +// writes each page as one image, so with the place and encoding of each, a page can later be read +// from the file on its own without the PDF ever being held or parsed whole. + +/** + * How an image's bytes are encoded, in our own names: `jpeg` is a JPEG file as it stands, + * `lossless` is deflated rows of samples that {@link PngLayout} describes, and the rest are + * encodings recognized but not read. + */ +export type PdfImageEncoding = "jpeg" | "lossless" | "fax" | "jbig2" | "jpeg2000" | "other"; + +/** What a PNG wrapped around a `lossless` image's data must say of it. */ +export type PngLayout = { + /** PNG's colour type: 0 for grey, 2 for RGB. */ + colorType: 0 | 2; + bitDepth: number; + /** Bytes in one row of samples. */ + rowBytes: number; + /** Whether each row already begins with a PNG filter byte, making the data a PNG's as it is. */ + filtered: boolean; +}; + +/** One image large enough to be a page, and where its data sits in the file. */ +export type PdfPageImage = { + offset: number; + length: number; + width: number; + height: number; + encoding: PdfImageEncoding; + /** Present when `encoding` is `lossless`. */ + png?: PngLayout; + /** The file did not hold the data where and as long as it said, so `length` is not reliable. */ + damaged: boolean; +}; + +/** What a scan found once the file has ended. */ +export type PdfImages = { + /** Page images in the order the file stores them, at most {@link MAX_PAGE_IMAGES}. */ + images: PdfPageImage[]; + /** Whether the file held more page images than were kept. */ + more: boolean; + /** Whether the file is encrypted, which makes every stream in it unreadable as it is. */ + encrypted: boolean; + /** How many bytes the file had. */ + bytes: number; +}; + +/** The most page images a scan keeps. */ +export const MAX_PAGE_IMAGES = 2000; +/** The shorter side, in pixels, below which an image is an icon, thumbnail or stamp, not a page. */ +const MIN_PAGE_SIDE = 200; +/** Characters of one object held while looking for its end; an image's dictionary is far smaller. */ +const MAX_OBJECT_CHARS = 64 * 1024; +/** Entries kept of each kind of small object an image can refer to. */ +const MAX_NOTED = 20_000; + +const WHITESPACE = "\0\t\n\f\r "; +const DELIMITERS = "()<>[]{}/%"; +const OBJECT_START = /(\d{1,10})[\0\t\n\f\r ]+\d{1,5}[\0\t\n\f\r ]+obj(?![A-Za-z0-9])/g; +const OBJECT_END = /(?; + +/** An image as its dictionary described it, before what it refers to is known. */ +type Candidate = { + id: number; + offset: number; + width: number; + height: number; + dict: PdfDict; + /** The length the dictionary gave, or the object it named for it. */ + declared: number | { ref: number } | undefined; + /** + * When the length was unknown: bytes from the data's start to the `endstream` found by + * searching, and how many of the last of them are a line end. + */ + searched?: { bytes: number; lineEnd: number }; + /** The data's length once it is settled. */ + length?: number; + damaged: boolean; +}; + +/** + * Reads a PDF fed to it in order, in chunks cut anywhere, and reports its page images at the end. + * It walks the file's objects from the front rather than from its cross-reference table, which + * sits at the end, and holds only the object it is reading (stream data is skipped, never kept) + * and the small objects an image may refer to: lengths, colour spaces and colour profiles' sizes. + * Objects kept inside compressed object streams are not seen, so a length stored in one is found + * by looking for the end of the data instead. + * + * An image is taken for a page when both its sides are at least {@link MIN_PAGE_SIDE} pixels and it is + * neither a mask nor another image's mask. Nothing here follows the page tree, so the order is + * the file's, which is page order for scanner output but is not guaranteed. + */ +export class PdfImageScan { + /** Bytes read but not yet passed, as one character each. */ + #pending = ""; + /** How many bytes came before `#pending`. */ + #passed = 0; + #state: "scan" | "object" | "data" | "afterData" | "search" = "scan"; + /** The object being read: its number, and how much of it has been searched for its end. */ + #object = 0; + #searched = 0; + /** Bytes of the stream being skipped that are still to come. */ + #remaining = 0; + /** The image whose data is being passed, if it is one. */ + #image: Candidate | undefined; + #ended = false; + + #candidates: Candidate[] = []; + #more = false; + #encrypted = false; + /** Objects that are a whole number, by number: a stream's length is often one. */ + #integers = new Map(); + /** Objects that are an array, by number: a colour space can be one. */ + #arrays = new Map(); + /** Component counts of colour profile streams, by number. */ + #profiles = new Map(); + /** Images other images use as their mask. */ + #masks = new Set(); + + /** Reads the next bytes of the file. */ + feed(chunk: Uint8Array): void { + for (;;) { + if (this.#state === "data") { + // Stream data is counted past without being decoded or kept. + let held = Math.min(this.#remaining, this.#pending.length); + this.#drop(held); + this.#remaining -= held; + let skipped = Math.min(this.#remaining, chunk.byteLength); + chunk = chunk.subarray(skipped); + this.#passed += skipped; + this.#remaining -= skipped; + if (this.#remaining > 0) return; + this.#state = "afterData"; + } + if (chunk.byteLength > 0) { + this.#pending += latin1(chunk); + chunk = chunk.subarray(chunk.byteLength); + } + if (!this.#step()) return; + } + } + + /** Reads the end of the file and reports what was found. Call once, after the last `feed`. */ + end(): PdfImages { + this.#ended = true; + this.feed(new Uint8Array(0)); + // Data the file ended inside was not all there. + if (this.#image && this.#state !== "scan") this.#image.damaged = true; + let images: PdfPageImage[] = []; + for (let candidate of this.#candidates) { + if (this.#masks.has(candidate.id)) continue; + let { offset, width, height } = candidate; + let length = this.#settledLength(candidate); + let { encoding, png } = this.#encoding(candidate); + images.push({ + offset, length: length ?? 0, width, height, encoding, ...(png ? { png } : {}), + damaged: candidate.damaged || length === undefined || length === 0, + }); + } + return { + images, more: this.#more, encrypted: this.#encrypted, + bytes: this.#passed + this.#pending.length, + }; + } + + #drop(count: number): void { + this.#pending = this.#pending.slice(count); + this.#passed += count; + } + + /** Advances through `#pending` once. False when it can go no further without more bytes. */ + #step(): boolean { + switch (this.#state) { + case "scan": return this.#findObject(); + case "object": return this.#readObject(); + case "afterData": return this.#endData(); + case "search": return this.#findDataEnd(); + case "data": return true; + } + } + + /** Between objects: looks for the next one, and for a trailer that names an encryption. */ + #findObject(): boolean { + OBJECT_START.lastIndex = 0; + let match = OBJECT_START.exec(this.#pending); + let between = match ? this.#pending.slice(0, match.index) : this.#pending; + if (between.includes("/Encrypt")) this.#encrypted = true; + if (!match) { + this.#drop(Math.max(0, this.#pending.length - SCAN_TAIL)); + return false; + } + this.#object = Number(match[1]); + this.#drop(match.index + match[0].length); + this.#searched = 0; + this.#state = "object"; + return true; + } + + /** + * Inside an object: looks for the `endobj` or `stream` that ends what was read so far as one + * whole value, which a keyword inside a string does not. + */ + #readObject(): boolean { + let text = this.#pending; + OBJECT_END.lastIndex = this.#searched; + for (let match; (match = OBJECT_END.exec(text));) { + let after = match.index + match[0].length; + if (match[1] === "endobj") { + let value: PdfValue; + try { + value = parseValue(text.slice(0, match.index)); + } catch { + continue; + } + this.#noteObject(value); + this.#drop(after); + this.#state = "scan"; + return true; + } + // Stream data begins after the keyword's line end, which may not have arrived yet. + if (after + 2 > text.length && !this.#ended) { + this.#searched = match.index; + return false; + } + let lineEnd = text.startsWith("\r\n", after) ? 2 : text[after] === "\n" || text[after] === "\r" ? 1 : 0; + if (lineEnd === 0) continue; + let dict: PdfValue; + try { + dict = parseValue(text.slice(0, match.index)); + } catch { + continue; + } + if (!(dict instanceof Map)) continue; + this.#drop(after + lineEnd); + this.#beginStream(dict); + return true; + } + if (text.length > MAX_OBJECT_CHARS) { + // Too large to be an image's dictionary, or not an object at all: look for the next one. + this.#state = "scan"; + return true; + } + // The longest keyword may be cut off at the end. + this.#searched = Math.max(0, text.length - 8); + return false; + } + + /** Keeps a small object an image may refer to. */ + #noteObject(value: PdfValue): void { + if (typeof value === "number" && Number.isSafeInteger(value) && value >= 0) { + if (this.#integers.size < MAX_NOTED) this.#integers.set(this.#object, value); + } else if (Array.isArray(value) && value.length <= 8) { + if (this.#arrays.size < MAX_NOTED) this.#arrays.set(this.#object, value); + } + } + + /** Notes what a stream's dictionary says, with `#pending` at the first byte of its data. */ + #beginStream(dict: PdfDict): void { + let type = dict.get("/Type"); + // A cross-reference stream carries the trailer's entries. + if (type === "/XRef" && dict.has("/Encrypt")) this.#encrypted = true; + let declared = dict.get("/Length"); + let length = typeof declared === "number" ? declared + : isRef(declared) ? this.#integers.get(declared.ref) : undefined; + this.#image = undefined; + if (dict.get("/Subtype") === "/Image") { + this.#image = this.#candidate(dict); + } else if (type !== "/XRef" && type !== "/ObjStm") { + let components = dict.get("/N"); + if (typeof components === "number" && this.#profiles.size < MAX_NOTED) { + this.#profiles.set(this.#object, components); + } + } + if (length !== undefined && Number.isSafeInteger(length) && length >= 0) { + this.#remaining = length; + this.#state = "data"; + } else { + this.#state = "search"; + } + } + + /** The image a dictionary describes, kept if it could be a page. */ + #candidate(dict: PdfDict): Candidate | undefined { + for (let key of ["/SMask", "/Mask"]) { + let mask = dict.get(key); + if (isRef(mask)) this.#masks.add(mask.ref); + } + let width = dict.get("/Width"), height = dict.get("/Height"); + if (typeof width !== "number" || typeof height !== "number" || + !Number.isSafeInteger(width) || !Number.isSafeInteger(height) || + Math.min(width, height) < MIN_PAGE_SIDE || dict.get("/ImageMask") === true) { + return undefined; + } + if (this.#candidates.length >= MAX_PAGE_IMAGES) { + this.#more = true; + return undefined; + } + let declared = dict.get("/Length"); + let candidate: Candidate = { + id: this.#object, offset: this.#passed, width, height, dict, damaged: false, + declared: typeof declared === "number" || isRef(declared) ? declared : undefined, + }; + this.#candidates.push(candidate); + return candidate; + } + + /** After a stream's data: the file must say `endstream` here, or the length was wrong. */ + #endData(): boolean { + let text = this.#pending; + if (text.length < "\r\nendstream".length && !this.#ended) return false; + let end = /^(?:\r\n|\n|\r)?endstream/.exec(text); + if (end) { + this.#drop(end[0].length); + if (this.#image) this.#image.length = this.#passed - end[0].length - this.#image.offset; + } else if (this.#image) { + this.#image.damaged = true; + } + this.#image = undefined; + this.#state = "scan"; + return true; + } + + /** In stream data of unknown length: looks for the `endstream` that closes it. */ + #findDataEnd(): boolean { + let text = this.#pending; + let end = text.indexOf("endstream"); + if (end < 0) { + // Keeps a cut-off keyword and the line end that may come before it. + this.#drop(Math.max(0, text.length - 10)); + return false; + } + if (this.#image) { + let before = text.slice(Math.max(0, end - 2), end); + this.#image.searched = { + bytes: this.#passed + end - this.#image.offset, + lineEnd: before.endsWith("\r\n") ? 2 : /[\r\n]$/.test(before) ? 1 : 0, + }; + } + this.#drop(end + "endstream".length); + this.#image = undefined; + this.#state = "scan"; + return true; + } + + /** + * How long an image's data is: what the file said when that proved right, or else the distance + * to the `endstream` found, less the line end before it. A length object read after the data + * is believed over that when the two agree to within a line end, since data can itself end in one. + */ + #settledLength(candidate: Candidate): number | undefined { + if (candidate.length !== undefined) return candidate.length; + let { searched, declared } = candidate; + if (searched === undefined) return undefined; + let named = isRef(declared) ? this.#integers.get(declared.ref) : undefined; + if (named !== undefined && searched.bytes - named >= 0 && searched.bytes - named <= 2) return named; + return Math.max(0, searched.bytes - searched.lineEnd); + } + + #encoding(candidate: Candidate): { encoding: PdfImageEncoding; png?: PngLayout } { + let { dict } = candidate; + let filter = dict.get("/Filter"); + let filters = Array.isArray(filter) ? filter : filter === undefined ? [] : [filter]; + if (filters.length !== 1) return { encoding: "other" }; + switch (filters[0]) { + case "/DCTDecode": return { encoding: "jpeg" }; + case "/CCITTFaxDecode": return { encoding: "fax" }; + case "/JBIG2Decode": return { encoding: "jbig2" }; + case "/JPXDecode": return { encoding: "jpeg2000" }; + case "/FlateDecode": { + let png = this.#pngLayout(candidate); + return png ? { encoding: "lossless", png } : { encoding: "other" }; + } + default: return { encoding: "other" }; + } + } + + /** + * How a deflated image's rows map onto a PNG's, or `undefined` when they do not: PNG holds grey + * and RGB samples only, as plain rows or rows with PNG's own filter bytes. A palette, CMYK, a + * TIFF predictor or a colour space that cannot be resolved is left unread. A `/Decode` array + * that inverts the samples is not applied, so such an image reads as a negative. + */ + #pngLayout({ dict, width }: Candidate): PngLayout | undefined { + let components = this.#components(dict.get("/ColorSpace"), 0); + let bitDepth = dict.get("/BitsPerComponent"); + if (typeof bitDepth !== "number" || (components !== 1 && components !== 3) || + !(components === 1 ? [1, 2, 4, 8, 16] : [8, 16]).includes(bitDepth)) { + return undefined; + } + let parms = dict.get("/DecodeParms"); + if (Array.isArray(parms)) parms = parms[0]; + let predictor = parms instanceof Map ? parms.get("/Predictor") ?? 1 : 1; + let filtered = predictor !== 1; + if (filtered) { + if (!(parms instanceof Map) || typeof predictor !== "number" || predictor < 10 || predictor > 15 || + (parms.get("/Colors") ?? 1) !== components || (parms.get("/Columns") ?? 1) !== width || + (parms.get("/BitsPerComponent") ?? 8) !== bitDepth) { + return undefined; + } + } + return { + colorType: components === 1 ? 0 : 2, bitDepth, filtered, + rowBytes: Math.ceil(width * components * bitDepth / 8), + }; + } + + /** How many components a colour space has, when it is grey or RGB and can be told. */ + #components(space: PdfValue, depth: number): number | undefined { + if (isRef(space)) return depth > 2 ? undefined : this.#components(this.#arrays.get(space.ref), depth + 1); + let family = Array.isArray(space) ? space[0] : space; + if (family === "/DeviceGray" || family === "/CalGray") return 1; + if (family === "/DeviceRGB" || family === "/CalRGB") return 3; + if (family === "/ICCBased" && Array.isArray(space) && isRef(space[1])) { + return this.#profiles.get(space[1].ref); + } + return undefined; + } +} + +function isRef(value: PdfValue): value is { ref: number } { + return typeof value === "object" && !Array.isArray(value) && !(value instanceof Map); +} + +/** `bytes` as a string of one character per byte, so that indexes into it are byte offsets. */ +function latin1(bytes: Uint8Array): string { + let text = ""; + for (let i = 0; i < bytes.byteLength; i += 8192) { + text += String.fromCharCode(...bytes.subarray(i, i + 8192)); + } + return text; +} + +function fail(): never { + throw new Error("not one PDF value"); +} + +/** The one value that `text` holds, with nothing but whitespace around it. Throws when it is not that. */ +function parseValue(text: string): PdfValue { + let at = 0; + let skip = () => { + for (; at < text.length; at++) { + if (text[at] === "%") { + while (at < text.length && text[at] !== "\n" && text[at] !== "\r") at++; + } else if (!WHITESPACE.includes(text[at]!)) { + return; + } + } + }; + let value = (depth: number): PdfValue => { + if (depth > 32) fail(); + skip(); + let ch = text[at]; + if (ch === undefined) return fail(); + if (text.startsWith("<<", at)) { + at += 2; + let dict: PdfDict = new Map(); + for (;;) { + skip(); + if (text.startsWith(">>", at)) { + at += 2; + return dict; + } + let key = value(depth + 1); + if (typeof key !== "string") return fail(); + dict.set(key, value(depth + 1)); + } + } + if (ch === "<") { + at = text.indexOf(">", at) + 1; + return at === 0 ? fail() : undefined; + } + if (ch === "(") { + for (let open = 0; at < text.length; at++) { + if (text[at] === "\\") at++; + else if (text[at] === "(") open++; + else if (text[at] === ")" && --open === 0) { + at++; + return undefined; + } + } + return fail(); + } + if (ch === "[") { + at++; + let items: PdfValue[] = []; + for (;;) { + skip(); + if (text[at] === "]") { + at++; + return items; + } + items.push(value(depth + 1)); + } + } + if (ch === "/") { + let start = at++; + while (at < text.length && !WHITESPACE.includes(text[at]!) && !DELIMITERS.includes(text[at]!)) at++; + return text.slice(start, at); + } + NUMBER.lastIndex = at; + let number = NUMBER.exec(text); + if (number) { + at += number[0].length; + REFERENCE_REST.lastIndex = at; + let rest = /^\d+$/.test(number[0]) ? REFERENCE_REST.exec(text) : null; + if (!rest) return Number(number[0]); + at += rest[0].length; + return { ref: Number(number[0]) }; + } + KEYWORD.lastIndex = at; + let keyword = KEYWORD.exec(text)?.[0]; + at += keyword?.length ?? 0; + if (keyword === "true" || keyword === "false") return keyword === "true"; + return keyword === "null" ? undefined : fail(); + }; + let result = value(0); + skip(); + return at === text.length ? result : fail(); +} diff --git a/packages/gatekeeper-google/src/pdf-png.ts b/packages/gatekeeper-google/src/pdf-png.ts new file mode 100644 index 0000000000..9ecb2e6423 --- /dev/null +++ b/packages/gatekeeper-google/src/pdf-png.ts @@ -0,0 +1,92 @@ +// Wraps a PDF's losslessly compressed image as a PNG, as its bytes stream past. A PDF keeps such +// an image as deflated rows of samples, which is what a PNG's image data is, so the PNG is the +// PDF's own bytes inside PNG's framing whenever the rows already carry PNG's filter bytes, and is +// re-deflated with one added to each row when they do not. No pixel is decoded, and the image is +// never held whole. + +import type { PngLayout } from "./pdf-images"; + +const SIGNATURE = Uint8Array.of(0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a); + +const CRC_TABLE = Uint32Array.from({ length: 256 }, (_, n) => { + let c = n; + for (let bit = 0; bit < 8; bit++) c = c & 1 ? 0xedb88320 ^ (c >>> 1) : c >>> 1; + return c; +}); + +/** One PNG chunk: its length, type, data and the checksum of the last two. */ +function chunk(type: string, data: Uint8Array): Uint8Array { + let out = new Uint8Array(12 + data.byteLength); + let view = new DataView(out.buffer); + view.setUint32(0, data.byteLength); + for (let i = 0; i < 4; i++) out[4 + i] = type.charCodeAt(i); + out.set(data, 8); + let crc = 0xffffffff; + for (let i = 4; i < 8 + data.byteLength; i++) crc = CRC_TABLE[(crc ^ out[i]!) & 0xff]! ^ (crc >>> 8); + view.setUint32(8 + data.byteLength, ~crc >>> 0); + return out; +} + +/** Puts PNG's "no filter" byte before each row of `rowBytes` samples. */ +function rowFilters(rowBytes: number): TransformStream { + /** Bytes of the current row already passed on. */ + let inRow = 0; + return new TransformStream({ + transform(samples, controller) { + let out = new Uint8Array(samples.byteLength + Math.ceil(samples.byteLength / rowBytes) + 1); + let length = 0; + for (let at = 0; at < samples.byteLength;) { + if (inRow === 0) out[length++] = 0; + let take = Math.min(rowBytes - inRow, samples.byteLength - at); + out.set(samples.subarray(at, at + take), length); + length += take; + at += take; + inRow = (inRow + take) % rowBytes; + } + controller.enqueue(out.subarray(0, length)); + }, + }); +} + +/** + * A PNG of the image whose deflated data is `data`, produced as it is read. Cancelling it cancels + * `data`, and `data` failing fails it. Data that does not inflate, or holds other than `height` + * rows, makes a PNG its reader will refuse; nothing here checks it. + */ +export function pngFromDeflated( + data: ReadableStream, size: { width: number; height: number }, layout: PngLayout, +): ReadableStream { + let rows = layout.filtered ? data : data + .pipeThrough(new DecompressionStream("deflate") as TransformStream) + .pipeThrough(rowFilters(layout.rowBytes)) + .pipeThrough(new CompressionStream("deflate") as TransformStream); + let reader = rows.getReader(); + return new ReadableStream({ + start(controller) { + let header = new Uint8Array(13); + let view = new DataView(header.buffer); + view.setUint32(0, size.width); + view.setUint32(4, size.height); + header[8] = layout.bitDepth; + header[9] = layout.colorType; + controller.enqueue(SIGNATURE); + controller.enqueue(chunk("IHDR", header)); + }, + async pull(controller) { + // A pull that enqueues nothing is not called again, so an empty piece is read past. + for (;;) { + let piece = await reader.read(); + if (piece.done) { + controller.enqueue(chunk("IEND", new Uint8Array(0))); + controller.close(); + return; + } + if (piece.value.byteLength > 0) { + controller.enqueue(chunk("IDAT", piece.value)); + return; + } + } + }, + cancel: reason => reader.cancel(reason), + }, { highWaterMark: 0 }); +} From 9e3c5a3bb4716e407e260d453304938b0e64d217 Mon Sep 17 00:00:00 2001 From: Maximo Guk <62088388+Maximo-Guk@users.noreply.github.com> Date: Mon, 9 Nov 2026 16:33:57 -0600 Subject: [PATCH 4/4] Transcribe scanned PDFs with readPdf(fileId, { ocr: true }) A scanned or image-only PDF has no text layer, so readPdf() returned hasText: false and nothing else. With ocr set, the PDF's page images are located as it streams into the conversion, and when the conversion finds no text the pages are transcribed by the vision model instead, under the same "### Page N" headings, with transcribed: true. Nothing is held and nothing is done early: a page's image is read from Drive by range and transcribed only as the text is read, one page at a time, and cancelling the text stops it. Each page read is a content read of its own, checked against the binding's scope, re-proved on a folder binding, reported as drive/content/, and refused if the file's size, or on a folder binding its modification time, has moved since it was converted. A PDF with text of its own is returned as before and no model is called. A scanned PDF that is encrypted, or whose pages are fax, JBIG2 or JPEG 2000 images, is refused in our words before any model call. Co-Authored-By: Claude Fable 5.2 --- packages/gatekeeper-google/README.md | 4 +- .../__tests__/drive-pdf.test.ts | 270 +++++++++++++++++- .../gatekeeper-google/__tests__/types.test.ts | 16 +- .../__tests__/workerd/native-sessions.test.ts | 192 ++++++++++++- packages/gatekeeper-google/src/drive-pdf.ts | 183 +++++++++++- .../gatekeeper-google/src/drive-types.d.ts | 39 ++- packages/gatekeeper-google/src/google.ts | 51 +++- .../gatekeeper-google/src/observability.ts | 8 + 8 files changed, 727 insertions(+), 36 deletions(-) diff --git a/packages/gatekeeper-google/README.md b/packages/gatekeeper-google/README.md index 4d8fa12cca..7a4e910f2a 100644 --- a/packages/gatekeeper-google/README.md +++ b/packages/gatekeeper-google/README.md @@ -660,7 +660,9 @@ A folder URL carrying `?resourcekey=` is not supported: the key is dropped, and The agent-facing `GoogleDriveReadSession` covers account and exact-file bindings. `GoogleDriveFolderSession` is positioned at the selected root and exposes only its current folder's direct children: `list()`, provider-side structured `search()`, `getEntry()`, native Doc/Sheet/Slides opens, `readPdf()` and `readFile()`, and `openFolder()` for one live direct child. Listing and search return disposable RPC cursors; child folders and native content sessions are independently disposable capabilities. There is no built-in recursive folder search, traversal pager, raw Drive `q`, file write, shortcut traversal, or export of Google-native files. -Both sessions offer `readPdf(fileId)`, which downloads an in-scope PDF and returns its text as Markdown, converted by Workers AI's `toMarkdown()`. It downloads with the Drive resources' `drive.readonly`; on an account connected before they requested it, `readPdf()` fails asking the user to reconnect. The file is checked against the binding's scope, and must be a PDF, before anything is downloaded, and a folder binding re-proves it is a live direct child after the download, as for a native read: the token's `drive.readonly` reaches any Drive file, so scope is enforced in code. Neither the PDF nor its text is held in memory: the download streams straight into the conversion request (through the `AI` binding's own `fetch`, since the binding's `toMarkdown()` takes a whole `Blob`), and `text` is a `ReadableStream` of UTF-8 Markdown read from the conversion's response as it is converted, so the agent bounds how much it reads. `readPdf()` reads the response only until the text begins before it returns, so `hasText` is already known. Pages are separated by `### Page N` headings, and pages before the first with text are left out. There is no download or text cap: a PDF Workers AI cannot convert, including one too large for it, fails. Nothing is OCR'd, so a scanned or image-only PDF returns `hasText: false` and an empty `text`. Conversion runs under a fixed file name with its images left undescribed, so no Drive file name reaches Workers AI and no billed model call is made. +Both sessions offer `readPdf(fileId)`, which downloads an in-scope PDF and returns its text as Markdown, converted by Workers AI's `toMarkdown()`. It downloads with the Drive resources' `drive.readonly`; on an account connected before they requested it, `readPdf()` fails asking the user to reconnect. The file is checked against the binding's scope, and must be a PDF, before anything is downloaded, and a folder binding re-proves it is a live direct child after the download, as for a native read: the token's `drive.readonly` reaches any Drive file, so scope is enforced in code. Neither the PDF nor its text is held in memory: the download streams straight into the conversion request (through the `AI` binding's own `fetch`, since the binding's `toMarkdown()` takes a whole `Blob`), and `text` is a `ReadableStream` of UTF-8 Markdown read from the conversion's response as it is converted, so the agent bounds how much it reads. `readPdf()` reads the response only until the text begins before it returns, so `hasText` is already known. Pages are separated by `### Page N` headings, and pages before the first with text are left out. There is no download or text cap: a PDF Workers AI cannot convert, including one too large for it, fails. A scanned or image-only PDF has no text to convert, so it returns `hasText: false` and an empty `text` unless the caller asks for it to be transcribed. Conversion runs under a fixed file name with its images left undescribed, so no Drive file name reaches Workers AI and the conversion makes no billed model call. + +`readPdf(fileId, { ocr: true })` transcribes a PDF that proves to have no text, and returns `transcribed: true` with the transcription as `text`; a PDF with any text of its own is returned as that text, and no model is called. Each page is one billed call to the Workers AI vision model `@cf/google/gemma-4-26b-a4b-it` (about 320 input and a few hundred output tokens for a page of prose), made with no budget or approval of its own, through the same AI Gateway as conversion when `CF_AI_GATEWAY` is set; a gateway that stores its logs then holds the page images and their transcriptions. Nothing is held here: as the PDF streams into the conversion, a scanner (`src/pdf-images.ts`) notes where each page image sits in the file, and a page is then read from Drive by range, base64-encoded into the model's request as it arrives, and its text passed on as the model produces it. That happens only as `text` is read, one page at a time, so an agent that stops reading, or cancels, stops paying; cancelling ends the page in progress. Every page read is a content read of its own, checked against the binding's scope, re-proved on a folder binding and reported as `drive/content/`, and it fails if the file is no longer the size it was converted at, or, on a folder binding, if its modification time has moved. Pages come under `### Page N` headings numbered in the order the file stores their images, which is page order for scanner output but is not guaranteed, since the page tree is not followed. Any image at least 200 pixels on each side that is not a mask counts as a page, so a PDF of pictures and no text has each picture read as one. JPEG page images are sent as they are stored, and losslessly compressed grey or RGB ones are wrapped as PNG on the way without being decoded (`src/pdf-png.ts`). A scanned PDF is refused, in the gatekeeper's words and before any model call, when it is encrypted, when a page is stored as fax (CCITT), JBIG2 or JPEG 2000, or as an image PNG cannot hold (a palette, CMYK), when a page image is over 12 MiB (Workers AI refuses a request somewhere between that and 16 MiB), is not where the file says it is, or when there are more than 2,000 of them. The model reads a whole page at a fixed resolution whatever the scan's, so small print is where it errs: in testing it got 98 to 100% of a page's words right, and its misses were codes, numbers and names in 6 to 8 point type, which is why the types tell the agent a transcription is not the document's own text. A page the model is stopped partway through ends with `[The transcription of this page was cut off.]`, a blank page has a heading and no text, and a page that fails errors the stream. Thinking is turned off for the model: it tripled a page's tokens and time and held the text back, without reading a poor scan any better. Logs carry page counts, page numbers, sizes, encodings and status codes, never text. The gatekeeper's `AI` binding is what `readPdf()` converts with. It runs remotely, so start the dev server with `pnpm dev-server -- --use-workers-ai-binding` (which needs a Cloudflare login) to read PDFs locally; without the binding `readPdf()` says PDF reading is not available, and everything else works. When `CF_AI_GATEWAY` is set on this Worker, conversion goes through the AI Gateway it names, which must be in the Worker's own account; otherwise it calls Workers AI directly. diff --git a/packages/gatekeeper-google/__tests__/drive-pdf.test.ts b/packages/gatekeeper-google/__tests__/drive-pdf.test.ts index 51e0c640e5..f361d46d98 100644 --- a/packages/gatekeeper-google/__tests__/drive-pdf.test.ts +++ b/packages/gatekeeper-google/__tests__/drive-pdf.test.ts @@ -1,8 +1,11 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; -import { pdfToText } from "../src/drive-pdf"; +import { deflateSync } from "node:zlib"; +import { pdfToText, type PdfPageRange, type PdfRangeReader } from "../src/drive-pdf"; +import type { AiFetcher } from "../src/workers-ai-markdown"; +import { imagePdf, type BuiltPdf, type PdfImage } from "./pdf-fixture"; import { - BYTES, byteStream, conversionJson, fakeAi, formatLastJson, markdownJson, splitResponse, - streamedResponse, + BYTES, byteStream, conversionJson, eventResponse, fakeAi, formatLastJson, markdownJson, + sentImage, splitResponse, streamedResponse, tokenEvents, } from "./workers-ai-fixture"; // What Workers AI's toMarkdown returned for generated PDFs with metadata off and images unconverted. @@ -342,3 +345,264 @@ describe("pdfToText's failures", () => { expect(warn).not.toHaveBeenCalled(); }); }); + +const TRANSCRIPTION_FAILED = "Google Drive's PDF could not be transcribed."; +/** What the model reads on each page of the scan. */ +const PAGES = [["Harbour Ledger", " Review"], ["Lot 55-0193-G", "\n\n18,442.07 florins"], ["Signatories"]]; +const TRANSCRIBED = + "### Page 1\nHarbour Ledger Review\n\n### Page 2\nLot 55-0193-G\n\n18,442.07 florins\n\n### Page 3\nSignatories"; + +function scanPage(seed: number, overrides: Partial = {}): PdfImage { + let data = Uint8Array.from({ length: 700 + seed }, (_, i) => (i * 7 + seed) % 256); + return { data, width: 1275, height: 1650, ...overrides }; +} + +/** A three-page scan, each page one JPEG. */ +const SCAN = imagePdf([scanPage(1), scanPage(2), scanPage(3)]); + +/** + * A binding that converts every PDF to `markdown` and answers the nth transcription with + * `pages[n]`: the text's pieces, or a response to give instead. + */ +function ocrAi(pages: (string[] | Response)[] = PAGES, markdown = SCANNED_PDF) { + let transcribed = 0; + let { ai, sent } = fakeAi(request => { + if (request.url.includes("markdown")) return streamedResponse(markdownJson(markdown), 64); + let page = pages[transcribed++]!; + return page instanceof Response ? page : eventResponse(tokenEvents(page)); + }); + let images = () => sent.filter(request => !request.url.includes("markdown")); + return { ai, sent, images }; +} + +/** A range reader over `pdf` that notes each range asked of it. */ +function ranges(pdf: BuiltPdf) { + let asked: PdfPageRange[] = []; + let readRange: PdfRangeReader = async range => { + asked.push(range); + return byteStream(pdf.bytes.slice(range.offset, range.offset + range.length), [100]).stream; + }; + return { asked, readRange }; +} + +/** Everything logged, as one string. */ +function everythingLogged(): string { + return JSON.stringify([console.info, console.warn].flatMap(log => vi.mocked(log).mock.calls)); +} + +describe("pdfToText transcribing a scanned PDF", () => { + it("transcribes each page image, in the file's order, under page headings", async () => { + let { ai, images } = ocrAi(); + let { asked, readRange } = ranges(SCAN); + + let result = await pdfToText(ai, "team-gateway", byteStream(SCAN.bytes, [333]).stream, readRange); + expect(result).toMatchObject({ hasText: false, transcribed: true }); + // Nothing is read or transcribed until the text is. + expect(asked).toEqual([]); + expect(await read(result)).toBe(TRANSCRIBED); + + expect(asked).toEqual(SCAN.images.map((image, i) => ({ + page: i + 1, ...image, fileSize: SCAN.bytes.byteLength, + }))); + expect(images().map(request => sentImage(request))).toEqual(SCAN.images.map(({ offset, length }) => ({ + mimeType: "image/jpeg", bytes: SCAN.bytes.slice(offset, offset + length), + }))); + expect(images().every(request => request.url.includes("/ai-gateway/run"))).toBe(true); + }); + + it("logs counts, sizes and encodings, and never the transcription", async () => { + let info = vi.mocked(console.info); + let { ai } = ocrAi(); + await read(await pdfToText(ai, undefined, byteStream(SCAN.bytes).stream, ranges(SCAN).readRange)); + + expect(info).toHaveBeenCalledWith(expect.objectContaining({ + event: "google.drive.pdf.transcribing", pdfBytes: SCAN.bytes.byteLength, ocrPages: 3, + })); + expect(info).toHaveBeenCalledWith(expect.objectContaining({ + event: "google.drive.pdf.page_transcribed", page: 2, imageBytes: 702, imageEncoding: "jpeg", + })); + for (let word of ["Harbour", "55-0193", "florins", "Signatories"]) expect(everythingLogged()).not.toContain(word); + }); + + it("sends a losslessly compressed page as a PNG", async () => { + let rows = Uint8Array.from({ length: 300 * 200 }, (_, i) => i % 251); + let pdf = imagePdf([{ + data: new Uint8Array(deflateSync(rows)), width: 300, height: 200, + filter: "/FlateDecode", colorSpace: "/DeviceGray", + }]); + let { ai, images } = ocrAi([["grey page"]]); + + let result = await pdfToText(ai, undefined, byteStream(pdf.bytes).stream, ranges(pdf).readRange); + expect(await read(result)).toBe("### Page 1\ngrey page"); + let { mimeType, bytes } = sentImage(images()[0]!); + expect(mimeType).toBe("image/png"); + expect(String.fromCharCode(...bytes.subarray(1, 4))).toBe("PNG"); + expect(new DataView(bytes.buffer).getUint32(16)).toBe(300); + }); + + it("returns a PDF's own text without transcribing anything", async () => { + let { ai, images } = ocrAi(PAGES, LARGE_PDF); + let { asked, readRange } = ranges(SCAN); + let result = await pdfToText(ai, undefined, byteStream(SCAN.bytes).stream, readRange); + expect(result).toMatchObject({ hasText: true, transcribed: false }); + expect(await read(result)).toBe("### Page 1\nLarge file probe text\n\n"); + expect(asked).toEqual([]); + expect(images()).toEqual([]); + }); + + it("transcribes nothing unless asked to", async () => { + let { ai, images } = ocrAi(); + let result = await pdfToText(ai, undefined, byteStream(SCAN.bytes).stream); + expect(result).toMatchObject({ hasText: false, transcribed: false }); + expect(await read(result)).toBe(""); + expect(images()).toEqual([]); + }); + + it("returns no text for a PDF with neither text nor page images", async () => { + let { ai, images } = ocrAi(); + let pdf = imagePdf([scanPage(1, { width: 64, height: 64 })]); + let result = await pdfToText(ai, undefined, byteStream(pdf.bytes).stream, ranges(pdf).readRange); + expect(result).toMatchObject({ hasText: false, transcribed: false }); + expect(await read(result)).toBe(""); + expect(images()).toEqual([]); + }); + + it.each([ + ["an encrypted PDF", imagePdf([scanPage(1)], { trailer: "/Encrypt 9 0 R" }), "it is encrypted", undefined], + ["fax-encoded pages", imagePdf([scanPage(1), scanPage(2, { filter: "/CCITTFaxDecode" })]), + "an image encoding that is not read yet, fax (CCITT); only JPEG and losslessly compressed pages are", "fax"], + ["JBIG2 pages", imagePdf([scanPage(1, { filter: "/JBIG2Decode" })]), "not read yet, JBIG2;", "jbig2"], + ["JPEG 2000 pages", imagePdf([scanPage(1, { filter: "/JPXDecode" })]), "not read yet, JPEG 2000;", "jpeg2000"], + ["pages in an unknown encoding", imagePdf([scanPage(1, { filter: "/LZWDecode" })]), "one this cannot read", "other"], + ["a page image that is not where the file says", imagePdf([scanPage(1, { declaredLength: 50 })]), + "a page image is not where the file says it is", "jpeg"], + ["a page image too large to send", imagePdf([scanPage(1, { data: new Uint8Array(12 * 1024 * 1024 + 1) })]), + "a page image is too large", "jpeg"], + ])("refuses %s before transcribing anything", async (_label, pdf, why, imageEncoding) => { + let { ai, images } = ocrAi(); + let { asked, readRange } = ranges(pdf); + await expect(pdfToText(ai, undefined, byteStream(pdf.bytes, [65_536]).stream, readRange)) + .rejects.toThrow(new RegExp(`^The PDF has no text of its own, and .*${ + why.replace(/[()]/g, "\\$&")}.*, so it cannot be transcribed\\.$`)); + expect(asked).toEqual([]); + expect(images()).toEqual([]); + expect(vi.mocked(console.info)).toHaveBeenCalledWith(expect.objectContaining({ + event: "google.drive.pdf.untranscribable", ...(imageEncoding ? { imageEncoding } : {}), + })); + }); + + it("refuses a PDF with more page images than it keeps", async () => { + let one = "<< /Subtype /Image /Width 600 /Height 800 /Filter /DCTDecode /Length 1 >>\nstream\nx\nendstream\nendobj\n"; + let text = "%PDF-1.4\n"; + for (let i = 1; i <= 2001; i++) text += `${i} 0 obj\n${one}`; + let { ai, images } = ocrAi(); + await expect(pdfToText(ai, undefined, byteStream(new TextEncoder().encode(text)).stream, ranges(SCAN).readRange)) + .rejects.toThrow("it has more than 2000 page images"); + expect(images()).toEqual([]); + }); + + it("refuses when the conversion answered before the PDF had all been sent", async () => { + let ai = { + fetch: async () => streamedResponse(markdownJson(SCANNED_PDF)), + } as unknown as AiFetcher; + await expect(pdfToText(ai, undefined, byteStream(SCAN.bytes, [100]).stream, ranges(SCAN).readRange)) + .rejects.toThrow(CONVERSION_FAILED); + }); + + it("reads a page from the file only once the text has been read up to it", async () => { + let { ai, images } = ocrAi(); + let { asked, readRange } = ranges(SCAN); + let result = await pdfToText(ai, undefined, byteStream(SCAN.bytes).stream, readRange); + let reader = result.text.getReader(); + let decoder = new TextDecoder(); + let text = ""; + let readUntil = async (end: string) => { + while (!text.endsWith(end)) text += decoder.decode((await reader.read()).value, { stream: true }); + }; + + await readUntil("### Page 1\n"); + expect(asked.map(range => range.page)).toEqual([1]); + await readUntil("Harbour Ledger Review"); + expect(asked.map(range => range.page)).toEqual([1]); + expect(images()).toHaveLength(1); + await readUntil("### Page 2\n"); + expect(asked.map(range => range.page)).toEqual([1, 2]); + + await reader.cancel(); + expect(asked.map(range => range.page)).toEqual([1, 2]); + expect(images()).toHaveLength(2); + }); + + it("stops the page in progress when cancelled", async () => { + let response = { cancelled: false }; + let endless = new Response(new ReadableStream({ + pull(controller) { + controller.enqueue(new TextEncoder().encode(tokenEvents(["more "]).split("data: [DONE]")[0])); + }, + cancel() { response.cancelled = true; }, + }, { highWaterMark: 0 })); + let { ai } = ocrAi([endless]); + let { asked, readRange } = ranges(SCAN); + let result = await pdfToText(ai, undefined, byteStream(SCAN.bytes).stream, readRange); + let reader = result.text.getReader(); + await reader.read(); + await reader.read(); + await reader.cancel(); + expect(response.cancelled).toBe(true); + expect(asked).toHaveLength(1); + }); + + it("errors the text in its own words when a page cannot be transcribed", async () => { + let warn = vi.mocked(console.warn); + let refused = new Response(JSON.stringify({ internalCode: 8006, message: "secret upstream text" }), { status: 400 }); + let { ai } = ocrAi([PAGES[0]!, refused]); + let result = await pdfToText(ai, undefined, byteStream(SCAN.bytes).stream, ranges(SCAN).readRange); + + let reader = result.text.getReader(); + let text = ""; + let error = await (async () => { + for (;;) { + let chunk = await reader.read(); + if (chunk.done) return undefined; + text += new TextDecoder().decode(chunk.value); + } + })().catch((caught: unknown) => caught); + expect(text).toBe("### Page 1\nHarbour Ledger Review"); + expect((error as Error).message).toBe(TRANSCRIPTION_FAILED); + expect(warn).toHaveBeenCalledWith(expect.objectContaining({ + event: "google.drive.pdf.transcription_failed", page: 2, imageBytes: 702, imageEncoding: "jpeg", + httpStatus: 400, providerCode: 8006, + })); + expect(everythingLogged()).not.toContain("secret"); + }); + + it("errors the text when the model's answer breaks off, without logging what it said", async () => { + let brokeOff = eventResponse(tokenEvents(["Lot 55-0193-G"]).split("data: [DONE]")[0]!); + let { ai } = ocrAi([PAGES[0]!, brokeOff]); + let result = await pdfToText(ai, undefined, byteStream(SCAN.bytes).stream, ranges(SCAN).readRange); + await expect(read(result)).rejects.toThrow(TRANSCRIPTION_FAILED); + expect(vi.mocked(console.warn)).toHaveBeenCalledWith(expect.objectContaining({ + event: "google.drive.pdf.transcription_failed", page: 2, + })); + expect(everythingLogged()).not.toContain("55-0193"); + }); + + it("passes on a page read's own failure, such as a refused or changed file, as it is", async () => { + let { ai, images } = ocrAi(); + let changed = new Error("The Drive file changed while it was being read; read it again."); + let { readRange } = ranges(SCAN); + let result = await pdfToText(ai, undefined, byteStream(SCAN.bytes).stream, + range => range.page === 2 ? Promise.reject(changed) : readRange(range)); + await expect(read(result)).rejects.toBe(changed); + expect(images()).toHaveLength(1); + expect(vi.mocked(console.warn)).not.toHaveBeenCalled(); + }); + + it("marks a page whose transcription the model was stopped partway through", async () => { + let { ai } = ocrAi([eventResponse(tokenEvents(["row | row"], "length")), ["next"]]); + let pdf = imagePdf([scanPage(1), scanPage(2)]); + let result = await pdfToText(ai, undefined, byteStream(pdf.bytes).stream, ranges(pdf).readRange); + expect(await read(result)).toBe( + "### Page 1\nrow | row\n\n[The transcription of this page was cut off.]\n\n### Page 2\nnext"); + }); +}); diff --git a/packages/gatekeeper-google/__tests__/types.test.ts b/packages/gatekeeper-google/__tests__/types.test.ts index c3289d378f..a1cc7c439c 100644 --- a/packages/gatekeeper-google/__tests__/types.test.ts +++ b/packages/gatekeeper-google/__tests__/types.test.ts @@ -304,7 +304,10 @@ const legacy: SheetCellFormat = { wrap: "OVERFLOW_CELL" }; it("streams PDF text from both Drive sessions", () => { const driveTypes = source("drive-types.d.ts"); expect(driveTypes).toContain("export type DrivePdfText = {"); - expect(driveTypes.split("readPdf(fileId: string): Promise;")).toHaveLength(3); + expect(driveTypes).toContain("export type DrivePdfReadOptions = {"); + expect(driveTypes.split( + "readPdf(fileId: string, options?: DrivePdfReadOptions): Promise;", + )).toHaveLength(3); expect(compileAgentTypes(driveBundle() + ` declare const drive: GoogleDriveReadSession; declare const folder: GoogleDriveFolderSession; @@ -314,12 +317,19 @@ async function read(): Promise { const reader = (await folder.readPdf("file-2")).text.getReader(); const first: Uint8Array | undefined = (await reader.read()).value; await reader.cancel(); + const scan = await folder.readPdf("file-3", { ocr: true }); + const read: boolean = scan.transcribed && !scan.hasText; + await drive.readPdf("file-4", {}); return new Response(text).text(); } // @ts-expect-error The text is a stream, not a string. -const whole: DrivePdfText = { text: "", hasText: true }; +const whole: DrivePdfText = { text: "", hasText: true, transcribed: false }; // @ts-expect-error Nothing is cut, so there is no truncation flag. -const cut: DrivePdfText = { text: new ReadableStream(), hasText: true, truncated: false }; +const cut: DrivePdfText = { text: new ReadableStream(), hasText: true, transcribed: false, truncated: false }; +// @ts-expect-error Whether the text was transcribed is always said. +const unsaid: DrivePdfText = { text: new ReadableStream(), hasText: true }; +// @ts-expect-error Transcribing is asked for with a boolean. +const asked: DrivePdfReadOptions = { ocr: "yes" }; `)).toEqual([]); }); diff --git a/packages/gatekeeper-google/__tests__/workerd/native-sessions.test.ts b/packages/gatekeeper-google/__tests__/workerd/native-sessions.test.ts index f1302a555f..4aa6fe178d 100644 --- a/packages/gatekeeper-google/__tests__/workerd/native-sessions.test.ts +++ b/packages/gatekeeper-google/__tests__/workerd/native-sessions.test.ts @@ -8,9 +8,9 @@ import { GoogleDocsApi } from "../../src/docs-api"; import { DriveApi } from "../../src/drive-api"; import { FILE_READ_IDLE_MS } from "../../src/drive-file-read"; import type { DriveBindingScope } from "../../src/drive-session"; -import type { PdfConverter } from "../../src/drive-pdf"; +import type { PdfConverter, PdfRangeReader } from "../../src/drive-pdf"; import { GoogleDriveSessionImpl } from "../../src/google"; -import type { DrivePdfText, DriveSessionSearchQuery } from "../../src/drive-types"; +import type { DrivePdfReadOptions, DrivePdfText, DriveSessionSearchQuery } from "../../src/drive-types"; import { GoogleSheetsApi } from "../../src/sheets-api"; import type { GoogleSpreadsheetSession } from "../../src/sheets-types"; import { GoogleSlidesApi } from "../../src/slides-api"; @@ -60,11 +60,43 @@ function fakeConverter(onConvert?: () => void) { }, cancel() { state.cancelled++; }, }, { highWaterMark: 0 }); - return { text: stream, hasText: true }; + return { text: stream, hasText: true, transcribed: false }; }; return { converted, convert, state }; } +/** + * A converter standing for a scanned PDF. Asked to transcribe, its text is what it reads of the + * file, a page as each piece is pulled: bytes 4-6, then bytes 8-10. + */ +function scanConverter() { + const given: (PdfRangeReader | undefined)[] = []; + const convert: PdfConverter = async (pdf, readRange) => { + const fileSize = (await new Response(pdf).arrayBuffer()).byteLength; + given.push(readRange); + let page = 0; + const pages = new ReadableStream({ + async pull(controller) { + if (!readRange || page === 2) return controller.close(); + page++; + const image = await readRange({ page, offset: page * 4, length: 3, fileSize }); + controller.enqueue(new Uint8Array(await new Response(image).arrayBuffer())); + }, + }, { highWaterMark: 0 }); + return { text: pages, hasText: false, transcribed: readRange !== undefined }; + }; + return { convert, given }; +} + +/** A download's answer to `range`: those bytes of PDF_BYTES as a 206, or the whole file to none. */ +function rangedDownload(range: string | null, total = PDF_BYTES.byteLength): Response { + const [, start, end] = /^bytes=(\d+)-(\d+)$/.exec(range ?? "")?.map(Number) ?? []; + if (start === undefined) return new Response(PDF_BYTES, { headers: { "Content-Type": PDF_MIME } }); + return new Response(PDF_BYTES.slice(start, end! + 1), { + status: 206, headers: { "Content-Range": `bytes ${start}-${end}/${total}` }, + }); +} + /** Reads a PDF read's text stream, as returned over RPC, to the end. */ async function readText(result: DrivePdfText): Promise<{ text: string; hasText: boolean }> { return { text: await new Response(result.text).text(), hasText: result.hasText }; @@ -97,8 +129,10 @@ let providerUrls: string[]; let providerTabs: unknown[]; /** Absent for a document the caller cannot edit, as Google returns it. */ let providerRevision: string | undefined; -/** What a file download answers. */ -let providerDownload: () => Response; +/** What a file download answers, given its `Range` header. */ +let providerDownload: (range: string | null) => Response; +/** The `Range` of each download {@link installProvider} served, in order. */ +let providerRanges: (string | null)[]; async function getAccessToken(): Promise { return "access-token"; @@ -106,14 +140,16 @@ async function getAccessToken(): Promise { class TestApprovalQueue extends RpcTarget implements ApprovalQueue { readonly observations: ObservationDescription[] = []; - /** Titles this queue refuses to authorize. */ + /** Titles, or whole descriptions, this queue refuses to authorize. */ readonly refuse = new Set(); /** Titles whose authorization waits for the promise mapped to them. */ readonly holds = new Map>(); async authorizeObservation(description: ObservationDescription): Promise { await this.holds.get(description.title); - if (this.refuse.has(description.title)) throw new Error("Observation refused"); + if (this.refuse.has(description.title) || this.refuse.has(description.description)) { + throw new Error("Observation refused"); + } this.observations.push(description); } @@ -202,14 +238,17 @@ function installProvider(tabs: unknown[] = [docTab("solo", "Solo", "")]) { providerDownload = () => new Response(PDF_BYTES, { headers: { "Content-Type": PDF_MIME, "Content-Length": String(PDF_BYTES.byteLength) }, }); - vi.stubGlobal("fetch", vi.fn(async (input: string | URL | Request) => { + providerRanges = []; + vi.stubGlobal("fetch", vi.fn(async (input: string | URL | Request, init?: RequestInit) => { const url = new URL(input instanceof Request ? input.url : input.toString()); urls.push(url.toString()); if (url.hostname === "www.googleapis.com" && url.pathname.endsWith("/drive/v3/files")) { return Response.json({ files: [providerFile("doc-1", DOC_MIME)] }); } if (url.hostname === "www.googleapis.com" && isDownload(url.toString())) { - return providerDownload(); + const range = new Headers(init?.headers).get("Range"); + providerRanges.push(range); + return providerDownload(range); } if (url.hostname === "www.googleapis.com" && url.pathname.includes("/drive/v3/files/")) { const id = decodeURIComponent(url.pathname.split("/").at(-1)!); @@ -480,6 +519,7 @@ describe("Drive PDF reads", () => { const converted = { cancelled: false }; const convert: PdfConverter = async () => ({ text: new ReadableStream({ cancel() { converted.cancelled = true; } }), hasText: true, + transcribed: false, }); const { queue, session } = newSession(convert); using owned = session; @@ -490,6 +530,94 @@ describe("Drive PDF reads", () => { expect(converted.cancelled).toBe(true); expect(download.cancelled).toBe(true); }); + + it("lets the converter read page images only when asked to transcribe", async () => { + const converter = scanConverter(); + using session = newSession(converter.convert).session; + + for (const options of [undefined, {}, { ocr: false }]) { + const result = await session.readPdf("pdf-1", options); + expect(result.transcribed).toBe(false); + expect(await readText(result)).toEqual({ text: "", hasText: false }); + } + expect(converter.given).toEqual([undefined, undefined, undefined]); + expect(providerRanges).toEqual([null, null, null]); + }); + + it("reads each page image as a content read of its own, as the text is read", async () => { + providerDownload = rangedDownload; + const converter = scanConverter(); + const { queue, session } = newSession(converter.convert); + using owned = session; + + const result = await owned.readPdf("pdf-1", { ocr: true }); + expect(result).toMatchObject({ hasText: false, transcribed: true }); + // RPC reads one piece ahead, so the first page may already have been asked for; the second + // is not until the text is read. + expect(providerRanges.slice(0, 2)).toEqual([null, "bytes=4-6"].slice(0, providerRanges.length)); + expect(await new Response(result.text).text()).toBe("-1.\n1 "); + expect(providerRanges).toEqual([null, "bytes=4-6", "bytes=8-10"]); + expect(queue.observations.map(({ title, description, subresources }) => + ({ title, description, subresources })).slice(1)).toEqual([ + { + title: "Read PDF from Google Drive", + description: "Read the text of Drive file pdf-1.", + subresources: ["drive/content/pdf-1"], + }, + { + title: "Read PDF from Google Drive", + description: "Read page image 1 of Drive file pdf-1, to transcribe it.", + subresources: ["drive/content/pdf-1"], + }, + { + title: "Read PDF from Google Drive", + description: "Read page image 2 of Drive file pdf-1, to transcribe it.", + subresources: ["drive/content/pdf-1"], + }, + ]); + }); + + it("fails the text when a page image's read is not authorized, releasing its download", async () => { + const download = { cancelled: 0 }; + providerDownload = range => range !== "bytes=8-10" ? rangedDownload(range) : new Response( + new ReadableStream({ cancel() { download.cancelled++; } }), + { status: 206, headers: { "Content-Range": `bytes 8-10/${PDF_BYTES.byteLength}` } }); + const converter = scanConverter(); + const { queue, session } = newSession(converter.convert); + using owned = session; + queue.refuse.add("Read page image 2 of Drive file pdf-1, to transcribe it."); + + const result = await owned.readPdf("pdf-1", { ocr: true }); + const reader = result.text.getReader(); + expect(new TextDecoder().decode((await reader.read()).value)).toBe("-1."); + await expect(reader.read()).rejects.toThrow(); + expect(download.cancelled).toBe(1); + expect(queue.observations.map(o => o.description)).not.toContain( + "Read page image 2 of Drive file pdf-1, to transcribe it."); + }); + + it("fails the text when the file is no longer the size it was converted at", async () => { + providerDownload = range => rangedDownload(range, PDF_BYTES.byteLength + 1); + const converter = scanConverter(); + const { queue, session } = newSession(converter.convert); + using owned = session; + + const result = await owned.readPdf("pdf-1", { ocr: true }); + await expect(new Response(result.text).text()).rejects.toThrow(); + expect(queue.observations.map(o => o.description)).not.toContain( + "Read page image 1 of Drive file pdf-1, to transcribe it."); + }); + + it("refuses an option that is not a boolean at the RPC boundary", async () => { + const converter = scanConverter(); + using session = newSession(converter.convert).session; + + await expect(Promise.resolve(session.readPdf( + "pdf-1", { ocr: "yes" } as unknown as DrivePdfReadOptions, + ))).rejects.toThrow(/readPdf\[1\]/); + expect(converter.given).toEqual([]); + expect(providerUrls).toEqual([]); + }); }); describe("Drive file reads", () => { @@ -1068,8 +1196,11 @@ describe("folder-scoped native sessions", () => { vi.stubGlobal("fetch", vi.fn(async (input: string | URL | Request, init?: RequestInit) => { const url = new URL(input instanceof Request ? input.url : input.toString()); if (isDownload(url.toString())) { - nativeCalls.push("download"); + const range = new Headers(init?.headers).get("Range"); + nativeCalls.push(range?.startsWith("bytes=0-") === false ? `download ${range}` : "download"); onNativeRead?.(); + // A range from past the start is answered as one; any other with the whole file. + if (range?.startsWith("bytes=0-") === false) return rangedDownload(range); const half = PDF_BYTES.byteLength >> 1; let sent = 0; return new Response(new ReadableStream({ @@ -1396,6 +1527,47 @@ describe("folder-scoped native sessions", () => { expect(converter.state.cancelled).toBe(1); }); + it("re-proves a PDF's place around each page image read", async () => { + const nodes = subtree(); + nodes.set("pdf-1", { id: "pdf-1", mimeType: PDF_MIME, parents: [ROOT], trashed: false }); + const nativeCalls = installFolderProvider(nodes); + const { queue, session } = folderSession(nodes, scanConverter().convert); + using owned = session; + + const result = await owned.readPdf("pdf-1", { ocr: true }); + expect(await new Response(result.text).text()).toBe("-1.\n1 "); + expect(nativeCalls).toEqual(["download", "download bytes=4-6", "download bytes=8-10"]); + expect(queue.observations.map(o => [o.description, o.subresources]).slice(1)).toEqual([ + "Read the text of Drive file pdf-1.", + "Read page image 1 of Drive file pdf-1, to transcribe it.", + "Read page image 2 of Drive file pdf-1, to transcribe it.", + ].map(description => + [description, [`drive/folder/${ROOT}`, "drive/file/pdf-1", "drive/content/pdf-1"]])); + }); + + it.each([ + ["moved out of the folder", { parents: ["elsewhere"] }], + ["rewritten", { modifiedTime: "2026-08-21T09:00:00Z" }], + ])("stops transcribing a PDF %s while a page image is read", async (_label, change) => { + const nodes = subtree(); + const pdf = { id: "pdf-1", mimeType: PDF_MIME, parents: [ROOT], trashed: false }; + nodes.set("pdf-1", pdf); + // The change lands as Google is asked for the second page image, the third download. + let downloads = 0; + installFolderProvider(nodes, () => { + if (++downloads === 3) nodes.set("pdf-1", { ...pdf, ...change }); + }); + const { queue, session } = folderSession(nodes, scanConverter().convert); + using owned = session; + + const result = await owned.readPdf("pdf-1", { ocr: true }); + const reader = result.text.getReader(); + expect(new TextDecoder().decode((await reader.read()).value)).toBe("-1."); + await expect(reader.read()).rejects.toThrow(); + expect(queue.observations.map(o => o.description)).not.toContain( + "Read page image 2 of Drive file pdf-1, to transcribe it."); + }); + it("refuses a PDF outside the folder without downloading it", async () => { const nodes = subtree(); nodes.set("pdf-1", { id: "pdf-1", mimeType: PDF_MIME, parents: ["elsewhere"], trashed: false }); diff --git a/packages/gatekeeper-google/src/drive-pdf.ts b/packages/gatekeeper-google/src/drive-pdf.ts index ad78e3f139..c248c37f49 100644 --- a/packages/gatekeeper-google/src/drive-pdf.ts +++ b/packages/gatekeeper-google/src/drive-pdf.ts @@ -1,20 +1,36 @@ // A Drive PDF's text, from Workers AI's markdown conversion, streamed in both directions: neither -// the PDF nor its text is ever held whole. +// the PDF nor its text is ever held whole. A scanned PDF has no text to convert, so when asked its +// page images are transcribed by a vision model instead, one at a time, as the text is read. import { PDF_MIME_TYPE } from "./drive-session"; import type { DrivePdfText } from "./drive-types"; import { obsContext } from "./observability"; +import { MAX_PAGE_IMAGES, PdfImageScan, type PdfImages, type PdfPageImage } from "./pdf-images"; +import { pngFromDeflated } from "./pdf-png"; import { convertToMarkdown, MarkdownConversionError, type AiFetcher, type MarkdownConversion, } from "./workers-ai-markdown"; +import { transcribeImage, TranscriptionError, type Transcription } from "./workers-ai-vision"; const logger = obsContext.createLogger({ component: "gatekeeper.google.drive.pdf", vendorId: "google" }); +/** One page image's place in a PDF: `length` bytes from `offset`, in a file of `fileSize` bytes. */ +export type PdfPageRange = { page: number; offset: number; length: number; fileSize: number }; + +/** + * Reads one page image's bytes from the PDF a conversion was given, some time after it. The file + * may have changed since, so the reader refuses a file that is no longer `fileSize` bytes long. + */ +export type PdfRangeReader = (range: PdfPageRange) => Promise>; + /** * Converts one PDF, read from a stream, to its text. It may resolve before reading `pdf` to its end, - * so a caller that checks the file after converting it must see the end itself. + * so a caller that checks the file after converting it must see the end itself. With `readRange`, + * a PDF that has no text has its page images transcribed, each read through it. */ -export type PdfConverter = (pdf: ReadableStream) => Promise; +export type PdfConverter = ( + pdf: ReadableStream, readRange?: PdfRangeReader, +) => Promise; /** * The name every PDF is converted under. The conversion heads its output with the input's name, so @@ -28,21 +44,43 @@ const PAGE_HEADING = /^### Page \d+$/; const MAX_HEADING_CHARS = 32; /** The one error a failed conversion is reported with, whatever the service said. */ export const CONVERSION_FAILED = "Google Drive's file could not be converted from PDF."; +/** The one error a failed transcription is reported with, whatever the service said. */ +export const TRANSCRIPTION_FAILED = "Google Drive's PDF could not be transcribed."; +/** What follows a page whose transcription the model was stopped partway through. */ +const CUT_OFF = "\n\n[The transcription of this page was cut off.]"; +/** + * The largest page image sent to be transcribed. Workers AI refuses a request body somewhere + * between a 12 MiB and a 16 MiB image, and a scanned page is a few megabytes at most. + */ +const MAX_IMAGE_BYTES = 12 * 1024 * 1024; +/** What the encodings that are not read are called in a refusal. */ +const UNREAD_ENCODINGS = { + fax: "fax (CCITT)", jbig2: "JBIG2", jpeg2000: "JPEG 2000", other: "one this cannot read", +}; /** * Converts a PDF with the Workers AI binding, through AI Gateway `gatewayId` when one is named. * The PDF is encoded into the request as it arrives, and the response is read only until its text * begins, which settles `hasText`; the text then streams from the response as it is read, without * the file heading, "## Contents" line, and pages before the first with text. Its images are not - * described (an image-only page yields no text; nothing is OCR'd), so the conversion makes no - * billed model call. A failed conversion is logged by status and error code only, never by the + * described (an image-only page yields no text), so the conversion makes no billed model call. A failed conversion is logged by status and error code only, never by the * service's text, and reported in our words, by a throw before the text begins or by erroring the * stream after; a failure of the PDF stream itself (such as the download breaking off) is thrown * as it is. + * + * With `readRange`, the PDF's page images are located as its bytes go by, and when it proves to + * have no text they are transcribed instead: see {@link transcription}. A PDF with any text of + * its own is returned as that text, and no model is called. One whose pages cannot be + * transcribed is refused before any is. */ export async function pdfToText( ai: AiFetcher, gatewayId: string | undefined, pdf: ReadableStream, + readRange?: PdfRangeReader, ): Promise { + let scan = readRange && new PdfImageScan(); + /** Whether the scan was shown the PDF to its end. */ + let scanned = false; + if (scan) pdf = watched(pdf, chunk => scan.feed(chunk), () => { scanned = true; }); let options = { ...(gatewayId ? { gateway: { id: gatewayId } } : {}), conversionOptions: { pdf: { metadata: false, images: { convert: false } } }, @@ -69,7 +107,17 @@ export async function pdfToText( event: "google.drive.pdf.converted", pdfBytes: conversion.sentBytes, hasText, }); if (!hasText) { - return { text: new ReadableStream({ start: controller => controller.close() }), hasText }; + // A conversion that answered before the PDF ended leaves page images unlocated. + if (scan && !scanned) throw new Error(CONVERSION_FAILED); + let pages = scan ? pageImages(scan.end()) : undefined; + if (!pages || pages.images.length === 0) { + let text = new ReadableStream({ start: controller => controller.close() }); + return { text, hasText, transcribed: false }; + } + logger.info("transcribing a scanned Drive PDF", { + event: "google.drive.pdf.transcribing", pdfBytes: pages.bytes, ocrPages: pages.images.length, + }); + return { text: transcription(ai, gatewayId, pages, readRange!), hasText, transcribed: true }; } let encoder = new TextEncoder(); @@ -93,7 +141,128 @@ export async function pdfToText( return conversion.cancel(reason); }, }, { highWaterMark: 0 }); - return { text, hasText }; + return { text, hasText, transcribed: false }; +} + +/** `pdf`, with each chunk shown to `onChunk` as it is read and `onEnd` called at its end. */ +function watched( + pdf: ReadableStream, onChunk: (chunk: Uint8Array) => void, onEnd: () => void, +): ReadableStream { + let reader = pdf.getReader(); + return new ReadableStream({ + async pull(controller) { + let chunk = await reader.read(); + if (chunk.done) { + onEnd(); + controller.close(); + } else { + onChunk(chunk.value); + controller.enqueue(chunk.value); + } + }, + cancel: reason => reader.cancel(reason), + }, { highWaterMark: 0 }); +} + +/** + * The page images of a scanned PDF as they can be transcribed, or a refusal in our words when + * they cannot all be: the file is encrypted, a page image is not where the file says, is too + * large, or is in an encoding that is not read. Only JPEG and losslessly compressed images are. + */ +function pageImages(found: PdfImages): PdfImages { + let refuse = (why: string, encoding?: PdfPageImage["encoding"]): never => { + logger.info("a scanned Drive PDF cannot be transcribed", { + event: "google.drive.pdf.untranscribable", pdfBytes: found.bytes, + ocrPages: found.images.length, ...(encoding ? { imageEncoding: encoding } : {}), + }); + throw new Error(`The PDF has no text of its own, and ${why}, so it cannot be transcribed.`); + }; + if (found.encrypted) refuse("it is encrypted"); + if (found.more) refuse(`it has more than ${MAX_PAGE_IMAGES} page images`); + for (let image of found.images) { + let { encoding } = image; + if (encoding !== "jpeg" && encoding !== "lossless") { + refuse(`its pages are stored in an image encoding that is not read yet, ${ + UNREAD_ENCODINGS[encoding]}; only JPEG and losslessly compressed pages are`, encoding); + } + if (image.damaged) refuse("a page image is not where the file says it is", encoding); + if (image.length > MAX_IMAGE_BYTES) refuse("a page image is too large", encoding); + } + return found; +} + +/** + * The transcription of a scanned PDF's `pages`, as UTF-8 Markdown under the "### Page N" headings + * a conversion gives, numbered in the order the file stores the images. Nothing is done until it + * is read, and then one page at a time: a page's image is read from the file through `readRange` + * and encoded into a vision model's request as it arrives, and the model's text is passed on as + * it is produced, so the next page is not asked for until this one has been read to its end, and + * neither an image nor a page's text is held whole. Cancelling it stops the page in progress and + * reads no more. + * + * A failed transcription is logged by page, size and status only, never by the service's text, + * and errors the stream in our words; a failure of a page's read is passed on as it is. + */ +function transcription( + ai: AiFetcher, gatewayId: string | undefined, { images, bytes: fileSize }: PdfImages, + readRange: PdfRangeReader, +): ReadableStream { + let encoder = new TextEncoder(); + /** The page being transcribed, from 0, and its transcription once the model has answered. */ + let index = 0; + let current: Transcription | undefined; + let cancelled = false; + let fields = () => ({ page: index + 1, imageEncoding: images[index]!.encoding }); + let begin = async (): Promise => { + let image = images[index]!; + let { offset, length } = image; + let data = await readRange({ page: index + 1, offset, length, fileSize }); + if (cancelled) { + await data.cancel().catch(() => undefined); + throw new Error("cancelled"); + } + return transcribeImage(ai, gatewayId, image.png + ? { mimeType: "image/png", body: pngFromDeflated(data, image, image.png) } + : { mimeType: "image/jpeg", body: data }); + }; + return new ReadableStream({ + async pull(controller) { + try { + // A pull that enqueues nothing is not called again, so go on until there is text. + for (;;) { + if (!current) { + if (index === images.length) return controller.close(); + let begun = await begin(); + if (cancelled) return await begun.cancel(); + current = begun; + return controller.enqueue(encoder.encode(`${index > 0 ? "\n\n" : ""}### Page ${index + 1}\n`)); + } + let piece = await current.next(); + if (piece !== undefined) return controller.enqueue(encoder.encode(piece)); + logger.info("transcribed a Drive PDF page", { + event: "google.drive.pdf.page_transcribed", imageBytes: current.sentBytes, ...fields(), + }); + let cutOff = current.truncated; + current = undefined; + index++; + if (cutOff) return controller.enqueue(encoder.encode(CUT_OFF)); + } + } catch (error) { + // Cancelling ends a read in progress as if the response had broken off. + if (cancelled) return; + if (!(error instanceof TranscriptionError)) return controller.error(error); + logger.warn("Workers AI failed to transcribe a PDF page", { + event: "google.drive.pdf.transcription_failed", imageBytes: error.sentBytes, + ...fields(), ...error.failure, + }); + controller.error(new Error(TRANSCRIPTION_FAILED)); + } + }, + cancel: reason => { + cancelled = true; + return current?.cancel(reason); + }, + }, { highWaterMark: 0 }); } /** diff --git a/packages/gatekeeper-google/src/drive-types.d.ts b/packages/gatekeeper-google/src/drive-types.d.ts index 85725a26a9..2e5d4d9b62 100644 --- a/packages/gatekeeper-google/src/drive-types.d.ts +++ b/packages/gatekeeper-google/src/drive-types.d.ts @@ -126,17 +126,38 @@ export type DriveFolderSearchQuery = Omit & export type DriveSessionSearchQuery = DriveSearchQuery & Pick; +/** How `readPdf()` reads a PDF. */ +export type DrivePdfReadOptions = { + /** + * Transcribe a scanned or image-only PDF, which has no text of its own, by having a vision + * model read its page images. Off by default, since each page is a billed model call. A PDF + * with any text of its own is returned as that text whether or not this is set. + */ + ocr?: boolean; +}; + /** A Drive PDF's text, streamed. */ export type DrivePdfText = { /** * The PDF's text as UTF-8 Markdown, read as it is converted: pages separated by "### Page N" * headings, and pages before the first with text left out. Read it with * `await new Response(text).text()`, or a reader to stop early; cancel it if unread. Empty when - * `hasText` is false. A failure partway through errors the stream. + * `hasText` is false and `transcribed` is too. A failure partway through errors the stream. + * + * When `transcribed`, each page is read from Drive and transcribed only as you read the text, so + * stop reading, and cancel, to stop paying for pages you do not need. Pages are numbered in the + * order the file stores their images, which is page order for a scanner's output. */ text: ReadableStream; - /** False for a scanned or image-only PDF, which has no text layer; it is not OCR'd. */ + /** False for a scanned or image-only PDF, which has no text layer. */ hasText: boolean; + /** + * True when `text` is a vision model's reading of the page images of a PDF with no text of its + * own, which `readPdf()` gives only with `ocr: true`. It is not the document's own text: the + * model can misread, most often small print and the codes, numbers and names in it, so check + * anything that matters against the file. + */ + transcribed: boolean; }; /** Which bytes of a Drive file `readFile()` reads. */ @@ -218,8 +239,13 @@ export interface GoogleDriveReadSession { /** * Read an in-scope PDF's text, converted to Markdown and streamed. Fails asking to reconnect * when the Google account was connected before Drive bindings could read file content. + * + * A scanned PDF has no text: `hasText` is false and `text` is empty. Pass `{ ocr: true }` to + * have its pages transcribed instead (see `DrivePdfText.transcribed`). That reads pages stored + * as JPEG or losslessly compressed images, and fails, before transcribing anything, for an + * encrypted PDF or one whose pages are stored as fax, JBIG2 or JPEG 2000 images. */ - readPdf(fileId: string): Promise; + readPdf(fileId: string, options?: DrivePdfReadOptions): Promise; /** * Read an in-scope file's own content as a stream of bytes: a CSV, image, PDF, Office or any @@ -261,10 +287,11 @@ export interface GoogleDriveFolderSession extends Pick; /** - * Read a live direct-child PDF's text, converted to Markdown and streamed. Fails asking to - * reconnect when the Google account was connected before Drive bindings could read file content. + * Read a live direct-child PDF's text, converted to Markdown and streamed, as + * `GoogleDriveReadSession.readPdf()` does. Fails asking to reconnect when the Google account + * was connected before Drive bindings could read file content. */ - readPdf(fileId: string): Promise; + readPdf(fileId: string, options?: DrivePdfReadOptions): Promise; /** * Read a live direct child's own content as a stream of bytes, as diff --git a/packages/gatekeeper-google/src/google.ts b/packages/gatekeeper-google/src/google.ts index 353044c7f2..7856e53fcf 100644 --- a/packages/gatekeeper-google/src/google.ts +++ b/packages/gatekeeper-google/src/google.ts @@ -41,7 +41,9 @@ import { type DocTabSnapshot, type EditableMarkdown, } from "./markdown-converter"; import { DriveApi, DriveApiRequestError, lacksFileContentAccess, MAX_BATCH_FILES } from "./drive-api"; -import { CONVERSION_FAILED, pdfToText, type PdfConverter } from "./drive-pdf"; +import { + CONVERSION_FAILED, pdfToText, type PdfConverter, type PdfRangeReader, +} from "./drive-pdf"; import { downloadableFile, exactRange, fileReadRange, FILE_CHANGED } from "./drive-file-read"; import type { AiFetcher } from "./workers-ai-markdown"; import { driveObserverTracker, type DriveObservation } from "./drive-observers"; @@ -55,9 +57,11 @@ import { DriveFolderSessionCore, DriveSessionCore, driveModifiedTime, GOOGLE_DOC_MIME_TYPE, GOOGLE_SHEET_MIME_TYPE, GOOGLE_SLIDES_MIME_TYPE, PDF_MIME_TYPE, requireDriveBindingScope, type DriveBindingScope, type DriveCore, type NativeObservation, type NativeRead, + type NativeReadOptions, } from "./drive-session"; import type { - DriveEntry, DriveFileBytes, DriveFileReadOptions, DriveListOptions, DrivePdfText, + DriveEntry, DriveFileBytes, DriveFileReadOptions, DriveListOptions, DrivePdfReadOptions, + DrivePdfText, DriveSessionSearchQuery, GoogleDriveFolderSession, GoogleDriveReadSession, GoogleDriveSession, } from "./drive-types"; import { ChatApi, ChatApiError } from "./chat-api"; @@ -3423,7 +3427,8 @@ export class GoogleDriveGatekeeperImpl // `Ai` declares no `fetch`, but the binding has one; conversion goes through it so it can stream. let ai = this.env.AI as unknown as AiFetcher | undefined; let gatewayId = this.env.CF_AI_GATEWAY; - let convertPdf: PdfConverter | undefined = ai && (pdf => pdfToText(ai, gatewayId, pdf)); + let convertPdf: PdfConverter | undefined = + ai && ((pdf, readRange) => pdfToText(ai, gatewayId, pdf, readRange)); return new GoogleDriveSessionImpl( new DriveApi(getDriveAccessToken), new GoogleDocsApi(getDriveAccessToken), @@ -3702,14 +3707,48 @@ export class GoogleDriveSessionImpl extends RpcTarget new GooglePresentationReadSessionImpl(this.#slidesApi, presentationId, queue, read)); } - async readPdf(fileId: string): Promise { + async readPdf(fileId: string, options: DrivePdfReadOptions = {}): Promise { // The text stream needs no queue of its own, so this reads through the session's, as getEntry. let core = this.#core; let id = await core.openNativeFile(fileId, PDF_MIME_TYPE, "PDF"); let convert = this.#convertPdf; if (!convert) throw new Error("PDF reading is not available on this deployment."); let converted: DrivePdfText | undefined; - return core.nativeRead(id, PDF_MIME_TYPE, {content: true})(async () => { + // A transcription reads the file again, page by page, at places found in the bytes converted, + // so its reads must all be of that file as it was. A folder binding's re-proofs see the + // modification time, which must not move from the first one on; a file of another size is + // refused on any binding. As for readFile(), an account or exact-file binding does not + // detect a rewrite that kept the size. + let modifiedTime: string | undefined; + let read: NativeReadOptions = options.ocr ? { + content: true, + unchanged: current => { + modifiedTime ??= current.modifiedTime; + if (current.modifiedTime !== modifiedTime) throw new Error(FILE_CHANGED); + }, + } : {content: true}; + // Each page read is a content read of its own, made as the text is read: checked against the + // binding's scope, re-proved on a folder binding, and reported, like the first. + let readRange: PdfRangeReader = async ({page, offset, length, fileSize}) => { + let opened: ReadableStream | undefined; + let download = await core.nativeRead(id, PDF_MIME_TYPE, read)(async () => { + let answer = await this.#driveApi.downloadRange(id, offset, length) + .catch((error: unknown) => { throw fileReadError(error); }); + opened = answer.content; + if ((answer.fileSize ?? fileSize) !== fileSize || (answer.length ?? length) < length) { + throw new Error(FILE_CHANGED); + } + return opened; + }, () => ({ + title: "Read PDF from Google Drive", + description: `Read page image ${page} of Drive file ${id}, to transcribe it.`, + })).catch(async (error: unknown) => { + await opened?.cancel().catch(() => undefined); + throw error; + }); + return exactRange(download, length); + }; + return core.nativeRead(id, PDF_MIME_TYPE, read)(async () => { let download = (await this.#driveApi.openDownload(id).catch((error: unknown) => { throw lacksFileContentAccess(error) ? new Error(contentAccessMissing("PDFs")) : error; })).getReader(); @@ -3727,7 +3766,7 @@ export class GoogleDriveSessionImpl extends RpcTarget } }, cancel: reason => download.cancel(reason), - }, { highWaterMark: 0 })); + }, { highWaterMark: 0 }), options.ocr ? readRange : undefined); if (!downloaded) { await download.cancel().catch(() => undefined); throw new Error(CONVERSION_FAILED); diff --git a/packages/gatekeeper-google/src/observability.ts b/packages/gatekeeper-google/src/observability.ts index 1c4d1bb843..35e52d3cbd 100644 --- a/packages/gatekeeper-google/src/observability.ts +++ b/packages/gatekeeper-google/src/observability.ts @@ -9,9 +9,17 @@ export type GoogleObservabilityFields = { /** Whether a converted PDF had a text layer. */ hasText: boolean; httpStatus: number; + /** Size of a PDF page image sent for transcription, in bytes as stored in the PDF. */ + imageBytes: number; + /** How a PDF page image is encoded, in the gatekeeper's own names. */ + imageEncoding: string; messageId: string; method: string; + /** How many page images a scanned PDF has to transcribe. */ + ocrPages: number; operation: string; + /** Which page image of a PDF, from 1, in the order the file stores them. */ + page: number; /** Size of a PDF sent for conversion, in bytes. */ pdfBytes: number; provider: string;