Merge pull request #460 from mediar-ai/pr/event-flush-20260119-163051

fix: Add event flush for reliable delivery before runner returns

Author: m13v

packages/workflow/src/events.ts

@@ -191,6 +191,43 @@ class EventTransport {
         }
     }
 
+    /**
+     * Flush pending writes to ensure all events are delivered
+     * Returns a promise that resolves when the stream is drained
+     */
+    async flush(): Promise<void> {
+        if (this.useStderr || !this.pipeStream) {
+            return;
+        }
+
+        return new Promise<void>((resolve) => {
+            if (!this.pipeStream) {
+                resolve();
+                return;
+            }
+
+            // If stream is already drained, resolve immediately
+            if (this.pipeStream.writableNeedDrain === false) {
+                resolve();
+                return;
+            }
+
+            // Wait for drain event
+            const onDrain = () => {
+                resolve();
+            };
+
+            this.pipeStream.once('drain', onDrain);
+
+            // Also resolve on error or close to avoid hanging
+            this.pipeStream.once('error', () => resolve());
+            this.pipeStream.once('close', () => resolve());
+
+            // Timeout fallback - don't wait forever
+            setTimeout(() => resolve(), 100);
+        });
+    }
+
     /**
      * Close the transport
      */
@@ -334,6 +371,14 @@ export const emit = {
     raw(event: Omit<WorkflowEvent, '__mcp_event__' | 'timestamp'>): void {
         transport.send(event);
     },
+
+    /**
+     * Flush all pending events to ensure they are delivered
+     * Call this before workflow exits to ensure all events are sent
+     */
+    async flush(): Promise<void> {
+        await transport.flush();
+    },
 };
 
 /**

packages/workflow/src/runner.ts

@@ -140,6 +140,7 @@ export class WorkflowRunner {
           this.state.context.data = result.result;
           this.state.lastStepId = step.config.id;
           this.state.lastStepIndex = i;
+          await emit.flush(); // Ensure events are delivered before returning
           return {
             status: 'executed_without_error',
             lastStepId: this.state.lastStepId,
@@ -210,6 +211,7 @@ export class WorkflowRunner {
         this.state.lastStepIndex = i;
 
         // Return error result
+        await emit.flush(); // Ensure events are delivered before returning
         return {
           status: 'executed_with_error',
           lastStepId: this.state.lastStepId,
@@ -219,6 +221,7 @@ export class WorkflowRunner {
       }
     }
 
+    await emit.flush(); // Ensure events are delivered before returning
     return {
       status: 'executed_without_error',
       lastStepId: this.state.lastStepId,