fix(knowledge-fs): harden long document ingestion

knowledge-fs/apps/api/package.json

@@ -22,7 +22,8 @@
     "@knowledge/generation": "workspace:*",
     "@knowledge/parsers": "workspace:*",
     "hono": "^4.12.25",
-    "sharp": "^0.35.3"
+    "sharp": "^0.35.3",
+    "undici": "^8.10.0"
   },
   "devDependencies": {
     "@types/node": "^22.10.2",

knowledge-fs/apps/api/src/parser-options.test.ts

@@ -1,10 +1,51 @@
 import { describe, expect, it } from "vitest";
 
-import { createApiDocumentParser } from "./parser-options";
+import type { Dispatcher } from "undici";
+
+import { createApiDocumentParser, createNodeUnstructuredFetch } from "./parser-options";
 
 const encoder = new TextEncoder();
 
 describe("createApiDocumentParser", () => {
+  it("translates native requests onto a matching Node transport with aligned timeouts", async () => {
+    let dispatcherOptions: Readonly<{ bodyTimeout: number; headersTimeout: number }> | undefined;
+    let requestDispatcher: Dispatcher | undefined;
+    let receivedBody: BodyInit | null | undefined;
+    let receivedInput: RequestInfo | URL | undefined;
+    let receivedMethod: string | undefined;
+    const dispatcher = {} as Dispatcher;
+    const nodeFetch = createNodeUnstructuredFetch({
+      createDispatcher: (options) => {
+        dispatcherOptions = options;
+        return dispatcher;
+      },
+      fetch: async (input, init) => {
+        receivedBody = init?.body;
+        receivedInput = input;
+        receivedMethod = init?.method;
+        requestDispatcher = (init as (RequestInit & { dispatcher?: Dispatcher }) | undefined)
+          ?.dispatcher;
+        return new Response("[]");
+      },
+      requestTimeoutMs: 600_000,
+    });
+    const request = new Request("https://unstructured.example.test/general/v0/general", {
+      body: "document",
+      method: "POST",
+    });
+
+    await nodeFetch(request);
+
+    expect(dispatcherOptions).toEqual({
+      bodyTimeout: 600_000,
+      headersTimeout: 600_000,
+    });
+    expect(receivedInput).toBe(request.url);
+    expect(receivedMethod).toBe("POST");
+    expect(receivedBody).toBe(request.body);
+    expect(requestDispatcher).toBe(dispatcher);
+  });
+
   it("keeps Markdown and structured data on native parsers", async () => {
     let fetchCalls = 0;
     const parser = createApiDocumentParser({
@@ -162,8 +203,11 @@ describe("createApiDocumentParser", () => {
     ).toThrow("UNSTRUCTURED_MAX_CONCURRENCY must be between 1 and 32");
     expect(() =>
       createApiDocumentParser({
-        env: { UNSTRUCTURED_API_URL: "http://parser", UNSTRUCTURED_REQUEST_TIMEOUT_MS: "600001" },
+        env: {
+          UNSTRUCTURED_API_URL: "http://parser",
+          UNSTRUCTURED_REQUEST_TIMEOUT_MS: "1800001",
+        },
       }),
-    ).toThrow("UNSTRUCTURED_REQUEST_TIMEOUT_MS must be between 1 and 600000");
+    ).toThrow("UNSTRUCTURED_REQUEST_TIMEOUT_MS must be between 1 and 1800000");
   });
 });

knowledge-fs/apps/api/src/parser-options.ts

@@ -6,6 +6,23 @@ import {
   createParserRouter,
   createUnstructuredParserClient,
 } from "@knowledge/parsers";
+import { Agent, type Dispatcher, fetch as undiciFetch } from "undici";
+
+const defaultUnstructuredRequestTimeoutMs = 120_000;
+const maxUnstructuredRequestTimeoutMs = 1_800_000;
+
+interface NodeUnstructuredFetchOptions {
+  readonly createDispatcher?: (
+    options: Readonly<{ bodyTimeout: number; headersTimeout: number }>,
+  ) => Dispatcher;
+  readonly fetch?: typeof fetch;
+  readonly requestTimeoutMs: number;
+}
+
+type DispatcherRequestInit = RequestInit & {
+  readonly dispatcher: Dispatcher;
+  readonly duplex?: "half";
+};
 
 export interface ApiParserEnv {
   readonly NODE_ENV?: string | undefined;
@@ -60,13 +77,26 @@ function createApiUnstructuredParser({
     };
   }
 
+  const requestTimeoutMs =
+    env.UNSTRUCTURED_REQUEST_TIMEOUT_MS === undefined
+      ? defaultUnstructuredRequestTimeoutMs
+      : parseBoundedPositiveInteger(
+          env.UNSTRUCTURED_REQUEST_TIMEOUT_MS,
+          "UNSTRUCTURED_REQUEST_TIMEOUT_MS",
+          maxUnstructuredRequestTimeoutMs,
+        );
+
   return createUnstructuredParserClient({
     ...(env.UNSTRUCTURED_DEFAULT_LANGUAGE?.trim()
       ? { defaultLanguage: env.UNSTRUCTURED_DEFAULT_LANGUAGE.trim() }
       : {}),
     endpoint,
     ...(env.UNSTRUCTURED_API_KEY?.trim() ? { apiKey: env.UNSTRUCTURED_API_KEY.trim() } : {}),
-    ...(fetchImpl ? { fetch: fetchImpl } : {}),
+    fetch:
+      fetchImpl ??
+      createNodeUnstructuredFetch({
+        requestTimeoutMs,
+      }),
     ...(env.UNSTRUCTURED_MAX_CONCURRENCY !== undefined
       ? {
           maxConcurrency: parseBoundedPositiveInteger(
@@ -100,18 +130,42 @@ function createApiUnstructuredParser({
           ),
         }
       : {}),
-    ...(env.UNSTRUCTURED_REQUEST_TIMEOUT_MS !== undefined
-      ? {
-          requestTimeoutMs: parseBoundedPositiveInteger(
-            env.UNSTRUCTURED_REQUEST_TIMEOUT_MS,
-            "UNSTRUCTURED_REQUEST_TIMEOUT_MS",
-            600_000,
-          ),
-        }
-      : {}),
+    requestTimeoutMs,
   });
 }
 
+export function createNodeUnstructuredFetch({
+  createDispatcher = (options) => new Agent(options),
+  fetch: fetchImpl = undiciFetch as unknown as typeof fetch,
+  requestTimeoutMs,
+}: NodeUnstructuredFetchOptions): typeof fetch {
+  const dispatcher = createDispatcher({
+    bodyTimeout: requestTimeoutMs,
+    headersTimeout: requestTimeoutMs,
+  });
+
+  return (input, init) => {
+    if (input instanceof Request) {
+      const body = init?.body ?? input.body;
+
+      return fetchImpl(input.url, {
+        ...init,
+        body,
+        dispatcher,
+        ...(body ? { duplex: "half" } : {}),
+        headers: init?.headers ?? input.headers,
+        method: init?.method ?? input.method,
+        signal: init?.signal ?? input.signal,
+      } as DispatcherRequestInit);
+    }
+
+    return fetchImpl(input, {
+      ...init,
+      dispatcher,
+    } as DispatcherRequestInit);
+  };
+}
+
 function resolveUnstructuredApiUrl(env: ApiParserEnv): string | undefined {
   const configured = env.UNSTRUCTURED_API_URL?.trim();
   if (configured) {

knowledge-fs/docs/production-deployment.md

@@ -64,7 +64,7 @@ the service:
 | `UNSTRUCTURED_API_URL` | Parser endpoint for complex formats. |
 | `UNSTRUCTURED_API_KEY` | Optional parser authentication. |
 | `UNSTRUCTURED_MAX_CONCURRENCY` | Process-wide parser request limit; defaults to `2`. |
-| `UNSTRUCTURED_REQUEST_TIMEOUT_MS` | Total timeout for one parser request and response body; defaults to `120000`. |
+| `UNSTRUCTURED_REQUEST_TIMEOUT_MS` | Total timeout for one parser request and response body; defaults to `120000` and accepts up to `1800000` for long OCR documents. |
 | `UNSTRUCTURED_MAX_RESPONSE_BYTES` | Maximum parser response body; defaults to `33554432` (32 MiB). |
 
 Compose injects `DIFY_INNER_API_URL` and `DIFY_INNER_API_KEY`; do not duplicate them in the

knowledge-fs/packages/api/src/document-compilation-worker.test.ts

@@ -375,7 +375,7 @@ describe("createDocumentCompilationWorker lease integration", () => {
     let smokeCalls = 0;
     const outlines = createInMemoryDocumentOutlineRepository({ maxOutlines: 2 });
     const knowledgePaths = createInMemoryKnowledgePathRepository({
-      maxBatchSize: 10,
+      maxBatchSize: 2,
       maxListLimit: 10,
       maxPaths: 10,
     });

knowledge-fs/packages/api/src/document-compilation-worker.ts

@@ -618,7 +618,8 @@ export function createDocumentCompilationWorker({
               documentOutlineIds = [persistedOutline.id];
               if (knowledgePaths && generateKnowledgePathId) {
                 await assertWritable();
-                const persistedPaths = await knowledgePaths.upsertMany(
+                const persistedPaths = await upsertKnowledgePathsInBatches(
+                  knowledgePaths,
                   buildCompilationKnowledgePaths({
                     asset: activeAsset,
                     generateId: generateKnowledgePathId,
@@ -729,7 +730,8 @@ export function createDocumentCompilationWorker({
             documentOutlineIds = [persistedOutline.id];
             if (knowledgePaths && generateKnowledgePathId) {
               await assertWritable();
-              const persistedPaths = await knowledgePaths.upsertMany(
+              const persistedPaths = await upsertKnowledgePathsInBatches(
+                knowledgePaths,
                 buildCompilationKnowledgePaths({
                   asset: activeAsset,
                   generateId: generateKnowledgePathId,
@@ -1353,6 +1355,21 @@ function buildCompilationKnowledgePaths({
   ];
 }
 
+async function upsertKnowledgePathsInBatches(
+  repository: KnowledgePathRepository,
+  paths: readonly KnowledgePath[],
+): Promise<KnowledgePath[]> {
+  const persisted: KnowledgePath[] = [];
+
+  for (let offset = 0; offset < paths.length; offset += repository.maxBatchSize) {
+    persisted.push(
+      ...(await repository.upsertMany(paths.slice(offset, offset + repository.maxBatchSize))),
+    );
+  }
+
+  return persisted;
+}
+
 function componentReferences(
   componentKeys: readonly string[],
   generationId: string,

knowledge-fs/packages/api/src/graph-index-repository.ts

@@ -56,6 +56,7 @@ export interface GraphRelation {
 }
 
 export interface GraphIndexRepository {
+  readonly maxBatchSize: number;
   deleteComponentsBySourceNodesAcrossGenerations(
     input: DeleteGraphComponentsBySourceNodesAcrossGenerationsInput,
   ): Promise<DeleteGraphComponentsBySourceNodesResult>;
@@ -179,6 +180,7 @@ export function createInMemoryGraphIndexRepository({
   const relations = new Map<string, GraphRelation>();
 
   return {
+    maxBatchSize,
     deleteComponentsBySourceNodesAcrossGenerations: async (input) => {
       validateGraphPruneSourceNodesAcrossGenerationsInput(input);
       return deleteInMemoryGraphComponentsBySourceNodes({
@@ -519,6 +521,7 @@ export function createDatabaseGraphIndexRepository({
   });
 
   return {
+    maxBatchSize,
     deleteComponentsBySourceNodesAcrossGenerations: async (input) =>
       database.transaction((transaction) =>
         deleteDatabaseGraphComponentsBySourceNodesAcrossGenerations(database, transaction, input),

knowledge-fs/packages/api/src/graph-index-writer.ts

@@ -179,7 +179,10 @@ export function createGraphIndexWriter({
             updatedAt: timestamp,
           }),
         );
-      const storedEntities = await graph.upsertEntities(entityInputs);
+      const graphBatchSize = Math.min(maxBatchSize, graph.maxBatchSize);
+      const storedEntities = await persistGraphBatches(entityInputs, graphBatchSize, (batch) =>
+        graph.upsertEntities(batch),
+      );
       // Back-reference: record on each source node the graph entity ids it now maps to, so
       // retrieval can seed graph expansion from a node's matched entities. `updateMetadataMany`
       // replaces metadata, so merge onto the node's current metadata. This runs as the last
@@ -294,7 +297,9 @@ export function createGraphIndexWriter({
             updatedAt: timestamp,
           }),
         );
-      const storedRelations = await graph.upsertRelations(relationInputs);
+      const storedRelations = await persistGraphBatches(relationInputs, graphBatchSize, (batch) =>
+        graph.upsertRelations(batch),
+      );
 
       return {
         entities: storedEntities.map(cloneGraphEntity),
@@ -364,6 +369,20 @@ export function createGraphIndexWriter({
   return writer;
 }
 
+async function persistGraphBatches<T>(
+  input: readonly T[],
+  maxBatchSize: number,
+  persist: (batch: readonly T[]) => Promise<readonly T[]>,
+): Promise<T[]> {
+  const persisted: T[] = [];
+
+  for (let offset = 0; offset < input.length; offset += maxBatchSize) {
+    persisted.push(...(await persist(input.slice(offset, offset + maxBatchSize))));
+  }
+
+  return persisted;
+}
+
 type GraphQualityFlag = {
   readonly graphEligible: boolean;
   readonly reason?: "budget" | "confidence-threshold" | "duplicate" | undefined;

knowledge-fs/packages/api/src/graph-index.test.ts

@@ -315,14 +315,14 @@ function createFakeGraphTraversalExecutor() {
 describe("graph index persistence", () => {
   const graphTimestamp = "2026-05-12T12:00:00.000Z";
 
-  it("indexes graph-eligible entities and relations without writing ineligible outputs", async () => {
+  it("indexes eligible graph outputs in repository-sized batches", async () => {
     const nodes = createInMemoryKnowledgeNodeRepository({
       maxBatchSize: 4,
       maxListLimit: 4,
       maxNodes: 4,
     });
     const graph = createInMemoryGraphIndexRepository({
-      maxBatchSize: 8,
+      maxBatchSize: 1,
       maxEntities: 8,
       maxRelations: 8,
       now: () => "2026-05-12T12:00:00.000Z",
@@ -360,6 +360,13 @@ describe("graph index persistence", () => {
             subject: "Acme Corp",
             type: "mentions",
           },
+          {
+            confidence: 0.9,
+            object: "Refund Policy",
+            quality: { graphEligible: true },
+            subject: "Acme Corp",
+            type: "references",
+          },
           {
             confidence: 0.42,
             object: "weak concept",
@@ -405,7 +412,7 @@ describe("graph index persistence", () => {
     expect(result.missingNodeIds).toEqual(["018f0d60-7a49-7cc2-9c1b-5b36f18f2c99"]);
     expect(result.stats).toEqual({
       entitiesIndexed: 2,
-      relationsIndexed: 1,
+      relationsIndexed: 2,
       skippedEntities: 1,
       skippedRelations: 1,
     });

knowledge-fs/packages/api/src/knowledge-path-repository.ts

@@ -73,6 +73,7 @@ export interface DeleteKnowledgePathsByDocumentAssetInput {
 }
 
 export interface KnowledgePathRepository {
+  readonly maxBatchSize: number;
   create(input: KnowledgePath): Promise<KnowledgePath>;
   deleteByDocumentAsset(input: DeleteKnowledgePathsByDocumentAssetInput): Promise<number>;
   deleteSemanticView(input: DeleteSemanticViewPathsInput): Promise<number>;
@@ -129,6 +130,7 @@ export function createInMemoryKnowledgePathRepository({
   const paths = new Map<string, KnowledgePath>();
 
   return {
+    maxBatchSize,
     create: async (input) => {
       const path = cloneKnowledgePath(KnowledgePathSchema.parse(input));
       const key = knowledgePathKey(
@@ -343,6 +345,7 @@ export function createDatabaseKnowledgePathRepository({
   const tableName = "knowledge_paths";
 
   return {
+    maxBatchSize,
     create: async (input) => {
       const path = cloneKnowledgePath(KnowledgePathSchema.parse(input));
       return path.publicationGenerationId

knowledge-fs/packages/api/src/retrieval-paths-coverage.test.ts

@@ -126,6 +126,7 @@ function fakeGraph(
   traverse: (startEntityId: string) => GraphTraversalResult,
 ): GraphIndexRepository {
   return {
+    maxBatchSize: 1,
     deleteComponentsBySourceNodesAcrossGenerations: async () => {
       throw new Error(
         "deleteComponentsBySourceNodesAcrossGenerations is not used by graph expansion",

knowledge-fs/packages/parsers/src/index.ts

@@ -193,6 +193,7 @@ const defaultMaxResponseBytes = 32 * 1024 * 1024;
 const defaultMaxConcurrency = 2;
 const defaultMaxRetries = 0;
 const defaultRequestTimeoutMs = 120_000;
+const maxRequestTimeoutMs = 1_800_000;
 const defaultMaxRows = 20_000;
 const defaultRetryDelayMs = 100;
 const defaultNow = () => new Date().toISOString();
@@ -3600,10 +3601,10 @@ function validateUnstructuredResourceOptions({
   if (
     !Number.isSafeInteger(requestTimeoutMs) ||
     requestTimeoutMs < 1 ||
-    requestTimeoutMs > 600_000
+    requestTimeoutMs > maxRequestTimeoutMs
   ) {
     throw new ProviderInputError(
-      "Unstructured parser requestTimeoutMs must be an integer between 1 and 600000",
+      `Unstructured parser requestTimeoutMs must be an integer between 1 and ${maxRequestTimeoutMs}`,
     );
   }
 }

knowledge-fs/packages/parsers/src/parser.test.ts

@@ -3128,13 +3128,13 @@ describe("parser adapters", () => {
       "maxConcurrency must be an integer between 1 and 32",
     );
     expect(() => createUnstructuredParserClient({ ...base, requestTimeoutMs: 0 })).toThrow(
-      "requestTimeoutMs must be an integer between 1 and 600000",
+      "requestTimeoutMs must be an integer between 1 and 1800000",
     );
-    expect(() => createUnstructuredParserClient({ ...base, requestTimeoutMs: 600_001 })).toThrow(
-      "requestTimeoutMs must be an integer between 1 and 600000",
+    expect(() => createUnstructuredParserClient({ ...base, requestTimeoutMs: 1_800_001 })).toThrow(
+      "requestTimeoutMs must be an integer between 1 and 1800000",
     );
     expect(() => createUnstructuredParserClient({ ...base, requestTimeoutMs: 1.5 })).toThrow(
-      "requestTimeoutMs must be an integer between 1 and 600000",
+      "requestTimeoutMs must be an integer between 1 and 1800000",
     );
   });
 

knowledge-fs/pnpm-lock.yaml

@@ -104,6 +104,9 @@ importers:
       sharp:
         specifier: 0.35.3
         version: 0.35.3(@types/node@22.19.18)
+      undici:
+        specifier: ^8.10.0
+        version: 8.10.0
     devDependencies:
       '@types/node':
         specifier: ^22.10.2
@@ -2207,6 +2210,10 @@ packages:
   undici-types@6.21.0:
     resolution: {integrity: sha512-iwDZqg0QAGrg9Rav5H4n0M64c3mkR59cJ6wQp+7C4nI0gsmExaedaYLNO44eT4AtBBwjbTiGPMlt2Md0T9H9JQ==}
 
+  undici@8.10.0:
+    resolution: {integrity: sha512-HvltHd7avK13QIw/oLe4qoOLyoVSoafqJ2jYOrtMRBkbYT31eiBQ8O0ehRKZiEZCMEyLFQNIADpgCWC5fALvYQ==}
+    engines: {node: '>=22.19.0'}
+
   unicode-segmenter@0.15.0:
     resolution: {integrity: sha512-Xmvwqx4F8nGuCv2eGPJVJq73NMTfpqx2Xe9/v5hQoyAUnERVhX+sRkyYVdYoBUbnTok2FTBOlstUeQ5sRleXSA==}
 
@@ -3883,6 +3890,8 @@ snapshots:
 
   undici-types@6.21.0: {}
 
+  undici@8.10.0: {}
+
   unicode-segmenter@0.15.0: {}
 
   unpipe@1.0.0: {}