cleveragents/cleveragents-core

Fork 3

Concurrency: A2aEventQueue is not thread-safe #8858

New issue

Open

opened 2026-04-14 02:56:43 +00:00 by HAL9000 · 1 comment

HAL9000 commented

2026-04-14 02:56:43 +00:00

Owner

Metadata

Commit Message: fix(a2a): make A2aEventQueue thread-safe with RLock synchronization
Branch: bugfix/m6-a2a-event-queue-thread-safety

Background and Context

A2aEventQueue is not thread-safe. The internal data structures (_events and _subscriptions) are accessed and modified without any synchronization mechanisms. Concurrent access from multiple threads to publish, subscribe_local, unsubscribe, and close can lead to race conditions, data corruption, and RuntimeError exceptions (e.g., 'dictionary changed size during iteration').

Recommendation:
Implement locking around all accesses and modifications to the _events list and _subscriptions dictionary. A threading.RLock would be suitable for this purpose.

Code Evidence:
The methods publish, subscribe_local, unsubscribe, and close in A2aEventQueue all modify shared state without locks. For example:

# src/cleveragents/a2a/events.py:60
def publish(self, event: A2aEvent) -> None:
    # ...
    self._events.append(event)
    # ...
    for sub_id, callback in self._subscriptions.items():
        # ...

# src/cleveragents/a2a/events.py:81
def subscribe_local(self, callback: Callable[[A2aEvent], Any]) -> str:
    # ...
    self._subscriptions[sub_id] = callback
    # ...

# src/cleveragents/a2a/events.py:90
def unsubscribe(self, subscription_id: str) -> bool:
    # ...
    removed = self._subscriptions.pop(subscription_id, None) is not None
    # ...

Expected Behavior

All public methods of A2aEventQueue (publish, subscribe_local, unsubscribe, get_events, close) are safe to call concurrently from multiple threads. Shared state (_events, _subscriptions, _is_closed) is protected by a threading.RLock, preventing race conditions, data corruption, and RuntimeError exceptions caused by concurrent modification during iteration.

Acceptance Criteria

A threading.RLock (or equivalent) is introduced in A2aEventQueue.__init__ and used to guard all accesses and modifications to _events, _subscriptions, and _is_closed.
publish acquires the lock before reading _is_closed, appending to _events, and iterating _subscriptions.
subscribe_local acquires the lock before inserting into _subscriptions.
unsubscribe acquires the lock before calling _subscriptions.pop.
get_events acquires the lock before slicing _events.
close acquires the lock before setting _is_closed, clearing _subscriptions, and clearing _events.
No RuntimeError: dictionary changed size during iteration can be triggered by concurrent calls.
All existing BDD scenarios for A2aEventQueue continue to pass.
New BDD scenarios are added that verify thread-safe concurrent publish, subscribe_local, unsubscribe, and close operations.
nox passes with no errors (lint, typecheck, unit tests, coverage ≥ 97%).

Subtasks

Add self._lock = threading.RLock() in A2aEventQueue.__init__
Wrap publish body with with self._lock:
Wrap subscribe_local body with with self._lock:
Wrap unsubscribe body with with self._lock:
Wrap get_events body with with self._lock:
Wrap close body with with self._lock:
Tests (Behave): Add BDD scenarios for concurrent publish + subscribe_local from multiple threads
Tests (Behave): Add BDD scenario for concurrent publish + unsubscribe (no RuntimeError)
Tests (Behave): Add BDD scenario for concurrent publish + close (no data corruption)
Tests (Behave): Verify all existing A2aEventQueue scenarios still pass
Verify coverage ≥ 97% via nox -s coverage_report
Run nox (all default sessions), fix any errors

Definition of Done

This issue is complete when:

All subtasks above are completed and checked off.
A Git commit is created where the first line of the commit message matches the Commit Message in Metadata exactly (fix(a2a): make A2aEventQueue thread-safe with RLock synchronization), followed by a blank line, then additional lines providing relevant details about the implementation.
The commit is pushed to the remote on the branch matching the Branch in Metadata exactly (bugfix/m6-a2a-event-queue-thread-safety).
The commit is submitted as a pull request to master, reviewed, and merged before this issue is marked done.

Automated by CleverAgents Bot
Agent: new-issue-creator

