fix(a2a): A2A facade missing standard message/send and message/stream operations

Implemented _STANDARD_OPERATIONS = ['message/send', 'message/stream'] in src/cleveragents/a2a/facade.py. Added _handle_message_send to route to SessionService.append_message() and return a Task-like response. Added _handle_message_stream to route to SessionService.append_message(), publish a TaskStatusUpdateEvent to the event queue, and return streaming: true. Added _publish_task_status_event helper. Updated _SUPPORTED_OPERATIONS to include the new standard operations. Updated _handlers() dispatch map to wire in both new operations. Added features/a2a_message_operations.feature with 17 scenarios. Added features/steps/a2a_message_operations_steps.py with step definitions. ISSUES CLOSED: #9088

features/a2a_message_operations.feature

@@ -0,0 +1,125 @@
+@mock_only
+Feature: A2A standard message/send and message/stream operations
+  As a client of the A2A local facade
+  I want to use the standard message/send and message/stream operations
+  So that I can interact with the agent using the A2A protocol (ADR-047)
+
+  # -----------------------------------------------------------------------
+  # message/send — list_operations includes it
+  # -----------------------------------------------------------------------
+
+  Scenario: list_operations includes message/send and message/stream
+    Given a message-ops facade with no services
+    Then the message-ops facade list_operations should include "message/send"
+    And the message-ops facade list_operations should include "message/stream"
+
+  # -----------------------------------------------------------------------
+  # message/send — stub path (no session service wired)
+  # -----------------------------------------------------------------------
+
+  Scenario: message/send returns stub response when no session service wired
+    Given a message-ops facade with no services
+    When I dispatch message-ops operation "message/send" with params {"message": {"role": "user", "parts": [{"text": "Hello agent"}]}}
+    Then the message-ops response status should be "ok"
+    And the message-ops response data should contain key "id"
+    And the message-ops response data key "stub" should be true
+    And the message-ops response data status state should be "completed"
+
+  Scenario: message/send with session service appends message and returns task id
+    Given a message-ops facade with a mock SessionService
+    When I dispatch message-ops operation "message/send" with params {"message": {"role": "user", "parts": [{"text": "Refactor the auth module"}]}, "sessionId": "SESS-001"}
+    Then the message-ops response status should be "ok"
+    And the message-ops response data should contain key "id"
+    And the message-ops response data should contain key "message_id"
+    And the message-ops response data key "sessionId" should equal "SESS-001"
+    And the message-ops response data status state should be "completed"
+
+  Scenario: message/send with session service but no sessionId returns task without message_id
+    Given a message-ops facade with a mock SessionService
+    When I dispatch message-ops operation "message/send" with params {"message": {"role": "user", "parts": [{"text": "Hello"}]}}
+    Then the message-ops response status should be "ok"
+    And the message-ops response data should contain key "id"
+    And the message-ops response data should not contain key "message_id"
+    And the message-ops response data status state should be "completed"
+
+  Scenario: message/send with missing message param returns error
+    Given a message-ops facade with no services
+    When I dispatch message-ops operation "message/send" with params {}
+    Then the message-ops response status should be "error"
+
+  Scenario: message/send with empty parts returns error
+    Given a message-ops facade with no services
+    When I dispatch message-ops operation "message/send" with params {"message": {"role": "user", "parts": []}}
+    Then the message-ops response status should be "error"
+
+  Scenario: message/send with parts containing no text returns error
+    Given a message-ops facade with no services
+    When I dispatch message-ops operation "message/send" with params {"message": {"role": "user", "parts": [{"image": "data:..."}]}}
+    Then the message-ops response status should be "error"
+
+  Scenario: message/send with unknown role falls back to user role
+    Given a message-ops facade with a mock SessionService
+    When I dispatch message-ops operation "message/send" with params {"message": {"role": "unknown_role", "parts": [{"text": "Hello"}]}, "sessionId": "SESS-001"}
+    Then the message-ops response status should be "ok"
+    And the message-ops response data should contain key "id"
+
+  # -----------------------------------------------------------------------
+  # message/stream — stub path (no session service wired)
+  # -----------------------------------------------------------------------
+
+  Scenario: message/stream returns stub response with streaming flag when no service wired
+    Given a message-ops facade with no services
+    When I dispatch message-ops operation "message/stream" with params {"message": {"role": "user", "parts": [{"text": "Stream this"}]}}
+    Then the message-ops response status should be "ok"
+    And the message-ops response data should contain key "id"
+    And the message-ops response data key "stub" should be true
+    And the message-ops response data key "streaming" should be true
+    And the message-ops response data status state should be "working"
+
+  Scenario: message/stream with session service appends message and returns streaming task
+    Given a message-ops facade with a mock SessionService
+    When I dispatch message-ops operation "message/stream" with params {"message": {"role": "user", "parts": [{"text": "Stream this task"}]}, "sessionId": "SESS-002"}
+    Then the message-ops response status should be "ok"
+    And the message-ops response data should contain key "id"
+    And the message-ops response data should contain key "message_id"
+    And the message-ops response data key "sessionId" should equal "SESS-002"
+    And the message-ops response data key "streaming" should be true
+    And the message-ops response data status state should be "working"
+
+  Scenario: message/stream publishes SSE event to event queue when wired
+    Given a message-ops facade with a mock SessionService and event queue
+    When I dispatch message-ops operation "message/stream" with params {"message": {"role": "user", "parts": [{"text": "Trigger SSE"}]}}
+    Then the message-ops response status should be "ok"
+    And the message-ops response data key "streaming" should be true
+    And the message-ops event queue should have received a TaskStatusUpdateEvent
+
+  Scenario: message/stream with event queue but no session service publishes SSE event
+    Given a message-ops facade with only an event queue
+    When I dispatch message-ops operation "message/stream" with params {"message": {"role": "user", "parts": [{"text": "Stub stream"}]}}
+    Then the message-ops response status should be "ok"
+    And the message-ops response data key "stub" should be true
+    And the message-ops event queue should have received a TaskStatusUpdateEvent
+
+  Scenario: message/stream with missing message param returns error
+    Given a message-ops facade with no services
+    When I dispatch message-ops operation "message/stream" with params {}
+    Then the message-ops response status should be "error"
+
+  Scenario: message/stream with empty parts returns error
+    Given a message-ops facade with no services
+    When I dispatch message-ops operation "message/stream" with params {"message": {"role": "user", "parts": []}}
+    Then the message-ops response status should be "error"
+
+  # -----------------------------------------------------------------------
+  # Validation — both operations reject whitespace-only text
+  # -----------------------------------------------------------------------
+
+  Scenario: message/send with whitespace-only text returns error
+    Given a message-ops facade with no services
+    When I dispatch message-ops operation "message/send" with params {"message": {"role": "user", "parts": [{"text": "   "}]}}
+    Then the message-ops response status should be "error"
+
+  Scenario: message/stream with whitespace-only text returns error
+    Given a message-ops facade with no services
+    When I dispatch message-ops operation "message/stream" with params {"message": {"role": "user", "parts": [{"text": "   "}]}}
+    Then the message-ops response status should be "error"

