Implement workflow chain execution to share results between workflows (#1872)

# READ CAREFULLY THEN REMOVE

Remove bullet points that are not relevant.

PLEASE REFRAIN FROM USING AI TO WRITE YOUR CODE AND PR DESCRIPTION. IF YOU DO USE AI TO WRITE YOUR CODE PLEASE PROVIDE A DESCRIPTION AND REVIEW IT CAREFULLY. MAKE SURE YOU UNDERSTAND THE CODE YOU ARE SUBMITTING USING AI.

- Pull requests that do not follow these guidelines will be closed without review or comment.
- If you use AI to write your PR description your pr will be close without review or comment.
- If you are unsure about anything, feel free to ask for clarification.

## Description

Please provide a clear description of your changes.

---

## Type of Change

Please delete options that are not relevant.

- [ ] 🐛 Bug fix (non-breaking change which fixes an issue)
- [ ] ✨ New feature (non-breaking change which adds functionality)
- [ ] 💥 Breaking change (fix or feature with breaking changes)
- [ ] 📝 Documentation update
- [ ] 🎨 UI/UX improvement
- [ ] 🔒 Security enhancement
- [ ] ⚡ Performance improvement

## Areas Affected

Please check all that apply:

- [ ] Email Integration (Gmail, IMAP, etc.)
- [ ] User Interface/Experience
- [ ] Authentication/Authorization
- [ ] Data Storage/Management
- [ ] API Endpoints
- [ ] Documentation
- [ ] Testing Infrastructure
- [ ] Development Workflow
- [ ] Deployment/Infrastructure

## Testing Done

Describe the tests you've done:

- [ ] Unit tests added/updated
- [ ] Integration tests added/updated
- [ ] Manual testing performed
- [ ] Cross-browser testing (if UI changes)
- [ ] Mobile responsiveness verified (if UI changes)

## Security Considerations

For changes involving data or authentication:

- [ ] No sensitive data is exposed
- [ ] Authentication checks are in place
- [ ] Input validation is implemented
- [ ] Rate limiting is considered (if applicable)

## Checklist

- [ ] I have read the [CONTRIBUTING](https://github.com/Mail-0/Zero/blob/staging/.github/CONTRIBUTING.md) document
- [ ] My code follows the project's style guidelines
- [ ] I have performed a self-review of my code
- [ ] I have commented my code, particularly in complex areas
- [ ] I have updated the documentation
- [ ] My changes generate no new warnings
- [ ] I have added tests that prove my fix/feature works
- [ ] All tests pass locally
- [ ] Any dependent changes are merged and published

## Additional Notes

Add any other context about the pull request here.

## Screenshots/Recordings

Add screenshots or recordings here if applicable.

---

_By submitting this pull request, I confirm that my contribution is made under the terms of the project's license._

    
<!-- This is an auto-generated description by cubic. -->
---

## Summary by cubic
Added workflow chain execution so results can be shared between workflows, enabling workflows to pass data to each other during execution.

- **New Features**
  - Introduced `executeWorkflowChain` to run multiple workflows in sequence and share results.
  - Updated workflow runner to use the new chain execution method.

<!-- End of auto-generated description by cubic. -->



<!-- This is an auto-generated comment: release notes by coderabbit.ai -->

## Summary by CodeRabbit

* **Bug Fixes**
  * Improved accuracy in counting successfully synced threads during workflow execution.
  * Workflows now proceed regardless of certain environment variable settings.

* **New Features**
  * Added support for executing multiple workflows in sequence with consolidated result and error reporting.

* **Refactor**
  * Simplified and streamlined workflow execution logic for better maintainability.

* **Style**
  * Enhanced code formatting for improved readability.

* **Documentation**
  * Updated method signatures to clarify optional parameters.

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: MrgSub

apps/server/src/main.ts

@@ -187,10 +187,7 @@ export class DbRpcDO extends RpcTarget {
     return await this.mainDo.deleteEmailTemplate(this.userId, templateId);
   }
 
-  async updateEmailTemplate(
-    templateId: string,
-    data: Partial<typeof emailTemplate.$inferInsert>,
-  ) {
+  async updateEmailTemplate(templateId: string, data: Partial<typeof emailTemplate.$inferInsert>) {
     return await this.mainDo.updateEmailTemplate(this.userId, templateId, data);
   }
 }
@@ -522,7 +519,10 @@ class ZeroDB extends DurableObject<Env> {
     });
   }
 
