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"

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}"
+    )

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: <event_type>
+        id: <event_id>
+        data: <json payload>
+
+    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",
 ]