features/steps/a2a_message_operations_steps.py

@@ -0,0 +1,223 @@
+"""Step definitions for a2a_message_operations.feature.
+
+Tests the standard A2A ``message/send`` and ``message/stream`` operations
+added to :class:`A2aLocalFacade` per ADR-047 and issue #9088.
+
+Both operations follow the A2A wire format::
+
+    {
+        "message": {
+            "role": "user",
+            "parts": [{"text": "..."}]
+        },
+        "sessionId": "<optional>"
+    }
+
+``message/send`` routes to ``SessionService.append_message()`` and returns
+a Task-like response with ``status.state = "completed"``.
+
+``message/stream`` does the same but also publishes a
+``TaskStatusUpdateEvent`` to the event queue (when wired) and returns
+``streaming: true`` with ``status.state = "working"``.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from behave import given, then, use_step_matcher, when
+from behave.runner import Context
+
+try:
+    from cleveragents.a2a.events import A2aEventQueue
+    from cleveragents.a2a.facade import A2aLocalFacade
+    from cleveragents.a2a.models import A2aRequest
+except ImportError:
+    A2aEventQueue = None  # type: ignore[assignment,misc]
+    A2aLocalFacade = None  # type: ignore[assignment,misc]
+    A2aRequest = None  # type: ignore[assignment,misc]
+
+use_step_matcher("re")
+
+
+# ---------------------------------------------------------------------------
+# Mock helpers
+# ---------------------------------------------------------------------------
+
+
+class _MockMessage:
+    """Minimal SessionMessage stub."""
+
+    def __init__(self, message_id: str = "MSG-001") -> None:
+        self.message_id = message_id
+
+
+class _MockSessionService:
+    """Minimal SessionService stub that records append_message calls."""
+
+    def __init__(self) -> None:
+        self._calls: list[dict[str, Any]] = []
+
+    def append_message(
+        self,
+        session_id: str,
+        role: Any,
+        content: str,
+        metadata: dict[str, Any] | None = None,
+    ) -> _MockMessage:
+        self._calls.append(
+            {"session_id": session_id, "role": role, "content": content}
+        )
+        return _MockMessage()
+
+
+def _build_mock_session_service() -> _MockSessionService:
+    return _MockSessionService()
+
+
+def _build_real_event_queue() -> A2aEventQueue:
+    return A2aEventQueue()
+
+
+# ---------------------------------------------------------------------------
+# Given steps
+# ---------------------------------------------------------------------------
+
+
+@given(r"a message-ops facade with no services")
+def step_msg_ops_no_services(context: Context) -> None:
+    context.msg_ops_facade = A2aLocalFacade()
+    context.msg_ops_event_queue = None
+
+
+@given(r"a message-ops facade with a mock SessionService")
+def step_msg_ops_session_service(context: Context) -> None:
+    svc = _build_mock_session_service()
+    context.msg_ops_facade = A2aLocalFacade(services={"session_service": svc})
+    context.msg_ops_event_queue = None
+
+
+@given(r"a message-ops facade with a mock SessionService and event queue")
+def step_msg_ops_session_and_queue(context: Context) -> None:
+    svc = _build_mock_session_service()
+    queue = _build_real_event_queue()
+    context.msg_ops_facade = A2aLocalFacade(
+        services={"session_service": svc, "event_queue": queue}
+    )
+    context.msg_ops_event_queue = queue
+
+
+@given(r"a message-ops facade with only an event queue")
+def step_msg_ops_only_queue(context: Context) -> None:
+    queue = _build_real_event_queue()
+    context.msg_ops_facade = A2aLocalFacade(services={"event_queue": queue})
+    context.msg_ops_event_queue = queue
+
+
+# ---------------------------------------------------------------------------
+# When steps
+# ---------------------------------------------------------------------------
+
+
+@when(
+    r'I dispatch message-ops operation "(?P<operation>[^"]+)" '
+    r"with params (?P<params_json>.+)"
+)
+def step_msg_ops_dispatch(
+    context: Context, operation: str, params_json: str
+) -> None:
+    params: dict[str, Any] = json.loads(params_json)
+    request = A2aRequest(method=operation, params=params)
+    context.msg_ops_response = context.msg_ops_facade.dispatch(request)
+
+
+# ---------------------------------------------------------------------------
+# Then steps
+# ---------------------------------------------------------------------------
+
+
+@then(r'the message-ops facade list_operations should include "(?P<op>[^"]+)"')
+def step_msg_ops_list_includes(context: Context, op: str) -> None:
+    ops = context.msg_ops_facade.list_operations()
+    assert op in ops, f"Expected '{op}' in list_operations(), got: {ops}"
+
+
+@then(r'the message-ops response status should be "(?P<status>[^"]+)"')
+def step_msg_ops_status(context: Context, status: str) -> None:
+    if status == "ok":
+        assert context.msg_ops_response.result is not None, (
+            f"Expected ok response but got error: {context.msg_ops_response.error}"
+        )
+    else:
+        assert context.msg_ops_response.error is not None, (
+            "Expected error response but got result: "
+            f"{context.msg_ops_response.result}"
+        )
+
+
+@then(r'the message-ops response data should contain key "(?P<key>[^"]+)"')
+def step_msg_ops_data_has_key(context: Context, key: str) -> None:
+    data = context.msg_ops_response.result or {}
+    assert key in data, (
+        f"Expected key '{key}' in response data, got: {list(data.keys())}"
+    )
+
+
+@then(r'the message-ops response data should not contain key "(?P<key>[^"]+)"')
+def step_msg_ops_data_no_key(context: Context, key: str) -> None:
+    data = context.msg_ops_response.result or {}
+    assert key not in data, (
+        f"Expected key '{key}' NOT in response data, but found it"
+    )
+
+
+@then(r'the message-ops response data key "(?P<key>[^"]+)" should be true')
+def step_msg_ops_data_key_true(context: Context, key: str) -> None:
+    data = context.msg_ops_response.result or {}
+    actual = data.get(key)
+    assert actual is True, f"Expected data['{key}'] to be True, got {actual!r}"
+
+
+@then(
+    r'the message-ops response data key "(?P<key>[^"]+)" '
+    r'should equal "(?P<value>[^"]+)"'
+)
+def step_msg_ops_data_key_equals(context: Context, key: str, value: str) -> None:
+    data = context.msg_ops_response.result or {}
+    actual = data.get(key)
+    assert str(actual) == value, (
+        f"Expected data['{key}'] = '{value}', got '{actual}'"
+    )
+
+
+@then(
+    r'the message-ops response data status state should be "(?P<state>[^"]+)"'
+)
+def step_msg_ops_status_state(context: Context, state: str) -> None:
+    data = context.msg_ops_response.result or {}
+    status = data.get("status", {})
+    actual_state = status.get("state") if isinstance(status, dict) else None
+    assert actual_state == state, (
+        f"Expected status.state = '{state}', got '{actual_state}'"
+    )
+
+
+@then(
+    r"the message-ops event queue should have received a TaskStatusUpdateEvent"
+)
+def step_msg_ops_event_received(context: Context) -> None:
+    queue: A2aEventQueue | None = context.msg_ops_event_queue
+    assert queue is not None, "No event queue was wired to the facade"
+    events = queue.get_events()
+    task_status_events = [
+        e for e in events if e.event_type == "TaskStatusUpdateEvent"
+    ]
+    assert task_status_events, (
+        f"Expected at least one TaskStatusUpdateEvent in queue, "
+        f"got: {[e.event_type for e in events]}"
+    )
+
+
+# Reset step matcher to parse (default) so subsequent step files are not affected
+use_step_matcher("parse")

src/cleveragents/a2a/facade.py

@@ -8,11 +8,14 @@ constructor.  Expected keys:
 
 | Key                          | Type                        | Used by               |
 |------------------------------|-----------------------------|-----------------------|
-| ``session_service``          | ``SessionService``          | session.create/close  |
+| ``session_service``          | ``SessionService``          | session.create/close, |
+|                              |                             | message/send,         |
+|                              |                             | message/stream        |
 | ``plan_lifecycle_service``   | ``PlanLifecycleService``    | plan.* operations     |
 | ``tool_registry``            | ``ToolRegistry``            | registry.list_tools   |
 | ``resource_registry_service``| ``ResourceRegistryService`` | registry.list_resources|
-| ``event_queue``              | ``A2aEventQueue``           | event.subscribe       |
+| ``event_queue``              | ``A2aEventQueue``           | event.subscribe,      |
+|                              |                             | message/stream SSE    |
 
 When a service is absent the handler falls back to a safe stub
 response so the facade never crashes due to missing wiring.
@@ -32,6 +35,7 @@ from cleveragents.a2a.errors import (
 )
 from cleveragents.a2a.models import (
     A2aErrorDetail,
+    A2aEvent,
     A2aRequest,
     A2aResponse,
 )
@@ -53,6 +57,13 @@ logger: structlog.stdlib.BoundLogger = structlog.get_logger(__name__)
 # Supported operations
 # ---------------------------------------------------------------------------
 
+# Standard A2A operations (defined by the external A2A specification, ADR-047).
+# These handle the core agent interaction lifecycle: messaging and task management.
+_STANDARD_OPERATIONS: list[str] = [
+    "message/send",
+    "message/stream",
+]
+
 # Spec-aligned _cleveragents/ extension method names (ADR-047).
 _EXTENSION_OPERATIONS: list[str] = [
     # Plan lifecycle
@@ -109,7 +120,9 @@ _LEGACY_OPERATIONS: list[str] = [
     "event.subscribe",
 ]
 
-_SUPPORTED_OPERATIONS: list[str] = _EXTENSION_OPERATIONS + _LEGACY_OPERATIONS
+_SUPPORTED_OPERATIONS: list[str] = (
+    _STANDARD_OPERATIONS + _EXTENSION_OPERATIONS + _LEGACY_OPERATIONS
+)
 
 
 class A2aLocalFacade:
@@ -242,6 +255,9 @@ class A2aLocalFacade:
         """
         if self._handler_map is None:
             self._handler_map = {
+                # --- Standard A2A operations (spec-defined, ADR-047) ---
+                "message/send": self._handle_message_send,
+                "message/stream": self._handle_message_stream,
                 # --- _cleveragents/ extension methods (spec-aligned) ---
                 # Plan lifecycle
                 "_cleveragents/plan/use": self._handle_plan_create,
@@ -297,6 +313,221 @@ class A2aLocalFacade:
             }
         return self._handler_map
 
