feat: stream BrainBar brain bus events#295
Conversation
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
There was a problem hiding this comment.
Your free trial has ended. If you'd like to continue receiving code reviews, you can add a payment method here.
|
@greptileai review |
|
@coderabbitai review |
|
@codex review |
|
You need to increase your spend limit or enable usage-based billing to run background agents. Go to Cursor |
|
Warning Rate limit exceeded
You’ve run out of usage credits. Purchase more in the billing tab. ⌛ How to resolve this issue?After the wait time has elapsed, a review can be triggered using the We recommend that you space out your commits to avoid hitting the rate limit. 🚦 How do rate limits work?CodeRabbit enforces hourly rate limits for each developer per organization. Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout. Please see our FAQ for further information. ℹ️ Review info⚙️ Run configurationConfiguration used: Organization UI Review profile: ASSERTIVE Plan: Pro Run ID: 📒 Files selected for processing (8)
✨ Finishing Touches🧪 Generate unit tests (beta)
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. Comment |
There was a problem hiding this comment.
Your free trial has ended. If you'd like to continue receiving code reviews, you can add a payment method here.
✅ Actions performedReview triggered.
|
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: b65df049fb
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| return true | ||
| } | ||
|
|
||
| let delivered = self.writeFramedData(fd: fd, data: framed) |
There was a problem hiding this comment.
When a Content-Length client uses watch-brain-bus on the same socket as normal MCP calls, this write runs from the hub's per-subscriber drain queue while ordinary RPC responses still write the same fd on the server queue. Because the stream socket has no message boundaries and there is no per-client write serialization here, a notification can interleave with sendResponse and corrupt both Content-Length frames for that client.
Useful? React with 👍 / 👎.
| if toolCall.name == "brain_enrich", !isToolError(response) { | ||
| brainBus.publish(.enrichStatus("running")) |
There was a problem hiding this comment.
For a tools/call to brain_enrich, router.handle(request) runs handleBrainEnrich synchronously before this branch is reached, so subscribers only see enrich_status: running after the enrichment call has completed, and there is no matching idle event afterward. Watchers using the new brain bus status will therefore miss the actual busy period and can remain stuck showing enrichment as running after a successful call.
Useful? React with 👍 / 👎.
![macroscopeapp[bot]](https://avatars.githubusercontent.com/in/900172?s=60&v=4){kind=small}
| return true | ||
| } | ||
|
|
||
| let delivered = self.writeFramedData(fd: fd, data: framed) |
There was a problem hiding this comment.
🟡 Medium BrainBar/BrainBarServer.swift:578
writeFramedData(fd:clientFD:data:) at line 578 is invoked from the brain bus subscriber's drain queue (created by BrainBusEventHub.drainNext), while sendResponse(fd:response:useContentLength:) writes to the same file descriptor from self.queue. Concurrent writes from different queues can interleave bytes, corrupting the JSON-RPC stream. Dispatch the writeFramedData call to self.queue to serialize all socket I/O.
🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file brain-bar/Sources/BrainBar/BrainBarServer.swift around line 578:
`writeFramedData(fd:clientFD:data:)` at line 578 is invoked from the brain bus subscriber's drain queue (created by `BrainBusEventHub.drainNext`), while `sendResponse(fd:response:useContentLength:)` writes to the same file descriptor from `self.queue`. Concurrent writes from different queues can interleave bytes, corrupting the JSON-RPC stream. Dispatch the `writeFramedData` call to `self.queue` to serialize all socket I/O.
Evidence trail:
BrainBarServer.swift:95 — `self.queue` defined as serial DispatchQueue
BrainBarServer.swift:392 — read source dispatches on `self.queue`
BrainBarServer.swift:436 — `sendResponse` called from read handler (on `self.queue`)
BrainBarServer.swift:484-525 — `sendResponse` performs raw `write()` at line 500
BrainBarServer.swift:528-552 — `writeFramedData` performs raw `write()` at line 533
BrainBarServer.swift:562-585 — subscriber closure calls `writeFramedData` at line 578
BrainBusEvent.swift:125 — per-subscriber `drainQueue` created as separate DispatchQueue
BrainBusEvent.swift:179 — writer callback dispatched on `drainQueue.async`
| } | ||
|
|
||
| private func publishStoredChunk(stored: StoreResultPayload, content: String, tags: [String], importance: Int) { | ||
| brainBusQueueDepthEstimate += 1 |
There was a problem hiding this comment.
🟡 Medium BrainBar/BrainBarServer.swift:849
brainBusQueueDepthEstimate is incremented at line 849 but never decremented, causing the queueDepth value to grow monotonically and report increasingly incorrect values to brain-bus subscribers. Consider decrementing the estimate after chunks are processed or delivered, or switch to tracking actual in-flight count.
🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file brain-bar/Sources/BrainBar/BrainBarServer.swift around line 849:
`brainBusQueueDepthEstimate` is incremented at line 849 but never decremented, causing the `queueDepth` value to grow monotonically and report increasingly incorrect values to brain-bus subscribers. Consider decrementing the estimate after chunks are processed or delivered, or switch to tracking actual in-flight count.
Evidence trail:
brain-bar/Sources/BrainBar/BrainBarServer.swift:105 (declaration: `private var brainBusQueueDepthEstimate = 0`), line 849 (only increment: `+= 1`), lines 346, 590, 850 (read-only usages publishing the value). git_grep for `brainBusQueueDepthEstimate -=` returned no results, confirming no decrement exists anywhere in the codebase.
There was a problem hiding this comment.
🟠 High
brainlayer/brain-bar/Sources/BrainBar/Dashboard/StatsCollector.swift
Lines 63 to 79 in b65df04
When brainBusEvents is provided, start() skips the initial refresh(force: true) call. If the first events are .healthTick, handleBrainBusEvent only samples daemon/agent activity and never calls refresh(), so stats remains at zeroed initial values until a different event type arrives. Consider calling refresh(force: true) unconditionally before starting the event stream.
guard !isRunning else { return }
isRunning = true
installDarwinObserver()
+ refresh(force: true)
if let brainBusEvents {🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file brain-bar/Sources/BrainBar/Dashboard/StatsCollector.swift around lines 63-79:
When `brainBusEvents` is provided, `start()` skips the initial `refresh(force: true)` call. If the first events are `.healthTick`, `handleBrainBusEvent` only samples daemon/agent activity and never calls `refresh()`, so `stats` remains at zeroed initial values until a different event type arrives. Consider calling `refresh(force: true)` unconditionally before starting the event stream.
Evidence trail:
brain-bar/Sources/BrainBar/Dashboard/StatsCollector.swift lines 47-58 (zeroed initial stats), lines 63-80 (start() method — brainBusEvents branch skips refresh(force:true)), lines 131-139 (handleBrainBusEvent — .healthTick case never calls refresh()), line 35 (lastDataVersion initialized to nil). Reviewed at commit REVIEWED_COMMIT.
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes and found 4 potential issues.
❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.
Reviewed by Cursor Bugbot for commit b65df04. Configure here.
|
|
||
| private func publishStoredChunk(stored: StoreResultPayload, content: String, tags: [String], importance: Int) { | ||
| brainBusQueueDepthEstimate += 1 | ||
| brainBus.publish(.queueDepth(brainBusQueueDepthEstimate)) |
There was a problem hiding this comment.
Queue depth estimate only increments, never decrements
Medium Severity
brainBusQueueDepthEstimate is incremented in publishStoredChunk but never decremented anywhere. This means the "queue depth" reported to subscribers grows monotonically with every stored chunk and never decreases, making it a total-chunks-stored counter rather than an actual queue depth metric. Subscribers receive a meaningless ever-growing number.
Additional Locations (1)
Reviewed by Cursor Bugbot for commit b65df04. Configure here.
| self?.disconnectClient(fd: fd) | ||
| } | ||
| } | ||
| return delivered |
There was a problem hiding this comment.
Concurrent writes to same fd from different queues
Medium Severity
The brain bus subscriber writer calls writeFramedData from the subscriber's drainQueue, while publishStoredChunk calls sendResponse on the server's queue for the same fd. A client that uses both watch-brain-bus and brain_subscribe on one connection gets concurrent unsynchronized write() calls from two threads, corrupting data on the wire.
Additional Locations (1)
Reviewed by Cursor Bugbot for commit b65df04. Configure here.
| } | ||
| if toolCall.name == "brain_enrich", !isToolError(response) { | ||
| brainBus.publish(.enrichStatus("running")) | ||
| } |
There was a problem hiding this comment.
Enrich status published as "running" after completion
Low Severity
The .enrichStatus("running") event is published after router.handle(request) returns — meaning the synchronous enrichment work has already completed. The status is never subsequently reset to "idle", so it remains permanently stuck at "running" for existing subscribers after the first successful brain_enrich call.
Reviewed by Cursor Bugbot for commit b65df04. Configure here.
| brainBus.publish(.dbBusy(database == nil)) | ||
| brainBus.publish(.queueDepth(brainBusQueueDepthEstimate)) | ||
| brainBus.publish(.enrichStatus("idle")) | ||
| brainBus.publish(.healthTick(openConnections: clients.count)) |
There was a problem hiding this comment.
Initial state events broadcast to all subscribers globally
Medium Severity
When a new watch-brain-bus client connects, handleWatchBrainBus publishes four initial-state events (including .enrichStatus("idle")) to the global brainBus, which delivers them to ALL existing subscribers, not just the newly-joined one. This incorrectly resets state for pre-existing subscribers — for example, forcing their enrich status back to "idle" even if enrichment is active — every time any new client connects.
Reviewed by Cursor Bugbot for commit b65df04. Configure here.
1 participant
Summary
watch-brain-busIPC method on the existing BrainBar Unix socket, streaming newline-delimited JSON notifications for queue depth, enrichment status, last stored chunk, DB busy state, and health ticks.BrainBusClientsubscription; Darwin DB notifications remain a fallback path.Stacking
This PR is stacked on Phase D IPC foundation PR #293 (
fix/brainbar-hybrid-parity) and follows its helper-subprocess IPC conventions: one existing UDS, long-lived raw JSON watch connection, existing Content-Length framing preserved for framed callers.Before / After
Before:
StatsCollector.start()created apollTask, slept for one second in a loop, and refreshed dashboard state directly.After:
StatsCollectorsubscribes toBrainBusClientover/tmp/brainbar.sock; dashboard state changes are driven by pushedwatch-brain-busevents, with no UI polling loop.Tailscale Safesocket Pattern
Test Plan
BrainBusEventHubTests,SocketIntegrationTests/testWatchBrainBusStreamsStoreEventsOverRawUnixSocket, andDashboardTests/testStatsCollectorSubscribesToBrainBusWithoutPollingDelay.swift test --package-path brain-bar-> 360 tests passed.Notes
The first push hook run hit
tests/test_arbitration.py::test_drain_daemon_serializes_three_concurrent_producers; the single test passed immediately on rerun and the full hook passed on the second push attempt.Note
Medium Risk
Adds a new long-lived socket subscription path and buffered event fanout, which affects IPC framing/write behavior and could impact client stability or server queueing under load.
Overview
Adds a new
watch-brain-busJSON-RPC method on the BrainBar Unix socket that streamsnotifications/brain-busevents (DB busy, queue depth, enrich status, last stored chunk ID, and periodic health ticks) to subscribed clients, with per-subscriber buffering/backpressure via the newBrainBusEventHub.Updates the dashboard
StatsCollectorto consume these pushed events via a newBrainBusClient(replacing the 1s polling loop when available) and wires this client into app startup. Socket write handling is tweaked (higher EAGAIN retry cap + shared write helper), and new unit/integration tests cover hub ordering/backpressure, dashboard subscription, and end-to-end brain bus streaming.Reviewed by Cursor Bugbot for commit b65df04. Bugbot is set up for automated code reviews on this repo. Configure here.
Note
Stream real-time brain bus events from BrainBar server to dashboard clients
watch-brain-busJSON-RPC method to BrainBarServer.swift that pushesnotifications/brain-busevents (db_busy,queue_depth,enrich_status,last_chunk_id,health_tick) over the existing UNIX socket connection.BrainBusEventHubfor sequenced, backpressure-aware fan-out to multiple subscribers, and BrainBusClient.swift as the client-side consumer that reconnects on disconnect.BrainBusEventSourceis provided.maxWriteRetriesfrom 10 to 50 to reduce disconnects under transient EAGAIN conditions.StatsCollectorno longer polls on a fixed interval when wired to aBrainBusClient; dashboard updates are now event-driven.📊 Macroscope summarized b65df04. 5 files reviewed, 9 issues evaluated, 0 issues filtered, 3 comments posted
🗂️ Filtered Issues