The A2aEventQueue class had no thread-safety guarantees for its internal
shared mutable state (_events, _subscriptions, _is_closed). When multiple
threads accessed the queue concurrently — e.g. one thread publishing events
while another iterated over subscriptions or read event lists — the lack of
locking could cause RuntimeError: dictionary changed size during iteration
and similar crashes.

This fix adds a threading.RLock to A2aEventQueue that guards all mutating
and reading operations (__init__, is_closed, publish, subscribe_local,
unsubscribe, get_events, close). The EventBusBridge class also receives
its own RLock for thread-safe start/stop operations.

Signed-off-by: CleverThis <hal9000@cleverthis.com>

@@ -503,0 +521,4 @@
+    def _publisher():
+        try:
+            for i in range(8):
+                if context.event_queue._is_closed:

@@ -503,0 +635,4 @@
+@then("all originally subscribed callbacks should have been invoked")
+def step_assert_all_original_callbacks_invoked(context):
+    """All pre-existing subscriptions remain valid (no dictionary corruption)."""
+    assert context.event_queue is not None

@@ -117,0 +129,4 @@
+            self._events.clear()
+            _count = count
+        if _count:
+            logger.info("a2a.event_queue.closed", subscription_count=_count)

@@ -503,0 +518,4 @@
+    context.concurrency_error = None
+    errors: list[Exception] = []
+
+    def _publisher():

@@ -503,0 +559,4 @@
+    errors: list[Exception] = []
+
+    def _publisher():
+        try:

@@ -503,0 +633,4 @@
+
+
+@then("all originally subscribed callbacks should have been invoked")
+def step_assert_all_original_callbacks_invoked(context):

@@ -47,6 +53,7 @@ class A2aEventQueue:
         self._events: list[A2aEvent] = []
         self._subscriptions: dict[str, Callable[[A2aEvent], Any]] = {}
         self._is_closed: bool = False
+        self._lock = threading.RLock()

@@ -67,0 +77,4 @@
+                try:
+                    callback(event)
+                except Exception:
+                    logger.exception(

@@ -117,0 +128,4 @@
+            self._subscriptions.clear()
+            self._events.clear()
+            _count = count
+        if _count: