Introduce AgentProtocolEventBus for pluggable SSE event handling - #2634
Merged
chickenlj merged 1 commit intoAug 9, 2026
Merged
Conversation
…ble SSE event bus
|
|
Contributor
There was a problem hiding this comment.
Pull request overview
This PR introduces a new AgentProtocolEventBus abstraction so Agent Protocol SSE streaming can be backed by pluggable/shared implementations (e.g., cross-instance event buses), while keeping the existing in-memory replay behavior as the default.
Changes:
- Added
AgentProtocolEventBusinterface (publish/subscribe/complete) for pluggable SSE event handling. - Updated default in-memory bus (
AgentProtocolTaskEventBus) to implement the interface and updated wiring (AgentProtocolAutoConfiguration,AgentProtocolTaskStore) to depend on the interface. - Added tests to validate auto-configuration override behavior and to validate sequencing behavior.
Reviewed changes
Copilot reviewed 6 out of 6 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agent-protocol/src/main/java/io/agentscope/extensions/agentprotocol/AgentProtocolEventBus.java | New event bus interface defining the contract for publish/subscribe/complete. |
| agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agent-protocol/src/main/java/io/agentscope/extensions/agentprotocol/AgentProtocolTaskEventBus.java | Default in-memory implementation updated to implement the new interface. |
| agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agent-protocol/src/main/java/io/agentscope/extensions/agentprotocol/AgentProtocolAutoConfiguration.java | Registers AgentProtocolEventBus conditionally to allow user-provided implementations to override. |
| agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agent-protocol/src/main/java/io/agentscope/extensions/agentprotocol/AgentProtocolTaskStore.java | Switches from concrete bus dependency to the interface. |
| agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agent-protocol/src/test/java/io/agentscope/extensions/agentprotocol/AgentProtocolTaskEventBusTest.java | Adds contract/concurrency-related sequencing tests. |
| agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agent-protocol/src/test/java/io/agentscope/extensions/agentprotocol/AgentProtocolAutoConfigurationTest.java | New test verifying default bus creation and custom-bean override behavior. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| @@ -53,7 +53,8 @@ | |||
| public class AgentProtocolAutoConfiguration { | |||
|
|
|||
| @Bean | |||
Comment on lines
+62
to
+81
| @Test | ||
| void publish_assignsUniqueSequencesForConcurrentPublishers() { | ||
| int eventCount = 200; | ||
|
|
||
| List<RemoteAgentEvent> published = | ||
| LongStream.range(0, eventCount) | ||
| .parallel() | ||
| .mapToObj( | ||
| ignored -> | ||
| bus.publish( | ||
| "task-concurrent", event(RemoteEventType.STATUS))) | ||
| .toList(); | ||
|
|
||
| assertEquals( | ||
| eventCount, published.stream().map(RemoteAgentEvent::getSeq).distinct().count()); | ||
| assertEquals( | ||
| LongStream.rangeClosed(1, eventCount).boxed().toList(), | ||
| published.stream().map(RemoteAgentEvent::getSeq).sorted().toList()); | ||
| assertTrue(published.stream().allMatch(e -> "task-concurrent".equals(e.getTaskId()))); | ||
| } |
Comment on lines
91
to
95
| public AgentProtocolTaskStore( | ||
| AgentFactory agentFactory, | ||
| ProtocolTaskRepository taskRepository, | ||
| AgentProtocolTaskEventBus eventBus, | ||
| AgentProtocolEventBus eventBus, | ||
| AgentProtocolProperties properties) { |
Codecov Report✅ All modified and coverable lines are covered by tests. 📢 Thoughts on this report? Let us know! |
chickenlj
approved these changes
Aug 9, 2026
wzl521
pushed a commit
to wzl521/agentscope-java
that referenced
this pull request
Aug 12, 2026
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
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.
AgentScope-Java Version
2.0.3-SNAPSHOT（pom.xml 中的 ）
Description
背景： 当前 Agent Protocol 的 SSE 事件流由 AgentProtocolTaskEventBus 直接提供，回放缓冲保存在进程内存中，且 TaskStore、自动配置与具体实现类强耦合。对于需要跨实例（多副本）流式输出的场景（例如基于 Redis Streams 的共享事件总线），只能复制代码改造，无法通过配置替换实现。
改动内容：
新增 AgentProtocolEventBus 接口，抽象事件总线能力（publish / subscribe / complete），由实现方负责单调递增 seq 的分配与回放
AgentProtocolTaskEventBus 实现该接口，保持原有进程内回放缓冲行为不变（默认实现）
AgentProtocolAutoConfiguration 改为声明 AgentProtocolEventBus bean，并加上 @ConditionalOnMissingBean，允许应用提供自定义实现（如 Redis Streams 适配器）自动覆盖默认实现
AgentProtocolTaskStore 依赖接口 AgentProtocolEventBus 而非具体类
新增测试：
AgentProtocolAutoConfigurationTest：默认创建内存实现；用户提供自定义 bean 时保留自定义实现
AgentProtocolTaskEventBusTest 新增用例：验证并发 publish 时 seq 单调唯一
如何测试：
格式检查
mvn spotless:check
相关测试（注意：本机需 JDK 17+ 运行 Maven，默认 JDK 8 与 spotless 3.4.0 不兼容）
mvn test -pl agentscope-extensions/agentscope-extensions-protocol/agentscope-extensions-agent-protocol
-Dtest='AgentProtocolTaskEventBusTest,AgentProtocolAutoConfigurationTest'
验证结果：spotless:check 通过；7 个测试（AgentProtocolTaskEventBusTest 5 + AgentProtocolAutoConfigurationTest 2）全部通过，BUILD SUCCESS。
破坏性变更： 无。仅新增接口与抽象，默认行为不变。
Checklist
Code has been formatted with mvn spotless:apply（已验证 spotless:check 通过）
All tests are passing (mvn test)（已跑改动模块相关测试，7 个全部通过；全量 mvn test 未执行）
Javadoc comments are complete and follow project conventions（新增接口含完整 Javadoc）
Related documentation has been updated (e.g. links, examples, etc.)（本次未更新文档；如需可补充自定义 EventBus 的使用说明）
Code is ready for review