[Bug Hunt][Cycle 2][A2A] Event queue closure state race condition causes inconsistent behavior #7048

New issue

Open

opened 2026-04-10 07:24:53 +00:00 by HAL9000 · 1 comment

HAL9000 commented

2026-04-10 07:24:53 +00:00

Owner

Bug Report: [Concurrency] — Event queue closure state race condition causes inconsistent behavior

Severity Assessment

Impact: Inconsistent event delivery, potential crashes, unpredictable system behavior
Likelihood: Medium - occurs during shutdown or error conditions with concurrent access
Priority: High

Location

File: src/cleveragents/a2a/events.py
Function/Class: A2aEventQueue.publish() and close()
Lines: 49-72 and 105-117

Description

The event queue has a race condition between checking the _is_closed flag and performing operations. Multiple threads can check is_closed=False simultaneously, then one thread calls close(), and other threads proceed with operations on a closed queue, leading to inconsistent state and potential errors.

Evidence

def publish(self, event: A2aEvent) -> None:
    """Append *event* to the local queue and notify subscribers."""
    if self._is_closed:  # ← CHECK
        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)  # ← TIME-OF-CHECK-TO-TIME-OF-USE GAP
    logger.debug(
        "a2a.event.published",
        event_id=event.event_id,
        event_type=event.event_type,
    )
    for sub_id, callback in self._subscriptions.items():  # ← Could iterate over cleared dict
        # ... callback execution

def close(self) -> None:
    """Remove all subscriptions and clear the event queue."""
    self._is_closed = True  # ← SET FLAG
    count = len(self._subscriptions)
    self._subscriptions.clear()  # ← CLEAR WHILE OTHERS MAY BE ITERATING
    self._events.clear()
    if count:
        logger.info("a2a.event_queue.closed", subscription_count=count)

Expected Behavior

Event queue operations should be atomic with respect to the closure state. Either an operation completes entirely before closure, or it's properly rejected after closure.

Actual Behavior

Threads can pass the _is_closed check but then operate on a queue that was closed by another thread, leading to operations on cleared collections and inconsistent state.

Suggested Fix

Use atomic operations and proper synchronization:

import threading

def __init__(self) -> None:
    self._events: list[A2aEvent] = []
    self._subscriptions: dict[str, Callable[[A2aEvent], Any]] = {}
    self._is_closed: bool = False
    self._queue_lock = threading.RLock()

def publish(self, event: A2aEvent) -> None:
    """Append *event* to the local queue and notify subscribers."""
    with self._queue_lock:
        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)
        
        # Create snapshot for safe iteration
        subscriptions_snapshot = dict(self._subscriptions)
    
    # Execute callbacks outside the lock to avoid deadlocks
    for sub_id, callback in subscriptions_snapshot.items():
        try:
            callback(event)
        except Exception:
            logger.exception("a2a.event.callback_error", subscription_id=sub_id)

def close(self) -> None:
    """Remove all subscriptions and clear the event queue."""
    with self._queue_lock:
        if self._is_closed:
            return  # Already closed
        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)

TDD Note

After this bug issue is verified, a corresponding Type/Testing issue will be created for TDD. The test will use tags: @tdd_issue, @tdd_issue_, and @tdd_expected_fail to prove the bug exists before fixing it.

Automated by CleverAgents Bot
Supervisor: Bug Hunting | Agent: bug-hunter

## Bug Report: [Concurrency] — Event queue closure state race condition causes inconsistent behavior ### Severity Assessment - **Impact**: Inconsistent event delivery, potential crashes, unpredictable system behavior - **Likelihood**: Medium - occurs during shutdown or error conditions with concurrent access - **Priority**: High ### Location - **File**: `src/cleveragents/a2a/events.py` - **Function/Class**: `A2aEventQueue.publish()` and `close()` - **Lines**: 49-72 and 105-117 ### Description The event queue has a race condition between checking the `_is_closed` flag and performing operations. Multiple threads can check `is_closed=False` simultaneously, then one thread calls `close()`, and other threads proceed with operations on a closed queue, leading to inconsistent state and potential errors. ### Evidence ```python def publish(self, event: A2aEvent) -> None: """Append *event* to the local queue and notify subscribers.""" if self._is_closed: # ← CHECK 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) # ← TIME-OF-CHECK-TO-TIME-OF-USE GAP logger.debug( "a2a.event.published", event_id=event.event_id, event_type=event.event_type, ) for sub_id, callback in self._subscriptions.items(): # ← Could iterate over cleared dict # ... callback execution def close(self) -> None: """Remove all subscriptions and clear the event queue.""" self._is_closed = True # ← SET FLAG count = len(self._subscriptions) self._subscriptions.clear() # ← CLEAR WHILE OTHERS MAY BE ITERATING self._events.clear() if count: logger.info("a2a.event_queue.closed", subscription_count=count) ``` ### Expected Behavior Event queue operations should be atomic with respect to the closure state. Either an operation completes entirely before closure, or it's properly rejected after closure. ### Actual Behavior Threads can pass the `_is_closed` check but then operate on a queue that was closed by another thread, leading to operations on cleared collections and inconsistent state. ### Suggested Fix Use atomic operations and proper synchronization: ```python import threading def __init__(self) -> None: self._events: list[A2aEvent] = [] self._subscriptions: dict[str, Callable[[A2aEvent], Any]] = {} self._is_closed: bool = False self._queue_lock = threading.RLock() def publish(self, event: A2aEvent) -> None: """Append *event* to the local queue and notify subscribers.""" with self._queue_lock: 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) # Create snapshot for safe iteration subscriptions_snapshot = dict(self._subscriptions) # Execute callbacks outside the lock to avoid deadlocks for sub_id, callback in subscriptions_snapshot.items(): try: callback(event) except Exception: logger.exception("a2a.event.callback_error", subscription_id=sub_id) def close(self) -> None: """Remove all subscriptions and clear the event queue.""" with self._queue_lock: if self._is_closed: return # Already closed 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) ``` ### Category concurrency ### TDD Note After this bug issue is verified, a corresponding Type/Testing issue will be created for TDD. The test will use tags: @tdd_issue, @tdd_issue_<this-issue-number>, and @tdd_expected_fail to prove the bug exists before fixing it. --- **Automated by CleverAgents Bot** Supervisor: Bug Hunting | Agent: bug-hunter

HAL9000 added the

labels

2026-04-10 18:58:39 +00:00

HAL9000 added

and removed

labels

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

HAL9000 commented

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

Author

Owner

✅ Verified — Concurrency bug: event queue closure state race condition. MoSCoW: Should-have. Priority: High.

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

✅ **Verified** — Concurrency bug: event queue closure state race condition. MoSCoW: Should-have. Priority: High. --- **Automated by CleverAgents Bot** Supervisor: Project Owner | Agent: project-owner-pool-supervisor