fix(a2a): JSON-RPC conformance for a2a-sdk 1.1.0 compatibility

Two bugs reported by gfdsa (PR #41711 comment, Jul 12) that break the
official a2a-sdk 1.1.0 Python client:

1. Task objects serialized non-spec createdAt/lastModified fields.
   The A2A v1.0 Task proto (lf.a2a.v1.Task) only has id, contextId,
   status, artifacts, history, metadata. Strict ProtoJSON parsers
   reject unknown fields with ParseError. Removed both fields from
   build_task(); created_at param kept for call-site compatibility.

2. SSE streaming frames were not JSON-RPC wrapped. A2A v1.0 §9.4
   requires data: {"jsonrpc":"2.0","id":...,"result":{StreamResponse}}.
   sse_data() now accepts req_id and wraps in JSON-RPC envelope.
   sse_done() changed from 'data: {}' to SSE comment ': done' so
   SDK doesn't try to parse an empty JSON-RPC response.

All call sites in adapter.py (_emit_terminal, _rpc_message_stream,
_rpc_tasks_subscribe) updated to thread req_id through.

Tests updated: 153 pass (151 unit + 17 integration, including 2 new
tests for JSON-RPC envelope wrapping and fallback behavior).

Refs: gfdsa/a2a-hermes reproduction repo

plugins/platforms/a2a/adapter.py

@@ -964,21 +964,25 @@ class A2AAdapter(BasePlatformAdapter):
         handler.wfile.write(chunk.encode("utf-8"))
         handler.wfile.flush()
 
-    def _emit_terminal(self, handler, task_id: str, context_id: str, state: str, reply: str) -> None:
+    def _emit_terminal(self, handler, task_id: str, context_id: str, state: str, reply: str,
+                       req_id: Any = None) -> None:
         """Emit the final artifact/status events and close the stream (v1.0:
-        closure signals terminal state, no ``final`` field)."""
+        closure signals terminal state, no ``final`` field).
+
+        ``req_id`` is threaded into JSON-RPC-wrapped SSE frames per §9.4."""
         if reply and state == protocol.STATE_COMPLETED:
             self._sse_write(handler, protocol.sse_data(
-                protocol.artifact_update(task_id, context_id, reply)))
+                protocol.artifact_update(task_id, context_id, reply), req_id))
             self._sse_write(handler, protocol.sse_data(
-                protocol.status_update(task_id, context_id, state)))
+                protocol.status_update(task_id, context_id, state), req_id))
         else:
             self._sse_write(handler, protocol.sse_data(
-                protocol.status_update(task_id, context_id, state, reply)))
+                protocol.status_update(task_id, context_id, state, reply), req_id))
         self._sse_write(handler, protocol.sse_done())
 
     def _rpc_message_stream(self, handler, req_id: Any, params: dict, peer: str, agent: Optional[dict] = None) -> None:
-        """Handle message/stream as an SSE response of StreamResponse events."""
+        """Handle message/stream as an SSE response of JSON-RPC-wrapped
+        StreamResponse events (A2A v1.0 §9.4)."""
         protocol.metrics.streams_started += 1
         self._sse_headers(handler)
 
@@ -989,19 +993,21 @@ class A2AAdapter(BasePlatformAdapter):
                     handler, terminal["id"], terminal["contextId"],
                     terminal["status"]["state"],
                     protocol.extract_text(terminal.get("status", {}).get("message", {}) or {}),
+                    req_id=req_id,
                 )
                 return
 
             task_id, context_id = pending["task_id"], pending["context_id"]
             self._sse_write(handler, protocol.sse_data(protocol.stream_task(
-                protocol.build_task(task_id, context_id, protocol.STATE_SUBMITTED, created_at=pending["created_iso"]))))
+                protocol.build_task(task_id, context_id, protocol.STATE_SUBMITTED, created_at=pending["created_iso"])),
+                req_id))
             self._sse_write(handler, protocol.sse_data(
-                protocol.status_update(task_id, context_id, protocol.STATE_WORKING)))
+                protocol.status_update(task_id, context_id, protocol.STATE_WORKING), req_id))
 
             state, reply = self._await_reply(
                 pending, keepalive=lambda: self._sse_write(handler, ": keepalive\n\n"))
             state, reply = self._finalize_task(pending, state, reply)
-            self._emit_terminal(handler, task_id, context_id, state, reply)
+            self._emit_terminal(handler, task_id, context_id, state, reply, req_id=req_id)
         except (BrokenPipeError, ConnectionResetError):
             logger.debug("A2A: stream client disconnected")
 
@@ -1030,7 +1036,7 @@ class A2AAdapter(BasePlatformAdapter):
                         state, reply = rec["state"], rec.get("reply", "")
                         break
                     self._sse_write(handler, ": keepalive\n\n")
-            self._emit_terminal(handler, task_id, rec["context_id"], state, reply)
+            self._emit_terminal(handler, task_id, rec["context_id"], state, reply, req_id=req_id)
         except (BrokenPipeError, ConnectionResetError):
             logger.debug("A2A: subscribe client disconnected")
 

plugins/platforms/a2a/protocol.py

@@ -372,14 +372,19 @@ def build_task(
     *,
     created_at: str = "",
 ) -> dict:
-    """Build an A2A v1.0 Task object for a message/send result."""
+    """Build an A2A v1.0 Task object for a message/send result.
+
+    ``created_at`` is accepted for call-site compatibility but not serialized —
+    the A2A v1.0 ``Task`` proto (``lf.a2a.v1.Task``) has no ``createdAt`` or
+    ``lastModified`` field.  Strict ProtoJSON parsers (e.g. a2a-sdk 1.1.0)
+    reject unknown fields, so we must not include them.  The spec's §5.6.1
+    timestamp-format example mentions them but they are not in the proto.
+    """
     now = now_iso()
     task: dict[str, Any] = {
         "id": task_id,
         "contextId": context_id,
         "status": {"state": state, "timestamp": now},
-        "createdAt": created_at or now,
-        "lastModified": now,
     }
     if agent_text:
         task["status"]["message"] = text_message(ROLE_AGENT, agent_text, context_id)
@@ -417,14 +422,29 @@ def artifact_update(task_id: str, context_id: str, text: str) -> dict:
     }
 
 
-def sse_data(payload: dict) -> str:
-    """Encode one StreamResponse as an SSE data frame (member-name discriminated)."""
-    return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"
+def sse_data(payload: dict, req_id: Any = None) -> str:
+    """Encode one StreamResponse as a JSON-RPC-wrapped SSE data frame.
+
+    A2A v1.0 §9.4 requires each SSE frame to be a full JSON-RPC response:
+    ``{"jsonrpc":"2.0","id":<req_id>,"result":{StreamResponse}}``.  Emitting a
+    bare StreamResponse (the REST binding shape) breaks JSON-RPC clients that
+    expect the envelope, including the official a2a-sdk.
+    """
+    if req_id is not None:
+        envelope = jsonrpc_result(req_id, payload)
+    else:
+        envelope = payload  # legacy/fallback — no envelope
+    return f"data: {json.dumps(envelope, ensure_ascii=False)}\n\n"
 
 
 def sse_done() -> str:
-    """SSE stream-closure marker — terminal state is implied by closure in v1.0."""
-    return "event: done\ndata: {}\n\n"
+    """SSE stream-closure marker — a comment, not a parseable data frame.
+
+    A2A v1.0 signals terminal state by closing the stream.  Emitting
+    ``data: {}`` causes JSON-RPC clients to try parsing an empty response and
+    fail.  An SSE comment line (``: done``) is ignored by all SSE parsers.
+    """
+    return ": done\n\n"
 
 
 # --------------------------------------------------------------------------

tests/plugins/test_a2a_phase23.py

@@ -55,7 +55,8 @@ def _make_live_adapter(monkeypatch, reply_fn=None):
 
 def _post_sse(url, body):
     """POST a JSON-RPC request and return the parsed SSE stream as