-  async createEmailTemplate(userId: string, payload: Omit<typeof emailTemplate.$inferInsert, 'userId'>) {
+  async createEmailTemplate(
+    userId: string,
+    payload: Omit<typeof emailTemplate.$inferInsert, 'userId'>,
+  ) {
     return await this.db
       .insert(emailTemplate)
       .values({

apps/server/src/pipelines.ts

@@ -393,7 +393,7 @@ export class WorkflowRunner extends DurableObject<Env> {
             { concurrency: 6 }, // Limit concurrency to avoid rate limits
           );
 
-          const syncedCount = syncResults.length;
+          const syncedCount = syncResults.filter((result) => result.result.success).length;
           const failedCount = threadWorkflowParams.length - syncedCount;
 
           if (failedCount > 0) {
@@ -627,37 +627,15 @@ export class WorkflowRunner extends DurableObject<Env> {
         // Execute configured workflows using the workflow engine
         const workflowResults = yield* Effect.tryPromise({
           try: async () => {
-            const allResults = new Map<string, any>();
-            const allErrors = new Map<string, Error>();
-
             // Execute all workflows registered in the engine
             const workflowNames = workflowEngine.getWorkflowNames();
 
-            for (const workflowName of workflowNames) {
-              console.log(`[THREAD_WORKFLOW] Executing workflow: ${workflowName}`);
-
-              try {
-                const { results, errors } = await workflowEngine.executeWorkflow(
-                  workflowName,
-                  workflowContext,
-                );
-
-                // Merge results and errors using efficient Map operations
-                results.forEach((value, key) => allResults.set(key, value));
-                errors.forEach((value, key) => allErrors.set(key, value));
-
-                console.log(`[THREAD_WORKFLOW] Completed workflow: ${workflowName}`);
-              } catch (error) {
-                console.error(
-                  `[THREAD_WORKFLOW] Failed to execute workflow ${workflowName}:`,
-                  error,
-                );
-                const errorObj = error instanceof Error ? error : new Error(String(error));
-                allErrors.set(workflowName, errorObj);
-              }
-            }
+            const { results, errors } = await workflowEngine.executeWorkflowChain(
+              workflowNames,
+              workflowContext,
+            );
 
-            return { results: allResults, errors: allErrors };
+            return { results, errors };
           },
           catch: (error) => ({ _tag: 'WorkflowCreationFailed' as const, error }),
         });

apps/server/src/routes/agent/index.ts

@@ -1779,7 +1779,7 @@ export class ZeroAgent extends AIChatAgent<typeof env> {
   private getDataStreamResponse(
     onFinish: StreamTextOnFinishCallback<{}>,
     options?: {
-      abortSignal: AbortSignal | undefined;
+      abortSignal?: AbortSignal;
       connectionId?: string;
     },
   ) {
@@ -1822,7 +1822,10 @@ export class ZeroAgent extends AIChatAgent<typeof env> {
           onError: (error) => {
             console.error('Error in streamText', error);
           },
-          system: await getPrompt(getPromptName(connectionId, EPrompts.Chat), AiChatPrompt(currentThreadId)),
+          system: await getPrompt(
+            getPromptName(connectionId, EPrompts.Chat),
+            AiChatPrompt(currentThreadId),
+          ),
         });
 
         result.mergeIntoDataStream(dataStream);
@@ -1912,7 +1915,9 @@ export class ZeroAgent extends AIChatAgent<typeof env> {
                 await this.persistMessages(finalMessages, [connection.id]);
                 this.removeAbortController(chatMessageId);
               },
-              abortSignal ? { abortSignal, connectionId: connection.id } : { connectionId: connection.id },
+              abortSignal
+                ? { abortSignal, connectionId: connection.id }
+                : { connectionId: connection.id },
             );
 
             if (response) {
@@ -1955,7 +1960,9 @@ export class ZeroAgent extends AIChatAgent<typeof env> {
         }
         case IncomingMessageType.ThreadIdUpdate: {
           this.connectionThreadIds.set(connection.id, data.threadId);
-          console.log(`[ZeroAgent] Updated threadId for connection ${connection.id}: ${data.threadId}`);
+          console.log(
+            `[ZeroAgent] Updated threadId for connection ${connection.id}: ${data.threadId}`,
+          );
           break;
         }
         // case IncomingMessageType.Mail_List: {
@@ -2023,7 +2030,7 @@ export class ZeroAgent extends AIChatAgent<typeof env> {
   async onChatMessage(
     onFinish: StreamTextOnFinishCallback<{}>,
     options?: {
-      abortSignal: AbortSignal | undefined;
+      abortSignal?: AbortSignal;
       connectionId?: string;
     },
   ) {

apps/server/src/thread-workflow-utils/workflow-engine.ts

@@ -42,13 +42,14 @@ export class WorkflowEngine {
   async executeWorkflow(
     workflowName: string,
     context: WorkflowContext,
+    existingResults?: Map<string, any>,
   ): Promise<{ results: Map<string, any>; errors: Map<string, Error> }> {
     const workflow = this.workflows.get(workflowName);
     if (!workflow) {
       throw new Error(`Workflow "${workflowName}" not found`);
     }
 
-    const results = new Map<string, any>();
+    const results = new Map<string, any>(existingResults || []);
     const errors = new Map<string, Error>();
 
     for (const step of workflow.steps) {
@@ -83,6 +84,45 @@ export class WorkflowEngine {
     return { results, errors };
   }
 
+  async executeWorkflowChain(
+    workflowNames: string[],
+    context: WorkflowContext,
+  ): Promise<{ results: Map<string, any>; errors: Map<string, Error> }> {
+    let sharedResults = new Map<string, any>();
+    let allErrors = new Map<string, Error>();
+
+    for (const workflowName of workflowNames) {
+      console.log(`[WORKFLOW_ENGINE] Executing workflow in chain: ${workflowName}`);
+      try {
+        const { results, errors } = await this.executeWorkflow(
+          workflowName,
+          context,
+          sharedResults,
+        );
+
+        // Merge results
+        for (const [key, value] of results) {
+          sharedResults.set(key, value);
+        }
+
+        // Merge errors
+        for (const [key, error] of errors) {
+          allErrors.set(key, error);
+        }
+
+        console.log(
+          `[WORKFLOW_ENGINE] Completed workflow: ${workflowName}, total results: ${sharedResults.size}`,
+        );
+      } catch (error) {
+        const errorObj = error instanceof Error ? error : new Error(String(error));
+        console.error(`[WORKFLOW_ENGINE] Failed to execute workflow ${workflowName}:`, errorObj);
+        allErrors.set(workflowName, errorObj);
+      }
+    }
+
+    return { results: sharedResults, errors: allErrors };
+  }
+
   clearContext(context: WorkflowContext): void {
     if (context.results) {
       context.results.clear();
@@ -158,7 +198,7 @@ export const createDefaultWorkflows = (): WorkflowEngine => {
     ],
   };
 
-  const _vectorizationWorkflow: WorkflowDefinition = {
+  const vectorizationWorkflow: WorkflowDefinition = {
     name: 'message-vectorization',
     description: 'Vectorizes thread messages for search and analysis',
     steps: [
@@ -272,7 +312,7 @@ export const createDefaultWorkflows = (): WorkflowEngine => {
   };
 
   engine.registerWorkflow(autoDraftWorkflow);
-  //   engine.registerWorkflow(vectorizationWorkflow);
+  engine.registerWorkflow(vectorizationWorkflow);
   engine.registerWorkflow(threadSummaryWorkflow);
   engine.registerWorkflow(labelGenerationWorkflow);