fix(worker): two-level fan-out for description generation (fixes #488)

apps/worker/worker.ts

@@ -109,13 +109,15 @@ async function startWorkerProcess() {
         { deleteGraphFiles },
         { processFile, processFiles },
         { updateDescriptions },
+        { processDescriptionsGroups },
     ] = await Promise.all([
         import("."),
         import("./env"),
         import("./workflows/delete-file"),
         import("./workflows/delete-graph-files"),
         import("./workflows/process-file"),
         import("./workflows/update-descriptions"),
+        import("./workflows/process-descriptions-group"),
     ]);
 
     const parentPid = process.ppid;
@@ -126,6 +128,7 @@ async function startWorkerProcess() {
     ow.implementWorkflow(deleteProjectFile.spec, deleteProjectFile.fn);
     ow.implementWorkflow(deleteGraphFiles.spec, deleteGraphFiles.fn);
     ow.implementWorkflow(updateDescriptions.spec, updateDescriptions.fn);
+    ow.implementWorkflow(processDescriptionsGroups.spec, processDescriptionsGroups.fn);
 
     const worker = ow.newWorker({ concurrency });
     let shuttingDown = false;

apps/worker/workflows/process-descriptions-group-spec.ts

@@ -0,0 +1,26 @@
+import { defineWorkflowSpec } from "openworkflow";
+import z from "zod";
+
+/**
+ * Maximum number of update-descriptions sub-workflow calls per group.
+ * Keep this low enough that the group workflow itself stays well under
+ * the OpenWorkflow step limit (1000), while ensuring the number of groups
+ * spawned by process-files also stays bounded.
+ */
+export const DESCRIPTION_BATCHES_PER_GROUP = 50;
+
+export const processDescriptionsGroupsSpec = defineWorkflowSpec({
+    name: "process-descriptions-groups",
+    version: "1.0.0",
+    retryPolicy: {
+        initialInterval: "1s",
+        backoffCoefficient: 2,
+        maximumInterval: "30s",
+        maximumAttempts: 3,
+    },
+    schema: z.object({
+        graphId: z.string(),
+        entityIds: z.array(z.string()),
+        relationshipIds: z.array(z.string()),
+    }),
+});

apps/worker/workflows/process-descriptions-group.ts

@@ -0,0 +1,41 @@
+import { defineWorkflow } from "openworkflow";
+import { updateDescriptionsSpec } from "./update-descriptions-spec";
+import { processDescriptionsGroupsSpec } from "./process-descriptions-group-spec";
+import { chunkItems } from "../lib/chunk";
+import { DESCRIPTION_BATCH_SIZE } from "../lib/description-workflow";
+
+/**
+ * Spawns update-descriptions sub-workflows for a slice of entity/relationship IDs.
+ * Each process-descriptions-group call handles at most DESCRIPTION_BATCHES_PER_GROUP
+ * worth of IDs (pre-sliced by process-files), so this simply batches them for
+ * update-descriptions without additional grouping.
+ */
+export const processDescriptionsGroups = defineWorkflow(
+    processDescriptionsGroupsSpec,
+    async ({ input, step }) => {
+        const entityIdBatches = chunkItems(input.entityIds, DESCRIPTION_BATCH_SIZE);
+        const relationshipIdBatches = chunkItems(input.relationshipIds, DESCRIPTION_BATCH_SIZE);
+
+        const entityPromises = entityIdBatches.map((entityIds) =>
+            step.runWorkflow(updateDescriptionsSpec, {
+                graphId: input.graphId,
+                entityIds,
+                relationshipIds: [],
+            })
+        );
+
+        const relationshipPromises = relationshipIdBatches.map((relationshipIds) =>
+            step.runWorkflow(updateDescriptionsSpec, {
+                graphId: input.graphId,
+                entityIds: [],
+                relationshipIds,
+            })
+        );
+
+        const results = await Promise.allSettled([...entityPromises, ...relationshipPromises]);
+        const failures = results.filter((r) => r.status === "rejected");
+        if (failures.length > 0) {
+            throw new Error(`${failures.length} of ${results.length} description batches failed`);
+        }
+    }
+);

apps/worker/workflows/process-file.ts

@@ -35,7 +35,7 @@ import { chunkItems } from "../lib/chunk";
 import { processFilesSpec } from "./process-files-spec";
 import { deleteGraphFileProcessingArtifacts, getGraphFileArtifactPaths } from "../lib/derived-files";
 import { buildMetadata, buildMetadataExcerpt } from "../lib/metadata";
-import { updateDescriptionsSpec } from "./update-descriptions-spec";
+import { processDescriptionsGroupsSpec, DESCRIPTION_BATCHES_PER_GROUP } from "./process-descriptions-group-spec";
 import { DESCRIPTION_BATCH_SIZE } from "../lib/description-workflow";
 import { getFileTypeProcessingConfig } from "../lib/file-type-config";
 import { toTextUnitRows } from "../lib/text-unit-rows";
@@ -200,20 +200,42 @@ export const processFiles = defineWorkflow(processFilesSpec, async ({ input, ste
             };
         });
 
-        await Promise.all([
-            ...chunkItems(descriptions.entityIds, DESCRIPTION_BATCH_SIZE).map((entityIds) =>
-                step.runWorkflow(updateDescriptionsSpec, {
+        // Two-level fan-out: spawn groups of description batches to avoid exceeding
+        // the OpenWorkflow step limit (1000) on large projects with many entities/relationships.
+        // Each group spawns update-descriptions sub-workflows internally, so the parent workflow
+        // only counts one step per group instead of one step per batch.
+        const totalEntityBatches = chunkItems(descriptions.entityIds, DESCRIPTION_BATCH_SIZE).length;
+        const totalRelationshipBatches = chunkItems(descriptions.relationshipIds, DESCRIPTION_BATCH_SIZE).length;
+        const totalBatches = totalEntityBatches + totalRelationshipBatches;
+        const groupsCount = Math.ceil(totalBatches / DESCRIPTION_BATCHES_PER_GROUP);
+
+        const entityIdBatches = chunkItems(descriptions.entityIds, DESCRIPTION_BATCH_SIZE);
+        const relationshipIdBatches = chunkItems(descriptions.relationshipIds, DESCRIPTION_BATCH_SIZE);
+        const allBatches = [
+            ...entityIdBatches.map((ids) => ({ entityIds: ids, relationshipIds: [] as string[] })),
+            ...relationshipIdBatches.map((ids) => ({ entityIds: [] as string[], relationshipIds: ids })),
+        ];
+
+        const groupPromises = [];
+        for (let i = 0; i < groupsCount; i++) {
+            const groupBatches = allBatches.slice(i * DESCRIPTION_BATCHES_PER_GROUP, (i + 1) * DESCRIPTION_BATCHES_PER_GROUP);
+            const groupEntityIds = groupBatches.flatMap((b) => b.entityIds);
+            const groupRelationshipIds = groupBatches.flatMap((b) => b.relationshipIds);
+
+            groupPromises.push(
+                step.runWorkflow(processDescriptionsGroupsSpec, {
                     graphId: input.graphId,
-                    entityIds,
+                    entityIds: groupEntityIds,
+                    relationshipIds: groupRelationshipIds,
                 })
-            ),
-            ...chunkItems(descriptions.relationshipIds, DESCRIPTION_BATCH_SIZE).map((relationshipIds) =>
-                step.runWorkflow(updateDescriptionsSpec, {
-                    graphId: input.graphId,
-                    relationshipIds,
-                })
-            ),
-        ]);
+            );
+        }
+
+        const groupResults = await Promise.allSettled(groupPromises);
+        const failedGroups = groupResults.filter((r) => r.status === "rejected");
+        if (failedGroups.length > 0) {
+            throw new Error(`${failedGroups.length} of ${groupResults.length} description groups failed`);
+        }
 
         await step.run({ name: "finalize-project-status" }, async () => {
             await db.update(graphTable).set({ state: "ready" }).where(eq(graphTable.id, input.graphId));