+    # ------------------------------------------------------------------
+    # Operation handlers — standard A2A messaging (ADR-047)
+    # ------------------------------------------------------------------
+
+    def _handle_message_send(self, params: dict[str, Any]) -> dict[str, Any]:
+        """Handle ``message/send`` — standard A2A operation (ADR-047).
+
+        Routes to :meth:`SessionService.append_message` when a session
+        service is wired.  The ``params.message`` field follows the A2A
+        wire format::
+
+            {
+                "message": {
+                    "role": "user",
+                    "parts": [{"text": "Hello, agent!"}]
+                },
+                "sessionId": "<optional-session-id>"
+            }
+
+        Returns a Task-like response with ``id``, ``status``, and
+        ``message_id`` so callers can track the resulting task.
+
+        When no session service is wired a stub response is returned so
+        the facade never crashes due to missing wiring.
+        """
+        message_payload: dict[str, Any] = params.get("message") or {}
+        if not message_payload:
+            raise ValueError("params.message is required for message/send")
+
+        role: str = message_payload.get("role", "user")
+        parts: list[dict[str, Any]] = message_payload.get("parts", [])
+        text_content: str = " ".join(
+            part.get("text", "") for part in parts if isinstance(part, dict)
+        ).strip()
+
+        if not text_content:
+            raise ValueError(
+                "params.message.parts must contain at least one text part"
+            )
+
+        task_id: str = str(ULID())
+        session_id: str | None = (
+            params.get("sessionId") or params.get("session_id")
+        )
+
+        svc = self._session_service
+        if svc is None:
+            # Stub response — no session service wired.
+            logger.debug(
+                "a2a.message_send.stub",
+                task_id=task_id,
+                role=role,
+            )
+            return {
+                "id": task_id,
+                "status": {"state": "completed"},
+                "stub": True,
+            }
+
+        # Append the message to the session (or create a transient record).
+        from cleveragents.domain.models.core.session import MessageRole
+
+        try:
+            msg_role = MessageRole(role)
+        except ValueError:
+            msg_role = MessageRole.USER
+
+        message_id: str | None = None
+        if session_id:
+            msg = svc.append_message(
+                session_id=session_id,
+                role=msg_role,
+                content=text_content,
+            )
+            message_id = msg.message_id
+
+        logger.info(
+            "a2a.message_send",
+            task_id=task_id,
+            session_id=session_id,
+            role=role,
+        )
+        result: dict[str, Any] = {
+            "id": task_id,
+            "status": {"state": "completed"},
+        }
+        if message_id is not None:
+            result["message_id"] = message_id
+        if session_id is not None:
+            result["sessionId"] = session_id
+        return result
+
+    def _handle_message_stream(self, params: dict[str, Any]) -> dict[str, Any]:
+        """Handle ``message/stream`` — standard A2A operation (ADR-047).
+
+        Behaves like :meth:`_handle_message_send` but additionally
+        publishes a :class:`~cleveragents.a2a.models.A2aEvent` to the
+        event queue (when wired) so that SSE subscribers receive a
+        ``TaskStatusUpdateEvent`` for the new task.
+
+        The ``streaming`` key in the response signals to callers that
+        SSE events will follow on the subscribed event channel.
+
+        Wire format is identical to ``message/send``::
+
+            {
+                "message": {
+                    "role": "user",
+                    "parts": [{"text": "Hello, agent!"}]
+                },
+                "sessionId": "<optional-session-id>"
+            }
+
+        Returns a Task-like response with ``id``, ``status``,
+        ``streaming: true``, and optionally ``message_id`` and
+        ``sessionId``.
+        """
+        message_payload: dict[str, Any] = params.get("message") or {}
+        if not message_payload:
+            raise ValueError("params.message is required for message/stream")
+
+        role: str = message_payload.get("role", "user")
+        parts: list[dict[str, Any]] = message_payload.get("parts", [])
+        text_content: str = " ".join(
+            part.get("text", "") for part in parts if isinstance(part, dict)
+        ).strip()
+
+        if not text_content:
+            raise ValueError(
+                "params.message.parts must contain at least one text part"
+            )
+
+        task_id: str = str(ULID())
+        session_id: str | None = (
+            params.get("sessionId") or params.get("session_id")
+        )
+
+        svc = self._session_service
+        message_id: str | None = None
+
+        if svc is None:
+            # Stub response — no session service wired.
+            logger.debug(
+                "a2a.message_stream.stub",
+                task_id=task_id,
+                role=role,
+            )
+            # Still publish SSE event if event queue is available.
+            self._publish_task_status_event(task_id=task_id, state="working")
+            return {
+                "id": task_id,
+                "status": {"state": "working"},
+                "streaming": True,
+                "stub": True,
+            }
+
+        # Append the message to the session.
+        from cleveragents.domain.models.core.session import MessageRole
+
+        try:
+            msg_role = MessageRole(role)
+        except ValueError:
+            msg_role = MessageRole.USER
+
+        if session_id:
+            msg = svc.append_message(
+                session_id=session_id,
+                role=msg_role,
+                content=text_content,
+            )
+            message_id = msg.message_id
+
+        # Publish SSE TaskStatusUpdateEvent so subscribers receive the update.
+        self._publish_task_status_event(task_id=task_id, state="working")
+
+        logger.info(
+            "a2a.message_stream",
+            task_id=task_id,
+            session_id=session_id,
+            role=role,
+        )
+        result: dict[str, Any] = {
+            "id": task_id,
+            "status": {"state": "working"},
+            "streaming": True,
+        }
+        if message_id is not None:
+            result["message_id"] = message_id
+        if session_id is not None:
+            result["sessionId"] = session_id
+        return result
+
+    def _publish_task_status_event(self, task_id: str, state: str) -> None:
+        """Publish a ``TaskStatusUpdateEvent`` to the event queue if wired.
+
+        Failures are logged but never propagated so the caller always
+        succeeds regardless of event queue availability.
+        """
+        queue = self._event_queue
+        if queue is None:
+            return
+        try:
+            event = A2aEvent(
+                event_type="TaskStatusUpdateEvent",
+                plan_id=task_id,
+                data={"state": state},
+            )
+            queue.publish(event)
+        except Exception:
+            logger.warning(
+                "a2a.message_stream.event_publish_failed",
+                task_id=task_id,
+                exc_info=True,
+            )
+
     # ------------------------------------------------------------------
     # Operation handlers — session
     # ------------------------------------------------------------------
@@ -452,7 +683,9 @@ class A2aLocalFacade:
             "tools": [{"name": s.name, "description": s.description} for s in specs],
         }
 
-    def _handle_registry_list_resources(self, params: dict[str, Any]) -> dict[str, Any]:
+    def _handle_registry_list_resources(
+        self, params: dict[str, Any]
+    ) -> dict[str, Any]:
         svc = self._resource_registry_service
         if svc is None:
             return {"resources": []}