diff --git a/packages/engine/src/__tests__/workflow-graph-executor-handlers.test.ts b/packages/engine/src/__tests__/workflow-graph-executor-handlers.test.ts index 8ef04e0232..637655a72f 100644 --- a/packages/engine/src/__tests__/workflow-graph-executor-handlers.test.ts +++ b/packages/engine/src/__tests__/workflow-graph-executor-handlers.test.ts @@ -159,6 +159,58 @@ describe("WorkflowGraphExecutor traversal", () => { ); }); + it("keeps projection writes to safe task metadata fields", async () => { + const ir: WorkflowIr = { + version: "v1", + name: "safe-projection", + nodes: [ + { id: "start", kind: "start" }, + { id: "a", kind: "prompt" }, + { id: "end", kind: "end" }, + ], + edges: [ + { from: "start", to: "a" }, + { from: "a", to: "end", condition: "success" }, + ], + }; + const publishTaskProjection = vi.fn(); + const executor = new WorkflowGraphExecutor({ + handlers: { + prompt: async () => ({ + outcome: "success", + contextPatch: { + modifiedFiles: ["src/index.ts"], + mergeDetails: { + commitSha: "engine-owned", + mergeConfirmed: true, + filesChanged: 3, + insertions: 12.8, + deletions: 1, + }, + status: "done", + error: "bypass", + review: {}, + reviewState: {}, + workflowStepResults: [{}], + tokenUsage: {}, + }, + }), + }, + publishTaskProjection, + }); + + await executor.run(task, settingsOn(), ir); + + expect(publishTaskProjection).toHaveBeenCalledWith( + task.id, + { + modifiedFiles: ["src/index.ts"], + mergeDetails: { filesChanged: 3, insertions: 12, deletions: 1 }, + }, + { nodeId: "a", nodeKind: "prompt" }, + ); + }); + it("publishes projections from loop template nodes", async () => { const ir: WorkflowIr = { version: "v2", @@ -210,6 +262,41 @@ describe("WorkflowGraphExecutor traversal", () => { ); }); + it("does not retry an already-executed node when projection publishing fails", async () => { + const ir: WorkflowIr = { + version: "v1", + name: "projection-failure", + nodes: [ + { id: "start", kind: "start" }, + { id: "a", kind: "prompt" }, + { id: "end", kind: "end" }, + ], + edges: [ + { from: "start", to: "a" }, + { from: "a", to: "end", condition: "failure" }, + ], + }; + const handler = vi.fn(async () => ({ + outcome: "success" as const, + contextPatch: { modifiedFiles: ["src/once.ts"] }, + })); + const publishTaskProjection = vi.fn(async () => { + throw new Error("store unavailable"); + }); + const executor = new WorkflowGraphExecutor({ + handlers: { prompt: handler }, + maxRetriesPerNode: 3, + publishTaskProjection, + }); + + const result = await executor.run(task, settingsOn(), ir); + + expect(handler).toHaveBeenCalledTimes(1); + expect(publishTaskProjection).toHaveBeenCalledTimes(1); + expect(result.outcome).toBe("failure"); + expect(result.context["node:a:projectionError"]).toBe("store unavailable"); + }); + it("caps retries and converts exceptions to failure", async () => { const ir: WorkflowIr = { version: "v1", diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index 91900c4be5..17592439b5 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -3759,15 +3759,9 @@ export class TaskExecutor { if (merged.length > 0) update.modifiedFiles = merged; } if (patch.mergeDetails) { - update.mergeDetails = { ...(liveTask?.mergeDetails ?? {}), ...patch.mergeDetails } as Task["mergeDetails"]; + update.mergeDetails = { ...(liveTask?.mergeDetails ?? {}), ...patch.mergeDetails }; } if (patch.summary !== undefined) update.summary = patch.summary; - if (patch.review !== undefined) update.review = patch.review as unknown as Task["review"]; - if (patch.reviewState !== undefined) update.reviewState = patch.reviewState as unknown as Task["reviewState"]; - if (patch.workflowStepResults !== undefined) update.workflowStepResults = patch.workflowStepResults as Task["workflowStepResults"]; - if (patch.tokenUsage !== undefined) update.tokenUsage = patch.tokenUsage as unknown as Task["tokenUsage"]; - if (patch.error !== undefined) update.error = patch.error; - if (patch.status !== undefined) update.status = patch.status; if (Object.keys(update).length > 0) { await this.store.updateTask(taskId, update); } diff --git a/packages/engine/src/workflow-graph-executor.ts b/packages/engine/src/workflow-graph-executor.ts index 34f77b226d..9d2dd03360 100644 --- a/packages/engine/src/workflow-graph-executor.ts +++ b/packages/engine/src/workflow-graph-executor.ts @@ -40,14 +40,12 @@ export interface WorkflowNodeResult { export interface WorkflowTaskProjection { modifiedFiles?: string[]; - mergeDetails?: Record; + mergeDetails?: { + filesChanged?: number; + insertions?: number; + deletions?: number; + }; summary?: string; - review?: Record; - reviewState?: Record; - workflowStepResults?: unknown[]; - tokenUsage?: Record; - error?: string | null; - status?: string | null; } export interface WorkflowNodeExecutionContext { @@ -187,6 +185,12 @@ function objectRecord(value: unknown): Record | undefined { : undefined; } +function finiteCount(value: unknown): number | undefined { + return typeof value === "number" && Number.isFinite(value) && value >= 0 + ? Math.floor(value) + : undefined; +} + function extractTaskProjection(contextPatch: Record | undefined): WorkflowTaskProjection { if (!contextPatch) return {}; const patch: WorkflowTaskProjection = {}; @@ -194,21 +198,22 @@ function extractTaskProjection(contextPatch: Record | undefined if (files.length > 0) patch.modifiedFiles = files; const mergeDetails = objectRecord(contextPatch.mergeDetails); - if (mergeDetails) patch.mergeDetails = mergeDetails; - if (typeof contextPatch.filesChanged === "number" && Number.isFinite(contextPatch.filesChanged)) { - patch.mergeDetails = { ...(patch.mergeDetails ?? {}), filesChanged: contextPatch.filesChanged }; + const safeMergeDetails = { + filesChanged: finiteCount(contextPatch.filesChanged ?? mergeDetails?.filesChanged), + insertions: finiteCount(mergeDetails?.insertions), + deletions: finiteCount(mergeDetails?.deletions), + }; + if ( + safeMergeDetails.filesChanged !== undefined + || safeMergeDetails.insertions !== undefined + || safeMergeDetails.deletions !== undefined + ) { + patch.mergeDetails = Object.fromEntries( + Object.entries(safeMergeDetails).filter(([, value]) => value !== undefined), + ) as NonNullable; } if (typeof contextPatch.summary === "string") patch.summary = contextPatch.summary; - const review = objectRecord(contextPatch.review); - if (review) patch.review = review; - const reviewState = objectRecord(contextPatch.reviewState); - if (reviewState) patch.reviewState = reviewState; - if (Array.isArray(contextPatch.workflowStepResults)) patch.workflowStepResults = contextPatch.workflowStepResults; - const tokenUsage = objectRecord(contextPatch.tokenUsage); - if (tokenUsage) patch.tokenUsage = tokenUsage; - if (typeof contextPatch.error === "string" || contextPatch.error === null) patch.error = contextPatch.error; - if (typeof contextPatch.status === "string" || contextPatch.status === null) patch.status = contextPatch.status; return patch; } @@ -681,15 +686,13 @@ export class WorkflowGraphExecutor { try { const pluginResult = await this.executePluginNodeHandler(node, task, workflow, context, signal); if (pluginResult) { - await this.publishTaskProjectionFromResult(task.id, node, pluginResult); - return pluginResult; + return await this.publishTaskProjectionFromResult(task.id, node, pluginResult); } if (!handler) { throw new WorkflowIrError(`No handler registered for node kind: ${node.kind}`); } const result = await handler(node, { task, settings, context, signal }); - await this.publishTaskProjectionFromResult(task.id, node, result); - return result; + return await this.publishTaskProjectionFromResult(task.id, node, result); } catch (error) { lastError = error; } @@ -708,13 +711,25 @@ export class WorkflowGraphExecutor { taskId: string, node: WorkflowIrNode, result: WorkflowNodeResult, - ): Promise { + ): Promise { const patch = extractTaskProjection(result.contextPatch); - if (!hasTaskProjection(patch)) return; + if (!hasTaskProjection(patch)) return result; const source = { nodeId: node.id, nodeKind: node.kind }; - await this.deps.publishTaskProjection?.(taskId, patch, source); - if (patch.modifiedFiles && patch.modifiedFiles.length > 0) { - await this.deps.publishTouchedFiles?.(taskId, patch.modifiedFiles, source); + try { + await this.deps.publishTaskProjection?.(taskId, patch, source); + if (patch.modifiedFiles && patch.modifiedFiles.length > 0) { + await this.deps.publishTouchedFiles?.(taskId, patch.modifiedFiles, source); + } + } catch (error) { + return { + outcome: "failure", + value: "projection-error", + contextPatch: { + ...(result.contextPatch ?? {}), + [`node:${node.id}:projectionError`]: error instanceof Error ? error.message : String(error), + }, + }; } + return result; } }