feat: add EventThrottleMiddleware for stream notification throttling - #1423
Merged
Conversation
…ractAgent Add configurable notification throttling to AbstractAgent that batches subscriber notifications during high-frequency event streams. Includes notifySubscribers helper for safe, deduplicated subscriber dispatch, throttle config types, and comprehensive tests. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
|
The latest updates on your projects. Learn more about Vercel for GitHub.
|
Contributor
Python Preview PackagesVersion
Install with uvAdd the TestPyPI index to your [[tool.uv.index]]
name = "testpypi"
url = "https://test.pypi.org/simple/"
explicit = trueThen install the packages you need: # Core SDK
uv add 'ag-ui-protocol==0.0.0.dev1775160689' --index testpypi
# Integrations (each already depends on the matching ag-ui-protocol preview)
uv add 'ag-ui-langgraph==0.0.0.dev1775160689' --index testpypi
uv add 'ag-ui-crewai==0.0.0.dev1775160689' --index testpypi
# NOTE: ag-ui-agent-spec depends on pyagentspec (git-only, not on PyPI).
# You will need to install pyagentspec separately from its git repo.
uv add 'ag-ui-agent-spec==0.0.0.dev1775160689' --index testpypi
uv add 'ag_ui_adk==0.0.0.dev1775160689' --index testpypi
uv add 'ag_ui_strands==0.0.0.dev1775160689' --index testpypiInstall with pippip install \
--index-url https://test.pypi.org/simple/ \
--extra-index-url https://pypi.org/simple/ \
ag-ui-protocol==0.0.0.dev1775160689
Commit: 98b5377 |
@ag-ui/a2a-middleware
@ag-ui/a2ui-middleware
@ag-ui/event-throttle-middleware
@ag-ui/mcp-apps-middleware
@ag-ui/middleware-starter
@ag-ui/a2a
@ag-ui/adk
@ag-ui/ag2
@ag-ui/agno
@ag-ui/aws-strands
@ag-ui/claude-agent-sdk
@ag-ui/crewai
@ag-ui/langchain
@ag-ui/langgraph
@ag-ui/langroid
@ag-ui/llamaindex
@ag-ui/mastra
@ag-ui/pydantic-ai
@ag-ui/server-starter
@ag-ui/server-starter-all-features
@ag-ui/vercel-ai-sdk
create-ag-ui-app
@ag-ui/client
@ag-ui/core
@ag-ui/encoder
@ag-ui/proto
commit: |
…er snapshots - Apply notifySubscribers to onInitialize/onError/onFinalize (6 bare forEach calls → isolated per-subscriber error handling) - Replace streamErrored flag with streamCompleted (track success, not failure — fixes edge case where error inside tap.next bypasses tap.error) - Remove broad catch from notify() — bookkeeping errors are programming bugs that should propagate; notifySubscribers already isolates subscriber errors - Add .catch() to void this.onFinalize() to prevent unhandled rejections - Snapshot subscriber list in addMessage/addMessages/setMessages/setState to prevent mid-iteration mutation - Add .catch() to all fire-and-forget async IIFEs - Wrap scheduleTrailing timer callback defensively - Add debug logs for disposed guard, stream error discard, and trailing timer failure Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
…ix JSDoc - Add ResolvedNotificationThrottleConfig with required minChunkSize (mirrors AgentDebugConfig/ResolvedAgentDebugConfig pattern) - Extract resolveNotificationThrottleConfig() to types.ts, co-located with the type (replaces inline validation in constructor) - Add readonly modifiers to interface fields - Fix intervalMs JSDoc: "only the latest state" → "single coalesced notification reflecting all accumulated mutations" - Fix minChunkSize JSDoc: "active assistant message" → "trailing assistant message (last in array, with string content)" - Add finalize flush clarification to interface JSDoc - Re-export ResolvedNotificationThrottleConfig and resolver from index.ts Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
- Combined messages + state mutations under single throttle window - intervalMs: 0 with minChunkSize streaming behavior - Multiple sequential runAgent on same throttled agent - Non-string content fallback with minChunkSize Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
- Updated agent message addition to use 'as const' for role type safety. - Introduced utility functions to extract assistant message content, reducing duplication in notification throttle tests. - Refactored message handling in tests to utilize new helper functions for clarity and maintainability.
…rottle tests Eliminate ~50 `as BaseEvent` casts via centralized event factories and extract repeated setup/lifecycle code into startRun/runToCompletion helpers. 700 lines → 439 lines with identical test semantics. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
…rove docs - notifySubscribers now handles MaybePromise<void> rejections (critical: async subscribers previously caused unhandled promise rejections) - Wrap throttle finalize in try/finally so disposed=true is always set - Add disposed guard in trailing timer callback - Add debug logger to fire-and-forget .catch() handlers for consistency - Fix non-string content test to use MESSAGES_SNAPSHOT with tool-call-only msg - Add tests: disposed suppression, timer cleanup on error, async rejections - Improve JSDoc: leading-edge semantics, intervalMs:0 behavior, undefined content Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
…-based tests Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
The throttle logic now lives in @ag-ui/event-throttle-middleware instead. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
…t, extend coalescing, add type guards - Invert IMMEDIATE_EVENT_TYPES → BUFFERABLE_EVENT_TYPES allowlist so new protocol event types default to immediate passthrough (safer for lifecycle events) - Extend coalescing to TOOL_CALL_CHUNK and REASONING_MESSAGE_CHUNK (previously buffered but not coalesced, defeating the middleware's purpose) - Replace `as any` casts with typed guards (isTextChunk, isToolCallChunk, isReasoningChunk) and prevent coalescing chunks with undefined IDs - Fix incorrect comment: "leading-edge treatment" → "throttle window restarts" - Wrap timer-triggered flush() in try-catch to prevent unhandled global exceptions - Add teardown tests: unsubscribe with pending timer, error with pending timer - Fix error test: assert error propagation instead of swallowing with catch() - Relax rxjs peer dep from exact "7.8.1" to "^7.0.0" Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
mme
approved these changes
Apr 2, 2026
mxmzb
changed the title
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
EventThrottleMiddleware— a new middleware that coalesces rapid subscriber notifications during streamingintervalMs: time-based throttle cap (e.g.16= ~60fps)minChunkSize: character-based cap (e.g.20= batch single-char tokens until 20 chars accumulate)intervalMs: 0with nominChunkSizeis a no-op passthrough (zero overhead)Replaces #1420 — clean branch without #1385 contamination.
Test plan
intervalMs: fewer notifications than chunks, final delta content completeminChunkSize: holds notifications until N chars accumulate, final delta content complete🤖 Generated with Claude Code