Have Pear subscribe directly to Relaycast's observer stream for full chat coverage#392
Have Pear subscribe directly to Relaycast's observer stream for full chat coverage#392willwashburn wants to merge 2 commits into
Conversation
…r full chat coverage The local broker's own dashboard WS only surfaces chat messages via two narrow paths (the broker's own send handler, and node-controlled PTY delivery), so it misses anything an agent's MCP server posts directly to Relaycast (bypassing the broker send path) and anything in a channel with no locally-attached worker (node delivery to this broker never fires). Pear's main process now opens a second, direct-to-Relaycast /v1/ws subscription per open project — the same mechanism the hosted observer dashboard already uses — scoped to an observer token minted via the existing mintObserverToken() path. Matching message-class frames are synthesized into the same relay_inbound shape and pushed through the existing publishBrokerEvent ingestion point, so the renderer's dedup/store logic handles them exactly like local-broker events, unchanged. Lifecycle is tied to the same session bookkeeping mintObserverToken's cache already uses (dropSession, project-scoped so a project's local+cloud sessions share one stream, torn down only once neither remains). Reconnect/ backoff comes for free from @relaycast/sdk's WsClient, which is what the hosted observer dashboard build on already. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
📝 WalkthroughWalkthroughAdds a per-project Relaycast observer WebSocket subscription to BrokerManager that forwards observer chat frames (message.created, thread.reply) into the existing broker event ingestion path as relay_inbound payloads, with lifecycle wiring across session start/attach/teardown paths, plus corresponding tests. ChangesRelaycast Observer Stream Integration
Estimated code review effort: 3 (Moderate) | ~30 minutes Sequence Diagram(s)sequenceDiagram
participant Relaycast as Relaycast WS
participant BrokerManager
participant Validator as validateIngressBrokerEvent
participant Renderer
BrokerManager->>BrokerManager: ensureObserverStream(projectId)
BrokerManager->>Relaycast: connect(WsClient, mintObserverToken)
Relaycast-->>BrokerManager: emit observer frame (message.created/thread.reply)
BrokerManager->>BrokerManager: handleObserverStreamEvent(projectId, event)
BrokerManager->>Validator: validate synthesized relay_inbound payload
Validator-->>BrokerManager: validated payload
BrokerManager->>Renderer: publishBrokerEvent(relay_inbound)
Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 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 |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request integrates a direct-to-Relaycast observer stream into the BrokerManager to capture and forward messages sent directly to Relaycast, along with comprehensive tests. The feedback suggests broadcasting live events to all active sibling sessions in multi-session environments, wrapping the lazy token minting function in a try/catch block for better error observability, and using strict undefined checks instead of falsy checks to prevent dropping empty message bodies.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| const [session] = this.sessionsForProject(projectId) | ||
| if (!session) return | ||
| this.publishBrokerEvent(sessionKeyFor(session), projectId, session.window, forwarded) |
There was a problem hiding this comment.
In a multi-session environment (e.g., when both a local broker and a cloud sandbox session are active for the same project), sessionsForProject(projectId) will return multiple sessions. By only destructuring the first session (const [session] = ...), the live observer stream events will only be published to the first session's window. Sibling sessions (such as the cloud sandbox session) will miss these live events on their respective windows.
To ensure all active sessions for the project receive the live events, we should record the event once in history and then dispatch the IPC event to all active windows.
const sessions = this.sessionsForProject(projectId)
if (sessions.length === 0) return
const firstSession = sessions[0]
const record = this.publishBrokerEvent(sessionKeyFor(firstSession), projectId, firstSession.window, forwarded)
for (let i = 1; i < sessions.length; i++) {
const sibling = sessions[i]
const targetWindow = this.windowForSession(sessionKeyFor(sibling), sibling.window)
if (targetWindow && !targetWindow.isDestroyed()) {
targetWindow.webContents.send('broker:event', {
...forwarded,
projectId,
observedAt: record.timestamp,
historyId: record.id
})
}
}| // the same cached/coalesced `mintObserverToken` path the "Join as | ||
| // observer" link uses, rather than one token baked in at construction | ||
| // time. | ||
| token: async () => (await this.mintObserverToken(projectId)).token, |
There was a problem hiding this comment.
If mintObserverToken rejects (e.g., due to network issues, broker timeouts, or version mismatches), the error will propagate up. Wrapping this call in a try/catch block to log the failure to console.error ensures better observability and prevents potential unhandled promise rejections in the Electron main process.
token: async () => {
try {
return (await this.mintObserverToken(projectId)).token
} catch (err) {
console.error(`[broker] Failed to mint observer token for project ${projectId}:`, err)
throw err
}
},| const id = typeof raw.id === 'string' ? raw.id : undefined | ||
| const from = typeof raw.agentName === 'string' ? raw.agentName : undefined | ||
| const body = typeof raw.text === 'string' ? raw.text : undefined | ||
| if (!id || !from || !body) return null |
There was a problem hiding this comment.
Using falsy checks (!id || !from || !body) can lead to unexpected bugs if any of these fields are empty strings (""). For example, if an agent sends an empty message body, body will be "", which is falsy, causing the entire message to be dropped. It is safer to strictly check for undefined to ensure robust handling of empty strings.
if (id === undefined || from === undefined || body === undefined) return null![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/main/broker.ts (1)
4418-4443: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winStop residual observer streams even when the session is already gone.
Line 4419 returns before the observer teardown check, so
shutdown()’s no-live-session cleanup path cannot clear a leftover project observer stream. Move the project-scoped stream cleanup into the no-session branch too.Suggested fix
const session = this.sessions.get(sessionKey) - if (!session) return + if (!session) { + if (!this.sessionsForProject(droppedProjectId).length) { + this.stopObserverStream(droppedProjectId) + } + return + }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/main/broker.ts` around lines 4418 - 4443, The session cleanup path in broker.ts returns immediately when `this.sessions.get(sessionKey)` is missing, which skips project-scoped observer stream teardown. Update the session removal logic around the `session` lookup in `stopSession`/the shutdown cleanup path so the `stopObserverStream(droppedProjectId)` check runs even when the session is already gone, while still preserving the existing `unsubEvent`, `leaseTimer`, `stopSessionFleetSidecar`, and `agentSessions` cleanup for live sessions.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/main/broker.ts`:
- Around line 1515-1516: The local startup flow in broker.ts currently waits for
ensureLocalFleetSidecar() before opening the direct Relaycast observer stream,
which can miss messages during the sidecar timeout window. Update the startup
ordering in the existing-session and reused-existing-broker paths so the session
is registered and ensureObserverStream(normalizedProjectId) is called before
awaiting ensureLocalFleetSidecar(existing), keeping the same sequencing wherever
those paths appear.
---
Outside diff comments:
In `@src/main/broker.ts`:
- Around line 4418-4443: The session cleanup path in broker.ts returns
immediately when `this.sessions.get(sessionKey)` is missing, which skips
project-scoped observer stream teardown. Update the session removal logic around
the `session` lookup in `stopSession`/the shutdown cleanup path so the
`stopObserverStream(droppedProjectId)` check runs even when the session is
already gone, while still preserving the existing `unsubEvent`, `leaseTimer`,
`stopSessionFleetSidecar`, and `agentSessions` cleanup for live sessions.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 98aaade0-f3f4-4f00-8f9b-2c99b8bf1ee7
📒 Files selected for processing (3)
src/main/broker.test.tssrc/main/broker.tssrc/renderer/src/stores/agent-store.stress.test.ts
| await this.ensureLocalFleetSidecar(existing) | ||
| this.ensureObserverStream(normalizedProjectId) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Start the observer stream before waiting on the fleet sidecar.
These local startup paths wait for ensureLocalFleetSidecar() before opening the direct Relaycast observer stream. Since sidecar registration can wait until its timeout, messages posted directly to Relaycast during that window are still missed. Register the session, then call ensureObserverStream() before the sidecar await.
Suggested ordering
- await this.ensureLocalFleetSidecar(session)
this.ensureObserverStream(normalizedProjectId)
+ await this.ensureLocalFleetSidecar(session)Apply the same ordering in the existing-session and reused-existing-broker paths.
Also applies to: 1565-1568, 1637-1638
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/main/broker.ts` around lines 1515 - 1516, The local startup flow in
broker.ts currently waits for ensureLocalFleetSidecar() before opening the
direct Relaycast observer stream, which can miss messages during the sidecar
timeout window. Update the startup ordering in the existing-session and
reused-existing-broker paths so the session is registered and
ensureObserverStream(normalizedProjectId) is called before awaiting
ensureLocalFleetSidecar(existing), keeping the same sequencing wherever those
paths appear.
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
3 issues found across 3 files
Prompt for AI agents (unresolved issues)
Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.
<file name="src/main/broker.ts">
<violation number="1" location="src/main/broker.ts:735">
P2: The falsy check `!body` will drop messages where an agent sends an empty text body (`""`), since empty string is falsy. While an empty `id` or `from` is arguably always invalid, an empty message body is a legitimate edge case (e.g., an attachment-only message or a whitespace-trimmed message). Consider using strict `undefined` checks instead:
```ts
if (id === undefined || from === undefined || body === undefined) return null
```</violation>
<violation number="2" location="src/main/broker.ts:1516">
P2: The observer stream is opened after `await this.ensureLocalFleetSidecar(session)`, which can block until its timeout. Messages arriving on Relaycast during the sidecar startup window will be missed since the observer stream isn't yet connected. Since `ensureObserverStream` is non-blocking (fires off a WS connection), calling it before the sidecar await would reduce the window of missed coverage.
This same ordering appears in all four session-start paths.</violation>
<violation number="3" location="src/main/broker.ts:1985">
P1: In a multi-session scenario (e.g., both a local broker and a cloud sandbox active for the same project), `sessionsForProject(projectId)` can return multiple sessions, but only the first one is destructured here. Observer stream events will only be dispatched to that first session's window — sibling sessions will miss live messages entirely.
Since the observer stream is intentionally project-scoped and shared across session types (as noted in the `observerStreams` map comment), the dispatch should iterate all sessions for the project, publishing to each active window.</violation>
</file>
Reply with feedback, questions, or to request a fix.
Re-trigger cubic
| const [session] = this.sessionsForProject(projectId) | ||
| if (!session) return | ||
| this.publishBrokerEvent(sessionKeyFor(session), projectId, session.window, forwarded) |
There was a problem hiding this comment.
P1: In a multi-session scenario (e.g., both a local broker and a cloud sandbox active for the same project), sessionsForProject(projectId) can return multiple sessions, but only the first one is destructured here. Observer stream events will only be dispatched to that first session's window — sibling sessions will miss live messages entirely.
Since the observer stream is intentionally project-scoped and shared across session types (as noted in the observerStreams map comment), the dispatch should iterate all sessions for the project, publishing to each active window.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At src/main/broker.ts, line 1985:
<comment>In a multi-session scenario (e.g., both a local broker and a cloud sandbox active for the same project), `sessionsForProject(projectId)` can return multiple sessions, but only the first one is destructured here. Observer stream events will only be dispatched to that first session's window — sibling sessions will miss live messages entirely.
Since the observer stream is intentionally project-scoped and shared across session types (as noted in the `observerStreams` map comment), the dispatch should iterate all sessions for the project, publishing to each active window.</comment>
<file context>
@@ -1830,6 +1940,61 @@ export class BrokerManager {
+ const forwarded = this.validateIngressBrokerEvent(payload as unknown as BrokerEvent)
+ if (!forwarded) return
+
+ const [session] = this.sessionsForProject(projectId)
+ if (!session) return
+ this.publishBrokerEvent(sessionKeyFor(session), projectId, session.window, forwarded)
</file context>
| const [session] = this.sessionsForProject(projectId) | |
| if (!session) return | |
| this.publishBrokerEvent(sessionKeyFor(session), projectId, session.window, forwarded) | |
| const sessions = this.sessionsForProject(projectId) | |
| if (sessions.length === 0) return | |
| for (const session of sessions) { | |
| this.publishBrokerEvent(sessionKeyFor(session), projectId, session.window, forwarded) | |
| } |
| await this.syncChannels(normalizedProjectId, nextChannels) | ||
| await this.refreshEventStream(normalizedProjectId, 'existing-session-start', win) | ||
| await this.ensureLocalFleetSidecar(existing) | ||
| this.ensureObserverStream(normalizedProjectId) |
There was a problem hiding this comment.
P2: The observer stream is opened after await this.ensureLocalFleetSidecar(session), which can block until its timeout. Messages arriving on Relaycast during the sidecar startup window will be missed since the observer stream isn't yet connected. Since ensureObserverStream is non-blocking (fires off a WS connection), calling it before the sidecar await would reduce the window of missed coverage.
This same ordering appears in all four session-start paths.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At src/main/broker.ts, line 1516:
<comment>The observer stream is opened after `await this.ensureLocalFleetSidecar(session)`, which can block until its timeout. Messages arriving on Relaycast during the sidecar startup window will be missed since the observer stream isn't yet connected. Since `ensureObserverStream` is non-blocking (fires off a WS connection), calling it before the sidecar await would reduce the window of missed coverage.
This same ordering appears in all four session-start paths.</comment>
<file context>
@@ -1407,6 +1513,7 @@ export class BrokerManager {
await this.syncChannels(normalizedProjectId, nextChannels)
await this.refreshEventStream(normalizedProjectId, 'existing-session-start', win)
await this.ensureLocalFleetSidecar(existing)
+ this.ensureObserverStream(normalizedProjectId)
this.sendStatus(normalizedProjectId, 'connected')
return false
</file context>
| const id = typeof raw.id === 'string' ? raw.id : undefined | ||
| const from = typeof raw.agentName === 'string' ? raw.agentName : undefined | ||
| const body = typeof raw.text === 'string' ? raw.text : undefined | ||
| if (!id || !from || !body) return null |
There was a problem hiding this comment.
P2: The falsy check !body will drop messages where an agent sends an empty text body (""), since empty string is falsy. While an empty id or from is arguably always invalid, an empty message body is a legitimate edge case (e.g., an attachment-only message or a whitespace-trimmed message). Consider using strict undefined checks instead:
if (id === undefined || from === undefined || body === undefined) return nullPrompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At src/main/broker.ts, line 735:
<comment>The falsy check `!body` will drop messages where an agent sends an empty text body (`""`), since empty string is falsy. While an empty `id` or `from` is arguably always invalid, an empty message body is a legitimate edge case (e.g., an attachment-only message or a whitespace-trimmed message). Consider using strict `undefined` checks instead:
```ts
if (id === undefined || from === undefined || body === undefined) return null
```</comment>
<file context>
@@ -672,6 +673,104 @@ function normalizeRelayMessageForChat(
+ const id = typeof raw.id === 'string' ? raw.id : undefined
+ const from = typeof raw.agentName === 'string' ? raw.agentName : undefined
+ const body = typeof raw.text === 'string' ? raw.text : undefined
+ if (!id || !from || !body) return null
+ return { id, from, body }
+}
</file context>
1 participant
Summary
Pear's dashboard has been missing chat messages that never touch the local broker's own dashboard WS — specifically:
This PR closes that gap the way the hosted observer dashboard already does: Pear's main process opens a second, direct-to-Relaycast
/v1/wssubscription per open project, scoped to an observer token (ot_live_...,stream:read), and feeds matching chat-message frames into the same renderer ingestion path that already handlesrelay_inboundevents from the local broker WS.This is purely additive — the existing local broker WS keeps doing everything else it's uniquely responsible for (agent spawn/exit, terminal state, delivery acks), and the existing reconciliation/active-room-polling fallback is unchanged and still serves as the catch-up safety net for any gap.
What changed (
src/main/broker.ts)ensureObserverStream(projectId)— mints an observer token via the existingmintObserverToken()path and opens a@relaycast/sdkWsClientconnection to/v1/ws. Called from every session-start path (local spawn, attach-to-existing-broker, cloud sandbox attach, and the "existing session" re-start branch), idempotently.handleObserverStreamEvent/observerStreamRelayInboundPayload— classifies message-class wire frames (message.created,thread.reply,dm.received,group_dm.received, plus the broader forward-compat alias set relay'sclassify_fleet_deliveryuses) and synthesizes the samerelay_inbound-shaped payload the local broker's own events use, using the Relaycast message's own canonical id asevent_id. Routes through the samevalidateIngressBrokerEvent+publishBrokerEventpath ordinary broker events use — no parallel dispatch path, no renderer changes.stopObserverStream— torn down indropSession(), but only once no session (local or cloud) remains for a project, since the stream is shared across a project's sibling sessions.Why this transport/mechanism
Checked
@relaycast/sdk's exports before hand-rolling anything against rawws: it already exports a genericWsClient(token as string-or-function, built-in exponential-backoff reconnect, resync/replay-gap handling, ping) that's exactly what the hosted observer dashboard itself uses under@relaycast/react'sRelayProvider(confirmed in therelaycastrepo'spackages/react/src/provider.tsxandpackages/observer-dashboard).AgentClient(the higher-level SDK class) is agent-identity-shaped (requires an API key, does presence/heartbeat) and doesn't fit an observer token, so this usesWsClientdirectly — same as the dashboard's ownwsTokenusage. Confirmed observer tokens are workspace-scoped server-side (wsAuth.ts'sscope: 'workspace'), so no per-channel.subscribe()call is needed (the dashboard doesn't call it either when it wants "everything").Reconnect/backoff comes for free from
WsClientitself — no bespoke replay-gap machinery was added, per the plan; the existing reconciliation/active-room-polling fallback remains the safety net for any gap.Dashboard-own-send dedup
Verified the existing
isCanonicalEchoOfLocalHumancontent-based dedup (matches onlocal: trueoptimistic records, regardless of which stream produced the echo) already covers a human's own send arriving twice — once via the local broker's own echo, once via this new observer stream, each under a differentevent_id. Added a renderer test (agent-store.stress.test.ts) proving this explicitly rather than assuming it. No additional self-skip logic was needed.One known scope limitation: live
dm.received/group_dm.receivedframes only carry aconversationId, not participant names (unlike the REST-basedreconcileMessageshistory fetch, which resolves "alice, bob"-style targets). This PR falls back to the raw conversation id as the DM target so the message still surfaces rather than being dropped — a follow-up could resolve participant names via an extra lookup if that DM room-title UX matters in practice. Channel messages (the primary repro case in the plan) are fully resolved to#channeltargets.Test plan
npm run typecheck— cleannpm test— 123/123 node tests pass (baseline unchanged)npx vitest run— 505/505 tests pass (498 baseline + 7 new)broker.test.tscoverage (BrokerManager observer stream): message-class frame → singlerelay_inbounddispatch with canonical id;thread.replycarriesthread_id; non-message-class frames are not forwarded; stream disconnects on session drop and a fresh one opens on restart without leaking the old connection's events; sibling cloud-session drop does NOT tear down a still-live local session's stream; token function reuses the cachedmintObserverTokenpath.agent-store.stress.test.tscoverage: a human's own send does not double-render when it arrives a second time via this new stream under a different id.🤖 Generated with Claude Code