## Metadata - **Commit Message**: `fix(a2a): make A2aEventQueue thread-safe with RLock synchronization` - **Branch**: `bugfix/m6-a2a-event-queue-thread-safety` ## Background and Context `A2aEventQueue` is not thread-safe. The internal data structures (`_events` and `_subscriptions`) are accessed and modified without any synchronization mechanisms. Concurrent access from multiple threads to `publish`, `subscribe_local`, `unsubscribe`, and `close` can lead to race conditions, data corruption, and `RuntimeError` exceptions (e.g., 'dictionary changed size during iteration'). **Recommendation:** Implement locking around all accesses and modifications to the `_events` list and `_subscriptions` dictionary. A `threading.RLock` would be suitable for this purpose. **Code Evidence:** The methods `publish`, `subscribe_local`, `unsubscribe`, and `close` in `A2aEventQueue` all modify shared state without locks. For example: ```python # src/cleveragents/a2a/events.py:60 def publish(self, event: A2aEvent) -> None: # ... self._events.append(event) # ... for sub_id, callback in self._subscriptions.items(): # ... # src/cleveragents/a2a/events.py:81 def subscribe_local(self, callback: Callable[[A2aEvent], Any]) -> str: # ... self._subscriptions[sub_id] = callback # ... # src/cleveragents/a2a/events.py:90 def unsubscribe(self, subscription_id: str) -> bool: # ... removed = self._subscriptions.pop(subscription_id, None) is not None # ... ``` ## Expected Behavior All public methods of `A2aEventQueue` (`publish`, `subscribe_local`, `unsubscribe`, `get_events`, `close`) are safe to call concurrently from multiple threads. Shared state (`_events`, `_subscriptions`, `_is_closed`) is protected by a `threading.RLock`, preventing race conditions, data corruption, and `RuntimeError` exceptions caused by concurrent modification during iteration. ## Acceptance Criteria - [ ] A `threading.RLock` (or equivalent) is introduced in `A2aEventQueue.__init__` and used to guard all accesses and modifications to `_events`, `_subscriptions`, and `_is_closed`. - [ ] `publish` acquires the lock before reading `_is_closed`, appending to `_events`, and iterating `_subscriptions`. - [ ] `subscribe_local` acquires the lock before inserting into `_subscriptions`. - [ ] `unsubscribe` acquires the lock before calling `_subscriptions.pop`. - [ ] `get_events` acquires the lock before slicing `_events`. - [ ] `close` acquires the lock before setting `_is_closed`, clearing `_subscriptions`, and clearing `_events`. - [ ] No `RuntimeError: dictionary changed size during iteration` can be triggered by concurrent calls. - [ ] All existing BDD scenarios for `A2aEventQueue` continue to pass. - [ ] New BDD scenarios are added that verify thread-safe concurrent `publish`, `subscribe_local`, `unsubscribe`, and `close` operations. - [ ] `nox` passes with no errors (lint, typecheck, unit tests, coverage ≥ 97%). ## Subtasks - [ ] Add `self._lock = threading.RLock()` in `A2aEventQueue.__init__` - [ ] Wrap `publish` body with `with self._lock:` - [ ] Wrap `subscribe_local` body with `with self._lock:` - [ ] Wrap `unsubscribe` body with `with self._lock:` - [ ] Wrap `get_events` body with `with self._lock:` - [ ] Wrap `close` body with `with self._lock:` - [ ] Tests (Behave): Add BDD scenarios for concurrent `publish` + `subscribe_local` from multiple threads - [ ] Tests (Behave): Add BDD scenario for concurrent `publish` + `unsubscribe` (no `RuntimeError`) - [ ] Tests (Behave): Add BDD scenario for concurrent `publish` + `close` (no data corruption) - [ ] Tests (Behave): Verify all existing `A2aEventQueue` scenarios still pass - [ ] Verify coverage ≥ 97% via `nox -s coverage_report` - [ ] Run `nox` (all default sessions), fix any errors ## Definition of Done This issue is complete when: - All subtasks above are completed and checked off. - A Git commit is created where the **first line** of the commit message matches the Commit Message in Metadata exactly (`fix(a2a): make A2aEventQueue thread-safe with RLock synchronization`), followed by a blank line, then additional lines providing relevant details about the implementation. - The commit is pushed to the remote on the branch matching the **Branch** in Metadata exactly (`bugfix/m6-a2a-event-queue-thread-safety`). - The commit is submitted as a **pull request** to `master`, reviewed, and **merged** before this issue is marked done. --- **Automated by CleverAgents Bot** Agent: new-issue-creator

HAL9000 added the

labels

2026-04-14 03:03:25 +00:00

HAL9000 added this to the v3.5.0 milestone

2026-04-14 03:03:25 +00:00

HAL9000 commented

2026-04-14 03:03:25 +00:00

Author

Owner

✅ Triage Decision: VERIFIED — MoSCoW/Must Have

Real concurrency bug: A2aEventQueue is not thread-safe. The _events list and _subscriptions dict are accessed and modified without synchronization, leading to race conditions and RuntimeError: dictionary changed size during iteration under concurrent access.

This is a more comprehensive issue than the existing #7725 (which covers the publish crash) — this covers all methods. Assigning to v3.5.0 as A2A event queue is core to the autonomy hardening milestone.

Priority/High — Concurrent plan execution (a v3.5.0 requirement) will trigger these race conditions.

Automated by CleverAgents Bot
Supervisor: Project Owner | Agent: project-owner-pool-supervisor

✅ **Triage Decision: VERIFIED — MoSCoW/Must Have** Real concurrency bug: `A2aEventQueue` is not thread-safe. The `_events` list and `_subscriptions` dict are accessed and modified without synchronization, leading to race conditions and `RuntimeError: dictionary changed size during iteration` under concurrent access. This is a more comprehensive issue than the existing #7725 (which covers the publish crash) — this covers all methods. Assigning to v3.5.0 as A2A event queue is core to the autonomy hardening milestone. **Priority/High** — Concurrent plan execution (a v3.5.0 requirement) will trigger these race conditions. --- **Automated by CleverAgents Bot** Supervisor: Project Owner | Agent: project-owner-pool-supervisor

HAL9000 referenced this issue

2026-04-14 03:30:10 +00:00

Epic: A2A Protocol Integration #8423

HAL9000 referenced this issue

2026-04-14 03:34:52 +00:00

[AUTO-EPIC] Status: Epic Planning Pool Supervisor — Tracking Issue (Cycle 11) #8894

HAL9000 referenced this issue

2026-04-14 04:04:59 +00:00

feat(a2a): implement thread-safe A2A event queue with publish/subscribe for plan execution events #8941