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

Added a threading.Lock to A2aEventQueue.__init__ to serialize access to internal state and prevent race conditions during concurrent subscribe/unsubscribe and publish.

Protected mutations of _subscriptions and _events with the lock in publish(), subscribe_local(), unsubscribe(), get_events(), and close() to ensure thread-safety.

In publish(), snapshot _subscriptions.items() inside the lock and iterate callbacks outside the lock to avoid deadlocks caused by user callbacks making further mutations.

This change ensures that modifications to the subscriptions/events are thread-safe and that publish callbacks are invoked without holding the lock.

Added BDD/Gherkin feature file features/a2a_event_queue_concurrency.feature with 6 scenarios covering concurrent subscribe, unsubscribe, and publish interactions.

Added step definitions features/steps/a2a_event_queue_concurrency_steps.py to wire the Gherkin scenarios to test code.

ISSUES CLOSED: #7604

features/a2a_event_queue_concurrency.feature

@@ -0,0 +1,62 @@
+@mock_only
+Feature: A2aEventQueue thread-safety (issue #7604)
+  As a CleverAgents developer
+  I want A2aEventQueue to be safe for concurrent use from multiple threads
+  So that publish/subscribe/unsubscribe operations never raise
+  RuntimeError: dictionary changed size during iteration
+
+  # -----------------------------------------------------------------------
+  # Regression: concurrent publish + subscribe must not raise RuntimeError
+  # -----------------------------------------------------------------------
+
+  Scenario: Concurrent publish and subscribe_local do not raise RuntimeError
+    Given a2aconcur a fresh A2aEventQueue
+    When a2aconcur I run concurrent publish and subscribe_local from multiple threads
+    Then a2aconcur no RuntimeError should have been raised
+    And a2aconcur all published events should be in the queue
+
+  # -----------------------------------------------------------------------
+  # Regression: concurrent publish + unsubscribe must not raise RuntimeError
+  # -----------------------------------------------------------------------
+
+  Scenario: Concurrent publish and unsubscribe do not raise RuntimeError
+    Given a2aconcur a fresh A2aEventQueue with subscribers
+    When a2aconcur I run concurrent publish and unsubscribe from multiple threads
+    Then a2aconcur no RuntimeError should have been raised
+
+  # -----------------------------------------------------------------------
+  # Regression: concurrent subscribe + unsubscribe must not raise RuntimeError
+  # -----------------------------------------------------------------------
+
+  Scenario: Concurrent subscribe_local and unsubscribe do not raise RuntimeError
+    Given a2aconcur a fresh A2aEventQueue
+    When a2aconcur I run concurrent subscribe_local and unsubscribe from multiple threads
+    Then a2aconcur no RuntimeError should have been raised
+
+  # -----------------------------------------------------------------------
+  # Lock is present on new instances
+  # -----------------------------------------------------------------------
+
+  Scenario: A2aEventQueue has a threading.Lock after construction
+    Given a2aconcur a fresh A2aEventQueue
+    Then a2aconcur the queue should have a threading Lock attribute
+
+  # -----------------------------------------------------------------------
+  # publish snapshots subscriptions so late-arriving unsubscribe is safe
+  # -----------------------------------------------------------------------
+
+  Scenario: Unsubscribing inside a callback does not affect the current publish iteration
+    Given a2aconcur a fresh A2aEventQueue
+    And a2aconcur a subscriber that unsubscribes itself during the callback
+    When a2aconcur I publish a valid A2aEvent
+    Then a2aconcur no RuntimeError should have been raised
+    And a2aconcur the event should be in the queue
+
+  # -----------------------------------------------------------------------
+  # get_events is thread-safe
+  # -----------------------------------------------------------------------
+
+  Scenario: Concurrent publish and get_events do not raise RuntimeError
+    Given a2aconcur a fresh A2aEventQueue
+    When a2aconcur I run concurrent publish and get_events from multiple threads
+    Then a2aconcur no RuntimeError should have been raised

features/steps/a2a_event_queue_concurrency_steps.py

@@ -0,0 +1,290 @@
+"""Step definitions for a2a_event_queue_concurrency.feature.
+
+Exercises thread-safety of A2aEventQueue (issue #7604):
+  - Concurrent publish + subscribe_local must not raise RuntimeError
+  - Concurrent publish + unsubscribe must not raise RuntimeError
+  - Concurrent subscribe_local + unsubscribe must not raise RuntimeError
+  - A2aEventQueue has a threading.Lock after construction
+  - Unsubscribing inside a callback does not affect the current publish iteration
+  - Concurrent publish + get_events must not raise RuntimeError
+"""
+
+from __future__ import annotations
+
+import threading
+from typing import Any
+
+from behave import given, then, when
+
+try:
+    from cleveragents.a2a.events import A2aEventQueue
+    from cleveragents.a2a.models import A2aEvent
+except ImportError:
+    pass  # module may be unavailable in isolated environments
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+_NUM_THREADS = 8
+_EVENTS_PER_THREAD = 10
+
+
+def _make_event(tag: str = "test") -> A2aEvent:
+    return A2aEvent(
+        event_type="TestEvent",
+        plan_id=f"plan-{tag}",
+        data={"tag": tag},
+    )
+
+
+# ---------------------------------------------------------------------------
+# Givens
+# ---------------------------------------------------------------------------
+
+
+@given("a2aconcur a fresh A2aEventQueue")
+def step_a2aconcur_fresh_queue(context: Any) -> None:
+    context.a2aconcur_queue = A2aEventQueue()
+    context.a2aconcur_errors: list[Exception] = []
+    context.a2aconcur_sub_ids: list[str] = []
+
+
+@given("a2aconcur a fresh A2aEventQueue with subscribers")
+def step_a2aconcur_fresh_queue_with_subs(context: Any) -> None:
+    context.a2aconcur_queue = A2aEventQueue()
+    context.a2aconcur_errors: list[Exception] = []
+    context.a2aconcur_sub_ids: list[str] = []
+    # Pre-register several subscribers so unsubscribe has something to remove
+    for i in range(_NUM_THREADS * 2):
+        sub_id = context.a2aconcur_queue.subscribe_local(lambda e, _i=i: None)
+        context.a2aconcur_sub_ids.append(sub_id)
+
+
+@given("a2aconcur a subscriber that unsubscribes itself during the callback")
+def step_a2aconcur_self_unsubscribing_subscriber(context: Any) -> None:
+    """Register a callback that removes its own subscription when called."""
+    queue: A2aEventQueue = context.a2aconcur_queue
+    holder: dict[str, str] = {}
+
+    def _self_unsubscribe(event: A2aEvent) -> None:
+        sub_id = holder.get("sub_id")
+        if sub_id:
+            queue.unsubscribe(sub_id)
+
+    sub_id = queue.subscribe_local(_self_unsubscribe)
+    holder["sub_id"] = sub_id
+    context.a2aconcur_sub_ids.append(sub_id)
+
+
+# ---------------------------------------------------------------------------
+# Whens
+# ---------------------------------------------------------------------------
+
+
+@when("a2aconcur I run concurrent publish and subscribe_local from multiple threads")
+def step_a2aconcur_concurrent_publish_subscribe(context: Any) -> None:
+    """Spawn threads that publish and subscribe concurrently."""
+    queue: A2aEventQueue = context.a2aconcur_queue
+    errors: list[Exception] = context.a2aconcur_errors
+    barrier = threading.Barrier(_NUM_THREADS * 2)
+
+    def _publisher(thread_id: int) -> None:
+        barrier.wait()
+        for i in range(_EVENTS_PER_THREAD):
+            try:
+                queue.publish(_make_event(f"pub-{thread_id}-{i}"))
+            except RuntimeError as exc:
+                errors.append(exc)
+
+    def _subscriber(thread_id: int) -> None:
+        barrier.wait()
+        for i in range(_EVENTS_PER_THREAD):
+            try:
+                sub_id = queue.subscribe_local(lambda e, _t=thread_id, _i=i: None)
+                context.a2aconcur_sub_ids.append(sub_id)
+            except RuntimeError as exc:
+                errors.append(exc)
+
+    threads = [
+        threading.Thread(target=_publisher, args=(t,), daemon=True)
+        for t in range(_NUM_THREADS)
+    ] + [
+        threading.Thread(target=_subscriber, args=(t,), daemon=True)
+        for t in range(_NUM_THREADS)
+    ]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join(timeout=10)
+
+
+@when("a2aconcur I run concurrent publish and unsubscribe from multiple threads")
+def step_a2aconcur_concurrent_publish_unsubscribe(context: Any) -> None:
+    """Spawn threads that publish and unsubscribe concurrently."""
+    queue: A2aEventQueue = context.a2aconcur_queue
+    errors: list[Exception] = context.a2aconcur_errors
+    sub_ids: list[str] = list(context.a2aconcur_sub_ids)
+    barrier = threading.Barrier(_NUM_THREADS * 2)
+
+    def _publisher(thread_id: int) -> None:
+        barrier.wait()
+        for i in range(_EVENTS_PER_THREAD):
+            try:
+                queue.publish(_make_event(f"pub-{thread_id}-{i}"))
+            except RuntimeError as exc:
+                errors.append(exc)
+
+    def _unsubscriber(thread_id: int) -> None:
+        barrier.wait()
+        # Each unsubscriber thread removes a slice of the pre-registered subs
+        start = thread_id * 2
+        for sub_id in sub_ids[start : start + 2]:
+            try:
+                queue.unsubscribe(sub_id)
+            except (RuntimeError, ValueError) as exc:
+                errors.append(exc)
+
+    threads = [
+        threading.Thread(target=_publisher, args=(t,), daemon=True)
+        for t in range(_NUM_THREADS)
+    ] + [
+        threading.Thread(target=_unsubscriber, args=(t,), daemon=True)
+        for t in range(_NUM_THREADS)
+    ]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join(timeout=10)
+
+
+@when(
+    "a2aconcur I run concurrent subscribe_local and unsubscribe from multiple threads"
+)
+def step_a2aconcur_concurrent_subscribe_unsubscribe(context: Any) -> None:
+    """Spawn threads that subscribe and unsubscribe concurrently."""
+    queue: A2aEventQueue = context.a2aconcur_queue
+    errors: list[Exception] = context.a2aconcur_errors
+    collected_ids: list[str] = []
+    lock = threading.Lock()
+    barrier = threading.Barrier(_NUM_THREADS * 2)
+
+    def _subscriber(thread_id: int) -> None:
+        barrier.wait()
+        for i in range(_EVENTS_PER_THREAD):
+            try:
+                sub_id = queue.subscribe_local(lambda e, _t=thread_id, _i=i: None)
+                with lock:
+                    collected_ids.append(sub_id)
+            except RuntimeError as exc:
+                errors.append(exc)
+
+    def _unsubscriber(thread_id: int) -> None:
+        barrier.wait()
+        for _ in range(_EVENTS_PER_THREAD):
+            try:
+                with lock:
+                    if collected_ids:
+                        sub_id = collected_ids.pop()
+                    else:
+                        continue
+                queue.unsubscribe(sub_id)
+            except (RuntimeError, ValueError) as exc:
+                errors.append(exc)
+
+    threads = [
+        threading.Thread(target=_subscriber, args=(t,), daemon=True)
+        for t in range(_NUM_THREADS)
+    ] + [
+        threading.Thread(target=_unsubscriber, args=(t,), daemon=True)
+        for t in range(_NUM_THREADS)
+    ]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join(timeout=10)
+
+
+@when("a2aconcur I publish a valid A2aEvent")
+def step_a2aconcur_publish_valid(context: Any) -> None:
+    context.a2aconcur_event = _make_event("self-unsub")
+    try:
+        context.a2aconcur_queue.publish(context.a2aconcur_event)
+    except Exception as exc:
+        context.a2aconcur_errors.append(exc)
+
+
+@when("a2aconcur I run concurrent publish and get_events from multiple threads")
+def step_a2aconcur_concurrent_publish_get_events(context: Any) -> None:
+    """Spawn threads that publish and call get_events concurrently."""
+    queue: A2aEventQueue = context.a2aconcur_queue
+    errors: list[Exception] = context.a2aconcur_errors
+    barrier = threading.Barrier(_NUM_THREADS * 2)
+
+    def _publisher(thread_id: int) -> None:
+        barrier.wait()
+        for i in range(_EVENTS_PER_THREAD):
+            try:
+                queue.publish(_make_event(f"pub-{thread_id}-{i}"))
+            except RuntimeError as exc:
+                errors.append(exc)
+
+    def _reader(thread_id: int) -> None:
+        barrier.wait()
+        for _ in range(_EVENTS_PER_THREAD):
+            try:
+                queue.get_events(limit=50)
+            except RuntimeError as exc:
+                errors.append(exc)
+
+    threads = [
+        threading.Thread(target=_publisher, args=(t,), daemon=True)
+        for t in range(_NUM_THREADS)
+    ] + [
+        threading.Thread(target=_reader, args=(t,), daemon=True)
+        for t in range(_NUM_THREADS)
+    ]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join(timeout=10)
+
+
+# ---------------------------------------------------------------------------
+# Thens
+# ---------------------------------------------------------------------------
+
+
+@then("a2aconcur no RuntimeError should have been raised")
+def step_a2aconcur_no_runtime_error(context: Any) -> None:
+    runtime_errors = [
+        e for e in context.a2aconcur_errors if isinstance(e, RuntimeError)
+    ]
+    assert not runtime_errors, (
+        f"Expected no RuntimeError, but got {len(runtime_errors)} error(s): "
+        + "; ".join(str(e) for e in runtime_errors)
+    )
+
+
+@then("a2aconcur all published events should be in the queue")
+def step_a2aconcur_events_in_queue(context: Any) -> None:
+    events = context.a2aconcur_queue.get_events(limit=10000)
+    assert len(events) > 0, "Expected at least one event in the queue"
+
+
+@then("a2aconcur the queue should have a threading Lock attribute")
+def step_a2aconcur_has_lock(context: Any) -> None:
+    queue: A2aEventQueue = context.a2aconcur_queue
+    assert hasattr(queue, "_lock"), "A2aEventQueue should have a _lock attribute"
+    assert isinstance(queue._lock, type(threading.Lock())), (
+        f"Expected threading.Lock, got {type(queue._lock)}"
+    )
+
+
+@then("a2aconcur the event should be in the queue")
+def step_a2aconcur_event_in_queue(context: Any) -> None:
+    events = context.a2aconcur_queue.get_events(limit=1000)
+    ids = [e.event_id for e in events]
+    assert context.a2aconcur_event.event_id in ids, (
+        f"Event {context.a2aconcur_event.event_id} not found in queue: {ids}"
+    )

src/cleveragents/a2a/events.py

@@ -16,6 +16,7 @@ publishes translated :class:`A2aEvent` instances to an event queue.
 from __future__ import annotations
 
 import json
+import threading
 from collections.abc import Callable
 from typing import Any, ClassVar
 
@@ -41,12 +42,24 @@ _REMOTE_MSG = (
 
 
 class A2aEventQueue:
-    """In-memory event queue with local pub/sub and remote stub."""
+    """In-memory event queue with local pub/sub and remote stub.
+
+    All mutations to ``_subscriptions`` and ``_events`` are protected by a
+    :class:`threading.Lock` so that concurrent calls to :meth:`publish`,
+    :meth:`subscribe_local`, and :meth:`unsubscribe` from different threads
+    do not raise ``RuntimeError: dictionary changed size during iteration``.
+
+    The :meth:`publish` method snapshots the subscriber dict *inside* the
+    lock and then invokes callbacks *outside* the lock.  This prevents
+    deadlocks when a callback itself calls :meth:`subscribe_local` or
+    :meth:`unsubscribe`.
+    """
 
     def __init__(self) -> None:
         self._events: list[A2aEvent] = []
         self._subscriptions: dict[str, Callable[[A2aEvent], Any]] = {}
         self._is_closed: bool = False
+        self._lock: threading.Lock = threading.Lock()
 
     # ------------------------------------------------------------------
     # Local-mode operations (working)
@@ -58,18 +71,27 @@ class A2aEventQueue:
         return self._is_closed
 
     def publish(self, event: A2aEvent) -> None:
-        """Append *event* to the local queue and notify subscribers."""
+        """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*
+        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")
-        self._events.append(event)
+        with self._lock:
+            self._events.append(event)
+            callbacks = list(self._subscriptions.items())
         logger.debug(
             "a2a.event.published",
             event_id=event.event_id,
             event_type=event.event_type,
         )
-        for sub_id, callback in self._subscriptions.items():
+        for sub_id, callback in callbacks:
             try:
                 callback(event)
             except Exception:
@@ -83,7 +105,8 @@ class A2aEventQueue:
         if not callable(callback):
             raise TypeError("callback must be callable")
         sub_id = str(ULID())
-        self._subscriptions[sub_id] = callback
+        with self._lock:
+            self._subscriptions[sub_id] = callback
         logger.debug("a2a.event.subscribed", subscription_id=sub_id)
         return sub_id
 
@@ -91,7 +114,8 @@ class A2aEventQueue:
         """Remove a subscription.  Returns ``True`` if it existed."""
         if not subscription_id or not isinstance(subscription_id, str):
             raise ValueError("subscription_id must be a non-empty string")
-        removed = self._subscriptions.pop(subscription_id, None) is not None
+        with self._lock:
+            removed = self._subscriptions.pop(subscription_id, None) is not None
         if removed:
             logger.debug("a2a.event.unsubscribed", subscription_id=subscription_id)
         return removed
@@ -100,7 +124,8 @@ class A2aEventQueue:
         """Return the most recent *limit* events from the queue."""
         if not isinstance(limit, int) or limit < 1:
             raise ValueError("limit must be a positive integer")
-        return list(self._events[-limit:])
+        with self._lock:
+            return list(self._events[-limit:])
 
     def close(self) -> None:
         """Remove all subscriptions and clear the event queue.
@@ -108,10 +133,11 @@ class A2aEventQueue:
         Logs the number of subscriptions that were active at the time of
         closing.  This method is safe to call multiple times.
         """
-        self._is_closed = True
-        count = len(self._subscriptions)
-        self._subscriptions.clear()
-        self._events.clear()
+        with self._lock:
+            self._is_closed = True
+            count = len(self._subscriptions)
+            self._subscriptions.clear()
+            self._events.clear()
         if count:
             logger.info("a2a.event_queue.closed", subscription_count=count)
 

test_reports/summary.txt

@@ -0,0 +1,87 @@
+Test Framework: generic
+Total Tests: 10
+Passed: 2
+Failed: 8
+
+--- Test Results ---
+✗ Error Block
+✗ Output Block 2
+✗ Output Block 3
+✓ Output Block 4
+✗ Output Block 5
+✗ Output Block 6
+✗ Output Block 7
+✗ Output Block 8
+✓ nox > Running session unit_tests-3.13
+✗ Error Output
+
+--- Failed Tests ---
+✗ Error Block
+  2026-04-13 06:19:25 [debug    ] detail_level_map_builder.created child_domain=uko-oo: parent_domain=uko-code:
+  2026-04-13 06:19:25 [debug    ] detail_level_map_builder.insert_after after_level=MEMBER_LISTING domain=uko-oo: new_level=CLASS_HIERARCHY
+  2026-04-13 06:19:25 [debug    ] detail_level_map_builder.insert_after after_level=SIGNATURES_WITH_DOCS domain=uko-oo: new_level=VISIBILITY_ANNOTATED
+  2026-04-13 06:19:25 [debug    ] detail_level_map_builder.built domain=uko-oo: max_depth=11 num_levels=12
+  @mock_only
+  Feature: A2aEventQueue thread-safety (issue #7604)
+    As a CleverAgents developer
+    I want A2aEventQueue to be safe for concurrent use from multiple threads
+    So that publish/subscribe/unsubscribe operations never raise
+    RuntimeError: dictionary changed size during iteration
+    Scenario: Concurrent publish and subscribe_local do not raise RuntimeError 
+      Given a2aconcur a fresh A2aEventQueue
+      When a2aconcur I run concurrent publish and subscribe_local from multiple threads
+      Then a2aconcur no RuntimeError should have been raised
+      And a2aconcur all published events should be in the queue
+
+✗ Output Block 2
+    Scenario: Concurrent publish and unsubscribe do not raise RuntimeError 
+      Given a2aconcur a fresh A2aEventQueue with subscribers
+      When a2aconcur I run concurrent publish and unsubscribe from multiple threads
+      Then a2aconcur no RuntimeError should have been raised
+
+✗ Output Block 3
+    Scenario: Concurrent subscribe_local and unsubscribe do not raise RuntimeError 
+      Given a2aconcur a fresh A2aEventQueue
+      When a2aconcur I run concurrent subscribe_local and unsubscribe from multiple threads
+      Then a2aconcur no RuntimeError should have been raised
+
+✗ Output Block 5
+    Scenario: Unsubscribing inside a callback does not affect the current publish iteration 
+      Given a2aconcur a fresh A2aEventQueue
+      And a2aconcur a subscriber that unsubscribes itself during the callback
+      When a2aconcur I publish a valid A2aEvent
+      Then a2aconcur no RuntimeError should have been raised
+      And a2aconcur the event should be in the queue
+
+✗ Output Block 6
+    Scenario: Concurrent publish and get_events do not raise RuntimeError 
+      Given a2aconcur a fresh A2aEventQueue
+      When a2aconcur I run concurrent publish and get_events from multiple threads
+      Then a2aconcur no RuntimeError should have been raised
+
+✗ Output Block 7
+  1 feature passed, 0 failed, 0 skipped
+  6 scenarios passed, 0 failed, 0 skipped
+  20 steps passed, 0 failed, 0 skipped
+  Took 0min 0.034s
+
+✗ Output Block 8
+  Overall summary:
+  1 features passed, 0 failed, 0 errored, 0 skipped
+  6 scenarios passed, 0 failed, 0 errored, 0 skipped
+  20 steps passed, 0 failed, 0 errored, 0 skipped
+  Took 0.034s
+  Wall time: 2m 7.653s
+
+✗ Error Output
+  nox > Running session unit_tests-3.13
+  nox > Reusing existing virtual environment at .nox/unit_tests-3-13.
+  nox > uv pip install -e '.[tests]'
+  nox > uv pip install setuptools wheel
+  nox > uv pip install .nox/unit_tests-3-13/tmp/behave-parallel-inprocess
+  nox > python scripts/create_template_db.py /tmp/impl-worker-7604-1776035000/cleveragents-core/build/.template-migrated.db
+  nox > python -m compileall -q features/
+  nox > /tmp/impl-worker-7604-1776035000/cleveragents-core/.nox/unit_tests-3-13/bin/behave-parallel -q features/a2a_event_queue_concurrency.feature --processes 1
+  nox > Session unit_tests-3.13 was successful in 2 minutes.
+  
+

test_reports/test_results.json

@@ -0,0 +1,146 @@
+{
+  "framework": "generic",
+  "tests": [
+    {
+      "name": "Error Block",
+      "passed": false,
+      "output": [
+        "2026-04-13 06:19:25 [debug    ] detail_level_map_builder.created child_domain=uko-oo: parent_domain=uko-code:",
+        "2026-04-13 06:19:25 [debug    ] detail_level_map_builder.insert_after after_level=MEMBER_LISTING domain=uko-oo: new_level=CLASS_HIERARCHY",
+        "2026-04-13 06:19:25 [debug    ] detail_level_map_builder.insert_after after_level=SIGNATURES_WITH_DOCS domain=uko-oo: new_level=VISIBILITY_ANNOTATED",
+        "2026-04-13 06:19:25 [debug    ] detail_level_map_builder.built domain=uko-oo: max_depth=11 num_levels=12",
+        "@mock_only",
+        "Feature: A2aEventQueue thread-safety (issue #7604)",
+        "  As a CleverAgents developer",
+        "  I want A2aEventQueue to be safe for concurrent use from multiple threads",
+        "  So that publish/subscribe/unsubscribe operations never raise",
+        "  RuntimeError: dictionary changed size during iteration",
+        "  Scenario: Concurrent publish and subscribe_local do not raise RuntimeError ",
+        "    Given a2aconcur a fresh A2aEventQueue",
+        "    When a2aconcur I run concurrent publish and subscribe_local from multiple threads",
+        "    Then a2aconcur no RuntimeError should have been raised",
+        "    And a2aconcur all published events should be in the queue"
+      ],
+      "rawOutput": "2026-04-13 06:19:25 [debug    ] detail_level_map_builder.created child_domain=uko-oo: parent_domain=uko-code:\n2026-04-13 06:19:25 [debug    ] detail_level_map_builder.insert_after after_level=MEMBER_LISTING domain=uko-oo: new_level=CLASS_HIERARCHY\n2026-04-13 06:19:25 [debug    ] detail_level_map_builder.insert_after after_level=SIGNATURES_WITH_DOCS domain=uko-oo: new_level=VISIBILITY_ANNOTATED\n2026-04-13 06:19:25 [debug    ] detail_level_map_builder.built domain=uko-oo: max_depth=11 num_levels=12\n@mock_only\nFeature: A2aEventQueue thread-safety (issue #7604)\n  As a CleverAgents developer\n  I want A2aEventQueue to be safe for concurrent use from multiple threads\n  So that publish/subscribe/unsubscribe operations never raise\n  RuntimeError: dictionary changed size during iteration\n  Scenario: Concurrent publish and subscribe_local do not raise RuntimeError \n    Given a2aconcur a fresh A2aEventQueue\n    When a2aconcur I run concurrent publish and subscribe_local from multiple threads\n    Then a2aconcur no RuntimeError should have been raised\n    And a2aconcur all published events should be in the queue"
+    },
+    {
+      "name": "Output Block 2",
+      "passed": false,
+      "output": [
+        "  Scenario: Concurrent publish and unsubscribe do not raise RuntimeError ",
+        "    Given a2aconcur a fresh A2aEventQueue with subscribers",
+        "    When a2aconcur I run concurrent publish and unsubscribe from multiple threads",
+        "    Then a2aconcur no RuntimeError should have been raised"
+      ],
+      "rawOutput": "  Scenario: Concurrent publish and unsubscribe do not raise RuntimeError \n    Given a2aconcur a fresh A2aEventQueue with subscribers\n    When a2aconcur I run concurrent publish and unsubscribe from multiple threads\n    Then a2aconcur no RuntimeError should have been raised"
+    },
+    {
+      "name": "Output Block 3",
+      "passed": false,
+      "output": [
+        "  Scenario: Concurrent subscribe_local and unsubscribe do not raise RuntimeError ",
+        "    Given a2aconcur a fresh A2aEventQueue",
+        "    When a2aconcur I run concurrent subscribe_local and unsubscribe from multiple threads",
+        "    Then a2aconcur no RuntimeError should have been raised"
+      ],
+      "rawOutput": "  Scenario: Concurrent subscribe_local and unsubscribe do not raise RuntimeError \n    Given a2aconcur a fresh A2aEventQueue\n    When a2aconcur I run concurrent subscribe_local and unsubscribe from multiple threads\n    Then a2aconcur no RuntimeError should have been raised"
+    },
+    {
+      "name": "Output Block 4",
+      "passed": true,
+      "output": [
+        "  Scenario: A2aEventQueue has a threading.Lock after construction ",
+        "    Given a2aconcur a fresh A2aEventQueue",
+        "    Then a2aconcur the queue should have a threading Lock attribute"
+      ],
+      "rawOutput": "  Scenario: A2aEventQueue has a threading.Lock after construction \n    Given a2aconcur a fresh A2aEventQueue\n    Then a2aconcur the queue should have a threading Lock attribute"
+    },
+    {
+      "name": "Output Block 5",
+      "passed": false,
+      "output": [
+        "  Scenario: Unsubscribing inside a callback does not affect the current publish iteration ",
+        "    Given a2aconcur a fresh A2aEventQueue",
+        "    And a2aconcur a subscriber that unsubscribes itself during the callback",
+        "    When a2aconcur I publish a valid A2aEvent",
+        "    Then a2aconcur no RuntimeError should have been raised",
+        "    And a2aconcur the event should be in the queue"
+      ],
+      "rawOutput": "  Scenario: Unsubscribing inside a callback does not affect the current publish iteration \n    Given a2aconcur a fresh A2aEventQueue\n    And a2aconcur a subscriber that unsubscribes itself during the callback\n    When a2aconcur I publish a valid A2aEvent\n    Then a2aconcur no RuntimeError should have been raised\n    And a2aconcur the event should be in the queue"
+    },
+    {
+      "name": "Output Block 6",
+      "passed": false,
+      "output": [
+        "  Scenario: Concurrent publish and get_events do not raise RuntimeError ",
+        "    Given a2aconcur a fresh A2aEventQueue",
+        "    When a2aconcur I run concurrent publish and get_events from multiple threads",
+        "    Then a2aconcur no RuntimeError should have been raised"
+      ],
+      "rawOutput": "  Scenario: Concurrent publish and get_events do not raise RuntimeError \n    Given a2aconcur a fresh A2aEventQueue\n    When a2aconcur I run concurrent publish and get_events from multiple threads\n    Then a2aconcur no RuntimeError should have been raised"
+    },
+    {
+      "name": "Output Block 7",
+      "passed": false,
+      "output": [
+        "1 feature passed, 0 failed, 0 skipped",
+        "6 scenarios passed, 0 failed, 0 skipped",
+        "20 steps passed, 0 failed, 0 skipped",
+        "Took 0min 0.034s"
+      ],
+      "rawOutput": "1 feature passed, 0 failed, 0 skipped\n6 scenarios passed, 0 failed, 0 skipped\n20 steps passed, 0 failed, 0 skipped\nTook 0min 0.034s"
+    },
+    {
+      "name": "Output Block 8",
+      "passed": false,
+      "output": [
+        "Overall summary:",
+        "1 features passed, 0 failed, 0 errored, 0 skipped",
+        "6 scenarios passed, 0 failed, 0 errored, 0 skipped",
+        "20 steps passed, 0 failed, 0 errored, 0 skipped",
+        "Took 0.034s",
+        "Wall time: 2m 7.653s"
+      ],
+      "rawOutput": "Overall summary:\n1 features passed, 0 failed, 0 errored, 0 skipped\n6 scenarios passed, 0 failed, 0 errored, 0 skipped\n20 steps passed, 0 failed, 0 errored, 0 skipped\nTook 0.034s\nWall time: 2m 7.653s"
+    },
+    {
+      "name": "nox > Running session unit_tests-3.13",
+      "passed": true,
+      "output": [
+        "nox > Running session unit_tests-3.13",
+        "nox > Reusing existing virtual environment at .nox/unit_tests-3-13.",
+        "nox > uv pip install -e '.[tests]'",
+        "nox > uv pip install setuptools wheel",
+        "nox > uv pip install .nox/unit_tests-3-13/tmp/behave-parallel-inprocess",
+        "nox > python scripts/create_template_db.py /tmp/impl-worker-7604-1776035000/cleveragents-core/build/.template-migrated.db",
+        "nox > python -m compileall -q features/",
+        "nox > /tmp/impl-worker-7604-1776035000/cleveragents-core/.nox/unit_tests-3-13/bin/behave-parallel -q features/a2a_event_queue_concurrency.feature --processes 1",
+        "nox > Session unit_tests-3.13 was successful in 2 minutes."
+      ],
+      "rawOutput": "nox > Running session unit_tests-3.13\nnox > Reusing existing virtual environment at .nox/unit_tests-3-13.\nnox > uv pip install -e '.[tests]'\nnox > uv pip install setuptools wheel\nnox > uv pip install .nox/unit_tests-3-13/tmp/behave-parallel-inprocess\nnox > python scripts/create_template_db.py /tmp/impl-worker-7604-1776035000/cleveragents-core/build/.template-migrated.db\nnox > python -m compileall -q features/\nnox > /tmp/impl-worker-7604-1776035000/cleveragents-core/.nox/unit_tests-3-13/bin/behave-parallel -q features/a2a_event_queue_concurrency.feature --processes 1\nnox > Session unit_tests-3.13 was successful in 2 minutes."
+    },
+    {
+      "name": "Error Output",
+      "passed": false,
+      "output": [
+        "nox > Running session unit_tests-3.13",
+        "nox > Reusing existing virtual environment at .nox/unit_tests-3-13.",
+        "nox > uv pip install -e '.[tests]'",
+        "nox > uv pip install setuptools wheel",
+        "nox > uv pip install .nox/unit_tests-3-13/tmp/behave-parallel-inprocess",
+        "nox > python scripts/create_template_db.py /tmp/impl-worker-7604-1776035000/cleveragents-core/build/.template-migrated.db",
+        "nox > python -m compileall -q features/",
+        "nox > /tmp/impl-worker-7604-1776035000/cleveragents-core/.nox/unit_tests-3-13/bin/behave-parallel -q features/a2a_event_queue_concurrency.feature --processes 1",
+        "nox > Session unit_tests-3.13 was successful in 2 minutes.",
+        ""
+      ],
+      "rawOutput": "nox > Running session unit_tests-3.13\nnox > Reusing existing virtual environment at .nox/unit_tests-3-13.\nnox > uv pip install -e '.[tests]'\nnox > uv pip install setuptools wheel\nnox > uv pip install .nox/unit_tests-3-13/tmp/behave-parallel-inprocess\nnox > python scripts/create_template_db.py /tmp/impl-worker-7604-1776035000/cleveragents-core/build/.template-migrated.db\nnox > python -m compileall -q features/\nnox > /tmp/impl-worker-7604-1776035000/cleveragents-core/.nox/unit_tests-3-13/bin/behave-parallel -q features/a2a_event_queue_concurrency.feature --processes 1\nnox > Session unit_tests-3.13 was successful in 2 minutes.\n"
+    }
+  ],
+  "summary": {
+    "total": 10,
+    "passed": 2,
+    "failed": 8
+  },
+  "rawOutput": "2026-04-13 06:19:25 [debug    ] detail_level_map_builder.created child_domain=uko-oo: parent_domain=uko-code:\n2026-04-13 06:19:25 [debug    ] detail_level_map_builder.insert_after after_level=MEMBER_LISTING domain=uko-oo: new_level=CLASS_HIERARCHY\n2026-04-13 06:19:25 [debug    ] detail_level_map_builder.insert_after after_level=SIGNATURES_WITH_DOCS domain=uko-oo: new_level=VISIBILITY_ANNOTATED\n2026-04-13 06:19:25 [debug    ] detail_level_map_builder.built domain=uko-oo: max_depth=11 num_levels=12\n@mock_only\nFeature: A2aEventQueue thread-safety (issue #7604)\n  As a CleverAgents developer\n  I want A2aEventQueue to be safe for concurrent use from multiple threads\n  So that publish/subscribe/unsubscribe operations never raise\n  RuntimeError: dictionary changed size during iteration\n  Scenario: Concurrent publish and subscribe_local do not raise RuntimeError \n    Given a2aconcur a fresh A2aEventQueue\n    When a2aconcur I run concurrent publish and subscribe_local from multiple threads\n    Then a2aconcur no RuntimeError should have been raised\n    And a2aconcur all published events should be in the queue\n\n  Scenario: Concurrent publish and unsubscribe do not raise RuntimeError \n    Given a2aconcur a fresh A2aEventQueue with subscribers\n    When a2aconcur I run concurrent publish and unsubscribe from multiple threads\n    Then a2aconcur no RuntimeError should have been raised\n\n  Scenario: Concurrent subscribe_local and unsubscribe do not raise RuntimeError \n    Given a2aconcur a fresh A2aEventQueue\n    When a2aconcur I run concurrent subscribe_local and unsubscribe from multiple threads\n    Then a2aconcur no RuntimeError should have been raised\n\n  Scenario: A2aEventQueue has a threading.Lock after construction \n    Given a2aconcur a fresh A2aEventQueue\n    Then a2aconcur the queue should have a threading Lock attribute\n\n  Scenario: Unsubscribing inside a callback does not affect the current publish iteration \n    Given a2aconcur a fresh A2aEventQueue\n    And a2aconcur a subscriber that unsubscribes itself during the callback\n    When a2aconcur I publish a valid A2aEvent\n    Then a2aconcur no RuntimeError should have been raised\n    And a2aconcur the event should be in the queue\n\n  Scenario: Concurrent publish and get_events do not raise RuntimeError \n    Given a2aconcur a fresh A2aEventQueue\n    When a2aconcur I run concurrent publish and get_events from multiple threads\n    Then a2aconcur no RuntimeError should have been raised\n\n1 feature passed, 0 failed, 0 skipped\n6 scenarios passed, 0 failed, 0 skipped\n20 steps passed, 0 failed, 0 skipped\nTook 0min 0.034s\n\nOverall summary:\n1 features passed, 0 failed, 0 errored, 0 skipped\n6 scenarios passed, 0 failed, 0 errored, 0 skipped\n20 steps passed, 0 failed, 0 errored, 0 skipped\nTook 0.034s\nWall time: 2m 7.653s\n\nnox > Running session unit_tests-3.13\nnox > Reusing existing virtual environment at .nox/unit_tests-3-13.\nnox > uv pip install -e '.[tests]'\nnox > uv pip install setuptools wheel\nnox > uv pip install .nox/unit_tests-3-13/tmp/behave-parallel-inprocess\nnox > python scripts/create_template_db.py /tmp/impl-worker-7604-1776035000/cleveragents-core/build/.template-migrated.db\nnox > python -m compileall -q features/\nnox > /tmp/impl-worker-7604-1776035000/cleveragents-core/.nox/unit_tests-3-13/bin/behave-parallel -q features/a2a_event_queue_concurrency.feature --processes 1\nnox > Session unit_tests-3.13 was successful in 2 minutes."
+}