diff --git a/integrations/langgraph/python/ag_ui_langgraph/agent.py b/integrations/langgraph/python/ag_ui_langgraph/agent.py
index c837426b8b..f61c0aea81 100644
--- a/integrations/langgraph/python/ag_ui_langgraph/agent.py
+++ b/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
 
diff --git a/integrations/langgraph/python/uv.lock b/integrations/langgraph/python/uv.lock
index 81128b2b61..f90ae383bd 100644
--- a/integrations/langgraph/python/uv.lock
+++ b/integrations/langgraph/python/uv.lock
@@ -1,10 +1,10 @@
 version = 1
-revision = 2
+revision = 3
 requires-python = ">=3.10, <3.14"
 
 [[package]]
 name = "ag-ui-langgraph"
-version = "0.0.25"
+version = "0.0.28"
 source = { editable = "." }
 dependencies = [
     { name = "ag-ui-protocol" },
@@ -28,7 +28,7 @@ dev = [
 requires-dist = [
     { name = "ag-ui-protocol", specifier = ">=0.1.10" },
     { name = "fastapi", marker = "extra == 'fastapi'", specifier = ">=0.115.12" },
-    { name = "langchain", specifier = ">=0.3.0" },
+    { name = "langchain", specifier = ">=1.2.0" },
     { name = "langchain-core", specifier = ">=0.3.0" },
     { name = "langgraph", specifier = ">=0.3.25,<1.1.0" },
     { name = "pydantic", specifier = ">=2.0.0" },
diff --git a/integrations/langgraph/typescript/src/agent.ts b/integrations/langgraph/typescript/src/agent.ts
index b5b2faebe1..b8b9f5b8ec 100644
--- a/integrations/langgraph/typescript/src/agent.ts
+++ b/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);
       }
diff --git a/integrations/langgraph/typescript/src/types.ts b/integrations/langgraph/typescript/src/types.ts
index fc51ea4af5..c63826e7d8 100644
--- a/integrations/langgraph/typescript/src/types.ts
+++ b/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>;
diff --git a/sdks/typescript/packages/client/src/agent/__tests__/agent-mutations.test.ts b/sdks/typescript/packages/client/src/agent/__tests__/agent-mutations.test.ts
index 5d18673a44..27d4a3f5f1 100644
--- a/sdks/typescript/packages/client/src/agent/__tests__/agent-mutations.test.ts
+++ b/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();
+    });
+  });
 });
