fix(core): clean up event listener properly (#9375)

Author: christian-bromann

libs/langchain-classic/src/chains/base.ts

@@ -106,16 +106,28 @@ export abstract class BaseChain<
     );
     let outputValues: RunOutput;
     try {
-      outputValues = await (fullValues.signal
-        ? (Promise.race([
-            this._call(fullValues as RunInput, runManager, config),
-            new Promise((_, reject) => {
-              fullValues.signal?.addEventListener("abort", () => {
-                reject(new Error("AbortError"));
-              });
-            }),
-          ]) as Promise<RunOutput>)
-        : this._call(fullValues as RunInput, runManager, config));
+      if (fullValues.signal) {
+        let listener: (() => void) | undefined;
+        outputValues = (await Promise.race([
+          this._call(fullValues as RunInput, runManager, config),
+          new Promise<never>((_, reject) => {
+            listener = () => {
+              reject(new Error("AbortError"));
+            };
+            fullValues.signal?.addEventListener("abort", listener);
+          }),
+        ]).finally(() => {
+          if (fullValues.signal && listener) {
+            fullValues.signal.removeEventListener("abort", listener);
+          }
+        })) as RunOutput;
+      } else {
+        outputValues = await this._call(
+          fullValues as RunInput,
+          runManager,
+          config
+        );
+      }
     } catch (e) {
       await runManager?.handleChainError(e);
       throw e;

libs/langchain-core/src/tools/index.ts

@@ -783,10 +783,19 @@ export function tool<
     schema,
     func: async (input, runManager, config) => {
       return new Promise<ToolOutputT>((resolve, reject) => {
+        let listener: (() => void) | undefined;
+        const cleanup = () => {
+          if (config?.signal && listener) {
+            config.signal.removeEventListener("abort", listener);
+          }
+        };
+
         if (config?.signal) {
-          config.signal.addEventListener("abort", () => {
-            return reject(getAbortSignalError(config.signal));
-          });
+          listener = () => {
+            cleanup();
+            reject(getAbortSignalError(config.signal));
+          };
+          config.signal.addEventListener("abort", listener);
         }
 
         const childConfig = patchConfig(config, {
@@ -805,11 +814,14 @@ export function tool<
                * as the promise is already rejected.
                */
               if (config?.signal?.aborted) {
+                cleanup();
                 return;
               }
 
+              cleanup();
               resolve(result);
             } catch (e) {
+              cleanup();
               reject(e);
             }
           }

libs/langchain-core/src/utils/async_caller.ts

@@ -140,14 +140,20 @@ export class AsyncCaller {
     // Note this doesn't cancel the underlying request,
     // when available prefer to use the signal option of the underlying call
     if (options.signal) {
+      let listener: (() => void) | undefined;
       return Promise.race([
         this.call<A, T>(callable, ...args),
         new Promise<never>((_, reject) => {
-          options.signal?.addEventListener("abort", () => {
+          listener = () => {
             reject(getAbortSignalError(options.signal));
-          });
+          };
+          options.signal?.addEventListener("abort", listener);
         }),
-      ]);
+      ]).finally(() => {
+        if (options.signal && listener) {
+          options.signal.removeEventListener("abort", listener);
+        }
+      });
     }
     return this.call<A, T>(callable, ...args);
   }