diff --git a/features/a2a_sse_streaming.feature b/features/a2a_sse_streaming.feature new file mode 100644 index 000000000..7c6aab42f --- /dev/null +++ b/features/a2a_sse_streaming.feature @@ -0,0 +1,52 @@ +@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" diff --git a/features/steps/a2a_sse_streaming_steps.py b/features/steps/a2a_sse_streaming_steps.py new file mode 100644 index 000000000..05df225bc --- /dev/null +++ b/features/steps/a2a_sse_streaming_steps.py @@ -0,0 +1,160 @@ +"""Step definitions for a2a_sse_streaming.feature.""" + +from __future__ import annotations + +import json +from typing import Any +from unittest.mock import MagicMock + +from behave import given, then, when + +from cleveragents.a2a.events import ( + TASK_ARTIFACT_UPDATE, + TASK_STATUS_UPDATE, + A2aEventQueue, + EventBusBridge, + SseEventFormatter, +) +from cleveragents.a2a.models import A2aEvent + + +@given('an A2aEvent with type "{event_type}" and plan_id "{plan_id}"') +def step_create_event(context: Any, event_type: str, plan_id: str) -> None: + context.event = A2aEvent( + event_type=event_type, + plan_id=plan_id, + data={"status": "working"}, + ) + + +@when("I format the event as SSE") +def step_format_sse(context: Any) -> None: + context.sse_output = SseEventFormatter.format(context.event) + + +@then('the SSE output should contain "{text}"') +def step_sse_contains(context: Any, text: str) -> None: + assert text in context.sse_output, ( + f"Expected '{text}' in SSE output: {context.sse_output!r}" + ) + + +@then("the SSE output should end with two newlines") +def step_sse_ends_newlines(context: Any) -> None: + assert context.sse_output.endswith("\n\n"), ( + f"SSE output should end with two newlines: {context.sse_output!r}" + ) + + +@when("I format a keepalive SSE message") +def step_format_keepalive(context: Any) -> None: + context.keepalive = SseEventFormatter.format_keepalive() + + +@then('the keepalive should start with ":"') +def step_keepalive_starts_colon(context: Any) -> None: + assert context.keepalive.startswith(":"), ( + f"Keepalive should start with ':', got: {context.keepalive!r}" + ) + + +@given("an A2aEventQueue for SSE testing") +def step_create_queue(context: Any) -> None: + context.queue = A2aEventQueue() + + +@given("a mock EventBus") +def step_create_mock_bus(context: Any) -> None: + context.bus = MagicMock() + context.bus_callback = None + + def fake_subscribe(callback: Any) -> MagicMock: + context.bus_callback = callback + return MagicMock() + + context.bus.subscribe.side_effect = fake_subscribe + + +@given("an EventBusBridge connecting bus to queue") +def step_create_bridge(context: Any) -> None: + context.bridge = EventBusBridge(context.bus, context.queue) + context.bridge.start() + + +@when("the bridge receives a PLAN_CREATED domain event") +def step_bridge_plan_created(context: Any) -> None: + domain_event = MagicMock() + domain_event.event_type = MagicMock(value="PLAN_CREATED") + domain_event.plan_id = "plan-test-001" + domain_event.details = {"action": "test"} + context.bridge_error = None + try: + context.bus_callback(domain_event) + except Exception as e: + context.bridge_error = e + + +@when("the bridge receives a CHECKPOINT_RESTORED domain event") +def step_bridge_checkpoint_restored(context: Any) -> None: + domain_event = MagicMock() + domain_event.event_type = MagicMock(value="CHECKPOINT_RESTORED") + domain_event.plan_id = "plan-test-002" + domain_event.details = {"checkpoint_id": "cp-001"} + context.bus_callback(domain_event) + + +@when("the queue is closed") +def step_close_queue(context: Any) -> None: + context.queue.close() + + +@then("the queue should contain a TaskStatusUpdateEvent") +def step_queue_has_status_event(context: Any) -> None: + events = context.queue.get_events() + types = [e.event_type for e in events] + assert TASK_STATUS_UPDATE in types, ( + f"Expected {TASK_STATUS_UPDATE} in queue, got: {types}" + ) + + +@then("the queue should contain a TaskArtifactUpdateEvent") +def step_queue_has_artifact_event(context: Any) -> None: + events = context.queue.get_events() + types = [e.event_type for e in events] + assert TASK_ARTIFACT_UPDATE in types, ( + f"Expected {TASK_ARTIFACT_UPDATE} in queue, got: {types}" + ) + + +@then("no error should be raised from the bridge") +def step_no_bridge_error(context: Any) -> None: + assert context.bridge_error is None, ( + f"Expected no error, got: {context.bridge_error}" + ) + + +@then('the TASK_STATUS_UPDATE constant should equal "{value}"') +def step_check_status_constant(context: Any, value: str) -> None: + assert value == TASK_STATUS_UPDATE + + +@then('the TASK_ARTIFACT_UPDATE constant should equal "{value}"') +def step_check_artifact_constant(context: Any, value: str) -> None: + assert value == TASK_ARTIFACT_UPDATE + + +@then("the SSE data line should contain valid JSON") +def step_sse_data_valid_json(context: Any) -> None: + for line in context.sse_output.splitlines(): + if line.startswith("data: "): + payload = line[6:] + context.sse_json = json.loads(payload) + return + raise AssertionError("No data: line found in SSE output") + + +@then('the JSON should have key "{key}" with value "{value}"') +def step_json_has_key_value(context: Any, key: str, value: str) -> None: + assert context.sse_json.get(key) == value, ( + f"Expected JSON['{key}'] == '{value}', got: {context.sse_json.get(key)!r}" + ) diff --git a/src/cleveragents/a2a/events.py b/src/cleveragents/a2a/events.py index c15c77614..1fdbdfebc 100644 --- a/src/cleveragents/a2a/events.py +++ b/src/cleveragents/a2a/events.py @@ -1,12 +1,19 @@ -"""A2A event streaming — local queue and remote stub. +"""A2A event streaming — local queue, SSE formatting, and remote stub. The :class:`A2aEventQueue` provides a working in-memory event queue for local mode and a stub for remote subscriptions that raises :class:`A2aNotAvailableError`. + +:class:`SseEventFormatter` converts :class:`A2aEvent` instances into +``text/event-stream`` formatted strings per the Server-Sent Events spec. + +:class:`EventBusBridge` subscribes to the internal ``EventBus`` and +publishes translated :class:`A2aEvent` instances to an event queue. """ from __future__ import annotations +import json from collections.abc import Callable from typing import Any @@ -16,6 +23,13 @@ from ulid import ULID from cleveragents.a2a.errors import A2aNotAvailableError from cleveragents.a2a.models import A2aEvent +# --------------------------------------------------------------------------- +# SSE event type constants (A2A protocol) +# --------------------------------------------------------------------------- + +TASK_STATUS_UPDATE: str = "TaskStatusUpdateEvent" +TASK_ARTIFACT_UPDATE: str = "TaskArtifactUpdateEvent" + logger: structlog.stdlib.BoundLogger = structlog.get_logger(__name__) _REMOTE_MSG = ( @@ -117,6 +131,151 @@ class A2aEventQueue: ) +# --------------------------------------------------------------------------- +# SSE event formatter +# --------------------------------------------------------------------------- + + +class SseEventFormatter: + """Convert :class:`A2aEvent` to ``text/event-stream`` format. + + Each event is serialized as:: + + event: + id: + data: + + Two trailing newlines terminate the event per the EventSource spec. + """ + + @staticmethod + def format(event: A2aEvent) -> str: + """Format an :class:`A2aEvent` as an SSE text block.""" + data_payload = json.dumps( + { + "event_id": event.event_id, + "event_type": event.event_type, + "plan_id": event.plan_id, + "data": event.data, + "timestamp": event.timestamp, + }, + default=str, + ) + lines = [ + f"event: {event.event_type}", + f"id: {event.event_id}", + f"data: {data_payload}", + "", + "", + ] + return "\n".join(lines) + + @staticmethod + def format_keepalive() -> str: + """Format a keep-alive comment for SSE connections.""" + return ": keepalive\n\n" + + +# --------------------------------------------------------------------------- +# EventBus -> A2aEventQueue bridge +# --------------------------------------------------------------------------- + + +class EventBusBridge: + """Subscribe to the internal :class:`EventBus` and forward events. + + Translates :class:`DomainEvent` instances from the reactive event + bus into :class:`A2aEvent` instances and publishes them to an + :class:`A2aEventQueue`. + + Usage:: + + bridge = EventBusBridge(event_bus, event_queue) + bridge.start() # subscribes to bus + bridge.stop() # unsubscribes + """ + + # Domain event types that map to SSE TaskStatusUpdateEvent + _STATUS_EVENT_TYPES: frozenset[str] = frozenset( + { + "PLAN_CREATED", + "PLAN_PHASE_CHANGED", + "PLAN_STATE_CHANGED", + "PLAN_APPLIED", + "PLAN_CANCELLED", + "PLAN_ERRORED", + } + ) + + # Domain event types that map to SSE TaskArtifactUpdateEvent + _ARTIFACT_EVENT_TYPES: frozenset[str] = frozenset( + { + "CHECKPOINT_RESTORED", + } + ) + + def __init__( + self, + event_bus: Any, + event_queue: A2aEventQueue, + ) -> None: + self._event_bus = event_bus + self._event_queue = event_queue + self._subscription: Any | None = None + + def start(self) -> None: + """Subscribe to the event bus and begin forwarding.""" + if hasattr(self._event_bus, "subscribe"): + self._subscription = self._event_bus.subscribe(self._on_domain_event) + logger.info("a2a.event_bridge.started") + + def stop(self) -> None: + """Unsubscribe from the event bus.""" + if self._subscription is not None: + if hasattr(self._subscription, "dispose"): + self._subscription.dispose() + self._subscription = None + logger.info("a2a.event_bridge.stopped") + + def _on_domain_event(self, domain_event: Any) -> None: + """Translate a domain event to an A2A event and publish.""" + event_type_name = getattr(domain_event, "event_type", None) + if event_type_name is None: + return + + # Normalize to string if it's an enum + type_str = ( + event_type_name.value + if hasattr(event_type_name, "value") + else str(event_type_name) + ) + + if type_str in self._STATUS_EVENT_TYPES: + sse_type = TASK_STATUS_UPDATE + elif type_str in self._ARTIFACT_EVENT_TYPES: + sse_type = TASK_ARTIFACT_UPDATE + else: + sse_type = type_str + + plan_id = getattr(domain_event, "plan_id", None) + details = getattr(domain_event, "details", {}) + + a2a_event = A2aEvent( + event_type=sse_type, + plan_id=plan_id, + data=dict(details) if details else {}, + ) + + import contextlib + + with contextlib.suppress(RuntimeError): + self._event_queue.publish(a2a_event) + + __all__ = [ + "TASK_ARTIFACT_UPDATE", + "TASK_STATUS_UPDATE", "A2aEventQueue", + "EventBusBridge", + "SseEventFormatter", ]