fix(core): Fix stream events bug when errors are thrown too quickly d…

…uring iteration (#7617)

langchain-core/src/runnables/base.ts

@@ -914,19 +914,48 @@ export abstract class Runnable<
       // eslint-disable-next-line no-param-reassign
       config.callbacks = copiedCallbacks;
     }
+    const abortController = new AbortController();
     // Call the runnable in streaming mode,
     // add each chunk to the output stream
     const outerThis = this;
     async function consumeRunnableStream() {
       try {
-        const runnableStream = await outerThis.stream(input, config);
+        let signal;
+        if (options?.signal) {
+          if ("any" in AbortSignal) {
+            // Use native AbortSignal.any() if available (Node 19+)
+            // eslint-disable-next-line @typescript-eslint/no-explicit-any
+            signal = (AbortSignal as any).any([
+              abortController.signal,
+              options.signal,
+            ]);
+          } else {
+            // Fallback for Node 18 and below - just use the provided signal
+            signal = options.signal;
+            // Ensure we still abort our controller when the parent signal aborts
+            options.signal.addEventListener(
+              "abort",
+              () => {
+                abortController.abort();
+              },
+              { once: true }
+            );
+          }
+        } else {
+          signal = abortController.signal;
+        }
+        const runnableStream = await outerThis.stream(input, {
+          ...config,
+          signal,
+        });
         const tappedStream = eventStreamer.tapOutputIterable(
           runId,
           runnableStream
         );
         // eslint-disable-next-line @typescript-eslint/no-unused-vars
         for await (const _ of tappedStream) {
           // Just iterate so that the callback handler picks up events
+          if (abortController.signal.aborted) break;
         }
       } finally {
         await eventStreamer.finish();
@@ -959,6 +988,7 @@ export abstract class Runnable<
         yield event;
       }
     } finally {
+      abortController.abort();
       await runnableStreamConsumePromise;
     }
   }

langchain-core/src/runnables/tests/runnable_stream_events_v2.test.ts

@@ -2263,3 +2263,23 @@ test("Runnable streamEvents method should respect passed signal", async () => {
     }
   }).rejects.toThrowError();
 });
+
+test("streamEvents method handles errors", async () => {
+  let caughtError: unknown;
+  const model = new FakeListChatModel({
+    responses: ["abc"],
+  });
+
+  try {
+    // eslint-disable-next-line no-unreachable-loop
+    for await (const _ of model.streamEvents("Hello! Tell me about yourself.", {
+      version: "v2",
+    })) {
+      throw new Error("should catch this error");
+    }
+  } catch (e) {
+    caughtError = e;
+  }
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  expect((caughtError as any)?.message).toEqual("should catch this error");
+});