fix(extraction): process large PDFs without truncating, timing out, or double-billing

containers/liteparse/server.mjs

@@ -11,9 +11,19 @@ import { Transform } from "node:stream";
 import { promisify } from "node:util";
 
 const port = 8080;
+// LiteParse defaults to 1000 pages and drops everything past that without reporting
+// it, which publishes a "ready" projection silently missing the tail of the document.
+// Parse one page past the supported ceiling so a document of exactly the ceiling is
+// distinguishable from a truncated longer one. Measured at roughly 0.57 MB resident
+// per page, so a full parse sits near 3 GB on the 8 GiB standard-2 instance — which
+// is why concurrent parses are also capped below.
+const supportedPages = 5000;
+const maxConcurrentParses = 2;
+let activeParses = 0;
 const parser = new LiteParse({
 	extractLinks: true,
 	imageMode: "placeholder",
+	maxPages: supportedPages + 1,
 	ocrEnabled: false,
 	outputFormat: "markdown",
 	quiet: true,
@@ -29,6 +39,7 @@ createServer(async (request, response) => {
 	let status = 500;
 	let errorType = null;
 	let errorMessage = null;
+	let holdsParseSlot = false;
 
 	try {
 		if (
@@ -53,9 +64,36 @@ createServer(async (request, response) => {
 			return sendJson(response, status, { error: "Not found." });
 		}
 
+		// Each parse can hold gigabytes resident, and one over-committed container
+		// dies taking every in-flight request with it. Shed load instead: 503 is
+		// retryable by the caller, an OOM crash is not.
+		if (activeParses >= maxConcurrentParses) {
+			status = 503;
+			return sendJson(response, status, {
+				code: "EXTRACTOR_BUSY",
+				error: "Extractor is at capacity, retry shortly.",
+			});
+		}
+
+		// Held until the whole request finishes, including streaming the result out:
+		// the parsed pages stay resident while the response drains, so releasing the
+		// slot any earlier would let admissions outrun actual memory use. A parse that
+		// outlives its timeout still runs to completion in the background holding
+		// memory — that zombie cannot be cancelled, only kept rare by the timeout.
+		activeParses += 1;
+		holdsParseSlot = true;
 		const bytes = await readPdfRequestBytes(request);
 		inputBytes = bytes.byteLength;
 		const result = await withTimeout(parser.parse(bytes), parseTimeoutMs);
+
+		if (result.pages.length > supportedPages) {
+			throw new PdfValidationError(
+				422,
+				"TOO_MANY_PAGES",
+				`PDFs longer than ${supportedPages} pages are not supported.`,
+			);
+		}
+
 		pageCount = result.pages.length;
 		status = 200;
 		response.writeHead(status, { "content-type": "application/x-ndjson; charset=utf-8" });
@@ -78,6 +116,9 @@ createServer(async (request, response) => {
 		}
 		return sendJson(response, status, { error: "PDF parsing failed." });
 	} finally {
+		if (holdsParseSlot) {
+			activeParses -= 1;
+		}
 		console.info(
 			JSON.stringify({
 				duration_ms: Date.now() - startedAt,

src/features/workspaces/content/workspace-read-observability.ts

@@ -0,0 +1,88 @@
+import type { WorkspaceContentReadResult } from "#/features/workspaces/content/workspace-content-contract";
+import { capturePostHogServerEvent } from "#/integrations/posthog/server";
+
+/**
+ * Records the readiness state every file read was served in, across every surface
+ * that reads — assistant tools and MCP both route through the same operation.
+ *
+ * This is the consumption side of the extraction pipeline's telemetry: the
+ * extraction event says how long the fast-pass window lasted, and this says whether
+ * anyone was actually in it — reads served `provisional`, reads that hit the
+ * pending spinner, and reads that found a stalled or failed document.
+ */
+export function recordWorkspaceFileReadOutcomes(input: {
+	operationId: string;
+	results: readonly WorkspaceContentReadResult[];
+	userId: string;
+	workspaceId: string;
+}) {
+	for (const result of input.results) {
+		if (result.status === "pending") {
+			capture(input, {
+				elapsed_seconds: result.elapsedSeconds,
+				empty_page_count: null,
+				failure_code: null,
+				item_id: null,
+				phase: result.phase,
+				provisional: null,
+				returned_page_count: null,
+				status: "pending",
+			});
+			continue;
+		}
+
+		if (result.status === "failed" && result.type === "file") {
+			capture(input, {
+				elapsed_seconds: null,
+				empty_page_count: null,
+				failure_code: result.code,
+				item_id: null,
+				phase: null,
+				provisional: null,
+				returned_page_count: null,
+				status: "failed",
+			});
+			continue;
+		}
+
+		if (result.status === "ready" && result.type === "file") {
+			capture(input, {
+				elapsed_seconds: null,
+				empty_page_count: result.emptyPages?.length ?? 0,
+				failure_code: null,
+				item_id: result.itemId,
+				phase: null,
+				provisional: result.provisional ?? false,
+				returned_page_count: result.location.returned.length,
+				status: "ready",
+			});
+		}
+	}
+}
+
+function capture(
+	input: { operationId: string; userId: string; workspaceId: string },
+	properties: {
+		elapsed_seconds: number | null;
+		empty_page_count: number | null;
+		failure_code: string | null;
+		item_id: string | null;
+		phase: "queued" | "extracting" | null;
+		provisional: boolean | null;
+		returned_page_count: number | null;
+		status: "ready" | "pending" | "failed";
+	},
+) {
+	capturePostHogServerEvent({
+		distinctId: input.userId,
+		event: "workspace_file_read_completed",
+		// Legitimate interest: operational readiness telemetry — ids and states only,
+		// no file names or content.
+		consentExempt: true,
+		properties: {
+			...properties,
+			operation_id: input.operationId,
+			workspace_id: input.workspaceId,
+		},
+	});
+}

src/features/workspaces/conversion/container-file-conversion.ts

@@ -57,7 +57,7 @@ export async function convertFileStreamWithContainer(input: {
 		sizeBytes: input.sizeBytes,
 	});
 
-	const [response] = await Promise.all([
+	const response = await multipart.awaitResponse(
 		input.container.fetch(
 			new Request(input.url, {
 				body: multipart.body,
@@ -66,8 +66,7 @@ export async function convertFileStreamWithContainer(input: {
 				method: "POST",
 			} as RequestInit & { duplex: "half" }),
 		),
-		multipart.done,
-	]);
+	);
 
 	if (!response.ok) {
 		throw input.error(await getConversionErrorMessage(response));

src/features/workspaces/extraction/liteparse-projection.ts

@@ -1,6 +1,10 @@
 import type { WorkflowStep } from "cloudflare:workers";
 
 import { extractPdfWithLiteParse } from "#/features/workspaces/extraction/providers/liteparse";
+import {
+	getWorkspaceExtractionStepConfig,
+	workspaceExtractionStepBudgets,
+} from "#/features/workspaces/extraction/workspace-extraction-budgets";
 import type {
 	LiteParseStageOutcome,
 	WorkspaceFileExtractionWorkflowParams,
@@ -28,10 +32,7 @@ export async function publishLiteParseProjection(
 	try {
 		return await step.do(
 			"publish fast LiteParse projection",
-			{
-				retries: { limit: 1, delay: "5 seconds", backoff: "constant" },
-				timeout: "2 minutes",
-			},
+			getWorkspaceExtractionStepConfig(workspaceExtractionStepBudgets.liteParse),
 			async () => {
 				const kernel = await getWorkspaceKernelFromEnv(env, params.workspaceId);
 				const { object, source } = await getWorkspaceFileSourceObject({
@@ -70,6 +71,9 @@ export async function publishLiteParseProjection(
 							markdownLength: projection.manifest.markdownLength,
 							pageCount: projection.manifest.pageCount,
 							provisional: true,
+							// Brand healing runs so the reconciler never picks this row up again:
+							// one upgrade attempt per document, bounded structurally.
+							...(params.healing ? { healed: true } : {}),
 						},
 						actorUserId: params.actorUserId,
 						clientMutationId: `${runId}:projection:liteparse-ready`,
@@ -104,6 +108,7 @@ export async function publishLiteParseProjection(
 		});
 		return {
 			durationMs: Date.now() - startedAt,
+			errorMessage: error instanceof Error ? error.message : String(error),
 			errorType: error instanceof Error ? error.name : "UnknownError",
 			outcome: "error",
 		};

src/features/workspaces/extraction/providers/liteparse.test.ts

@@ -33,6 +33,39 @@ describe("LiteParse response parsing", () => {
 		expect(() => parseLiteParsePage(payload)).toThrow("LiteParse returned an invalid");
 	});
 
+	// The workflow reads this error's name to decide whether to skip the paid tier, so
+	// mislabelling a refusal here means paying a provider to reach the same verdict on
+	// every reconciler sweep, forever.
+	it("reports a rejected document as unsupported rather than a retryable failure", async () => {
+		vi.mocked(requestWorkspaceFileProcessor).mockResolvedValue(
+			Response.json(
+				{ code: "TOO_MANY_PAGES", error: "PDFs longer than 4999 pages." },
+				{ status: 422 },
+			),
+		);
+		const pages = extractPdfWithLiteParse({} as Cloudflare.Env, {
+			body: new ReadableStream<Uint8Array>(),
+			fileName: "document.pdf",
+			sizeBytes: 1,
+		});
+
+		await expect(pages.next()).rejects.toMatchObject({
+			name: "WorkspaceDocumentUnsupportedError",
+			message: "PDFs longer than 4999 pages.",
+		});
+	});
+
+	it("treats any other processor failure as retryable", async () => {
+		vi.mocked(requestWorkspaceFileProcessor).mockResolvedValue(new Response("", { status: 500 }));
+		const pages = extractPdfWithLiteParse({} as Cloudflare.Env, {
+			body: new ReadableStream<Uint8Array>(),
+			fileName: "document.pdf",
+			sizeBytes: 1,
+		});
+
+		await expect(pages.next()).rejects.toMatchObject({ name: "Error" });
+	});
+
 	it("rejects an oversized NDJSON record and cancels the processor response", async () => {
 		const cancel = vi.fn();
 		const body = new ReadableStream<Uint8Array>({

src/features/workspaces/extraction/providers/liteparse.ts

@@ -1,5 +1,6 @@
 import type { MarkdownProjectionPage } from "#/features/workspaces/extraction/page-markdown-projection";
 import { parseLiteParsePage } from "#/features/workspaces/extraction/providers/liteparse-response";
+import { WorkspaceDocumentUnsupportedError } from "#/features/workspaces/extraction/types";
 import { requestWorkspaceFileProcessor } from "#/features/workspaces/files/workspace-file-processor";
 
 const maxNdjsonLineBytes = 8 * 1024 * 1024;
@@ -21,7 +22,12 @@ export async function* extractPdfWithLiteParse(
 	});
 
 	if (!response.ok) {
-		throw new Error(`LiteParse failed with status ${response.status}.`);
+		// The processor answers 422 only when it has read the file and found it
+		// unusable — too long, encrypted, or damaged. Every other status is an
+		// extraction that went wrong and may work next time.
+		throw response.status === 422
+			? new WorkspaceDocumentUnsupportedError(await getLiteParseErrorMessage(response))
+			: new Error(`LiteParse failed with status ${response.status}.`);
 	}
 
 	if (!response.body) {
@@ -39,6 +45,17 @@ export async function* extractPdfWithLiteParse(
 	}
 }
 
+async function getLiteParseErrorMessage(response: Response) {
+	const body: unknown = await response.json().catch(() => null);
+
+	return typeof body === "object" &&
+		body !== null &&
+		"error" in body &&
+		typeof body.error === "string"
+		? body.error
+		: "This document cannot be read.";
+}
+
 async function* readNdjsonLines(body: ReadableStream<Uint8Array>): AsyncGenerator<string> {
 	const reader = body.getReader();
 	const decoder = new TextDecoder();

src/features/workspaces/extraction/providers/llama-parse-credits.test.ts

@@ -0,0 +1,35 @@
+import { describe, expect, it } from "vitest";
+
+import { getLlamaParseDerivedCredits } from "#/features/workspaces/extraction/providers/llama-parse-credits";
+
+function createPages(total: number, optimized: number) {
+	return Array.from({ length: total }, (_, index) => ({ cost_optimized: index < optimized }));
+}
+
+describe("getLlamaParseDerivedCredits", () => {
+	it("bills every page at the requested tier when nothing was optimized", () => {
+		expect(getLlamaParseDerivedCredits({ pages: createPages(9, 0) }, "agentic")).toBe(90);
+	});
+
+	it("bills optimizer-downgraded pages at the cost_effective rate", () => {
+		// The production 1,527-page run: 717 pages downgraded, 810 left on agentic.
+		expect(getLlamaParseDerivedCredits({ pages: createPages(1527, 717) }, "agentic")).toBe(
+			717 * 3 + 810 * 10,
+		);
+	});
+
+	it("prices each tier from its own rate", () => {
+		expect(getLlamaParseDerivedCredits({ pages: createPages(2, 0) }, "cost_effective")).toBe(6);
+		expect(getLlamaParseDerivedCredits({ pages: createPages(2, 0) }, "agentic_plus")).toBe(90);
+	});
+
+	it("treats a missing cost_optimized flag as not downgraded", () => {
+		expect(getLlamaParseDerivedCredits({ pages: [{}, {}] }, "agentic")).toBe(20);
+	});
+
+	it("returns null when there is no per-page breakdown to price", () => {
+		expect(getLlamaParseDerivedCredits({}, "agentic")).toBeNull();
+		expect(getLlamaParseDerivedCredits({ pages: [] }, "agentic")).toBeNull();
+		expect(getLlamaParseDerivedCredits(null, "agentic")).toBeNull();
+	});
+});

src/features/workspaces/extraction/providers/llama-parse-credits.ts

@@ -0,0 +1,36 @@
+import type { LlamaParseTier } from "#/features/workspaces/extraction/types";
+import { getBooleanValue, getRecordArrayValue } from "#/integrations/llamaparse/client";
+
+// Published per-page rates, at $1.25 per 1,000 credits. The cost optimizer bills the
+// pages it downgrades at the cost_effective rate, so a job's real cost is a blend and
+// cannot be read off the requested tier alone.
+const llamaParseCreditsPerPage: Record<LlamaParseTier, number> = {
+	cost_effective: 3,
+	agentic: 10,
+	agentic_plus: 45,
+};
+
+/**
+ * Blended credit cost read off the per-page breakdown: pages the optimizer downgraded
+ * bill at the cost_effective rate, the rest at the requested tier.
+ *
+ * This is derived from published rates, not an invoice. LlamaParse v2 has never
+ * populated a billed figure on a parse response, and reporting null there understates
+ * spend on exactly the runs worth investigating — a wrong-by-a-few-percent number
+ * still catches a job that cost ten times what it should.
+ */
+export function getLlamaParseDerivedCredits(metadata: unknown, tier: LlamaParseTier) {
+	const pages = getRecordArrayValue(metadata, "pages");
+	if (pages.length === 0) {
+		return null;
+	}
+
+	return pages.reduce(
+		(total, page) =>
+			total +
+			(getBooleanValue(page, "cost_optimized")
+				? llamaParseCreditsPerPage.cost_effective
+				: llamaParseCreditsPerPage[tier]),
+		0,
+	);
+}

src/features/workspaces/extraction/providers/llama-parse.test.ts

@@ -0,0 +1,19 @@
+import { describe, expect, it } from "vitest";
+
+import { buildLlamaParseJobRequest } from "#/features/workspaces/extraction/providers/llama-parse";
+
+describe("buildLlamaParseJobRequest", () => {
+	it.each(["agentic", "agentic_plus"] as const)("enables the cost optimizer on %s", (tier) => {
+		expect(buildLlamaParseJobRequest({ fileId: "f", tier }).processing_options).toEqual({
+			cost_optimizer: { enable: true },
+		});
+	});
+
+	// LlamaParse rejects cost_optimizer + cost_effective with a 422 — sending it
+	// unconditionally once made every cost_effective parse fail outright.
+	it("omits the cost optimizer on cost_effective", () => {
+		expect(
+			buildLlamaParseJobRequest({ fileId: "f", tier: "cost_effective" }).processing_options,
+		).toEqual({});
+	});
+});