-    (data_payloads, event_names)."""
+    (data_payloads, event_names).  Unwraps the JSON-RPC envelope from
+    each data frame so callers see bare StreamResponse objects."""
     req = urllib.request.Request(
         url, data=json.dumps(body).encode(),
         headers={"Content-Type": "application/json"}, method="POST",
@@ -66,11 +67,17 @@ def _post_sse(url, body):
     for block in raw.split("\n\n"):
         for line in block.splitlines():
             if line.startswith("event: "):
-                events.append(line[len("event: "):].strip())
+                events.append(line[len("event:"):].strip())
             elif line.startswith("data: "):
                 data = line[len("data: "):].strip()
                 if data:
-                    payloads.append(json.loads(data))
+                    obj = json.loads(data)
+                    # Unwrap JSON-RPC envelope: {"jsonrpc":"2.0","id":...,"result":{...}}
+                    if isinstance(obj, dict) and "jsonrpc" in obj and "result" in obj:
+                        payloads.append(obj["result"])
+                    else:
+                        payloads.append(obj)
+            # SSE comment lines (": done") are ignored — not data frames.
     return payloads, events
 
 
@@ -127,8 +134,32 @@ class TestStreamResponseFormat:
         # No event-name line: v1.0 discriminates by member presence.
         assert "event:" not in chunk
 
+    def test_sse_data_jsonrpc_envelope(self):
+        """A2A v1.0 §9.4: SSE frames must be JSON-RPC-wrapped when req_id is
+        provided.  Bare StreamResponse (REST binding) breaks a2a-sdk clients."""
+        chunk = protocol.sse_data({"statusUpdate": {"taskId": "t"}}, req_id="42")
+        assert chunk.startswith("data: ")
+        obj = json.loads(chunk[len("data: "):].strip())
+        assert obj["jsonrpc"] == "2.0"
+        assert obj["id"] == "42"
+        assert "result" in obj
+        assert obj["result"]["statusUpdate"]["taskId"] == "t"
+
+    def test_sse_data_no_envelope_without_req_id(self):
+        """Without req_id, sse_data falls back to bare payload for legacy callers."""
+        chunk = protocol.sse_data({"statusUpdate": {"taskId": "t"}})
+        obj = json.loads(chunk[len("data: "):].strip())
+        assert "jsonrpc" not in obj
+        assert obj["statusUpdate"]["taskId"] == "t"
+
     def test_sse_done_marker(self):
-        assert protocol.sse_done() == "event: done\ndata: {}\n\n"
+        """v1.0 signals stream completion by closing the stream.  The done
+        marker is an SSE comment (``: done``), not a parseable data frame —
+        emitting ``data: {}`` breaks JSON-RPC clients that try to parse it."""
+        done = protocol.sse_done()
+        assert ": done" in done
+        assert "data:" not in done  # no data frame for SDK to parse
+        assert done.endswith("\n\n")
 
 
 @pytest.mark.integration
@@ -164,7 +195,7 @@ class TestStreamingEndToEnd:
             assert len(artifacts) == 1
             assert "ECHO:" in protocol.extract_text(artifacts[0]["artifact"])
 
-            assert events == ["done"]
+            assert events == []  # v1.0: stream closure is the terminal signal, no event frame
             await adapter.disconnect()
 
         asyncio.run(run())
@@ -189,7 +220,7 @@ class TestStreamingEndToEnd:
             artifacts = [p for p in payloads if "artifactUpdate" in p]
             assert artifacts and "ECHO:" in protocol.extract_text(
                 artifacts[0]["artifactUpdate"]["artifact"])
-            assert events == ["done"]
+            assert events == []  # v1.0: stream closure is the terminal signal, no event frame
             await adapter.disconnect()
 
         asyncio.run(run())

tests/plugins/test_a2a_plugin.py

@@ -381,8 +381,10 @@ class TestV1Task:
         assert task["status"]["state"] == "TASK_STATE_COMPLETED"
         assert task["artifacts"][0]["parts"][0] == {"text": "the answer", "mediaType": "text/plain"}
         assert "kind" not in task
-        assert task["createdAt"]
-        assert task["lastModified"]
+        # A2A v1.0 Task proto (lf.a2a.v1.Task) has no createdAt/lastModified.
+        # Strict ProtoJSON parsers (a2a-sdk) reject unknown fields.
+        assert "createdAt" not in task
+        assert "lastModified" not in task
 
     def test_failed_task_has_message_no_artifacts(self):
         task = protocol.build_task("t2", "c2", protocol.STATE_FAILED, "went wrong")