Add TypeScript implementation and expand Python documentation#22
Closed
pirate wants to merge 83 commits intobrowser-use:mainfrom
Closed
Add TypeScript implementation and expand Python documentation#22pirate wants to merge 83 commits intobrowser-use:mainfrom
pirate wants to merge 83 commits intobrowser-use:mainfrom
Conversation
Add support for middlewares to hook into event bus handler lifecycle
implement swappable EventHistory storage backend
Updated the description to clarify the library's functionality and similarities to JS event systems.
Revise README description for bubus library
…ead of process-until-event
…nGroup for multiple exceptions
Updated project description to reflect multi-language support.
Shortened references to Typescript and applications.
Removed 'library' from the project description.
Added badges for Python and TypeScript implementations, as well as NPM version.
Removed duplicate image and adjusted formatting.
Standalone higher-order function / TC39 decorator that adds configurable retry logic and semaphore-based concurrency limiting to any async function. Works independently of the event bus (on plain functions, class methods, or event handlers). Features: - max_attempts, retry_after, retry_backoff_factor, retry_on_errors, timeout - Global semaphore registry (semaphore_limit, semaphore_name, semaphore_lax) - AsyncLocalStorage-based re-entrancy tracking to prevent deadlocks when nested/recursive calls share the same semaphore - 30 tests covering retry logic, backoff, error filtering, timeouts, semaphore concurrency, re-entrancy, and event bus integration https://claude.ai/code/session_01TyuqFQFwDXa4h5QzQDCUsv
Switch from a separate node:async_hooks import to the existing createAsyncLocalStorage() factory from async_context.ts. This ensures browser compatibility by gracefully degrading to a no-op when AsyncLocalStorage is unavailable. https://claude.ai/code/session_01TyuqFQFwDXa4h5QzQDCUsv
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
14 issues found across 67 files
Prompt for AI agents (all issues)
Check if these issues are valid — if so, understand the root cause of each and fix them.
<file name="bubus/service.py">
<violation number="1" location="bubus/service.py:631">
P2: `total_pending` double-counts queued events because queued events are already in `event_history`, so adding `queue_size` to `pending_in_history` inflates the capacity check and can reject events prematurely.</violation>
<violation number="2" location="bubus/service.py:654">
P2: PENDING middleware notifications are fired via an unawaited task after the event is enqueued, so the run loop can process the event and emit STARTED/COMPLETED before the PENDING hook runs. This creates out-of-order lifecycle notifications for middleware.</violation>
<violation number="3" location="bubus/service.py:785">
P2: find() can miss in-flight events: it only scans completed history and registers a future handler after the scan, but handle_event snapshots handlers at start, so an event already executing won’t trigger notify_find_handler. If the matching event is in-flight during find(), the call can hang waiting for a future event that never comes.</violation>
</file>
<file name="bubus-ts/src/event_handler.ts">
<violation number="1" location="bubus-ts/src/event_handler.ts:106">
P2: Path redaction only handles forward slashes, so Windows stack traces with backslashes won’t be redacted, leaking usernames and producing inconsistent handler IDs.</violation>
</file>
<file name=".github/workflows/publish-npm.yml">
<violation number="1" location=".github/workflows/publish-npm.yml:50">
P2: Directly interpolating `inputs.tag` into a `run` command can lead to command injection if a crafted tag value is provided. Use an env variable to avoid expression substitution inside the script.</violation>
</file>
<file name="bubus-ts/src/types.ts">
<violation number="1" location="bubus-ts/src/types.ts:83">
P2: ZodLiteral in v4 stores literal values in `value`/`values` (Set), not an array. Using `Array.isArray(def.values)` makes `literal` undefined, so literal schemas are misidentified.</violation>
</file>
<file name="README.md">
<violation number="1" location="README.md:358">
P2: Missing `await` on async `bus.find(...)` in the debouncing example; this returns a coroutine and short-circuits the `or` expression, so `event` becomes a coroutine instead of an event.</violation>
<violation number="2" location="README.md:538">
P3: Documentation is inconsistent about the default `max_history_size` (100 vs 50). The Memory Management section claims 100, but the API docs and signature state default 50, which will confuse users.</violation>
</file>
<file name="bubus/models.py">
<violation number="1" location="bubus/models.py:352">
P2: `_process_self_on_all_buses` exits after 1000 iterations without ensuring the event is complete, so `await event` can return an incomplete event if the loop hits the limit.</violation>
</file>
<file name="bubus-ts/src/lock_manager.ts">
<violation number="1" location="bubus-ts/src/lock_manager.ts:71">
P2: AsyncSemaphore can exceed its size because release() decrements in_use before waking a waiter, and the waiter increments later after the await, allowing a new acquire() to slip in and push in_use past the limit.</violation>
</file>
<file name="bubus-ts/src/retry.ts">
<violation number="1" location="bubus-ts/src/retry.ts:234">
P2: retry() always returns an async wrapper, but the generic allows synchronous functions and then casts back to T, so a sync function becomes Promise-returning at runtime while type says otherwise.</violation>
</file>
<file name="tests/test_comprehensive_patterns.py">
<violation number="1" location="tests/test_comprehensive_patterns.py:587">
P2: The test claims to verify multi-bus forwarding, but no forwarding is configured between bus1 and bus2. As written, it only tests isolated buses, not the forwarding scenario described in the docstring.</violation>
</file>
<file name="tests/test_event_result_standalone.py">
<violation number="1" location="tests/test_event_result_standalone.py:46">
P2: Test docstrings claim EventBus is not required, but the tests instantiate and pass an EventBus, so they do not actually verify standalone behavior.</violation>
</file>
<file name="bubus-ts/src/event_result.ts">
<violation number="1" location="bubus-ts/src/event_result.ts:188">
P2: Serializing EventResult via toJSON drops Error details; a standard Error instance becomes `{}` in JSON, losing message/stack/name.</violation>
</file>
Reply with feedback, questions, or to request a fix. Tag @cubic-dev-ai to re-run a review.
Comment on lines
+631
to
633
| if queue_size + pending_in_history >= 100: | ||
| break | ||
| total_pending = queue_size + pending_in_history |
There was a problem hiding this comment.
P2: total_pending double-counts queued events because queued events are already in event_history, so adding queue_size to pending_in_history inflates the capacity check and can reject events prematurely.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At bubus/service.py, line 631:
<comment>`total_pending` double-counts queued events because queued events are already in `event_history`, so adding `queue_size` to `pending_in_history` inflates the capacity check and can reject events prematurely.</comment>
<file context>
@@ -546,7 +624,12 @@ def dispatch(self, event: T_ExpectedEvent) -> T_ExpectedEvent:
+ for existing_event in self.event_history.values():
+ if not self._is_event_complete_fast(existing_event):
+ pending_in_history += 1
+ if queue_size + pending_in_history >= 100:
+ break
total_pending = queue_size + pending_in_history
</file context>
Suggested change
| if queue_size + pending_in_history >= 100: | |
| break | |
| total_pending = queue_size + pending_in_history | |
| if pending_in_history >= 100: | |
| break | |
| total_pending = pending_in_history |
| self._active_event_ids.add(event.event_id) | ||
| if self.middlewares: | ||
| loop = asyncio.get_running_loop() | ||
| loop.create_task(self._on_event_change(event, EventStatus.PENDING)) |
There was a problem hiding this comment.
P2: PENDING middleware notifications are fired via an unawaited task after the event is enqueued, so the run loop can process the event and emit STARTED/COMPLETED before the PENDING hook runs. This creates out-of-order lifecycle notifications for middleware.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At bubus/service.py, line 654:
<comment>PENDING middleware notifications are fired via an unawaited task after the event is enqueued, so the run loop can process the event and emit STARTED/COMPLETED before the PENDING hook runs. This creates out-of-order lifecycle notifications for middleware.</comment>
<file context>
@@ -565,71 +648,244 @@ def dispatch(self, event: T_ExpectedEvent) -> T_ExpectedEvent:
+ self._active_event_ids.add(event.event_id)
+ if self.middlewares:
+ loop = asyncio.get_running_loop()
+ loop.create_task(self._on_event_change(event, EventStatus.PENDING))
+ if logger.isEnabledFor(logging.INFO):
+ logger.info(
</file context>
| events = list(self.event_history.values()) | ||
| for event in reversed(events): | ||
| # Only match completed events in history | ||
| if event.event_completed_at is None: |
There was a problem hiding this comment.
P2: find() can miss in-flight events: it only scans completed history and registers a future handler after the scan, but handle_event snapshots handlers at start, so an event already executing won’t trigger notify_find_handler. If the matching event is in-flight during find(), the call can hang waiting for a future event that never comes.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At bubus/service.py, line 785:
<comment>find() can miss in-flight events: it only scans completed history and registers a future handler after the scan, but handle_event snapshots handlers at start, so an event already executing won’t trigger notify_find_handler. If the matching event is in-flight during find(), the call can hang waiting for a future event that never comes.</comment>
<file context>
@@ -565,71 +648,244 @@ def dispatch(self, event: T_ExpectedEvent) -> T_ExpectedEvent:
+ events = list(self.event_history.values())
+ for event in reversed(events):
+ # Only match completed events in history
+ if event.event_completed_at is None:
+ continue
+ # Skip events older than cutoff (dispatched before the time window)
</file context>
| normalized = path | ||
| } | ||
| } | ||
| normalized = normalized.replace(/\/users\/[^/]+\//i, '~/').replace(/\/home\/[^/]+\//i, '~/') |
There was a problem hiding this comment.
P2: Path redaction only handles forward slashes, so Windows stack traces with backslashes won’t be redacted, leaking usernames and producing inconsistent handler IDs.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At bubus-ts/src/event_handler.ts, line 106:
<comment>Path redaction only handles forward slashes, so Windows stack traces with backslashes won’t be redacted, leaking usernames and producing inconsistent handler IDs.</comment>
<file context>
@@ -0,0 +1,191 @@
+ normalized = path
+ }
+ }
+ normalized = normalized.replace(/\/users\/[^/]+\//i, '~/').replace(/\/home\/[^/]+\//i, '~/')
+ return line_number ? `${normalized}:${line_number}` : normalized
+ }
</file context>
|
|
||
| - name: Publish manual tag | ||
| if: github.event_name == 'workflow_dispatch' | ||
| run: pnpm publish --access public --tag "${{ inputs.tag }}" --no-git-checks |
There was a problem hiding this comment.
P2: Directly interpolating inputs.tag into a run command can lead to command injection if a crafted tag value is provided. Use an env variable to avoid expression substitution inside the script.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At .github/workflows/publish-npm.yml, line 50:
<comment>Directly interpolating `inputs.tag` into a `run` command can lead to command injection if a crafted tag value is provided. Use an env variable to avoid expression substitution inside the script.</comment>
<file context>
@@ -0,0 +1,52 @@
+
+ - name: Publish manual tag
+ if: github.event_name == 'workflow_dispatch'
+ run: pnpm publish --access public --tag "${{ inputs.tag }}" --no-git-checks
+ env:
+ NODE_AUTH_TOKEN: ${{ secrets.NPM_TOKEN }}
</file context>
| } | ||
|
|
||
| Object.defineProperty(retryWrapper, 'name', { value: fn_name, configurable: true }) | ||
| return retryWrapper as unknown as T |
There was a problem hiding this comment.
P2: retry() always returns an async wrapper, but the generic allows synchronous functions and then casts back to T, so a sync function becomes Promise-returning at runtime while type says otherwise.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At bubus-ts/src/retry.ts, line 234:
<comment>retry() always returns an async wrapper, but the generic allows synchronous functions and then casts back to T, so a sync function becomes Promise-returning at runtime while type says otherwise.</comment>
<file context>
@@ -0,0 +1,307 @@
+ }
+
+ Object.defineProperty(retryWrapper, 'name', { value: fn_name, configurable: true })
+ return retryWrapper as unknown as T
+ }
+}
</file context>
|
|
||
| async def test_multi_bus_forwarding_with_queued_events(): | ||
| """ | ||
| Test queue jumping with multiple buses that have forwarding set up, |
There was a problem hiding this comment.
P2: The test claims to verify multi-bus forwarding, but no forwarding is configured between bus1 and bus2. As written, it only tests isolated buses, not the forwarding scenario described in the docstring.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At tests/test_comprehensive_patterns.py, line 587:
<comment>The test claims to verify multi-bus forwarding, but no forwarding is configured between bus1 and bus2. As written, it only tests isolated buses, not the forwarding scenario described in the docstring.</comment>
<file context>
@@ -244,10 +294,676 @@ def bad_handler(bad: BaseEvent[Any]) -> None:
+
+async def test_multi_bus_forwarding_with_queued_events():
+ """
+ Test queue jumping with multiple buses that have forwarding set up,
+ where both buses already have events queued.
+
</file context>
| async def handler(event: _StubEvent) -> str: | ||
| return 'ok' | ||
|
|
||
| test_bus = EventBus(name='StandaloneTest1') |
There was a problem hiding this comment.
P2: Test docstrings claim EventBus is not required, but the tests instantiate and pass an EventBus, so they do not actually verify standalone behavior.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At tests/test_event_result_standalone.py, line 46:
<comment>Test docstrings claim EventBus is not required, but the tests instantiate and pass an EventBus, so they do not actually verify standalone behavior.</comment>
<file context>
@@ -0,0 +1,92 @@
+ async def handler(event: _StubEvent) -> str:
+ return 'ok'
+
+ test_bus = EventBus(name='StandaloneTest1')
+ result_value = await event_result.execute(
+ cast(BaseEvent[Any], stub_event),
</file context>
| completed_at: this.completed_at, | ||
| completed_ts: this.completed_ts, | ||
| result: this.result, | ||
| error: this.error, |
There was a problem hiding this comment.
P2: Serializing EventResult via toJSON drops Error details; a standard Error instance becomes {} in JSON, losing message/stack/name.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At bubus-ts/src/event_result.ts, line 188:
<comment>Serializing EventResult via toJSON drops Error details; a standard Error instance becomes `{}` in JSON, losing message/stack/name.</comment>
<file context>
@@ -0,0 +1,252 @@
+ completed_at: this.completed_at,
+ completed_ts: this.completed_ts,
+ result: this.result,
+ error: this.error,
+ event_children: this.event_children.map((child) => child.event_id),
+ }
</file context>
|
|
||
| ```python | ||
| # Create a bus with memory limits (default: 50 events) | ||
| # Create a bus with memory limits (default: 100 events) |
There was a problem hiding this comment.
P3: Documentation is inconsistent about the default max_history_size (100 vs 50). The Memory Management section claims 100, but the API docs and signature state default 50, which will confuse users.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At README.md, line 538:
<comment>Documentation is inconsistent about the default `max_history_size` (100 vs 50). The Memory Management section claims 100, but the API docs and signature state default 50, which will confuse users.</comment>
<file context>
@@ -412,12 +467,75 @@ email_list = await event_bus.dispatch(FetchInboxEvent(account_id='124', ...)).ev
```python
-# Create a bus with memory limits (default: 50 events)
+# Create a bus with memory limits (default: 100 events)
bus = EventBus(max_history_size=100) # Keep max 100 events in history
</file context>
Suggested change
| # Create a bus with memory limits (default: 100 events) | |
| # Create a bus with memory limits (default: 50 events) |
retry_on_errors now accepts a mix of: - Error class constructors (instanceof check) - String error names (matched against error.name) - RegExp patterns (tested against String(error)) https://claude.ai/code/session_01TyuqFQFwDXa4h5QzQDCUsv
- 'global': all calls share one semaphore (default, existing behavior) - 'class': keyed by constructor.name — all instances of a class share one - 'instance': keyed by WeakMap identity — each object gets its own Falls back to 'global' when `this` is not an object (standalone calls). Multiprocess scope is not supported (single-process JS runtime). https://claude.ai/code/session_01TyuqFQFwDXa4h5QzQDCUsv
There was a problem hiding this comment.
1 issue found across 3 files (changes from recent commits).
Prompt for AI agents (all issues)
Check if these issues are valid — if so, understand the root cause of each and fix them.
<file name="bubus-ts/src/retry.ts">
<violation number="1" location="bubus-ts/src/retry.ts:193">
P2: Instance-scoped semaphores are stored in a global Map keyed by a unique instance id string, but entries are never removed. This causes unbounded growth in SEMAPHORE_REGISTRY as instances are created and destroyed in long-lived processes.</violation>
</file>
Reply with feedback, questions, or to request a fix. Tag @cubic-dev-ai to re-run a review.
| let semaphore_acquired = false | ||
|
|
||
| if (needs_semaphore && !is_reentrant) { | ||
| semaphore = getOrCreateSemaphore(scoped_key, semaphore_limit!) |
There was a problem hiding this comment.
P2: Instance-scoped semaphores are stored in a global Map keyed by a unique instance id string, but entries are never removed. This causes unbounded growth in SEMAPHORE_REGISTRY as instances are created and destroyed in long-lived processes.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At bubus-ts/src/retry.ts, line 193:
<comment>Instance-scoped semaphores are stored in a global Map keyed by a unique instance id string, but entries are never removed. This causes unbounded growth in SEMAPHORE_REGISTRY as instances are created and destroyed in long-lived processes.</comment>
<file context>
@@ -149,17 +177,20 @@ export function retry(options: RetryOptions = {}) {
if (needs_semaphore && !is_reentrant) {
- semaphore = getOrCreateSemaphore(sem_name, semaphore_limit!)
+ semaphore = getOrCreateSemaphore(scoped_key, semaphore_limit!)
const effective_sem_timeout =
</file context>
Tests verify: @Retry() works with native TC39 Stage 3 decorator syntax on class methods, preserves `this` context, composes with semaphore_scope (class/instance), works with bus.on() via .bind(), and class/instance scopes correctly fall back to global for standalone functions. https://claude.ai/code/session_01TyuqFQFwDXa4h5QzQDCUsv
Tests added (12 new, 51 total retry tests): - TC39 @Retry() decorator on class methods with all 3 scopes - @Retry + bus.on via .bind(this) for class/instance/global scopes - HOF retry()(fn).bind(instance) pattern (bind after wrap) - HOF retry()(fn.bind(instance)) → verifies scope falls back to global - Standalone functions with class/instance scope → fall back to global README updated: - TC39 decorator syntax examples with bus.on + .bind(this) - HOF .bind() ordering requirement documented - Note on scope fallback for standalone/unbound functions Also fixed flaky bus tests caused by handler ID collision (bus uses ms-precision timestamps in handler ID hash — added 2ms delay between same-millisecond handler registrations). https://claude.ai/code/session_01TyuqFQFwDXa4h5QzQDCUsv
…ents Verifies the pattern where retry() wraps the full bus.emit→event.done() cycle so each retry dispatches a fresh event, while other events race in parallel via Promise.all. https://claude.ai/code/session_01TyuqFQFwDXa4h5QzQDCUsv
… wrapping Rewrite README retry section to establish @Retry() on class methods as the primary recommended pattern. Explain why retry/timeout is a handler-level concern (handlers fail, events don't), why emit-level retry hurts replayability/determinism, and how retry semaphores are orthogonal to bus concurrency options. Mark the emit→done wrapping pattern as technically supported but not recommended, with clear rationale. Reorganize test section headers to reflect the recommended pattern hierarchy. https://claude.ai/code/session_01TyuqFQFwDXa4h5QzQDCUsv
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
This PR introduces a TypeScript/Node.js implementation of bubus alongside the existing Python library, significantly expands the README with comprehensive documentation, and updates the project configuration to support multi-language development.
Key Changes
New TypeScript Implementation
bubus-ts/directory with TypeScript event bus implementation.on(),.emit(),.find(), etc.).github/workflows/publish-npm.yml)Documentation Improvements
find()(new unified method) andexpect()(backwards-compatible wrapper)API Enhancements
find()method as the recommended unified approach for event lookup with flexiblepast/futureparametersquery()method for searching completed event historyevent_is_child_of()andevent_is_parent_of()helper methodsEventBusMiddlewaresystem for extensible handler lifecycle hooksSQLiteHistoryMirrorMiddlewarefor event persistence and auditingConfiguration Updates
.claude/settings.local.jsonto allow broader pytest and git operations.gitignoreto includenode_modules/and*.sqlite*fileswal_pathparameter to middleware-based approachBug Fixes & Improvements
Notable Implementation Details
find()method unifies history search and future waiting with intuitive boolean/float parametershttps://claude.ai/code/session_01TyuqFQFwDXa4h5QzQDCUsv
Summary by cubic
Adds a full TypeScript/ESM port of bubus (with a TC39-decorator-compatible retry) and upgrades the Python package with middleware, a pluggable history backend, ContextVar propagation, and a unified find/query API. Expands docs, adds npm publish CI, introduces semaphore_scope for retry, and adds a new test that wraps the full emit→done cycle in parallel with other events.
New Features
Migration
Written for commit df34d9d. Summary will update on new commits.