fix(a2a/events): guard A2aEventQueue with threading.Lock to prevent concurrent iteration crash

src/cleveragents/a2a/events.py

@@ -1,4 +1,4 @@
-"""A2A event streaming — local queue, SSE formatting, 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
@@ -62,30 +62,32 @@ class A2aEventQueue:
         self._lock: threading.Lock = threading.Lock()
 
     # ------------------------------------------------------------------
-    # Local-mode operations (working)
+    # Local-mode operations (thread-safe)
     # ------------------------------------------------------------------
 
     @property
     def is_closed(self) -> bool:
         """Whether this queue has been closed."""
-        return self._is_closed
+        with self._lock:
+            return self._is_closed
 
     def publish(self, event: A2aEvent) -> None:
         """Append *event* to the local queue and notify subscribers.
 
         The subscriber dict is snapshotted inside the lock so that concurrent
         :meth:`subscribe_local` / :meth:`unsubscribe` calls on other threads
-        cannot mutate it while we iterate.  Callbacks are invoked *outside*
+        cannot mutate it while we iterate.  The close check and type check
+        are also performed inside the lock. Callbacks are invoked *outside*
         the lock to avoid deadlocks when a callback itself modifies
         subscriptions.
         """
-        if self._is_closed:
-            raise RuntimeError("Cannot publish to a closed event queue")
-        if not isinstance(event, A2aEvent):
-            raise TypeError("event must be an A2aEvent instance")
         with self._lock:
-            self._events.append(event)
+            if self._is_closed:
+                raise RuntimeError("Cannot publish to a closed event queue")
+            if not isinstance(event, A2aEvent):
+                raise TypeError("event must be an A2aEvent instance")
             callbacks = list(self._subscriptions.items())
+            self._events.append(event)
         logger.debug(
             "a2a.event.published",
             event_id=event.event_id,
@@ -175,12 +177,12 @@ class SseEventFormatter:
 
     Two trailing newlines terminate the event per the EventSource spec.
     The data payload follows JSON-RPC 2.0 notification format per the A2A
-    protocol specification (§Streaming Architecture).
+    protocol specification (Streaming Architecture).
     """
 
     # Mapping from A2A event type names to JSON-RPC 2.0 method strings.
-    # Per spec §Streaming Architecture: TaskStatusUpdateEvent → task/statusUpdate,
-    # TaskArtifactUpdateEvent → task/artifactUpdate.
+    # Per spec Streaming Architecture: TaskStatusUpdateEvent -> task/statusUpdate,
+    # TaskArtifactUpdateEvent -> task/artifactUpdate.
     _EVENT_TYPE_TO_METHOD: ClassVar[dict[str, str]] = {
         "TaskStatusUpdateEvent": "task/statusUpdate",
         "TaskArtifactUpdateEvent": "task/artifactUpdate",