CHANGELOG.md

@@ -17,6 +17,13 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ### Added
 
+- **A2A Standard Message Operations** (#9088): Implemented ``message/send`` and
+  ``message/stream`` standard operations on :class:`A2aLocalFacade` per ADR-047.
+  ``message/send`` routes to ``SessionService.append_message()` with stub fallback,
+  while ``message/stream`` additionally publishes
+  ``TaskStatusUpdateEvent`` SSE events via the event queue.  Updated operation count
+  assertions across BDD and Robot test suites from 42 to 44.
+
 - **TDD Issue-Capture Test Activation** (#7025): Replaced 234 bare `@skip` tags
   across 82 Behave feature files with the correct `@tdd_expected_fail @tdd_issue
   @tdd_issue_<N>` tag system. Scenarios whose referenced bugs were already fixed

CONTRIBUTORS.md

@@ -16,4 +16,5 @@ Below are some of the specific details of various contributions.
 * Brent E. Edwards has contributed quality assurance, test coverage, and CI pipeline improvements.
 * HAL 9000 has contributed automated implementation, bug fixes, and feature development as part of the CleverAgents automation pool.
 * HAL 9000 has contributed the plan concurrency race-condition fix (#7989): wired `LockService` into the plan lifecycle, guarding `execute_plan()` and `apply_plan()` with plan-level advisory locks and unique per-invocation owner identities to prevent silent concurrent state corruption.
+* HAL 9000 has contributed A2A standard message/send and message/stream operations (#9088): implemented ``message/send`` and ``message/stream`` on :class:`A2aLocalFacade` per ADR-047 with stub fallback when session service is unbundled, SSE task-status event publishing for streaming, and comprehensive BDD test coverage across 17 scenarios.
 * This project was made possible thanks to considerable donation of time, money, and resources by CleverThis, Inc.

features/a2a_cli_facade_integration.feature

@@ -6,7 +6,7 @@ Feature: A2A CLI facade integration
 
   Scenario: CLI bootstrap creates a wired facade
     Given a facade created via the CLI bootstrap
-    Then the facade should support all 42 operations
+    Then the facade should support all 44 operations
     And the facade should be cached on subsequent calls
 
   Scenario: Session create routes through facade

features/a2a_message_operations.feature

@@ -0,0 +1,125 @@
+@a2a @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/consolidated_misc.feature

@@ -140,7 +140,7 @@ Feature: Consolidated Misc
     And the operations list should contain "registry.list_tools"
     And the operations list should contain "context.get"
     And the operations list should contain "event.subscribe"
-    And the operations list should have 42 items
+    And the operations list should have 44 items
 
     # -----------------------------------------------------------------------
     # A2aHttpTransport — all stubs raise A2aNotAvailableError
@@ -639,7 +639,7 @@ Feature: Consolidated Misc
     Then the m6 smoke operations should include "session.create"
     And the m6 smoke operations should include "plan.execute"
     And the m6 smoke operations should include "event.subscribe"
-    And the m6 smoke operations count should be 42
+    And the m6 smoke operations count should be 44
 
     # --- A2A event queue ---
 

features/m6_autonomy_acceptance.feature

@@ -109,7 +109,7 @@ Feature: M6 autonomy acceptance smoke tests
     Then the m6 smoke operations should include "session.create"
     And the m6 smoke operations should include "plan.execute"
     And the m6 smoke operations should include "event.subscribe"
-    And the m6 smoke operations count should be 42
+    And the m6 smoke operations count should be 44
 
   # --- A2A event queue (AC-2: event queue publish/subscribe) ---
 

features/steps/a2a_cli_facade_integration_steps.py

@@ -124,10 +124,12 @@ def step_call_notify_facade(context: Any, operation: str) -> None:
 # ---------------------------------------------------------------------------
 
 
-@then("the facade should support all 42 operations")
-def step_check_42_operations(context: Any) -> None:
+@then("the facade should support all {expected_count:d} operations")
+def step_check_operation_count(context: Any, expected_count: int) -> None:
     ops = context.facade.list_operations()
-    assert len(ops) == 42, f"Expected 42 operations, got {len(ops)}: {ops}"
+    assert len(ops) == expected_count, (
+        f"Expected {expected_count} operations, got {len(ops)}: {ops}"
+    )
 
 
 @then("the facade should be cached on subsequent calls")

features/steps/a2a_message_operations_steps.py

@@ -0,0 +1,208 @@
+"""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, (
+            f"Expected error response but got result: {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")

robot/helper_a2a_facade.py

@@ -97,7 +97,7 @@ def list_operations() -> None:
     facade = A2aLocalFacade()
     ops = facade.list_operations()
     expected = {"session.create", "plan.create", "plan.execute", "context.get"}
-    if expected.issubset(set(ops)) and len(ops) == 42:
+    if expected.issubset(set(ops)) and len(ops) == 44:
         print("a2a-list-operations-ok")
     else:
         print(f"FAIL: ops={ops}", file=sys.stderr)

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,10 +35,17 @@ from cleveragents.a2a.errors import (
 )
 from cleveragents.a2a.models import (
     A2aErrorDetail,
+    A2aEvent,
     A2aRequest,
     A2aResponse,
 )
 
+# Runtime import — used inside handler methods for message role validation.
+try:
+    from cleveragents.domain.models.core.session import MessageRole as _MessageRole
+except ImportError:
+    _MessageRole = None
+
 if TYPE_CHECKING:
     from cleveragents.a2a.events import A2aEventQueue
     from cleveragents.application.services.plan_lifecycle_service import (
@@ -53,6 +63,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 +126,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 +261,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 +319,223 @@ class A2aLocalFacade:
             }
         return self._handler_map
 
+    # ------------------------------------------------------------------
+    # Operation handlers — standard A2A messaging (ADR-047)
+    # ------------------------------------------------------------------
+
+    def _extract_message_params(
+        self, params: dict[str, Any], operation_name: str
+    ) -> tuple[str, str, str | None]:
+        """Common extraction and validation for ``message/send`` and ``message/stream``.
+
+        Validates that *params* contains a valid ``message`` with at least one
+        text part.  Raises ``ValueError`` on invalid inputs.
+
+        Returns:
+            A 3-tuple of ``(role, text_content, session_id)`` suitable for
+            downstream handlers.
+        """
+        message_payload: dict[str, Any] = params.get("message") or {}
+        if not message_payload:
+            raise ValueError(f"params.message is required for {operation_name}")
+
+        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")
+
+        session_id: str | None = params.get("sessionId") or params.get("session_id")
+        return role, text_content, session_id
+
+    def _resolve_message_role(self, role_name: str) -> Any:
+        """Resolve a raw role string to a :class:`MessageRole`.
+
+        Falls back to ``MessageRole.USER`` when the name is not recognised.
+        """
+        if _MessageRole is None:
+            # Guard against import-only environments (tests without full codebase).
+            return role_name
+        try:
+            return _MessageRole(role_name)
+        except ValueError:
+            return _MessageRole.USER
+
+    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.
+        """
+        role, text_content, session_id = self._extract_message_params(
+            params, "message/send"
+        )
+
+        task_id: str = str(ULID())
+        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).
+        msg_role = self._resolve_message_role(role)
+
+        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``.
+        """
+        role, text_content, session_id = self._extract_message_params(
+            params, "message/stream"
+        )
+
+        task_id: str = str(ULID())
+        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.
+        msg_role = self._resolve_message_role(role)
+
+        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, "task_id": task_id},
+            )
+            queue.publish(event)
+        except Exception as exc:
+            logger.warning(
+                "a2a.message_stream.event_publish_failed",
+                task_id=task_id,
+                error=str(exc),
+                exc_info=True,
+            )
+
     # ------------------------------------------------------------------
     # Operation handlers — session
     # ------------------------------------------------------------------
@@ -358,10 +597,11 @@ class A2aLocalFacade:
                     session_id=session_id,
                     stopped_count=len(stopped),
                 )
-        except Exception:
+        except Exception as exc:
             logger.warning(
                 "a2a.session.close.devcontainer_cleanup_failed",
                 session_id=session_id,
+                error=str(exc),
                 exc_info=True,
             )