diff --git a/sdks/typescript/packages/client/src/agent/__tests__/notification-throttle.test.ts b/sdks/typescript/packages/client/src/agent/__tests__/notification-throttle.test.ts
new file mode 100644
index 0000000000..1e90b57020
--- /dev/null
+++ b/sdks/typescript/packages/client/src/agent/__tests__/notification-throttle.test.ts
@@ -0,0 +1,583 @@
+import { describe, it, expect, vi } from "vitest";
+import { Observable, Subject } from "rxjs";
+import { AbstractAgent } from "../agent";
+import { BaseEvent, RunAgentInput, EventType } from "@ag-ui/core";
+
+class TestAgent extends AbstractAgent {
+  public subject = new Subject<BaseEvent>();
+
+  run(_input: RunAgentInput): Observable<BaseEvent> {
+    return this.subject.asObservable();
+  }
+}
+
+/** Wait one macrotask so runAgent's pipeline has subscribed to the subject */
+const tick = () => new Promise((r) => setTimeout(r, 0));
+
+describe("AbstractAgent notification throttle", () => {
+  // ── Baseline (no throttle) ──────────────────────────────────────────
+
+  it("without throttle config, onMessagesChanged fires for every chunk", async () => {
+    const agent = new TestAgent();
+    const calls: number[] = [];
+
+    agent.subscribe({
+      onMessagesChanged: ({ messages }) => {
+        calls.push(messages.length);
+      },
+    });
+
+    const runPromise = agent.runAgent();
+    await tick();
+
+    agent.subject.next({ type: EventType.RUN_STARTED } as BaseEvent);
+    for (let i = 0; i < 5; i++) {
+      agent.subject.next({
+        type: EventType.TEXT_MESSAGE_CHUNK,
+        messageId: "m1",
+        delta: `chunk${i} `,
+      } as BaseEvent);
+    }
+    agent.subject.next({ type: EventType.RUN_FINISHED } as BaseEvent);
+    agent.subject.complete();
+
+    await runPromise;
+
+    expect(calls.length).toBeGreaterThanOrEqual(5);
+  });
+
+  // ── Time-based throttle ─────────────────────────────────────────────
+
+  it("with intervalMs, fewer onMessagesChanged calls than chunks", async () => {
+    const agent = new TestAgent({ notificationThrottle: { intervalMs: 50 } });
+    const calls: string[] = [];
+
+    agent.subscribe({
+      onMessagesChanged: ({ messages }) => {
+        const msg = messages[0];
+        const content = msg?.role === "assistant" && typeof msg.content === "string" ? msg.content : "";
+        calls.push(content);
+      },
+    });
+
+    const runPromise = agent.runAgent();
+    await tick();
+
+    agent.subject.next({ type: EventType.RUN_STARTED } as BaseEvent);
+    for (let i = 0; i < 20; i++) {
+      agent.subject.next({
+        type: EventType.TEXT_MESSAGE_CHUNK,
+        messageId: "m1",
+        delta: String.fromCharCode(65 + i),
+      } as BaseEvent);
+    }
+    agent.subject.next({ type: EventType.RUN_FINISHED } as BaseEvent);
+    agent.subject.complete();
+
+    await runPromise;
+
+    expect(calls.length).toBeLessThan(20);
+    expect(calls[calls.length - 1]).toBe("ABCDEFGHIJKLMNOPQRST");
+  });
+
+  // ── Chunk-size throttle ─────────────────────────────────────────────
+
+  it("with minChunkSize, notifications wait until enough chars accumulate", async () => {
+    const agent = new TestAgent({
+      notificationThrottle: { intervalMs: 500, minChunkSize: 10 },
+    });
+    const calls: string[] = [];
+
+    agent.subscribe({
+      onMessagesChanged: ({ messages }) => {
+        const msg = messages[0];
+        const content = msg?.role === "assistant" && typeof msg.content === "string" ? msg.content : "";
+        calls.push(content);
+      },
+    });
+
+    const runPromise = agent.runAgent();
+    await tick();
+
+    agent.subject.next({ type: EventType.RUN_STARTED } as BaseEvent);
+    for (let i = 0; i < 20; i++) {
+      agent.subject.next({
+        type: EventType.TEXT_MESSAGE_CHUNK,
+        messageId: "m1",
+        delta: String.fromCharCode(65 + i),
+      } as BaseEvent);
+    }
+    agent.subject.next({ type: EventType.RUN_FINISHED } as BaseEvent);
+    agent.subject.complete();
+
+    await runPromise;
+
+    // minChunkSize=10: 20 single-char chunks → ~2-3 notifications
+    expect(calls.length).toBeLessThanOrEqual(4);
+    expect(calls.length).toBeGreaterThanOrEqual(1);
+    expect(calls[calls.length - 1]).toBe("ABCDEFGHIJKLMNOPQRST");
+  });
+
+  // ── Leading edge fires immediately ──────────────────────────────────
+
+  it("with large throttle window, coalesces into leading + trailing notifications", async () => {
+    const agent = new TestAgent({
+      notificationThrottle: { intervalMs: 5000 },
+    });
+    const calls: string[] = [];
+
+    agent.subscribe({
+      onMessagesChanged: ({ messages }) => {
+        const msg = messages[0];
+        const content = msg?.role === "assistant" && typeof msg.content === "string" ? msg.content : "";
+        calls.push(content);
+      },
+    });
+
+    const runPromise = agent.runAgent();
+    await tick();
+
+    agent.subject.next({ type: EventType.RUN_STARTED } as BaseEvent);
+    agent.subject.next({
+      type: EventType.TEXT_MESSAGE_CHUNK,
+      messageId: "m1",
+      delta: "hello",
+    } as BaseEvent);
+    agent.subject.next({
+      type: EventType.TEXT_MESSAGE_CHUNK,
+      messageId: "m1",
+      delta: " world",
+    } as BaseEvent);
+    agent.subject.next({ type: EventType.RUN_FINISHED } as BaseEvent);
+    agent.subject.complete();
+
+    await runPromise;
+
+    // With 5s window, all events land within it → leading edge + finalize flush
+    expect(calls.length).toBeGreaterThanOrEqual(1);
+    expect(calls.length).toBeLessThanOrEqual(3);
+    // Final notification must contain full content
+    expect(calls[calls.length - 1]).toBe("hello world");
+  });
+
+  // ── agent.messages stays current even when notification is deferred ─
+
+  it("agent.messages is always up-to-date even between throttled notifications", async () => {
+    const agent = new TestAgent({
+      notificationThrottle: { intervalMs: 5000 },
+    });
+    const notificationContents: string[] = [];
+
+    agent.subscribe({
+      onMessagesChanged: ({ messages }) => {
+        const msg = messages[0];
+        const content = msg?.role === "assistant" && typeof msg.content === "string" ? msg.content : "";
+        notificationContents.push(content);
+      },
+    });
+
+    const runPromise = agent.runAgent();
+    await tick();
+
+    agent.subject.next({ type: EventType.RUN_STARTED } as BaseEvent);
+    for (let i = 0; i < 10; i++) {
+      agent.subject.next({
+        type: EventType.TEXT_MESSAGE_CHUNK,
+        messageId: "m1",
+        delta: String.fromCharCode(65 + i),
+      } as BaseEvent);
+    }
+    agent.subject.next({ type: EventType.RUN_FINISHED } as BaseEvent);
+    agent.subject.complete();
+
+    await runPromise;
+
+    // Final notification must have all content
+    expect(notificationContents[notificationContents.length - 1]).toBe("ABCDEFGHIJ");
+    // agent.messages was current at the time of the finalize notification
+    expect(agent.messages[0]).toBeDefined();
+    const finalMsg = agent.messages[0];
+    const finalContent = finalMsg?.role === "assistant" && typeof finalMsg.content === "string" ? finalMsg.content : "";
+    expect(finalContent).toBe("ABCDEFGHIJ");
+  });
+
+  // ── State change notifications under throttle ───────────────────────
+
+  it("onStateChanged is throttled and flushed correctly", async () => {
+    const agent = new TestAgent({
+      notificationThrottle: { intervalMs: 50 },
+    });
+    const stateCalls: any[] = [];
+
+    agent.subscribe({
+      onStateChanged: ({ state }) => {
+        stateCalls.push(structuredClone(state));
+      },
+    });
+
+    const runPromise = agent.runAgent();
+    await tick();
+
+    agent.subject.next({ type: EventType.RUN_STARTED } as BaseEvent);
+    agent.subject.next({
+      type: EventType.STATE_SNAPSHOT,
+      snapshot: { count: 1 },
+    } as BaseEvent);
+    agent.subject.next({
+      type: EventType.STATE_SNAPSHOT,
+      snapshot: { count: 2 },
+    } as BaseEvent);
+    agent.subject.next({
+      type: EventType.STATE_SNAPSHOT,
+      snapshot: { count: 3 },
+    } as BaseEvent);
+    agent.subject.next({ type: EventType.RUN_FINISHED } as BaseEvent);
+    agent.subject.complete();
+
+    await runPromise;
+
+    // Should have coalesced, but final state must be { count: 3 }
+    expect(stateCalls.length).toBeGreaterThanOrEqual(1);
+    expect(stateCalls.length).toBeLessThanOrEqual(3);
+    expect(stateCalls[stateCalls.length - 1]).toEqual({ count: 3 });
+  });
+
+  // ── Subscriber error does not crash the pipeline ────────────────────
+
+  it("subscriber error in throttled path is caught and does not crash", async () => {
+    const agent = new TestAgent({
+      notificationThrottle: { intervalMs: 50 },
+    });
+    const goodCalls: string[] = [];
+    const consoleErrorSpy = vi.spyOn(console, "error").mockImplementation(() => {});
+
+    agent.subscribe({
+      onMessagesChanged: () => {
+        throw new Error("boom");
+      },
+    });
+    agent.subscribe({
+      onMessagesChanged: ({ messages }) => {
+        const msg = messages[0];
+        const content = msg?.role === "assistant" && typeof msg.content === "string" ? msg.content : "";
+        goodCalls.push(content);
+      },
+    });
+
+    const runPromise = agent.runAgent();
+    await tick();
+
+    agent.subject.next({ type: EventType.RUN_STARTED } as BaseEvent);
+    agent.subject.next({
+      type: EventType.TEXT_MESSAGE_CHUNK,
+      messageId: "m1",
+      delta: "hello",
+    } as BaseEvent);
+    agent.subject.next({ type: EventType.RUN_FINISHED } as BaseEvent);
+    agent.subject.complete();
+
+    await runPromise;
+
+    // The second (good) subscriber still received notifications
+    expect(goodCalls.length).toBeGreaterThanOrEqual(1);
+    expect(goodCalls[goodCalls.length - 1]).toBe("hello");
+    // The error was logged
+    expect(consoleErrorSpy).toHaveBeenCalledWith(
+      expect.stringContaining("AG-UI: Subscriber onMessagesChanged threw"),
+      expect.any(Error),
+    );
+
+    consoleErrorSpy.mockRestore();
+  });
+
+  // ── Clone preserves throttle config ─────────────────────────────────
+
+  it("clone() preserves notificationThrottle config", () => {
+    const agent = new TestAgent({
+      notificationThrottle: { intervalMs: 42, minChunkSize: 10 },
+    });
+
+    const cloned = agent.clone();
+
+    expect(cloned.notificationThrottle).toEqual({ intervalMs: 42, minChunkSize: 10 });
+    // Should be a separate object
+    expect(cloned.notificationThrottle).not.toBe(agent.notificationThrottle);
+  });
+
+  it("clone() preserves undefined notificationThrottle", () => {
+    const agent = new TestAgent();
+    const cloned = agent.clone();
+    expect(cloned.notificationThrottle).toBeUndefined();
+  });
+
+  // ── Input validation ────────────────────────────────────────────────
+
+  it("throws on negative intervalMs", () => {
+    expect(() => new TestAgent({ notificationThrottle: { intervalMs: -1 } })).toThrow(
+      "non-negative finite number",
+    );
+  });
+
+  it("throws on NaN intervalMs", () => {
+    expect(() => new TestAgent({ notificationThrottle: { intervalMs: NaN } })).toThrow(
+      "non-negative finite number",
+    );
+  });
+
+  it("throws on Infinity intervalMs", () => {
+    expect(() => new TestAgent({ notificationThrottle: { intervalMs: Infinity } })).toThrow(
+      "non-negative finite number",
+    );
+  });
+
+  it("throws on negative minChunkSize", () => {
+    expect(
+      () => new TestAgent({ notificationThrottle: { intervalMs: 16, minChunkSize: -5 } }),
+    ).toThrow("non-negative finite number");
+  });
+
+  it("intervalMs: 0 with no minChunkSize skips throttle activation", () => {
+    const agent = new TestAgent({ notificationThrottle: { intervalMs: 0 } });
+    // Zero-zero is a no-op — treated as unthrottled
+    expect(agent.notificationThrottle).toBeUndefined();
+  });
+
+  it("intervalMs: 0 with minChunkSize > 0 still activates throttle", () => {
+    const agent = new TestAgent({ notificationThrottle: { intervalMs: 0, minChunkSize: 10 } });
+    expect(agent.notificationThrottle).toEqual({ intervalMs: 0, minChunkSize: 10 });
+  });
+
+  // ── Trailing timer fires mid-stream (Issue 6) ───────────────────────
+
+  it("trailing timer fires pending notification mid-stream", async () => {
+    vi.useFakeTimers();
+    try {
+      const agent = new TestAgent({ notificationThrottle: { intervalMs: 50 } });
+      const calls: string[] = [];
+
+      agent.subscribe({
+        onMessagesChanged: ({ messages }) => {
+          const msg = messages[0];
+          const content =
+            msg?.role === "assistant" && typeof msg.content === "string" ? msg.content : "";
+          calls.push(content);
+        },
+      });
+
+      const runPromise = agent.runAgent();
+      await vi.advanceTimersByTimeAsync(0);
+
+      agent.subject.next({ type: EventType.RUN_STARTED } as BaseEvent);
+      agent.subject.next({
+        type: EventType.TEXT_MESSAGE_CHUNK,
+        messageId: "m1",
+        delta: "A",
+      } as BaseEvent);
+      agent.subject.next({
+        type: EventType.TEXT_MESSAGE_CHUNK,
+        messageId: "m1",
+        delta: "B",
+      } as BaseEvent);
+
+      const callsBeforeTimer = calls.length;
+
+      // Advance past the throttle window — trailing timer should fire
+      await vi.advanceTimersByTimeAsync(60);
+
+      // Trailing timer should have fired, producing at least one more notification
+      expect(calls.length).toBeGreaterThan(callsBeforeTimer);
+      expect(calls[calls.length - 1]).toBe("AB");
+
+      agent.subject.next({ type: EventType.RUN_FINISHED } as BaseEvent);
+      agent.subject.complete();
+      await vi.advanceTimersByTimeAsync(0);
+      await runPromise;
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
+  // ── onStateChanged subscriber error (Issue 7) ──────────────────────
+
+  it("onStateChanged subscriber error is caught and does not crash", async () => {
+    const agent = new TestAgent({
+      notificationThrottle: { intervalMs: 50 },
+    });
+    const goodCalls: any[] = [];
+    const consoleErrorSpy = vi.spyOn(console, "error").mockImplementation(() => {});
+
+    agent.subscribe({
+      onStateChanged: () => {
+        throw new Error("state boom");
+      },
+    });
+    agent.subscribe({
+      onStateChanged: ({ state }) => {
+        goodCalls.push(structuredClone(state));
+      },
+    });
+
+    const runPromise = agent.runAgent();
+    await tick();
+
+    agent.subject.next({ type: EventType.RUN_STARTED } as BaseEvent);
+    agent.subject.next({
+      type: EventType.STATE_SNAPSHOT,
+      snapshot: { count: 1 },
+    } as BaseEvent);
+    agent.subject.next({ type: EventType.RUN_FINISHED } as BaseEvent);
+    agent.subject.complete();
+
+    await runPromise;
+
+    expect(goodCalls.length).toBeGreaterThanOrEqual(1);
+    expect(goodCalls[goodCalls.length - 1]).toEqual({ count: 1 });
+    expect(consoleErrorSpy).toHaveBeenCalledWith(
+      expect.stringContaining("AG-UI: Subscriber onStateChanged threw"),
+      expect.any(Error),
+    );
+
+    consoleErrorSpy.mockRestore();
+  });
+
+  // ── Interleaved message IDs with minChunkSize (Issue 8) ─────────────
+
+  it("minChunkSize resets tracking when message identity changes", async () => {
+    const agent = new TestAgent({
+      notificationThrottle: { intervalMs: 5000, minChunkSize: 5 },
+    });
+    const calls: number[] = [];
+
+    agent.subscribe({
+      onMessagesChanged: () => {
+        calls.push(calls.length);
+      },
+    });
+
+    const runPromise = agent.runAgent();
+    await tick();
+
+    agent.subject.next({ type: EventType.RUN_STARTED } as BaseEvent);
+
+    // Emit 4 chars on m1 (below minChunkSize=5)
+    for (let i = 0; i < 4; i++) {
+      agent.subject.next({
+        type: EventType.TEXT_MESSAGE_CHUNK,
+        messageId: "m1",
+        delta: String.fromCharCode(65 + i),
+      } as BaseEvent);
+    }
+
+    // Switch to m2 — 6 chars (above minChunkSize=5, should trigger notification)
+    for (let i = 0; i < 6; i++) {
+      agent.subject.next({
+        type: EventType.TEXT_MESSAGE_CHUNK,
+        messageId: "m2",
+        delta: String.fromCharCode(75 + i),
+      } as BaseEvent);
+    }
+
+    agent.subject.next({ type: EventType.RUN_FINISHED } as BaseEvent);
+    agent.subject.complete();
+
+    await runPromise;
+
+    // Leading edge on first m1 chunk. After identity change to m2, tracking resets.
+    // After 5+ chars on m2, chunk threshold fires. Then finalize flush.
+    // Should see 2-4 notifications, not 10+ (one per event).
+    expect(calls.length).toBeGreaterThanOrEqual(2);
+    expect(calls.length).toBeLessThanOrEqual(5);
+  });
+
+  // ── No flush on stream error (Issue 9) ──────────────────────────────
+
+  it("does not flush pending notifications on stream error", async () => {
+    const agent = new TestAgent({
+      notificationThrottle: { intervalMs: 5000 },
+    });
+    const calls: string[] = [];
+    const consoleErrorSpy = vi.spyOn(console, "error").mockImplementation(() => {});
+
+    agent.subscribe({
+      onMessagesChanged: ({ messages }) => {
+        const msg = messages[0];
+        const content =
+          msg?.role === "assistant" && typeof msg.content === "string" ? msg.content : "";
+        calls.push(content);
+      },
+    });
+
+    const runPromise = agent.runAgent();
+    await tick();
+
+    agent.subject.next({ type: EventType.RUN_STARTED } as BaseEvent);
+    agent.subject.next({
+      type: EventType.TEXT_MESSAGE_CHUNK,
+      messageId: "m1",
+      delta: "hello",
+    } as BaseEvent);
+    agent.subject.next({
+      type: EventType.TEXT_MESSAGE_CHUNK,
+      messageId: "m1",
+      delta: " world",
+    } as BaseEvent);
+
+    // Snapshot how many notifications fired before the error
+    const callsBeforeError = calls.length;
+
+    // Error the stream — pending notifications should NOT be flushed
+    agent.subject.error(new Error("stream error"));
+    await runPromise.catch(() => {});
+
+    // No additional notification was flushed after the error
+    expect(calls.length).toBe(callsBeforeError);
+
+    consoleErrorSpy.mockRestore();
+  });
+
+  // ── Non-throttled subscriber error resilience ───────────────────────
+
+  it("without throttle, a throwing subscriber does not crash the pipeline", async () => {
+    const agent = new TestAgent();
+    const calls: number[] = [];
+
+    // First subscriber throws
+    agent.subscribe({
+      onMessagesChanged: () => {
+        throw new Error("subscriber boom");
+      },
+    });
+
+    // Second subscriber should still receive notifications
+    agent.subscribe({
+      onMessagesChanged: ({ messages }) => {
+        calls.push(messages.length);
+      },
+    });
+
+    const errorSpy = vi.spyOn(console, "error").mockImplementation(() => {});
+
+    const runPromise = agent.runAgent();
+    await tick();
+
+    agent.subject.next({ type: EventType.RUN_STARTED } as BaseEvent);
+    agent.subject.next({
+      type: EventType.TEXT_MESSAGE_CHUNK,
+      messageId: "m1",
+      delta: "hello",
+    } as BaseEvent);
+    agent.subject.next({ type: EventType.RUN_FINISHED } as BaseEvent);
+    agent.subject.complete();
+
+    await runPromise;
+
+    // Second subscriber was reached despite first throwing
+    expect(calls.length).toBeGreaterThanOrEqual(1);
+    expect(errorSpy).toHaveBeenCalledWith(
+      expect.stringContaining("AG-UI: Subscriber"),
+      expect.any(Error),
+    );
+
+    errorSpy.mockRestore();
+  });
+});
diff --git a/sdks/typescript/packages/client/src/agent/agent.ts b/sdks/typescript/packages/client/src/agent/agent.ts
index 122d5ab503..1ac952756f 100644
--- a/sdks/typescript/packages/client/src/agent/agent.ts
+++ b/sdks/typescript/packages/client/src/agent/agent.ts
@@ -12,6 +12,7 @@ import {
 import {
   AgentConfig,
   AgentDebugConfig,
+  NotificationThrottleConfig,
   RunAgentParameters,
   ResolvedAgentDebugConfig,
   resolveAgentDebugConfig,
@@ -53,6 +54,7 @@ export abstract class AbstractAgent {
   public state: State;
   private _debug: ResolvedAgentDebugConfig;
   private _debugLogger: DebugLogger | undefined;
+  public readonly notificationThrottle: NotificationThrottleConfig | undefined;
   public subscribers: AgentSubscriber[] = [];
   public isRunning: boolean = false;
   private middlewares: Middleware[] = [];
@@ -94,6 +96,7 @@ export abstract class AbstractAgent {
     initialMessages,
     initialState,
     debug,
+    notificationThrottle,
   }: AgentConfig = {}) {
     this.agentId = agentId;
     this.description = description ?? "";
@@ -103,6 +106,27 @@ export abstract class AbstractAgent {
     this._debug = resolveAgentDebugConfig(debug);
     this._debugLogger = createDebugLogger(this._debug);
 
+    if (notificationThrottle) {
+      const { intervalMs, minChunkSize } = notificationThrottle;
+      if (!Number.isFinite(intervalMs) || intervalMs < 0) {
+        throw new Error(
+          `notificationThrottle.intervalMs must be a non-negative finite number, got ${intervalMs}`,
+        );
+      }
+      if (minChunkSize !== undefined && (!Number.isFinite(minChunkSize) || minChunkSize < 0)) {
+        throw new Error(
+          `notificationThrottle.minChunkSize must be a non-negative finite number, got ${minChunkSize}`,
+        );
+      }
+      // If both thresholds are zero, throttling is a no-op; skip activation
+      if (intervalMs > 0 || (minChunkSize ?? 0) > 0) {
+        this.notificationThrottle = {
+          intervalMs,
+          minChunkSize: minChunkSize ?? 0,
+        };
+      }
+    }
+
     if (compareVersions(this.maxVersion, "0.0.39") <= 0) {
       this.middlewares.unshift(new BackwardCompatibility_0_0_39());
     }
@@ -331,34 +355,260 @@ export abstract class AbstractAgent {
     events$: Observable<AgentStateMutation>,
     subscribers: AgentSubscriber[],
   ): Observable<AgentStateMutation> {
-    return events$.pipe(
+    // Step 1: Always apply mutations immediately (agent.messages/state stay current)
+    const mutated$ = events$.pipe(
+      tap((event) => {
+        if (event.messages) this.messages = event.messages;
+        if (event.state) this.state = event.state;
+      }),
+    );
+
+    // Step 2: Notify subscribers — throttled when configured, immediate otherwise
+    if (this.notificationThrottle) {
+      return this.processThrottledNotifications(mutated$, input, subscribers);
+    }
+
+    return mutated$.pipe(
       tap((event) => {
         if (event.messages) {
-          this.messages = event.messages;
           subscribers.forEach((subscriber) => {
-            subscriber.onMessagesChanged?.({
-              messages: this.messages,
-              state: this.state,
-              agent: this,
-              input,
-            });
+            try {
+              subscriber.onMessagesChanged?.({
+                messages: this.messages,
+                state: this.state,
+                agent: this,
+                input,
+              });
+            } catch (err) {
+              console.error("AG-UI: Subscriber onMessagesChanged threw:", err);
+              this._debugLogger?.lifecycle("LIFECYCLE", "Subscriber onMessagesChanged error:", {
+                error: err instanceof Error ? err.message : String(err),
+              });
+            }
           });
         }
         if (event.state) {
-          this.state = event.state;
           subscribers.forEach((subscriber) => {
-            subscriber.onStateChanged?.({
-              state: this.state,
-              messages: this.messages,
-              agent: this,
-              input,
-            });
+            try {
+              subscriber.onStateChanged?.({
+                state: this.state,
+                messages: this.messages,
+                agent: this,
+                input,
+              });
+            } catch (err) {
+              console.error("AG-UI: Subscriber onStateChanged threw:", err);
+              this._debugLogger?.lifecycle("LIFECYCLE", "Subscriber onStateChanged error:", {
+                error: err instanceof Error ? err.message : String(err),
+              });
+            }
           });
         }
       }),
     );
   }
 
+  /**
+   * Throttled notification layer.
+   *
+   * The first event always fires immediately (leading edge). Subsequent
+   * notifications fire when any condition is met:
+   *   - `intervalMs` has elapsed since the last notification, OR
+   *   - `minChunkSize` new characters have accumulated on the active assistant message
+   *
+   * A trailing timer ensures pending notifications are flushed after each
+   * window. On normal stream completion, any remaining pending notification
+   * is delivered. On stream error, pending notifications are discarded to
+   * avoid delivering potentially inconsistent state.
+   */
+  private processThrottledNotifications(
+    mutated$: Observable<AgentStateMutation>,
+    input: RunAgentInput,
+    subscribers: AgentSubscriber[],
+  ): Observable<AgentStateMutation> {
+    const throttleMs = this.notificationThrottle!.intervalMs;
+    const minChunkSize = this.notificationThrottle!.minChunkSize ?? 0;
+
+    let lastNotifyTime = 0;
+    let charsSinceLastNotify = 0;
+    let lastContentLength = 0;
+    let lastTrackedMessageId: string | null = null;
+    let pendingMessages = false;
+    let pendingState = false;
+    let timerId: ReturnType<typeof setTimeout> | null = null;
+    let disposed = false;
+    let streamErrored = false;
+
+    const notify = (force = false) => {
+      if (disposed && !force) return;
+      try {
+        if (timerId !== null) {
+          clearTimeout(timerId);
+          timerId = null;
+        }
+        lastNotifyTime = Date.now();
+        charsSinceLastNotify = 0;
+
+        // Snapshot the content length of the current trailing assistant message
+        if (this.messages.length > 0) {
+          const lastMsg = this.messages[this.messages.length - 1];
+          if (lastMsg.role === "assistant" && typeof lastMsg.content === "string") {
+            lastContentLength = lastMsg.content.length;
+            lastTrackedMessageId = lastMsg.id;
+          }
+        }
+
+        if (pendingMessages) {
+          pendingMessages = false;
+          subscribers.forEach((subscriber) => {
+            try {
+              subscriber.onMessagesChanged?.({
+                messages: this.messages,
+                state: this.state,
+                agent: this,
+                input,
+              });
+            } catch (err) {
+              console.error(
+                "AG-UI: Subscriber onMessagesChanged threw during throttled notification:",
+                err,
+              );
+              this._debugLogger?.lifecycle(
+                "LIFECYCLE",
+                "Subscriber onMessagesChanged error:",
+                {
+                  error: err instanceof Error ? err.message : String(err),
+                },
+              );
+            }
+          });
+        }
+        if (pendingState) {
+          pendingState = false;
+          subscribers.forEach((subscriber) => {
+            try {
+              subscriber.onStateChanged?.({
+                state: this.state,
+                messages: this.messages,
+                agent: this,
+                input,
+              });
+            } catch (err) {
+              console.error(
+                "AG-UI: Subscriber onStateChanged threw during throttled notification:",
+                err,
+              );
+              this._debugLogger?.lifecycle(
+                "LIFECYCLE",
+                "Subscriber onStateChanged error:",
+                {
+                  error: err instanceof Error ? err.message : String(err),
+                },
+              );
+            }
+          });
+        }
+      } catch (err) {
+        console.error("AG-UI: Unexpected error in throttled notify():", err);
+        this._debugLogger?.lifecycle("LIFECYCLE", "Throttled notify error:", {
+          error: err instanceof Error ? err.message : String(err),
+        });
+      }
+    };
+
+    const scheduleTrailing = () => {
+      if (timerId !== null) return;
+      const elapsed = Date.now() - lastNotifyTime;
+      const remaining = Math.max(0, throttleMs - elapsed);
+      timerId = setTimeout(notify, remaining);
+    };
+
+    return mutated$.pipe(
+      tap({
+        next: (event) => {
+          if (event.messages) {
+            if (minChunkSize > 0 && this.messages.length > 0) {
+              const lastMsg = this.messages[this.messages.length - 1];
+              if (lastMsg.role === "assistant" && typeof lastMsg.content === "string") {
+                // Reset tracking when the message identity changes
+                if (lastMsg.id !== lastTrackedMessageId) {
+                  lastTrackedMessageId = lastMsg.id;
+                  lastContentLength = 0;
+                }
+                charsSinceLastNotify = Math.max(
+                  0,
+                  lastMsg.content.length - lastContentLength,
+                );
+              }
+            }
+            pendingMessages = true;
+          }
+          if (event.state) {
+            pendingState = true;
+          }
+
+          const now = Date.now();
+          // Sentinel: lastNotifyTime is 0 only before the very first notification
+          const isLeading = lastNotifyTime === 0;
+          const timeThresholdMet =
+            throttleMs > 0 && now - lastNotifyTime >= throttleMs;
+          const chunkThresholdMet =
+            minChunkSize > 0 && charsSinceLastNotify >= minChunkSize;
+
+          if (isLeading || timeThresholdMet || chunkThresholdMet) {
+            notify();
+          } else {
+            scheduleTrailing();
+          }
+        },
+        error: () => {
+          streamErrored = true;
+        },
+      }),
+      finalize(() => {
+        disposed = true;
+        if (timerId !== null) {
+          clearTimeout(timerId);
+          timerId = null;
+        }
+        // Only flush on normal completion; skip on error to avoid
+        // delivering potentially inconsistent state to subscribers.
+        if (!streamErrored && (pendingMessages || pendingState)) {
+          if (pendingMessages) {
+            pendingMessages = false;
+            subscribers.forEach((subscriber) => {
+              try {
+                subscriber.onMessagesChanged?.({
+                  messages: this.messages,
+                  state: this.state,
+                  agent: this,
+                  input,
+                });
+              } catch (err) {
+                console.error("AG-UI: Subscriber onMessagesChanged threw during finalize flush:", err);
+              }
+            });
+          }
+          if (pendingState) {
+            pendingState = false;
+            subscribers.forEach((subscriber) => {
+              try {
+                subscriber.onStateChanged?.({
+                  state: this.state,
+                  messages: this.messages,
+                  agent: this,
+                  input,
+                });
+              } catch (err) {
+                console.error("AG-UI: Subscriber onStateChanged threw during finalize flush:", err);
+              }
+            });
+          }
+        }
+      }),
+    );
+  }
+
   protected prepareRunAgentInput(parameters?: RunAgentParameters): RunAgentInput {
     const clonedMessages = structuredClone_(this.messages) as Message[];
     const messagesWithoutActivity = clonedMessages.filter((message) => message.role !== "activity");
@@ -509,6 +759,15 @@ export abstract class AbstractAgent {
     cloned.state = structuredClone_(this.state);
     cloned._debug = this._debug;
     cloned._debugLogger = this._debugLogger;
+    // notificationThrottle is readonly on the class; bypass via Object.defineProperty
+    Object.defineProperty(cloned, "notificationThrottle", {
+      value: this.notificationThrottle
+        ? { ...this.notificationThrottle }
+        : undefined,
+      writable: false,
+      enumerable: true,
+      configurable: true,
+    });
     cloned.isRunning = this.isRunning;
     cloned.subscribers = [...this.subscribers];
     cloned.middlewares = [...this.middlewares];
@@ -524,35 +783,47 @@ export abstract class AbstractAgent {
     (async () => {
       // Fire onNewMessage sequentially
       for (const subscriber of this.subscribers) {
-        await subscriber.onNewMessage?.({
-          message,
-          messages: this.messages,
-          state: this.state,
-          agent: this,
-        });
+        try {
+          await subscriber.onNewMessage?.({
+            message,
+            messages: this.messages,
+            state: this.state,
+            agent: this,
+          });
+        } catch (err) {
+          console.error("AG-UI: Subscriber onNewMessage threw:", err);
+        }
       }
 
       // Fire onNewToolCall if the message is from assistant and contains tool calls
       if (message.role === "assistant" && message.toolCalls) {
         for (const toolCall of message.toolCalls) {
           for (const subscriber of this.subscribers) {
-            await subscriber.onNewToolCall?.({
-              toolCall,
-              messages: this.messages,
-              state: this.state,
-              agent: this,
-            });
+            try {
+              await subscriber.onNewToolCall?.({
+                toolCall,
+                messages: this.messages,
+                state: this.state,
+                agent: this,
+              });
+            } catch (err) {
+              console.error("AG-UI: Subscriber onNewToolCall threw:", err);
+            }
           }
         }
       }
 
       // Fire onMessagesChanged sequentially
       for (const subscriber of this.subscribers) {
-        await subscriber.onMessagesChanged?.({
-          messages: this.messages,
-          state: this.state,
-          agent: this,
-        });
+        try {
+          await subscriber.onMessagesChanged?.({
+            messages: this.messages,
+            state: this.state,
+            agent: this,
+          });
+        } catch (err) {
+          console.error("AG-UI: Subscriber onMessagesChanged threw:", err);
+        }
       }
     })();
   }
@@ -567,24 +838,32 @@ export abstract class AbstractAgent {
       for (const message of messages) {
         // Fire onNewMessage sequentially
         for (const subscriber of this.subscribers) {
-          await subscriber.onNewMessage?.({
-            message,
-            messages: this.messages,
-            state: this.state,
-            agent: this,
-          });
+          try {
+            await subscriber.onNewMessage?.({
+              message,
+              messages: this.messages,
+              state: this.state,
+              agent: this,
+            });
+          } catch (err) {
+            console.error("AG-UI: Subscriber onNewMessage threw:", err);
+          }
         }
 
         // Fire onNewToolCall if the message is from assistant and contains tool calls
         if (message.role === "assistant" && message.toolCalls) {
           for (const toolCall of message.toolCalls) {
             for (const subscriber of this.subscribers) {
-              await subscriber.onNewToolCall?.({
-                toolCall,
-                messages: this.messages,
-                state: this.state,
-                agent: this,
-              });
+              try {
+                await subscriber.onNewToolCall?.({
+                  toolCall,
+                  messages: this.messages,
+                  state: this.state,
+                  agent: this,
+                });
+              } catch (err) {
+                console.error("AG-UI: Subscriber onNewToolCall threw:", err);
+              }
             }
           }
         }
@@ -592,11 +871,15 @@ export abstract class AbstractAgent {
 
       // Fire onMessagesChanged once at the end sequentially
       for (const subscriber of this.subscribers) {
-        await subscriber.onMessagesChanged?.({
-          messages: this.messages,
-          state: this.state,
-          agent: this,
-        });
+        try {
+          await subscriber.onMessagesChanged?.({
+            messages: this.messages,
+            state: this.state,
+            agent: this,
+          });
+        } catch (err) {
+          console.error("AG-UI: Subscriber onMessagesChanged threw:", err);
+        }
       }
     })();
   }
@@ -609,11 +892,15 @@ export abstract class AbstractAgent {
     (async () => {
       // Fire onMessagesChanged sequentially
       for (const subscriber of this.subscribers) {
-        await subscriber.onMessagesChanged?.({
-          messages: this.messages,
-          state: this.state,
-          agent: this,
-        });
+        try {
+          await subscriber.onMessagesChanged?.({
+            messages: this.messages,
+            state: this.state,
+            agent: this,
+          });
+        } catch (err) {
+          console.error("AG-UI: Subscriber onMessagesChanged threw:", err);
+        }
       }
     })();
   }
@@ -626,11 +913,15 @@ export abstract class AbstractAgent {
     (async () => {
       // Fire onStateChanged sequentially
       for (const subscriber of this.subscribers) {
-        await subscriber.onStateChanged?.({
-          messages: this.messages,
-          state: this.state,
-          agent: this,
-        });
+        try {
+          await subscriber.onStateChanged?.({
+            messages: this.messages,
+            state: this.state,
+            agent: this,
+          });
+        } catch (err) {
+          console.error("AG-UI: Subscriber onStateChanged threw:", err);
+        }
       }
     })();
   }
diff --git a/sdks/typescript/packages/client/src/agent/index.ts b/sdks/typescript/packages/client/src/agent/index.ts
index 9ff480c201..1ba3bf9d05 100644
--- a/sdks/typescript/packages/client/src/agent/index.ts
+++ b/sdks/typescript/packages/client/src/agent/index.ts
@@ -4,6 +4,7 @@ export { HttpAgent } from "./http";
 export type {
   AgentConfig,
   HttpAgentConfig,
+  NotificationThrottleConfig,
   RunAgentParameters,
   AgentDebugConfig,
   ResolvedAgentDebugConfig,
diff --git a/sdks/typescript/packages/client/src/agent/types.ts b/sdks/typescript/packages/client/src/agent/types.ts
index d85a95500e..b2f077ea95 100644
--- a/sdks/typescript/packages/client/src/agent/types.ts
+++ b/sdks/typescript/packages/client/src/agent/types.ts
@@ -30,6 +30,36 @@ export function resolveAgentDebugConfig(
   return { enabled: events || lifecycle, events, lifecycle, verbose };
 }
 
+/**
+ * Configuration for throttling subscriber notifications during streaming.
+ *
+ * Mutations are always applied immediately (`agent.messages`/`agent.state`
+ * stay current); only subscriber notifications (`onMessagesChanged`,
+ * `onStateChanged`) are coalesced.
+ *
+ * The first event always fires immediately (leading edge). Subsequent
+ * notifications fire when either threshold is met. A trailing timer
+ * ensures pending notifications are flushed. On stream completion,
+ * any remaining pending notification is always delivered.
+ */
+export interface NotificationThrottleConfig {
+  /**
+   * Time-based throttle window in milliseconds.
+   * Notifications are suppressed for this duration after each delivery;
+   * only the latest state is delivered when the window expires.
+   * Must be a non-negative finite number. Example: `16` ≈ 60 fps.
+   */
+  intervalMs: number;
+  /**
+   * Minimum new characters to accumulate before firing a notification.
+   * When set, a notification also fires when this many new characters
+   * have been appended to the active assistant message, even if the
+   * time window has not yet elapsed.
+   * Must be a non-negative finite number. Default: `0` (no minimum).
+   */
+  minChunkSize?: number;
+}
+
 export interface AgentConfig {
   agentId?: string;
   description?: string;
@@ -37,6 +67,11 @@ export interface AgentConfig {
   initialMessages?: Message[];
   initialState?: State;
   debug?: AgentDebugConfig;
+  /**
+   * Throttle subscriber notifications during streaming.
+   * When omitted, every mutation fires a notification immediately.
+   */
+  notificationThrottle?: NotificationThrottleConfig;
 }
 
 export interface HttpAgentConfig extends AgentConfig {