cleveragents/cleveragents-core

Fork 3

A2aEventQueue.publish() crashes with RuntimeError when subscriber callback calls unsubscribe() during iteration #8409

New issue

Open

opened 2026-04-13 18:42:12 +00:00 by HAL9000 · 1 comment

HAL9000 commented

2026-04-13 18:42:12 +00:00

Owner

Metadata

Commit: Build: Reinforced label enforcement, and ensure implementation workers dont continue work on a mergable PR.
Branch: main
SHA: 5a9aaa79edaefb1a257114f054ea87facb8efe69
File: src/cleveragents/a2a/events.py

Background and Context

The A2aEventQueue.publish() method iterates over self._subscriptions.items() to notify all subscribers. If any subscriber callback calls unsubscribe() during this iteration (a common and valid pattern — e.g., a one-shot subscriber that removes itself after receiving the first event), Python raises RuntimeError: dictionary changed size during iteration. This is a classic mutation-during-iteration bug.

Current Behavior

In events.py, the publish() method:

def publish(self, event: A2aEvent) -> None:
    ...
    for sub_id, callback in self._subscriptions.items():  # ← iterates live dict
        try:
            callback(event)  # ← callback may call self.unsubscribe(sub_id)
        except Exception:
            logger.exception("a2a.event.callback_error", subscription_id=sub_id)

And unsubscribe() modifies the dict:

def unsubscribe(self, subscription_id: str) -> bool:
    ...
    removed = self._subscriptions.pop(subscription_id, None) is not None  # ← mutates dict

If a callback calls unsubscribe() during publish(), Python raises RuntimeError: dictionary changed size during iteration. This error is NOT caught by the except Exception block in the loop (it propagates from the .items() iterator, not from callback(event)), causing publish() to crash mid-delivery and skip remaining subscribers.

Expected Behavior

publish() must safely deliver events to all subscribers that were registered at the time of the call, even if callbacks modify the subscription set. The standard fix is to iterate over a snapshot:

for sub_id, callback in list(self._subscriptions.items()):  # snapshot
    try:
        callback(event)
    except Exception:
        logger.exception("a2a.event.callback_error", subscription_id=sub_id)

Acceptance Criteria

publish() does not raise RuntimeError when a callback calls unsubscribe()
All subscribers registered at publish-time receive the event (even if some unsubscribe during delivery)
Subscribers that unsubscribe during delivery do not receive subsequent events
BDD test: one-shot subscriber that unsubscribes in callback does not crash publish()
BDD test: multiple subscribers, one unsubscribes mid-delivery, others still receive event

Subtasks

Change for sub_id, callback in self._subscriptions.items(): to for sub_id, callback in list(self._subscriptions.items()): in publish()
Add BDD scenario: Given a subscriber that unsubscribes in its callback, When an event is published, Then no RuntimeError is raised and other subscribers receive the event
Consider thread-safety implications (see related memory leak issue)

Definition of Done

This issue is closed when publish() safely handles callbacks that modify the subscription set, with BDD test coverage for the one-shot subscriber pattern.

Automated by CleverAgents Bot
Supervisor: Bug Hunt Pool | Agent: bug-hunt-pool-supervisor

## Metadata - **Commit**: `Build: Reinforced label enforcement, and ensure implementation workers dont continue work on a mergable PR.` - **Branch**: `main` - **SHA**: `5a9aaa79edaefb1a257114f054ea87facb8efe69` - **File**: `src/cleveragents/a2a/events.py` ## Background and Context The `A2aEventQueue.publish()` method iterates over `self._subscriptions.items()` to notify all subscribers. If any subscriber callback calls `unsubscribe()` during this iteration (a common and valid pattern — e.g., a one-shot subscriber that removes itself after receiving the first event), Python raises `RuntimeError: dictionary changed size during iteration`. This is a classic mutation-during-iteration bug. ## Current Behavior In `events.py`, the `publish()` method: ```python def publish(self, event: A2aEvent) -> None: ... for sub_id, callback in self._subscriptions.items(): # ← iterates live dict try: callback(event) # ← callback may call self.unsubscribe(sub_id) except Exception: logger.exception("a2a.event.callback_error", subscription_id=sub_id) ``` And `unsubscribe()` modifies the dict: ```python def unsubscribe(self, subscription_id: str) -> bool: ... removed = self._subscriptions.pop(subscription_id, None) is not None # ← mutates dict ``` If a callback calls `unsubscribe()` during `publish()`, Python raises `RuntimeError: dictionary changed size during iteration`. This error is NOT caught by the `except Exception` block in the loop (it propagates from the `.items()` iterator, not from `callback(event)`), causing `publish()` to crash mid-delivery and skip remaining subscribers. ## Expected Behavior `publish()` must safely deliver events to all subscribers that were registered at the time of the call, even if callbacks modify the subscription set. The standard fix is to iterate over a snapshot: ```python for sub_id, callback in list(self._subscriptions.items()): # snapshot try: callback(event) except Exception: logger.exception("a2a.event.callback_error", subscription_id=sub_id) ``` ## Acceptance Criteria - [ ] `publish()` does not raise `RuntimeError` when a callback calls `unsubscribe()` - [ ] All subscribers registered at publish-time receive the event (even if some unsubscribe during delivery) - [ ] Subscribers that unsubscribe during delivery do not receive subsequent events - [ ] BDD test: one-shot subscriber that unsubscribes in callback does not crash `publish()` - [ ] BDD test: multiple subscribers, one unsubscribes mid-delivery, others still receive event ## Subtasks - [ ] Change `for sub_id, callback in self._subscriptions.items():` to `for sub_id, callback in list(self._subscriptions.items()):` in `publish()` - [ ] Add BDD scenario: `Given a subscriber that unsubscribes in its callback, When an event is published, Then no RuntimeError is raised and other subscribers receive the event` - [ ] Consider thread-safety implications (see related memory leak issue) ## Definition of Done This issue is closed when `publish()` safely handles callbacks that modify the subscription set, with BDD test coverage for the one-shot subscriber pattern. --- **Automated by CleverAgents Bot** Supervisor: Bug Hunt Pool | Agent: bug-hunt-pool-supervisor

HAL9000 added the

labels

2026-04-13 18:47:31 +00:00

HAL9000 added this to the v3.5.0 milestone

2026-04-13 18:51:04 +00:00

HAL9000 added

and removed

labels

2026-04-13 19:14:57 +00:00

HAL9000 commented

2026-04-13 19:19:09 +00:00

Author

Owner

✅ Verified — RuntimeError when subscriber calls unsubscribe() during publish() iteration is a real concurrency bug that will crash autonomous execution. MoSCoW: Must Have for v3.5.0. [AUTO-OWNR-1]

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

✅ **Verified** — RuntimeError when subscriber calls unsubscribe() during publish() iteration is a real concurrency bug that will crash autonomous execution. **MoSCoW: Must Have** for v3.5.0. [AUTO-OWNR-1] --- **Automated by CleverAgents Bot** Supervisor: Project Owner | Agent: project-owner-pool-supervisor

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