diff --git a/src/features/workspaces/content/workspace-content-contract.ts b/src/features/workspaces/content/workspace-content-contract.ts index 6be179a9..f97b663b 100644 --- a/src/features/workspaces/content/workspace-content-contract.ts +++ b/src/features/workspaces/content/workspace-content-contract.ts @@ -8,6 +8,8 @@ const workspacePathSchema = z.string().min(1); const readWorkspaceItemsFailureCodes = [ "content_changed", + "extraction_failed", + "extraction_stalled", "invalid_cursor", "invalid_selection", "page_range_out_of_range", @@ -93,22 +95,48 @@ const workspaceContentReadResultSchema = z.union([ z.object({ assetKind: workspaceFileAssetKindSchema, content: z.string(), + emptyPages: z + .array(z.number().int().min(1)) + .optional() + .describe( + "Returned pages that extracted no text. Expect these to fill in later while provisional is true.", + ), format: z.literal("markdown"), itemId: z.string().min(1), location: workspaceReadPagesSchema.extend({ kind: z.literal("pages") }), nextCursor: z.string().optional(), path: workspacePathSchema, + provisional: z + .boolean() + .optional() + .describe( + "True when this content came from the fast extraction pass and a higher-quality pass is still running.", + ), relations: workspaceReadRelationsSchema.optional(), status: z.literal("ready"), type: z.literal("file"), }), z.object({ + elapsedSeconds: z + .number() + .int() + .nonnegative() + .describe("How long extraction has been running."), path: workspacePathSchema, + phase: z + .enum(["queued", "extracting"]) + .describe("Whether extraction has started yet for this file."), + retryAfterSeconds: z + .number() + .int() + .positive() + .describe("Suggested wait before reading this path again."), status: z.literal("pending"), type: z.literal("file"), }), z.object({ code: z.enum(readWorkspaceItemsFailureCodes), + message: z.string().optional().describe("Why extraction failed, when known."), path: workspacePathSchema, status: z.literal("failed"), type: z.literal("file").optional(), diff --git a/src/features/workspaces/content/workspace-content-reader.ts b/src/features/workspaces/content/workspace-content-reader.ts index deb51d61..433c571d 100644 --- a/src/features/workspaces/content/workspace-content-reader.ts +++ b/src/features/workspaces/content/workspace-content-reader.ts @@ -8,6 +8,10 @@ import type { DocumentMarkdownChunkReadResult, } from "#/features/workspaces/documents/document-markdown-chunk"; import { readWorkspacePageProjection } from "#/features/workspaces/extraction/workspace-page-projection"; +import { + resolveWorkspaceProjectionReadiness, + type WorkspaceProjectionReadiness, +} from "#/features/workspaces/extraction/workspace-projection-readiness"; import type { WorkspaceKernelClient } from "#/features/workspaces/kernel/workspace-kernel-access"; import { resolveWorkspaceFileTypeFromItem } from "#/features/workspaces/model/workspace-file"; import { serializeWorkspaceRelations } from "#/features/workspaces/operations/relations"; @@ -190,24 +194,12 @@ async function readFile(input: { return { code: "unsupported_item_type", path: input.path, status: "failed" }; } - const projection = await input.kernel.readFileProjection({ - itemId: input.item.id, - format: "pages", - }); - if ( - !projection || - projection.status === "not_started" || - projection.status === "queued" || - projection.status === "processing" - ) { - return { path: input.path, status: "pending", type: "file" }; - } - if ( - projection.status !== "ready" || - projection.objectKey === null || - projection.sourceHash === null - ) { - return { code: "projection_failed", path: input.path, status: "failed", type: "file" }; + const projection = resolveWorkspaceProjectionReadiness( + await input.kernel.readFileProjection({ itemId: input.item.id, format: "pages" }), + Date.now(), + ); + if (projection.state !== "ready") { + return describeUnreadableProjection(projection, input.path); } const encodedCursor = input.request.mode === "continue" ? input.request.cursor : undefined; @@ -223,7 +215,7 @@ async function readFile(input: { pageRead = await readWorkspacePageProjection({ bucket: input.bucket, expectedSourceHash: projection.sourceHash, - manifestObjectKey: projection.objectKey, + manifestObjectKey: projection.manifestObjectKey, pages: cursor?.kind === "file" ? String(cursor.nextPage) @@ -241,6 +233,7 @@ async function readFile(input: { return { assetKind: fileType.assetKind, content: pageRead.content, + ...(pageRead.emptyPages.length > 0 ? { emptyPages: pageRead.emptyPages } : {}), format: "markdown", itemId: input.item.id, location: { kind: "pages", ...pageRead.pages }, @@ -256,11 +249,51 @@ async function readFile(input: { }), }), path: input.path, + ...(projection.provisional ? { provisional: true } : {}), status: "ready", type: "file", }; } +/** + * Maps a non-ready projection onto the read result the model sees. + * + * @param projection - Readiness for a projection that is not serving content. + * @param path - Absolute workspace path that was read. + * @returns The pending or failed read result for that path. + */ +function describeUnreadableProjection( + projection: Exclude, + path: string, +): WorkspaceContentReadResult { + if (projection.state === "pending") { + return { + elapsedSeconds: projection.elapsedSeconds, + path, + phase: projection.phase, + retryAfterSeconds: projection.retryAfterSeconds, + status: "pending", + type: "file", + }; + } + + if (projection.state === "stalled") { + return { code: "extraction_stalled", path, status: "failed", type: "file" }; + } + + if (projection.state === "failed") { + return { + code: "extraction_failed", + ...(projection.message ? { message: projection.message } : {}), + path, + status: "failed", + type: "file", + }; + } + + return { code: "projection_failed", path, status: "failed", type: "file" }; +} + async function attachRelationPaths( kernel: WorkspaceKernelClient, readyResults: PendingReadyResult[], diff --git a/src/features/workspaces/content/workspace-read-references.test.ts b/src/features/workspaces/content/workspace-read-references.test.ts index 7fbf028d..185b1fa2 100644 --- a/src/features/workspaces/content/workspace-read-references.test.ts +++ b/src/features/workspaces/content/workspace-read-references.test.ts @@ -85,6 +85,89 @@ describe("workspace read references", () => { }); expect(modelOutput.results[0]).not.toHaveProperty("pageReferences"); }); + + it("stays silent when every read succeeded outright", () => { + const results = [documentResult(), fileResult()] satisfies WorkspaceContentReadResult[]; + + expect( + createWorkspaceReadItemsModelOutput({ + references: createWorkspaceReadReferences(results), + results, + }), + ).not.toHaveProperty("guidance"); + }); + + it("explains a pending read once however many paths are waiting", () => { + const results = [ + { + elapsedSeconds: 4, + path: "/A.pdf", + phase: "extracting", + retryAfterSeconds: 15, + status: "pending", + type: "file", + }, + { + elapsedSeconds: 0, + path: "/B.pdf", + phase: "queued", + retryAfterSeconds: 15, + status: "pending", + type: "file", + }, + ] satisfies WorkspaceContentReadResult[]; + const guidance = createWorkspaceReadItemsModelOutput({ references: [], results }).guidance; + + expect(guidance).toHaveLength(1); + expect(guidance?.[0]).toContain("Never sleep"); + }); + + it("separates failures that will never resolve from transient ones", () => { + const results = [ + { code: "extraction_failed", path: "/A.pdf", status: "failed", type: "file" }, + { code: "extraction_stalled", path: "/B.pdf", status: "failed", type: "file" }, + { code: "projection_failed", path: "/C.pdf", status: "failed", type: "file" }, + ] satisfies WorkspaceContentReadResult[]; + const guidance = createWorkspaceReadItemsModelOutput({ references: [], results }).guidance; + + expect(guidance).toHaveLength(2); + expect(guidance?.[0]).toContain("do not suggest re-uploading"); + expect(guidance?.[1]).toContain("One repeat read is reasonable"); + }); + + it("says nothing about failures that are the caller's own mistake", () => { + const results = [ + { code: "path_not_found", path: "/Missing.pdf", status: "failed" }, + ] satisfies WorkspaceContentReadResult[]; + + expect(createWorkspaceReadItemsModelOutput({ references: [], results })).not.toHaveProperty( + "guidance", + ); + }); + + it("warns that blank pages from the fast pass are not final", () => { + const results = [ + { ...fileResult(), emptyPages: [13], provisional: true }, + ] satisfies WorkspaceContentReadResult[]; + const guidance = createWorkspaceReadItemsModelOutput({ + references: createWorkspaceReadReferences(results), + results, + }).guidance; + + expect(guidance).toHaveLength(1); + expect(guidance?.[0]).toContain("still extracting"); + }); + + it("stays silent about blank pages on a final, non-provisional read", () => { + const results = [{ ...fileResult(), emptyPages: [13] }] satisfies WorkspaceContentReadResult[]; + + expect( + createWorkspaceReadItemsModelOutput({ + references: createWorkspaceReadReferences(results), + results, + }), + ).not.toHaveProperty("guidance"); + }); }); function documentResult(): Extract< diff --git a/src/features/workspaces/content/workspace-read-references.ts b/src/features/workspaces/content/workspace-read-references.ts index 72e54395..3478ec4b 100644 --- a/src/features/workspaces/content/workspace-read-references.ts +++ b/src/features/workspaces/content/workspace-read-references.ts @@ -51,12 +51,70 @@ export function createWorkspaceReadReferences( return createWorkspaceReferenceRecords(locations); } +/** + * Guidance for read outcomes the model has to react to, keyed by situation. + * + * These live here rather than in the tool description because none of them + * change how a read is issued, and background extraction states are rare enough + * that every request should not carry the instructions for handling them. + */ +const workspaceReadGuidance = { + pending: + "Some paths are still extracting. Never sleep, poll, or otherwise stall waiting for them, including inside compute, sandbox_bash, or orchestrate. Either do other work and read those paths again later in this reply, or tell the user they are still processing and to ask again in about retryAfterSeconds. Never read the same pending path more than twice in one reply.", + unrecoverable: + "Extraction will not finish for some paths. Report the code and any message to the user; do not retry those reads and do not suggest re-uploading the file.", + transient: + "Some paths failed on a transient storage problem. One repeat read is reasonable; if it fails again, tell the user.", + provisional: + "Some content came from a fast first pass. Pages listed in emptyPages are still extracting, so read them again later rather than reporting them as blank.", +} as const; + +/** + * Collects the handling guidance a set of read results calls for. + * + * Emitted once per situation rather than once per result so a batch of pending + * reads does not repeat the same paragraph for every path. + * + * @param results - Ordered workspace read results. + * @returns Guidance lines for the situations present, most actionable first. + */ +function createWorkspaceReadGuidance(results: readonly WorkspaceContentReadResult[]): string[] { + const situations = new Set(); + + for (const result of results) { + if (result.status === "pending") { + situations.add("pending"); + continue; + } + + if (result.status === "failed") { + if (result.code === "extraction_failed" || result.code === "extraction_stalled") { + situations.add("unrecoverable"); + } else if (result.code === "projection_failed") { + situations.add("transient"); + } + continue; + } + + // Gate on provisional only: emptyPages on a final (non-provisional) read + // describes genuinely blank pages, which must not be reported as still + // extracting or the model retries a completed read forever. + if (result.type === "file" && result.provisional) { + situations.add("provisional"); + } + } + + return (Object.keys(workspaceReadGuidance) as Array) + .filter((situation) => situations.has(situation)) + .map((situation) => workspaceReadGuidance[situation]); +} + /** * Projects a rich workspace read result into compact model-visible JSON. * * Raw item IDs and durable locations remain in the persisted tool result. * Model-visible content receives only opaque refs next to the content they - * identify. + * identify, plus guidance for any outcome that needs handling. * * @param output - Validated rich workspace read output. * @returns JSON-safe results annotated with short workspace refs. @@ -65,8 +123,10 @@ export function createWorkspaceReadItemsModelOutput(output: WorkspaceReadItemsOu const refsByLocation = new Map( output.references.map((record) => [getWorkspaceLocationKey(record.location), record.ref]), ); + const guidance = createWorkspaceReadGuidance(output.results); return { + ...(guidance.length > 0 ? { guidance } : {}), results: output.results.map((result) => { if (result.status !== "ready") { return result; diff --git a/src/features/workspaces/extraction/providers/firecrawl.ts b/src/features/workspaces/extraction/providers/firecrawl.ts deleted file mode 100644 index e58d4de2..00000000 --- a/src/features/workspaces/extraction/providers/firecrawl.ts +++ /dev/null @@ -1,137 +0,0 @@ -import type { - FirecrawlPdfMode, - MarkdownExtractionInput, - MarkdownExtractionProvider, - MarkdownExtractionResult, -} from "#/features/workspaces/extraction/types"; -import { createSingleMarkdownProjectionPage } from "#/features/workspaces/extraction/page-markdown-projection"; -import { - firecrawlJsonRequest, - getFirstArrayRecord, - getNumberValue, - getRecordValue, - getStringValue, -} from "#/integrations/firecrawl/client"; -import { createStreamingMultipartFile } from "#/lib/http/streaming-multipart"; - -const firecrawlParseTimeoutMs = 300_000; - -export function createFirecrawlPdfExtractionProvider( - env: Cloudflare.Env, -): MarkdownExtractionProvider { - return { - id: "firecrawl", - async extract(input) { - const mode = normalizeFirecrawlMode(input.mode); - const multipart = createStreamingMultipartFile({ - body: input.body, - contentType: input.contentType || "application/pdf", - fields: { - options: JSON.stringify({ - formats: ["markdown"], - parsers: [{ type: "pdf", mode }], - timeout: firecrawlParseTimeoutMs, - }), - }, - fileName: input.fileName, - formFieldName: "file", - sizeBytes: input.sizeBytes, - }); - - const [responseJson] = await Promise.all([ - firecrawlJsonRequest({ - env, - path: "/v2/parse", - operation: "Firecrawl PDF parsing", - method: "POST", - headers: { "content-type": multipart.contentType }, - body: multipart.body, - }), - multipart.done, - ]); - - const markdown = getFirecrawlMarkdown(responseJson); - - if (!markdown) { - throw new Error("Firecrawl PDF parsing completed without markdown output."); - } - - return { - pages: createSingleMarkdownProjectionPage(markdown), - provider: "firecrawl", - providerMode: mode, - metadata: getFirecrawlMetadata(responseJson), - } satisfies MarkdownExtractionResult; - }, - }; -} - -function normalizeFirecrawlMode(mode: MarkdownExtractionInput["mode"]): FirecrawlPdfMode { - if (mode === "fast" || mode === "ocr") { - return mode; - } - - return "auto"; -} - -function getFirecrawlMarkdown(value: unknown): string | null { - const candidates = [ - value, - getRecordValue(value, "data"), - getRecordValue(value, "document"), - getRecordValue(value, "result"), - getFirstArrayRecord(getRecordValue(value, "data")), - getFirstArrayRecord(getRecordValue(value, "documents")), - getFirstArrayRecord(getRecordValue(value, "results")), - ]; - - for (const candidate of candidates) { - const markdown = getRecordValue(candidate, "markdown"); - - if (typeof markdown === "string" && markdown.trim().length > 0) { - return markdown; - } - } - - return null; -} - -function getFirecrawlMetadata(value: unknown) { - const data = getRecordValue(value, "data"); - const usage = getRecordValue(value, "usage") ?? getRecordValue(data, "usage"); - const metadata = getRecordValue(data, "metadata") ?? getRecordValue(value, "metadata"); - const creditsUsed = - getNumberValue(metadata, "creditsUsed") ?? - getNumberValue(usage, "credits") ?? - getNumberValue(data, "creditsUsed") ?? - getNumberValue(value, "creditsUsed"); - const title = getStringValue(metadata, "title") ?? getStringValue(data, "title"); - const sourceFile = getStringValue(metadata, "sourceFile") ?? getStringValue(data, "sourceFile"); - const pageCount = - getNumberValue(metadata, "numPages") ?? - getNumberValue(metadata, "pageCount") ?? - getNumberValue(data, "numPages") ?? - getNumberValue(data, "pageCount") ?? - getNumberValue(value, "numPages") ?? - getNumberValue(value, "pageCount"); - const result: Record = {}; - - if (creditsUsed !== null) { - result.creditsUsed = creditsUsed; - } - - if (pageCount !== null) { - result.numPages = pageCount; - result.pageCount = pageCount; - } - - if (title !== null) { - result.title = title; - } - - if (sourceFile !== null) { - result.sourceFile = sourceFile; - } - - return result; -} diff --git a/src/features/workspaces/extraction/providers/index.ts b/src/features/workspaces/extraction/providers/index.ts index 452f0c29..d8545f26 100644 --- a/src/features/workspaces/extraction/providers/index.ts +++ b/src/features/workspaces/extraction/providers/index.ts @@ -1,24 +1,16 @@ -import { createFirecrawlPdfExtractionProvider } from "#/features/workspaces/extraction/providers/firecrawl"; import { createLlamaParseExtractionProvider } from "#/features/workspaces/extraction/providers/llama-parse"; -import { createStubMarkdownExtractionProvider } from "#/features/workspaces/extraction/providers/stubs"; import { createWorkersAiToMarkdownProvider } from "#/features/workspaces/extraction/providers/workers-ai-to-markdown"; -import type { - MarkdownExtractionProvider, - MarkdownExtractionProviderId, -} from "#/features/workspaces/extraction/types"; +import type { MarkdownExtractionProvider } from "#/features/workspaces/extraction/types"; +import type { WorkspaceFileExtractionProviderId } from "#/features/workspaces/model/workspace-file/types"; export function createMarkdownExtractionProvider( - providerId: MarkdownExtractionProviderId, + providerId: WorkspaceFileExtractionProviderId, env: Env, ): MarkdownExtractionProvider { switch (providerId) { - case "firecrawl": - return createFirecrawlPdfExtractionProvider(env); case "workers_ai_to_markdown": return createWorkersAiToMarkdownProvider(env); case "llama_parse": return createLlamaParseExtractionProvider(env); - case "mistral_ocr": - return createStubMarkdownExtractionProvider(providerId); } } diff --git a/src/features/workspaces/extraction/providers/stubs.ts b/src/features/workspaces/extraction/providers/stubs.ts deleted file mode 100644 index ca52b556..00000000 --- a/src/features/workspaces/extraction/providers/stubs.ts +++ /dev/null @@ -1,28 +0,0 @@ -import type { - MarkdownExtractionProvider, - MarkdownExtractionProviderId, -} from "#/features/workspaces/extraction/types"; - -const implementedMarkdownExtractionProviders = [ - "firecrawl", - "llama_parse", - "workers_ai_to_markdown", -] as const satisfies readonly MarkdownExtractionProviderId[]; - -type StubMarkdownExtractionProviderId = Exclude< - MarkdownExtractionProviderId, - (typeof implementedMarkdownExtractionProviders)[number] ->; - -export function createStubMarkdownExtractionProvider( - id: StubMarkdownExtractionProviderId, -): MarkdownExtractionProvider { - return { - id, - async extract() { - throw new Error( - `${id} markdown extraction is intentionally stubbed. Add credentials, pricing limits, and routing rules before routing uploads here.`, - ); - }, - }; -} diff --git a/src/features/workspaces/extraction/types.ts b/src/features/workspaces/extraction/types.ts index 5dbc9b86..fcb52fb4 100644 --- a/src/features/workspaces/extraction/types.ts +++ b/src/features/workspaces/extraction/types.ts @@ -3,16 +3,8 @@ import type { WorkspaceFileExtractionMode, WorkspaceFileExtractionProviderId, } from "#/features/workspaces/model/workspace-file/types"; -import { workspaceFileExtractionProviders } from "#/features/workspaces/model/workspace-file/types"; import type { MarkdownProjectionPage } from "#/features/workspaces/extraction/page-markdown-projection"; -export type MarkdownExtractionProviderId = WorkspaceFileExtractionProviderId; - -export type MarkdownExtractionProviderMode = WorkspaceFileExtractionMode; - -export { workspaceFileExtractionProviders as markdownExtractionProviders }; - -export type FirecrawlPdfMode = "fast" | "auto" | "ocr"; export type LlamaParseTier = "cost_effective" | "agentic" | "agentic_plus"; export interface WorkspaceFileExtractionWorkflowParams { @@ -41,17 +33,17 @@ export interface MarkdownExtractionInput { contentType: string; sizeBytes: number; sourceHash: string; - mode: MarkdownExtractionProviderMode; + mode: WorkspaceFileExtractionMode; } export interface MarkdownExtractionResult { pages: MarkdownProjectionPage[]; - provider: MarkdownExtractionProviderId; - providerMode: MarkdownExtractionProviderMode; + provider: WorkspaceFileExtractionProviderId; + providerMode: WorkspaceFileExtractionMode; metadata: Record; } export interface MarkdownExtractionProvider { - id: MarkdownExtractionProviderId; + id: WorkspaceFileExtractionProviderId; extract(input: MarkdownExtractionInput): Promise; } diff --git a/src/features/workspaces/extraction/workspace-file-extraction-observability.ts b/src/features/workspaces/extraction/workspace-file-extraction-observability.ts index 1d3d1364..036ce082 100644 --- a/src/features/workspaces/extraction/workspace-file-extraction-observability.ts +++ b/src/features/workspaces/extraction/workspace-file-extraction-observability.ts @@ -1,9 +1,11 @@ import type { LiteParseStageOutcome, - MarkdownExtractionProviderId, - MarkdownExtractionProviderMode, WorkspaceFileExtractionWorkflowParams, } from "#/features/workspaces/extraction/types"; +import type { + WorkspaceFileExtractionMode, + WorkspaceFileExtractionProviderId, +} from "#/features/workspaces/model/workspace-file/types"; import { logOperationalEvent, recordOperationalFailure, @@ -32,8 +34,8 @@ type WorkspaceFileExtractionOutcome = WorkspaceFileExtractionOutcomeBase & | { outcome: "partial" | "success"; pageCount: number; - provider: MarkdownExtractionProviderId | "liteparse"; - providerMode: MarkdownExtractionProviderMode; + provider: WorkspaceFileExtractionProviderId | "liteparse"; + providerMode: WorkspaceFileExtractionMode; routeReason: string; } ); diff --git a/src/features/workspaces/extraction/workspace-file-extraction-workflow.ts b/src/features/workspaces/extraction/workspace-file-extraction-workflow.ts index 44622600..e9abb120 100644 --- a/src/features/workspaces/extraction/workspace-file-extraction-workflow.ts +++ b/src/features/workspaces/extraction/workspace-file-extraction-workflow.ts @@ -3,11 +3,11 @@ import { WorkflowEntrypoint, type WorkflowEvent, type WorkflowStep } from "cloud import { publishLiteParseProjection } from "#/features/workspaces/extraction/liteparse-projection"; import { recordWorkspaceFileExtractionOutcome } from "#/features/workspaces/extraction/workspace-file-extraction-observability"; import { createMarkdownExtractionProvider } from "#/features/workspaces/extraction/providers/index"; +import type { WorkspaceFileExtractionWorkflowParams } from "#/features/workspaces/extraction/types"; import type { - MarkdownExtractionProviderId, - MarkdownExtractionProviderMode, - WorkspaceFileExtractionWorkflowParams, -} from "#/features/workspaces/extraction/types"; + WorkspaceFileExtractionMode, + WorkspaceFileExtractionProviderId, +} from "#/features/workspaces/model/workspace-file/types"; import { getWorkspaceFileSourceObject } from "#/features/workspaces/extraction/workspace-file-source"; import { writeWorkspacePageProjection } from "#/features/workspaces/extraction/workspace-page-projection"; import { getWorkspaceKernelFromEnv } from "#/features/workspaces/kernel/workspace-kernel-access"; @@ -42,8 +42,8 @@ export class WorkspaceFileExtractionWorkflow extends WorkflowEntrypoint< let extraction: StagedPageExtractionResult; let result: { pageCount: number; - provider: MarkdownExtractionProviderId; - providerMode: MarkdownExtractionProviderMode; + provider: WorkspaceFileExtractionProviderId; + providerMode: WorkspaceFileExtractionMode; status: "ready"; }; @@ -241,8 +241,8 @@ export class WorkspaceFileExtractionWorkflow extends WorkflowEntrypoint< interface StagedPageExtractionResult { manifestObjectKey: string; markdownLength: number; - provider: MarkdownExtractionProviderId; - providerMode: MarkdownExtractionProviderMode; + provider: WorkspaceFileExtractionProviderId; + providerMode: WorkspaceFileExtractionMode; metadata: Record; pageCount: number; routeReason: string; diff --git a/src/features/workspaces/extraction/workspace-page-projection.test.ts b/src/features/workspaces/extraction/workspace-page-projection.test.ts index 61139a0f..f2c98797 100644 --- a/src/features/workspaces/extraction/workspace-page-projection.test.ts +++ b/src/features/workspaces/extraction/workspace-page-projection.test.ts @@ -35,6 +35,7 @@ describe("workspace page projections", () => { expect(result).toEqual({ content: "## Page 2\n\nSecond\n\n## Page 3\n\nThird", + emptyPages: [], pages: { requested: "2-3", returned: [2, 3], total: 3 }, }); const prefix = reference.manifestObjectKey.slice(0, -"manifest.json".length); @@ -71,6 +72,7 @@ describe("workspace page projections", () => { }), ).resolves.toEqual({ content: "## Page 2", + emptyPages: [2], pages: { requested: "2", returned: [2], total: 3 }, }); }); @@ -186,6 +188,7 @@ describe("workspace page projections", () => { }), ).resolves.toEqual({ content: "## Page 1\n\nPage 1", + emptyPages: [], pages: { requested: "1", returned: [1], total: 1 }, }); }); @@ -213,6 +216,7 @@ describe("workspace page projections", () => { }), ).resolves.toEqual({ content: `## Page 1\n\n${markdown}`, + emptyPages: [], pages: { requested: "1", returned: [1], total: 1 }, }); }); diff --git a/src/features/workspaces/extraction/workspace-page-projection.ts b/src/features/workspaces/extraction/workspace-page-projection.ts index 72136fb0..19a55124 100644 --- a/src/features/workspaces/extraction/workspace-page-projection.ts +++ b/src/features/workspaces/extraction/workspace-page-projection.ts @@ -148,7 +148,7 @@ export async function readWorkspacePageProjection(input: { expectedSourceHash: string; manifestObjectKey: string; pages?: string; -}): Promise<{ content: string; pages: WorkspaceReadPages }> { +}): Promise<{ content: string; emptyPages: number[]; pages: WorkspaceReadPages }> { const manifest = await readWorkspacePageProjectionManifest(input.bucket, input.manifestObjectKey); if (manifest.sourceHash !== input.expectedSourceHash) { throw new Error("Workspace page projection source does not match its published revision."); @@ -200,6 +200,9 @@ export async function readWorkspacePageProjection(input: { return { content: pages.map(formatProjectionPage).join("\n\n"), + // Surfaced so callers can distinguish a genuinely blank page from one the + // fast extraction pass could not read yet. + emptyPages: pages.filter((page) => page.markdown.length === 0).map((page) => page.pageNumber), pages: { requested, returned: selectedPageNumbers, diff --git a/src/features/workspaces/extraction/workspace-projection-readiness.test.ts b/src/features/workspaces/extraction/workspace-projection-readiness.test.ts new file mode 100644 index 00000000..3670dbe9 --- /dev/null +++ b/src/features/workspaces/extraction/workspace-projection-readiness.test.ts @@ -0,0 +1,127 @@ +import { describe, expect, it } from "vitest"; + +import { resolveWorkspaceProjectionReadiness } from "#/features/workspaces/extraction/workspace-projection-readiness"; +import type { ReadWorkspaceKernelFileProjectionResult } from "#/features/workspaces/kernel/workspace-kernel-types"; + +const now = Date.parse("2026-07-29T12:00:00.000Z"); + +function createProjection( + overrides: Partial, +): ReadWorkspaceKernelFileProjectionResult { + return { + itemId: "item-1", + format: "pages", + status: "ready", + objectKey: "workspaces/w1/items/item-1/extractions/run-1/fast/manifest.json", + provider: "liteparse", + providerMode: "fast", + errorMessage: null, + sourceHash: "hash-1", + metadataJson: {}, + updatedAt: new Date(now).toISOString(), + ...overrides, + }; +} + +describe("resolveWorkspaceProjectionReadiness", () => { + it("treats a missing projection row as queued", () => { + expect(resolveWorkspaceProjectionReadiness(null, now)).toEqual({ + state: "pending", + phase: "queued", + elapsedSeconds: 0, + retryAfterSeconds: 15, + }); + }); + + it("reports how long a processing projection has been running", () => { + const projection = createProjection({ + status: "processing", + objectKey: null, + sourceHash: null, + updatedAt: new Date(now - 40_000).toISOString(), + }); + + expect(resolveWorkspaceProjectionReadiness(projection, now)).toEqual({ + state: "pending", + phase: "extracting", + elapsedSeconds: 40, + retryAfterSeconds: 40, + }); + }); + + it("keeps the retry hint within its bounds as the wait grows", () => { + const brief = resolveWorkspaceProjectionReadiness( + createProjection({ status: "processing", updatedAt: new Date(now - 2_000).toISOString() }), + now, + ); + const long = resolveWorkspaceProjectionReadiness( + createProjection({ status: "processing", updatedAt: new Date(now - 600_000).toISOString() }), + now, + ); + + expect(brief).toMatchObject({ retryAfterSeconds: 15 }); + expect(long).toMatchObject({ retryAfterSeconds: 120 }); + }); + + it("stalls a processing projection that outlived the retrying extraction budget", () => { + const projection = createProjection({ + status: "processing", + updatedAt: new Date(now - 46 * 60_000).toISOString(), + }); + + expect(resolveWorkspaceProjectionReadiness(projection, now)).toEqual({ + state: "stalled", + elapsedSeconds: 46 * 60, + }); + }); + + it("keeps a slow but healthy extraction pending rather than stalling it", () => { + const projection = createProjection({ + status: "processing", + updatedAt: new Date(now - 31 * 60_000).toISOString(), + }); + + expect(resolveWorkspaceProjectionReadiness(projection, now)).toMatchObject({ + state: "pending", + phase: "extracting", + }); + }); + + it("surfaces the recorded reason for a failed projection", () => { + const projection = createProjection({ + status: "failed", + errorMessage: "LiteParse failed with status 500.", + objectKey: null, + sourceHash: null, + }); + + expect(resolveWorkspaceProjectionReadiness(projection, now)).toEqual({ + state: "failed", + message: "LiteParse failed with status 500.", + }); + }); + + it("marks a ready projection missing its manifest as unreadable", () => { + expect( + resolveWorkspaceProjectionReadiness(createProjection({ sourceHash: null }), now), + ).toEqual({ state: "unreadable" }); + }); + + it("exposes the manifest for a ready projection", () => { + expect(resolveWorkspaceProjectionReadiness(createProjection({}), now)).toEqual({ + state: "ready", + manifestObjectKey: "workspaces/w1/items/item-1/extractions/run-1/fast/manifest.json", + sourceHash: "hash-1", + provisional: false, + }); + }); + + it("flags a fast-pass projection as provisional", () => { + const projection = createProjection({ metadataJson: { provisional: true } }); + + expect(resolveWorkspaceProjectionReadiness(projection, now)).toMatchObject({ + state: "ready", + provisional: true, + }); + }); +}); diff --git a/src/features/workspaces/extraction/workspace-projection-readiness.ts b/src/features/workspaces/extraction/workspace-projection-readiness.ts new file mode 100644 index 00000000..0a50de42 --- /dev/null +++ b/src/features/workspaces/extraction/workspace-projection-readiness.ts @@ -0,0 +1,92 @@ +import type { ReadWorkspaceKernelFileProjectionResult } from "#/features/workspaces/kernel/workspace-kernel-types"; + +/** + * How long a projection may sit in `processing` before it is treated as stalled. + * + * The enhanced extraction step allows a 10 minute timeout across 3 attempts with + * exponential backoff, and the projection row is not touched between attempts, so + * a healthy run can legitimately stay `processing` for a little over half an hour. + * The threshold sits above that worst case so slow retries are never mislabelled. + */ +const extractionStallThresholdMs = 45 * 60_000; + +const minimumRetryAfterSeconds = 15; +const maximumRetryAfterSeconds = 120; + +export type WorkspaceProjectionReadiness = + | { + state: "pending"; + phase: "queued" | "extracting"; + elapsedSeconds: number; + retryAfterSeconds: number; + } + | { state: "stalled"; elapsedSeconds: number } + | { state: "failed"; message: string | null } + | { state: "unreadable" } + | { + state: "ready"; + manifestObjectKey: string; + sourceHash: string; + provisional: boolean; + }; + +/** + * Classifies a page-projection row into the states a content read cares about. + * + * Extraction publishes a fast projection before the higher-quality pass finishes, + * so a `ready` projection may still be provisional, and a `processing` one may be + * either genuinely in flight or abandoned by a dead workflow. Keeping the + * classification pure means each of those cases is decided in one place rather + * than as extra branches inside the read path. + * + * @param projection - Projection row for the `pages` format, or null when extraction has not recorded one yet. + * @param now - Current time in epoch milliseconds, used to age `processing` rows. + * @returns How the projection should be treated by a content read. + */ +export function resolveWorkspaceProjectionReadiness( + projection: ReadWorkspaceKernelFileProjectionResult | null, + now: number, +): WorkspaceProjectionReadiness { + if (!projection) { + return { + state: "pending", + phase: "queued", + elapsedSeconds: 0, + retryAfterSeconds: minimumRetryAfterSeconds, + }; + } + + if (projection.status === "processing") { + const elapsedMs = Math.max(0, now - Date.parse(projection.updatedAt)); + if (elapsedMs > extractionStallThresholdMs) { + return { state: "stalled", elapsedSeconds: Math.round(elapsedMs / 1000) }; + } + + const elapsedSeconds = Math.round(elapsedMs / 1000); + return { + state: "pending", + phase: "extracting", + elapsedSeconds, + // Back off as the wait grows so a slow image extraction is not re-read every 15s. + retryAfterSeconds: Math.min( + maximumRetryAfterSeconds, + Math.max(minimumRetryAfterSeconds, elapsedSeconds), + ), + }; + } + + if (projection.status === "failed") { + return { state: "failed", message: projection.errorMessage }; + } + + if (!projection.objectKey || !projection.sourceHash) { + return { state: "unreadable" }; + } + + return { + state: "ready", + manifestObjectKey: projection.objectKey, + sourceHash: projection.sourceHash, + provisional: projection.metadataJson.provisional === true, + }; +} diff --git a/src/features/workspaces/kernel/workspace-kernel-types.ts b/src/features/workspaces/kernel/workspace-kernel-types.ts index 54425c54..d553e47b 100644 --- a/src/features/workspaces/kernel/workspace-kernel-types.ts +++ b/src/features/workspaces/kernel/workspace-kernel-types.ts @@ -180,12 +180,7 @@ export interface WorkspaceKernelFileSource { export type WorkspaceKernelFileProjectionFormat = "pages" | "preview"; -export type WorkspaceKernelFileProjectionStatus = - | "not_started" - | "queued" - | "processing" - | "ready" - | "failed"; +export type WorkspaceKernelFileProjectionStatus = "processing" | "ready" | "failed"; interface WorkspaceKernelFileProjectionMutationBase { itemId: string; diff --git a/src/features/workspaces/model/workspace-file/types.ts b/src/features/workspaces/model/workspace-file/types.ts index f3bab7de..87d3f004 100644 --- a/src/features/workspaces/model/workspace-file/types.ts +++ b/src/features/workspaces/model/workspace-file/types.ts @@ -5,21 +5,13 @@ export const workspaceFileAssetKindSchema = z.enum(workspaceFileAssetKinds); export type WorkspaceFileAssetKind = (typeof workspaceFileAssetKinds)[number]; -export const workspaceFileExtractionProviders = [ - "firecrawl", - "workers_ai_to_markdown", - "mistral_ocr", - "llama_parse", -] as const; +export const workspaceFileExtractionProviders = ["workers_ai_to_markdown", "llama_parse"] as const; export type WorkspaceFileExtractionProviderId = (typeof workspaceFileExtractionProviders)[number]; export type WorkspaceFileExtractionMode = | "fast" - | "auto" - | "ocr" | "default" - | "stub" | "cost_effective" | "agentic" | "agentic_plus";