forked from cleveragents/cleveragents-core
freemo
f138bab5ff
Add Server-Sent Events (SSE) streaming infrastructure to the A2A event system, enabling real-time delivery of task status updates and artifact notifications. Key changes: - Defined SSE event type constants: TASK_STATUS_UPDATE (TaskStatusUpdateEvent) and TASK_ARTIFACT_UPDATE (TaskArtifactUpdateEvent) per the A2A protocol specification. - Added SseEventFormatter class that converts A2aEvent instances to text/event-stream format with event, id, and data fields. Includes keepalive formatting for long-lived connections. - Added EventBusBridge class that subscribes to the internal EventBus (ReactiveEventBus) and translates DomainEvent instances into A2aEvent instances published to the A2aEventQueue. Maps plan lifecycle events (PLAN_CREATED, PLAN_PHASE_CHANGED, etc.) to TaskStatusUpdateEvent and checkpoint events to TaskArtifactUpdateEvent. - Bridge handles closed queue gracefully via contextlib.suppress. - Added 8 Behave scenarios covering SSE formatting, event type constants, EventBusBridge translation for both status and artifact events, closed queue handling, and JSON payload validation. ISSUES CLOSED: #875
53 lines
2.3 KiB
Gherkin
53 lines
2.3 KiB
Gherkin
@mock_only
|
|
Feature: A2A SSE streaming for task updates and artifacts
|
|
As a CleverAgents developer
|
|
I want the A2A event system to support SSE streaming
|
|
So that clients can receive real-time task status and artifact updates
|
|
|
|
Scenario: SSE event formatter produces valid text/event-stream output
|
|
Given an A2aEvent with type "TaskStatusUpdateEvent" and plan_id "plan-001"
|
|
When I format the event as SSE
|
|
Then the SSE output should contain "event: TaskStatusUpdateEvent"
|
|
And the SSE output should contain "id: "
|
|
And the SSE output should contain "data: "
|
|
And the SSE output should end with two newlines
|
|
|
|
Scenario: SSE keepalive produces a comment line
|
|
When I format a keepalive SSE message
|
|
Then the keepalive should start with ":"
|
|
|
|
Scenario: EventBusBridge translates plan status events
|
|
Given an A2aEventQueue for SSE testing
|
|
And a mock EventBus
|
|
And an EventBusBridge connecting bus to queue
|
|
When the bridge receives a PLAN_CREATED domain event
|
|
Then the queue should contain a TaskStatusUpdateEvent
|
|
|
|
Scenario: EventBusBridge translates artifact events
|
|
Given an A2aEventQueue for SSE testing
|
|
And a mock EventBus
|
|
And an EventBusBridge connecting bus to queue
|
|
When the bridge receives a CHECKPOINT_RESTORED domain event
|
|
Then the queue should contain a TaskArtifactUpdateEvent
|
|
|
|
Scenario: EventBusBridge handles closed queue gracefully
|
|
Given an A2aEventQueue for SSE testing
|
|
And a mock EventBus
|
|
And an EventBusBridge connecting bus to queue
|
|
When the queue is closed
|
|
And the bridge receives a PLAN_CREATED domain event
|
|
Then no error should be raised from the bridge
|
|
|
|
Scenario: TaskStatusUpdateEvent type constant is defined
|
|
Then the TASK_STATUS_UPDATE constant should equal "TaskStatusUpdateEvent"
|
|
|
|
Scenario: TaskArtifactUpdateEvent type constant is defined
|
|
Then the TASK_ARTIFACT_UPDATE constant should equal "TaskArtifactUpdateEvent"
|
|
|
|
Scenario: SseEventFormatter data payload is valid JSON
|
|
Given an A2aEvent with type "TaskStatusUpdateEvent" and plan_id "plan-002"
|
|
When I format the event as SSE
|
|
Then the SSE data line should contain valid JSON
|
|
And the JSON should have key "event_type" with value "TaskStatusUpdateEvent"
|
|
And the JSON should have key "plan_id" with value "plan-002"
|