fix: introduce thread-safe locks in EventBus handler operations

[DebasmitaBose0]

Summary

This PR introduces thread-safe synchronization to the EventBus operations to prevent data races and runtime errors in multi-threaded environments, resolving #134.

Problem

In multi-threaded contexts, concurrent operations on the EventBus (such as subscribing, unsubscribing, publishing events, or querying stats) can cause data races, inconsistent state updates, or concurrency-related runtime errors.

Proposed Changes

• Thread Synchronization: Integrated self._thread_lock = threading.Lock() within the EventBus constructor.
• Thread-Safe Mutation & Access: Wrapped subscription handler registration/unregistration, index lookups, event logging, and stats collection in thread-safe blocks using with self._thread_lock.
• Deadlock Prevention: Refactored the stats() method to count active subscribers directly using len(self._handlers) under the lock, rather than calling self.handler_count(). Since self._thread_lock is a standard Lock (and not an RLock), calling handler_count() (which acquires the lock again) from within another lock-acquired method would lead to a deadlock.

Verification Results

Ran all tests in the repository and verified that they pass with zero issues:

• tests/test_core.py (including TestEventBus test cases) passed successfully.
• The full test suite passed (440 tests total).

Closes #134

Summary by CodeRabbit

Release Notes

• Bug Fixes
  • Enhanced event system stability by improving thread-safety for handler registration, unregistration, event publishing, and state reads (recent events, stats, and handler counts), reducing the risk of race conditions during concurrent operations.

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: 619f5f15-eda9-40f7-a8ec-12033159f429

📥 Commits

Reviewing files that changed from the base of the PR and between 799a227 and 9738bb1.

📒 Files selected for processing (1)

• agentwatch/core/event_bus.py

🚧 Files skipped from review as they are similar to previous changes (1)

• agentwatch/core/event_bus.py

---

📝 Walkthrough

Walkthrough

EventBus in agentwatch/core/event_bus.py gains a threading.Lock (_thread_lock) to guard handler registration, unregistration, event logging, stats updates, handler snapshots in publish, and read operations in get_recent_events, stats, and handler_count.

Changes

EventBus Thread-Safety

Layer / File(s)	Summary
threading import and _thread_lock field agentwatch/core/event_bus.py	Adds import threading and declares _thread_lock = threading.Lock() on EventBus.
Subscribe/unsubscribe paths under _thread_lock agentwatch/core/event_bus.py	subscribe decorator and subscribe_fn perform stale-subscription cleanup and handler/index registration under _thread_lock. A new _unsubscribe_locked helper encapsulates index removal assuming the lock is held. unsubscribe acquires _thread_lock before mutating handler indexes.
publish snapshot and read paths under _thread_lock agentwatch/core/event_bus.py	publish acquires _thread_lock inside the existing async _lock to log the event, update _stats, and snapshot handler IDs and registrations before dispatch. get_recent_events, stats, and handler_count read their fields under _thread_lock; stats computes active_subscribers directly from handler count.

Estimated code review effort

🎯 3 (Moderate) | ⏱️ ~20 minutes

Possibly related PRs

• sreerevanth/AgentWatch#109: Modifies the same publish path in EventBus to serialize mutations of _event_log and _stats and snapshot handler registrations before dispatch—directly overlapping with this PR's publish-lock changes.

Suggested labels

bug, level: advanced, backend, infra, high-priority

Poem

🐇 Hop, hop, through threads I race,
No more data torn from place!
With _thread_lock in my paw,
Publish, subscribe — flawless law.
Safe at last, no race condition,
EventBus runs on a stable mission! 🔒

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title clearly summarizes the main change: adding thread-safe locking to EventBus handler operations.
Linked Issues check	✅ Passed	The PR addresses issue #134 by adding thread-safe locking around subscriptions, unsubscriptions, publishing, and related state access.
Out of Scope Changes check	✅ Passed	The changes stay focused on EventBus concurrency fixes and related state protection, with no clear unrelated additions.
Docstring Coverage	✅ Passed	Docstring coverage is 80.00% which is sufficient. The required threshold is 80.00%.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[github-actions]

🧪 PR Test Results

Check	Result
Tests (pytest tests/)	✅ success
Lint (ruff check .)	❌ failure
Coverage (agentwatch)	74.69%

_{Python 3.12 · commit 9738bb1}

[sreerevanth]

@DebasmitaBose0 please fix your pr's

[sreerevanth]

@DebasmitaBose0 pr is sitting past 1 week ???

[DebasmitaBose0]

@sreerevanth Sorry for the inconvenience caused....I am fixing this

[DebasmitaBose0]

@sreerevanth PR has no conflicts

[sreerevanth]

@DebasmitaBose0 check the ci also please

[sreerevanth]

@DebasmitaBose0 its a simple ci lint issue in case u got stuck im here to help u out just ping me anywhere is fyn

[sreerevanth]

@DebasmitaBose0 do u wanna work on this or no ?

[sreerevanth]

@DebasmitaBose0 please check this

[sreerevanth]

hey @DebasmitaBose0 please fix the linting issues its been sitting for 3 weeks