feat: add notificationThrottleMs and notificationMinChunkSize to AbstractAgent

integrations/langgraph/python/ag_ui_langgraph/agent.py

@@ -286,10 +286,16 @@ async def _handle_stream_events(self, input: RunAgentInput) -> AsyncGenerator[st
                             )
                         )
 
-                yield self._dispatch_event(
-                    RawEvent(type=EventType.RAW, event=event)
+                should_emit_raw = event.get("metadata", {}).get("emit-raw-events", True)
+                self.active_run["emit_raw_event_data"] = event.get("metadata", {}).get(
+                    "emit-raw-event-data", self.active_run.get("emit_raw_event_data", True)
                 )
 
+                if should_emit_raw:
+                    yield self._dispatch_event(
+                        RawEvent(type=EventType.RAW, event=event)
+                    )
+
                 async for single_event in self._handle_single_event(event, state):
                     yield single_event
 

integrations/langgraph/typescript/src/agent.ts

@@ -176,6 +176,12 @@ export class LangGraphAgent extends AbstractAgent {
   }
 
   dispatchEvent(event: ProcessedEvents) {
+    if (event.type !== EventType.RAW && event.rawEvent !== undefined) {
+      const emitRawData = this.activeRun?.emitRawEventData ?? true;
+      if (!emitRawData) {
+        delete event.rawEvent;
+      }
+    }
     this.subscriber.next(event);
     return true;
   }
@@ -609,10 +615,18 @@ export class LangGraphAgent extends AbstractAgent {
           );
         }
 
-        this.dispatchEvent({
-          type: EventType.RAW,
-          event: chunkData,
-        });
+        const shouldEmitRaw = chunkData.metadata?.["emit-raw-events"] ?? true;
+        const emitRawEventData = chunkData.metadata?.["emit-raw-event-data"];
+        if (emitRawEventData !== undefined) {
+          this.activeRun!.emitRawEventData = emitRawEventData;
+        }
+
+        if (shouldEmitRaw) {
+          this.dispatchEvent({
+            type: EventType.RAW,
+            event: chunkData,
+          });
+        }
 
         this.handleSingleEvent(chunkData);
       }

integrations/langgraph/typescript/src/types.ts

@@ -71,6 +71,8 @@ export interface RunMetadata {
   serverRunIdKnown?: boolean;
   // True after a PredictState event is emitted; cleared on OnToolEnd
   hasPredictState?: boolean;
+  // When false, rawEvent is stripped from non-RAW dispatched events
+  emitRawEventData?: boolean;
 }
 
 export type MessagesInProgressRecord = Record<string, MessageInProgress | null>;

sdks/typescript/packages/client/src/agent/__tests__/agent-mutations.test.ts

@@ -495,4 +495,37 @@ describe("Agent Mutations", () => {
       expect(subscriber3.onNewToolCall).toBeUndefined();
     });
   });
+
+  describe("subscriber error isolation", () => {
+    it("addMessage: throwing subscriber does not cause unhandled rejection and next subscriber still fires", async () => {
+      const calls: string[] = [];
+
+      agent.subscribe({
+        onMessagesChanged: () => {
+          throw new Error("subscriber boom");
+        },
+      });
+
+      agent.subscribe({
+        onMessagesChanged: () => {
+          calls.push("reached");
+        },
+      });
+
+      const errorSpy = vi.spyOn(console, "error").mockImplementation(() => {});
+
+      agent.addMessage({ id: "m1", role: "user", content: "hi" } as any);
+
+      // Wait for async IIFE to complete
+      await new Promise((r) => setTimeout(r, 10));
+
+      expect(calls).toContain("reached");
+      expect(errorSpy).toHaveBeenCalledWith(
+        expect.stringContaining("AG-UI: Subscriber"),
+        expect.any(Error),
+      );
+
+      errorSpy.mockRestore();
+    });
+  });
 });