diff --git a/features/a2a_event_queue_concurrency.feature b/features/a2a_event_queue_concurrency.feature new file mode 100644 index 000000000..a561a5f7d --- /dev/null +++ b/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 diff --git a/features/steps/a2a_event_queue_concurrency_steps.py b/features/steps/a2a_event_queue_concurrency_steps.py new file mode 100644 index 000000000..7bf31ff1f --- /dev/null +++ b/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}" + ) diff --git a/src/cleveragents/a2a/events.py b/src/cleveragents/a2a/events.py index f17ae21e7..f0fa976ed 100644 --- a/src/cleveragents/a2a/events.py +++ b/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) diff --git a/test_reports/summary.txt b/test_reports/summary.txt new file mode 100644 index 000000000..0bba941e4 --- /dev/null +++ b/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. + + diff --git a/test_reports/test_results.json b/test_reports/test_results.json new file mode 100644 index 000000000..e4a991412 --- /dev/null +++ b/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." +} \ No newline at end of file