feat: allow training multiple capture actions in one recording session

maxun-core/src/interpret.ts

@@ -45,6 +45,7 @@ interface InterpreterOptions {
   debugChannel: Partial<{
     activeId: Function,
     debugMessage: Function,
+    setActionType: Function,
   }>
 }
 
@@ -377,12 +378,20 @@ export default class Interpreter extends EventEmitter {
      */
     const wawActions: Record<CustomFunctions, (...args: any[]) => void> = {
       screenshot: async (params: PageScreenshotOptions) => {
+        if (this.options.debugChannel?.setActionType) {
+          this.options.debugChannel.setActionType('screenshot');
+        }
+
         const screenshotBuffer = await page.screenshot({
           ...params, path: undefined,
         });
         await this.options.binaryCallback(screenshotBuffer, 'image/png');
       },
       enqueueLinks: async (selector: string) => {
+        if (this.options.debugChannel?.setActionType) {
+          this.options.debugChannel.setActionType('enqueueLinks');
+        }
+
         const links: string[] = await page.locator(selector)
           .evaluateAll(
             // @ts-ignore
@@ -409,13 +418,21 @@ export default class Interpreter extends EventEmitter {
         await page.close();
       },
       scrape: async (selector?: string) => {
+        if (this.options.debugChannel?.setActionType) {
+          this.options.debugChannel.setActionType('scrape');
+        }
+
         await this.ensureScriptsLoaded(page);
 
         const scrapeResults: Record<string, string>[] = await page.evaluate((s) => window.scrape(s ?? null), selector);
         await this.options.serializableCallback(scrapeResults);
       },
 
       scrapeSchema: async (schema: Record<string, { selector: string; tag: string, attribute: string; shadow: string}>) => {
+        if (this.options.debugChannel?.setActionType) {
+          this.options.debugChannel.setActionType('scrapeSchema');
+        }
+
         await this.ensureScriptsLoaded(page);
 
         const scrapeResult = await page.evaluate((schemaObj) => window.scrapeSchema(schemaObj), schema);
@@ -458,6 +475,10 @@ export default class Interpreter extends EventEmitter {
       },
 
       scrapeList: async (config: { listSelector: string, fields: any, limit?: number, pagination: any }) => {
+        if (this.options.debugChannel?.setActionType) {
+          this.options.debugChannel.setActionType('scrapeList');
+        }
+
         await this.ensureScriptsLoaded(page);
         if (!config.pagination) {
           const scrapeResults: Record<string, any>[] = await page.evaluate((cfg) => window.scrapeList(cfg), config);
@@ -469,6 +490,10 @@ export default class Interpreter extends EventEmitter {
       },
 
       scrapeListAuto: async (config: { listSelector: string }) => {
+        if (this.options.debugChannel?.setActionType) {
+          this.options.debugChannel.setActionType('scrapeListAuto');
+        }
+
         await this.ensureScriptsLoaded(page);
 
         const scrapeResults: { selector: string, innerText: string }[] = await page.evaluate((listSelector) => {
@@ -479,6 +504,10 @@ export default class Interpreter extends EventEmitter {
       },
 
       scroll: async (pages?: number) => {
+        if (this.options.debugChannel?.setActionType) {
+          this.options.debugChannel.setActionType('scroll');
+        }
+
         await page.evaluate(async (pagesInternal) => {
           for (let i = 1; i <= (pagesInternal ?? 1); i += 1) {
             // @ts-ignore
@@ -488,6 +517,10 @@ export default class Interpreter extends EventEmitter {
       },
 
       script: async (code: string) => {
+        if (this.options.debugChannel?.setActionType) {
+          this.options.debugChannel.setActionType('script');
+        }
+
         const AsyncFunction: FunctionConstructor = Object.getPrototypeOf(
           async () => { },
         ).constructor;
@@ -496,6 +529,10 @@ export default class Interpreter extends EventEmitter {
       },
 
       flag: async () => new Promise((res) => {
+        if (this.options.debugChannel?.setActionType) {
+          this.options.debugChannel.setActionType('flag');
+        }
+
         this.emit('flag', page, res);
       }),
     };
@@ -526,6 +563,10 @@ export default class Interpreter extends EventEmitter {
         const params = !step.args || Array.isArray(step.args) ? step.args : [step.args];
         await wawActions[step.action as CustomFunctions](...(params ?? []));
       } else {
+        if (this.options.debugChannel?.setActionType) {
+          this.options.debugChannel.setActionType(String(step.action));
+        }
+
         // Implements the dot notation for the "method name" in the workflow
         const levels = String(step.action).split('.');
         const methodName = levels[levels.length - 1];

server/src/pgboss-worker.ts

@@ -255,7 +255,6 @@ async function processRunExecution(job: Job<ExecuteRunData>) {
         return { success: true };
       }
 
-      // Process the results
       const binaryOutputService = new BinaryOutputService('maxun-run-screenshots');
       const uploadedBinaryOutput = await binaryOutputService.uploadAndStoreBinaryOutput(run, interpretationInfo.binaryOutput);
 
@@ -264,36 +263,57 @@ async function processRunExecution(job: Job<ExecuteRunData>) {
         return { success: true };
       }
 
-      // Update the run record with results
+      const categorizedOutput = {
+        scrapeSchema: interpretationInfo.scrapeSchemaOutput || {},
+        scrapeList: interpretationInfo.scrapeListOutput || {},
+        other: interpretationInfo.otherOutput || {}
+      };
+
       await run.update({
         ...run,
         status: 'success',
         finishedAt: new Date().toLocaleString(),
         browserId: plainRun.browserId,
         log: interpretationInfo.log.join('\n'),
-        serializableOutput: interpretationInfo.serializableOutput,
+        serializableOutput: {
+          scrapeSchema: Object.values(categorizedOutput.scrapeSchema),
+          scrapeList: Object.values(categorizedOutput.scrapeList),
+          other: Object.values(categorizedOutput.other),
+        },
         binaryOutput: uploadedBinaryOutput,
       });
 
       // Track extraction metrics
-      let totalRowsExtracted = 0;
+      let totalSchemaItemsExtracted = 0;
+      let totalListItemsExtracted = 0;
       let extractedScreenshotsCount = 0;
-      let extractedItemsCount = 0;
-
-      if (run.dataValues.binaryOutput && run.dataValues.binaryOutput["item-0"]) {
-        extractedScreenshotsCount = 1;
+
+      if (categorizedOutput.scrapeSchema) {
+        Object.values(categorizedOutput.scrapeSchema).forEach((schemaResult: any) => {
+          if (Array.isArray(schemaResult)) {
+            totalSchemaItemsExtracted += schemaResult.length;
+          } else if (schemaResult && typeof schemaResult === 'object') {
+            totalSchemaItemsExtracted += 1;
+          }
+        });
       }
-
-      if (run.dataValues.serializableOutput && run.dataValues.serializableOutput["item-0"]) {
-        const itemsArray = run.dataValues.serializableOutput["item-0"];
-        extractedItemsCount = itemsArray.length;
-
-        totalRowsExtracted = itemsArray.reduce((total, item) => {
-          return total + Object.keys(item).length;
-        }, 0);
+
+      if (categorizedOutput.scrapeList) {
+        Object.values(categorizedOutput.scrapeList).forEach((listResult: any) => {
+          if (Array.isArray(listResult)) {
+            totalListItemsExtracted += listResult.length;
+          }
+        });
       }
-
-      console.log(`Extracted Items Count: ${extractedItemsCount}`);
+
+      if (uploadedBinaryOutput) {
+        extractedScreenshotsCount = Object.keys(uploadedBinaryOutput).length;
+      }
+
+      const totalRowsExtracted = totalSchemaItemsExtracted + totalListItemsExtracted;
+
+      console.log(`Extracted Schema Items Count: ${totalSchemaItemsExtracted}`);
+      console.log(`Extracted List Items Count: ${totalListItemsExtracted}`);
       console.log(`Extracted Screenshots Count: ${extractedScreenshotsCount}`);
       console.log(`Total Rows Extracted: ${totalRowsExtracted}`);
 
@@ -306,7 +326,8 @@ async function processRunExecution(job: Job<ExecuteRunData>) {
           created_at: new Date().toISOString(),
           status: 'success',
           totalRowsExtracted,
-          extractedItemsCount,
+          schemaItemsExtracted: totalSchemaItemsExtracted,
+          listItemsExtracted: totalListItemsExtracted,
           extractedScreenshotsCount,
         }
       );
@@ -339,7 +360,7 @@ async function processRunExecution(job: Job<ExecuteRunData>) {
         robotName: recording.recording_meta.name,
         status: 'success',
         finishedAt: new Date().toLocaleString()
-      });;
+      });
 
       // Check for and process queued runs before destroying the browser
       const queuedRunProcessed = await checkAndProcessQueuedRun(data.userId, plainRun.browserId);
@@ -458,7 +479,11 @@ async function abortRun(runId: string, userId: string): Promise<boolean> {
     }
 
     let currentLog = 'Run aborted by user';
-    let serializableOutput: Record<string, any> = {};
+    let categorizedOutput = {
+      scrapeSchema: {},
+      scrapeList: {},
+      other: {}
+    };
     let binaryOutput: Record<string, any> = {};
 
     try {
@@ -467,16 +492,16 @@ async function abortRun(runId: string, userId: string): Promise<boolean> {
           currentLog = browser.interpreter.debugMessages.join('\n') || currentLog;
         }
 
-        if (browser.interpreter.serializableData) {
-          browser.interpreter.serializableData.forEach((item, index) => {
-            serializableOutput[`item-${index}`] = item;
-          });
+        if (browser.interpreter.serializableDataByType) {
+          categorizedOutput = {
+            scrapeSchema: collectDataByType(browser.interpreter.serializableDataByType.scrapeSchema || []),
+            scrapeList: collectDataByType(browser.interpreter.serializableDataByType.scrapeList || []),
+            other: collectDataByType(browser.interpreter.serializableDataByType.other || [])
+          };
         }
 
         if (browser.interpreter.binaryData) {
-          browser.interpreter.binaryData.forEach((item, index) => {
-            binaryOutput[`item-${index}`] = item;
-          });
+          binaryOutput = collectBinaryData(browser.interpreter.binaryData);
         }
       }
     } catch (interpreterError) {
@@ -488,7 +513,11 @@ async function abortRun(runId: string, userId: string): Promise<boolean> {
       finishedAt: new Date().toLocaleString(),
       browserId: plainRun.browserId,
       log: currentLog,
-      serializableOutput,
+      serializableOutput: {
+        scrapeSchema: Object.values(categorizedOutput.scrapeSchema),
+        scrapeList: Object.values(categorizedOutput.scrapeList),
+        other: Object.values(categorizedOutput.other),
+      },
       binaryOutput,
     });
 
@@ -529,6 +558,30 @@ async function abortRun(runId: string, userId: string): Promise<boolean> {
   }
 }
 
+/**
+ * Helper function to collect data from arrays into indexed objects
+ * @param dataArray Array of data to be transformed into an object with indexed keys
+ * @returns Object with indexed keys
+ */
+function collectDataByType(dataArray: any[]): Record<string, any> {
+  return dataArray.reduce((result: Record<string, any>, item, index) => {
+    result[`item-${index}`] = item;
+    return result;
+  }, {});
+}
+
+/**
+ * Helper function to collect binary data (like screenshots)
+ * @param binaryDataArray Array of binary data objects to be transformed
+ * @returns Object with indexed keys
+ */
+function collectBinaryData(binaryDataArray: { mimetype: string, data: string, type?: string }[]): Record<string, any> {
+  return binaryDataArray.reduce((result: Record<string, any>, item, index) => {
+    result[`item-${index}`] = item;
+    return result;
+  }, {});
+}
+
 async function registerRunExecutionWorker() {
   try {
     const registeredUserQueues = new Map();

server/src/workflow-management/classes/Interpreter.ts

@@ -87,9 +87,22 @@ export class WorkflowInterpreter {
   public debugMessages: string[] = [];
 
   /**
-   * An array of all the serializable data extracted from the run.
+   * Storage for different types of serializable data
    */
-  public serializableData: string[] = [];
+  public serializableDataByType: {
+    scrapeSchema: any[],
+    scrapeList: any[],
+    other: any[]
+  } = {
+    scrapeSchema: [],
+    scrapeList: [],
+    other: []
+  };
+
+  /**
+   * Track the current action type being processed
+   */
+  private currentActionType: string | null = null;
 
   /**
    * An array of all the binary data extracted from the run.
@@ -181,6 +194,9 @@ export class WorkflowInterpreter {
           this.debugMessages.push(`[${new Date().toLocaleString()}] ` + msg);
           this.socket.emit('log', msg)
         },
+        setActionType: (type: string) => {
+          this.currentActionType = type;
+        }
       },
       serializableCallback: (data: any) => {
         this.socket.emit('serializableCallback', data);
@@ -246,7 +262,12 @@ export class WorkflowInterpreter {
     this.interpreter = null;
     this.breakpoints = [];
     this.interpretationResume = null;
-    this.serializableData = [];
+    this.currentActionType = null;
+    this.serializableDataByType = {
+      scrapeSchema: [],
+      scrapeList: [],
+      other: []
+    };
     this.binaryData = [];
   }
 
@@ -278,9 +299,19 @@ export class WorkflowInterpreter {
           this.debugMessages.push(`[${new Date().toLocaleString()}] ` + msg);
           this.socket.emit('debugMessage', msg)
         },
+        setActionType: (type: string) => {
+          this.currentActionType = type;
+        }
       },
       serializableCallback: (data: any) => {
-        this.serializableData.push(data);
+        if (this.currentActionType === 'scrapeSchema') {
+          this.serializableDataByType.scrapeSchema.push(data);
+        } else if (this.currentActionType === 'scrapeList') {
+          this.serializableDataByType.scrapeList.push(data);
+        } else {
+          this.serializableDataByType.other.push(data);
+        }
+
         this.socket.emit('serializableCallback', data);
       },
       binaryCallback: async (data: string, mimetype: string) => {
@@ -311,14 +342,23 @@ export class WorkflowInterpreter {
 
     const status = await interpreter.run(page, params);
 
-    const lastArray = this.serializableData.length > 1
-    ? [this.serializableData[this.serializableData.length - 1]]
-    : this.serializableData;
-
+    // Structure the output to maintain separate data for each action type
     const result = {
       log: this.debugMessages,
       result: status,
-      serializableOutput: lastArray.reduce((reducedObject, item, index) => {
+      scrapeSchemaOutput: this.serializableDataByType.scrapeSchema.reduce((reducedObject, item, index) => {
+        return {
+          [`schema-${index}`]: item,
+          ...reducedObject,
+        }
+      }, {}),
+      scrapeListOutput: this.serializableDataByType.scrapeList.reduce((reducedObject, item, index) => {
+        return {
+          [`list-${index}`]: item,
+          ...reducedObject,
+        }
+      }, {}),
+      otherOutput: this.serializableDataByType.other.reduce((reducedObject, item, index) => {
         return {
           [`item-${index}`]: item,
           ...reducedObject,