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

Fix a critical concurrency bug in A2aEventQueue.publish() where the method
iterates over _subscriptions without holding a lock. This causes RuntimeError:
dictionary changed size during iteration when subscribe_local/unsubscribe are
called concurrently from other threads.

The fix introduces a threading.Lock to protect all state mutations and
dictionary access in the A2aEventQueue class, while carefully ensuring callbacks
are invoked outside the lock to prevent potential deadlocks.

Changes:
- src/cleveragents/a2a/events.py: Added _lock attribute, protected
  __init__, is_closed, publish, subscribe_local, unsubscribe, get_events,
  and close methods with threading.Lock snapshot pattern for callbacks.

- features/a2a_event_queue_concurrency.feature: BDD feature file with 6
  scenarios covering concurrent publish/subscribe/unsubscribe safety.

- features/steps/a2a_event_queue_concurrency_steps.py: Step definitions
  implementing multi-threaded concurrency test harness.

ISSUES CLOSED: #7604

CHANGELOG.md

@@ -143,7 +143,7 @@ ensuring data is stored with proper parameter values.
 - **Structural Component Output Validation** (#8164): Replaces exact character matching with structural component checking for output validation. Implements three validators covering plan tree output, decision CLI dicts, and structured session snapshots. The `validate_plan_tree` function validates node dicts for required keys (`decision_id`, `type`, `sequence`, `question`, `children`), ULID format, correct types, and sibling ordering. The `validate_decision_dict` function validates decision CLI output against the `Decision.as_cli_dict()` schema with field presence, type, ULID pattern, confidence range [0..1], and boolean field checks. The `validate_structured_output` function validates the StructuredOutput envelope for `command`, `session_id` (ULID), status membership, `exit_code`, and elements integrity. A unified dispatcher (`validate_structured_component_output`) enables routing by target_type. BDD test coverage added in `features/structural_validation.feature`. [Epic #8137](https://git.cleverthis.com/cleveragents/cleveragents-core/issues/8137)
 - **Fixed `agents actor add --config` crash with nested `actors:` map and `config.actor` combined shorthand** (#11189): The
   CLI command `agents actor add --config` now correctly parses spec-canonical YAML using the
-  nested `actors:` map format with `config.actor: "provider/model"` combined shorthand.
+  nested `actors:` map format with `config.actor: "<provider>/<model>"` combined shorthand.
   Fixed two defects in `ActorConfiguration._extract_v3_actor()`: `type` is now detected at
   the actor-entry level (sibling of `config`), and `config.actor` is parsed as fallback
   when separate `provider`/`model` keys are absent. Added validation to reject malformed
@@ -405,6 +405,13 @@ ensuring data is stored with proper parameter values.
   are all populated for Strategize-phase decisions.
 
 - **`auto_debug` node functions return partial state updates instead of mutating state** (#10494, #10496): The `_analyze_error()`, `_generate_fix()`, `_validate_fix()`, and `_finalize()` methods in `src/cleveragents/agents/graphs/auto_debug.py` now return new `dict[str, Any]` objects containing only the keys they update rather than mutating the input state in-place or returning full state objects. This respects the LangGraph node contract (all `StateGraph` node functions are passed the current state and must return a partial dict representing only their incremental updates). Callers rely on LangGraph's built-in state delta-merge logic to combine these partial updates across successive nodes.
+- **A2aEventQueue thread-safety (#7604)**: Guarded ``_subscriptions`` and ``_events`` with a
+  ``threading.Lock`` to prevent ``RuntimeError: dictionary changed size during iteration`` when
+  ``publish()``, ``subscribe_local()``, and ``unsubscribe()`` are called concurrently from multiple
+  threads. The ``publish()`` method snapshots the subscriber dict inside the lock and invokes
+  callbacks outside the lock to prevent deadlocks. The ``_is_closed`` check in ``publish()`` is
+  performed inside the lock to eliminate a TOCTOU race with concurrent ``close()`` calls, and the
+  ``is_closed`` property reads ``_is_closed`` under the lock for memory-visibility guarantees.
 
 ### Changed
 
@@ -1298,4 +1305,4 @@ iteration` and data corruption under concurrent plan execution. All public
 - **TUI -- Permission Question Widget**: A new inline `PermissionQuestionWidget`
   renders permission requests directly in the conversation stream for single-key
   operations. Users can allow/reject with single-key shortcuts (`a`/`A`/`r`/`R`),
-  navigate with arrow keys, confirm with `Enter`, or press `v` to open the full
+  navigate with arrow keys, confirm with `Enter`, or press `v` to open the full

CONTRIBUTORS.md

@@ -61,6 +61,7 @@ Below are some of the specific details of various contributions.
 * HAL 9000 has contributed the git worktree TOCTOU race condition fix (PR #8178 / issue #7507): replaced the unsafe mkdtemp() + rmdir() pattern with a parent-directory approach to eliminate the race window in concurrent git worktree operations.
 * HAL 9000 has contributed CLI documentation for version, info, and diagnostics commands (PR #4211 / issue #7592): created a beginner-friendly showcase walkthrough covering fast-path eager flags, rich format output with Rich panels, machine-readable JSON envelope structure, and CI-friendly diagnostics health checks.
 
+* HAL 9000 has contributed automated bug fixes, thread-safety improvements, and concurrency hardening including the A2aEventQueue threading.Lock guard (#7604).
 * HAL 9000 has contributed the git_tools TOCTOU race condition fix (PR #8255 / issue #7619): eliminated the Time-Of-Check-To-Time-Of-Use race in `_get_base_env()` by adding double-checked locking with a module-level `threading.Lock`, preventing concurrent threads from writing conflicting environment snapshots.
 * HAL 9000 has contributed the ActorLoader.list_actors TOCTOU race condition fix (PR #8660 / issue #8588): moved the namespace filter inside the ``with self._lock:`` block in ``list_actors()`` so that dictionary reads and filtering are atomic, eliminating stale results and potential ``RuntimeError`` under concurrent access. Added concurrency BDD coverage via Behave test scenarios using ``threading.Barrier``.
 * HAL 9000 has contributed the mandatory PR compliance checklist to `implementation-supervisor.md` (#9824): added an 8-item checklist to the worker prompt body with concrete items covering CHANGELOG.md, CONTRIBUTORS.md, commit footer, CI verification, BDD tests, Epic reference, labels, and milestone assignment to eliminate systemic PR merge blockers.

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

@@ -1,4 +1,4 @@
-"""A2A event streaming — local queue, SSE formatting, and remote stub.
+"""A2A event streaming - local queue, SSE formatting, and remote stub.
 
 The :class:`A2aEventQueue` provides a working in-memory event queue for
 local mode and a stub for remote subscriptions that raises
@@ -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,35 +42,58 @@ _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)
+    # Local-mode operations (thread-safe)
     # ------------------------------------------------------------------
 
     @property
     def is_closed(self) -> bool:
         """Whether this queue has been closed."""
-        return self._is_closed
+        with self._lock:
+            return self._is_closed
 
     def publish(self, event: A2aEvent) -> None:
-        """Append *event* to the local queue and notify subscribers."""
-        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)
+        """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.  The close check and type check
+        are also performed inside the lock. Callbacks are invoked *outside*
+        the lock to avoid deadlocks when a callback itself modifies
+        subscriptions.
+        """
+        with self._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")
+            callbacks = list(self._subscriptions.items())
+            self._events.append(event)
         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 +107,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 +116,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 +126,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 +135,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)
 
@@ -149,12 +177,12 @@ class SseEventFormatter:
 
     Two trailing newlines terminate the event per the EventSource spec.
     The data payload follows JSON-RPC 2.0 notification format per the A2A
-    protocol specification (§Streaming Architecture).
+    protocol specification (Streaming Architecture).
     """
 
     # Mapping from A2A event type names to JSON-RPC 2.0 method strings.
-    # Per spec §Streaming Architecture: TaskStatusUpdateEvent → task/statusUpdate,
-    # TaskArtifactUpdateEvent → task/artifactUpdate.
+    # Per spec Streaming Architecture: TaskStatusUpdateEvent -> task/statusUpdate,
+    # TaskArtifactUpdateEvent -> task/artifactUpdate.
     _EVENT_TYPE_TO_METHOD: ClassVar[dict[str, str]] = {
         "TaskStatusUpdateEvent": "task/statusUpdate",
         "TaskArtifactUpdateEvent": "task/artifactUpdate",