fix(FN-6035): harden workflow projection publishing

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",

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);
           }

packages/engine/src/workflow-graph-executor.ts

@@ -40,14 +40,12 @@ export interface WorkflowNodeResult {
 
 export interface WorkflowTaskProjection {
   modifiedFiles?: string[];
-  mergeDetails?: Record<string, unknown>;
+  mergeDetails?: {
+    filesChanged?: number;
+    insertions?: number;
+    deletions?: number;
+  };
   summary?: string;
-  review?: Record<string, unknown>;
-  reviewState?: Record<string, unknown>;
-  workflowStepResults?: unknown[];
-  tokenUsage?: Record<string, unknown>;
-  error?: string | null;
-  status?: string | null;
 }
 
 export interface WorkflowNodeExecutionContext {
@@ -187,6 +185,12 @@ function objectRecord(value: unknown): Record<string, unknown> | 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<string, unknown> | undefined): WorkflowTaskProjection {
   if (!contextPatch) return {};
   const patch: WorkflowTaskProjection = {};
@@ -194,21 +198,22 @@ function extractTaskProjection(contextPatch: Record<string, unknown> | 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<WorkflowTaskProjection["mergeDetails"]>;
   }
 
   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<void> {
+  ): Promise<WorkflowNodeResult> {
     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;
   }
 }