feat: observer-stream live chat (flag-gated, cursor-resumable) with polling fallback#393
feat: observer-stream live chat (flag-gated, cursor-resumable) with polling fallback#393willwashburn wants to merge 2 commits into
Conversation
…ERVER_STREAM Add a feature-flagged (default OFF) main-process ObserverStreamManager that mints a scoped read-only observer token via the local broker's POST /api/observer-token, opens the engine observer WebSocket (wss://<relay base>/v1/ws?token=...), REST-backfills missed events from a persisted per-workspace cursor (GET /v1/workspace/events?since=<seq>), merges by seq (dedupe, ascending), and reconnects with exponential backoff, re-minting the token on auth failure. Message-class events (message.created, thread.reply, dm/group_dm.received) are translated into BrokerReconciledChatMessage batches and merged through the renderer's existing reconcileMessages store path — the same id-deduping merge the 3s REST polling reconciliation feeds — so both planes converge on canonical relaycast message ids with no parallel store. message.reacted / message.read are left as a TODO (the store has no incremental merge path for them yet). Every failure mode degrades silently to the untouched polling path: flag off, observer-token endpoint missing (broker 404), /v1/workspace/events 404 (older engine), or a WS that will not authenticate. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VASvErz2MxkQMCsSzTgsbe
|
Warning Review limit reached
Next review available in: 49 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (10)
✨ 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 |
![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 introduces a feature-flagged (PEAR_OBSERVER_STREAM) relaycast observer-stream consumer that streams chat updates via WebSockets and REST backfills, integrating with the existing renderer message state. Feedback on the changes highlights three key issues: a potential lost update bug in observer-chat.ts when processing multiple thread replies for the same parent in a single batch, a missing state check in observer-stream.ts that could lead to wasted network requests during backfill if the manager is stopped, and an issue in broker.ts where forceFresh fails to clear the in-flight token promise cache.
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.
| for (const entry of update.threadReplies) { | ||
| const parent = existingMessages.find((message) => message.id === entry.parentId) | ||
| // A reply whose parent we have never merged is skipped — the store only | ||
| // carries thread replies attached to a full parent record. The polling | ||
| // reconciliation cannot recover these either (it does not fetch thread | ||
| // replies), so a skipped reply reappears only when the parent lands. | ||
| if (!parent) continue | ||
| if (parent.threadReplies?.some((reply) => reply.id === entry.reply.id)) continue | ||
| incoming.push({ | ||
| ...toReconciled(parent), | ||
| threadReplies: [...(parent.threadReplies || []), entry.reply] | ||
| }) | ||
| } |
There was a problem hiding this comment.
There is a potential lost update bug when multiple thread replies for the same parent message are received in the same batch. Since existingMessages is not updated during the loop, each iteration retrieves the stale parent message from existingMessages (which does not contain the previously processed replies from the same batch). As a result, only the last reply in the batch for a given parent will survive, and previous ones will be overwritten and lost.
To fix this, group the incoming thread replies by parentId first, and then merge them with the parent's existing replies in a single update per parent.
const repliesByParent = new Map<string, typeof update.threadReplies[number]['reply'][]>()
for (const entry of update.threadReplies) {
const list = repliesByParent.get(entry.parentId) || []
list.push(entry.reply)
repliesByParent.set(entry.parentId, list)
}
for (const [parentId, newReplies] of repliesByParent.entries()) {
const parent = existingMessages.find((message) => message.id === parentId)
if (!parent) continue
const combinedReplies = [...(parent.threadReplies || [])]
for (const reply of newReplies) {
if (!combinedReplies.some((r) => r.id === reply.id)) {
combinedReplies.push(reply)
}
}
incoming.push({
...toReconciled(parent),
threadReplies: combinedReplies
})
}| private async backfill(token: string): Promise<void> { | ||
| if (this.cursor === undefined) { | ||
| // First run for this workspace: don't replay the entire history — adopt | ||
| // the current head and stream from here on. The REST reconciliation | ||
| // already hydrates canonical room history on demand. | ||
| const page = await this.fetchEventsPage(token, 0, 1) | ||
| this.setCursor(page.latestSeq ?? 0) | ||
| this.log(`observer stream cursor initialized at seq ${this.cursor}`) | ||
| return | ||
| } | ||
|
|
||
| const limit = Math.min(this.options.backfillPageLimit ?? DEFAULT_BACKFILL_PAGE_LIMIT, MAX_BACKFILL_PAGE_LIMIT) | ||
| let since = this.cursor | ||
| for (;;) { | ||
| const page = await this.fetchEventsPage(token, since, limit) | ||
| this.processEvents(page.events) | ||
| const pageMax = maxObserverSeq(page.events) | ||
| if (page.events.length < limit) break | ||
| if (pageMax === undefined || pageMax <= since) break | ||
| since = pageMax | ||
| } | ||
| } |
There was a problem hiding this comment.
If the manager is stopped or disabled while backfill is in progress, the loop will continue to fetch pages and process events unnecessarily. Adding a check for this.state !== 'running' inside the loop and after the first-run fetch will prevent wasted network requests and processing.
private async backfill(token: string): Promise<void> {
if (this.cursor === undefined) {
// First run for this workspace: don't replay the entire history — adopt
// the current head and stream from here on. The REST reconciliation
// already hydrates canonical room history on demand.
const page = await this.fetchEventsPage(token, 0, 1)
if (this.state !== 'running') return
this.setCursor(page.latestSeq ?? 0)
this.log(`observer stream cursor initialized at seq ${this.cursor}`)
return
}
const limit = Math.min(this.options.backfillPageLimit ?? DEFAULT_BACKFILL_PAGE_LIMIT, MAX_BACKFILL_PAGE_LIMIT)
let since = this.cursor
for (;;) {
if (this.state !== 'running') break
const page = await this.fetchEventsPage(token, since, limit)
this.processEvents(page.events)
const pageMax = maxObserverSeq(page.events)
if (page.events.length < limit) break
if (pageMax === undefined || pageMax <= since) break
since = pageMax
}
}| mintToken: async ({ forceFresh }) => { | ||
| if (forceFresh) this.observerTokens.delete(projectId) | ||
| try { |
There was a problem hiding this comment.
When forceFresh is true, deleting the cached token from this.observerTokens is not sufficient if there is an active in-flight promise in this.observerTokenMints. The manager would still return the existing in-flight promise (which resolves to the old/stale token) instead of minting a fresh one. Deleting the entry from this.observerTokenMints as well ensures a truly fresh token is minted.
mintToken: async ({ forceFresh }) => {
if (forceFresh) {
this.observerTokens.delete(projectId)
this.observerTokenMints.delete(projectId)
}
try {CI's verify:mcp-resources-drift gate compares against the merge with main, where @relayfile/client entered the dependency tree; the generated resources list needed the corresponding entry. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VASvErz2MxkQMCsSzTgsbe
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
5 issues found and verified against the latest diff
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/observer-stream.ts">
<violation number="1" location="src/main/observer-stream.ts:375">
P2: A stale socket callback can still emit events after reconnect because the handler only checks `state`, not socket identity. Adding a `this.socket === socket` guard makes reconnect behavior match the repo’s stale-listener hardening pattern.</violation>
<violation number="2" location="src/main/observer-stream.ts:488">
P2: Auth-expired live sockets can reconnect with the same stale token because post-open close codes are treated as a normal close. Handling auth close codes in the settled-open branch and marking `nextMintForceFresh` would keep re-auth behavior consistent with the pre-open auth-failure path.</violation>
<violation number="3" location="src/main/observer-stream.ts:544">
P2: The backfill pagination loop doesn't check whether the manager has been stopped between page fetches. If `stop()` is called while backfill is in progress (e.g., a session is dropped), this loop will continue issuing network requests and processing events until pagination naturally terminates. Adding a `if (this.state !== 'running') break` at the top of the loop (and after the first-run fetch) would avoid wasted work.</violation>
</file>
<file name="src/renderer/src/lib/observer-chat.ts">
<violation number="1" location="src/renderer/src/lib/observer-chat.ts:61">
P1: Thread replies can be lost when multiple `update.threadReplies` entries target the same parent (or when parent arrives in `update.messages` in the same batch). The loop resolves from `existingMessages` only, so it doesn't accumulate against already-prepared `incoming` parent state before `reconcileMessages` merges.</violation>
</file>
<file name="src/main/broker.ts">
<violation number="1" location="src/main/broker.ts:1867">
P2: Cloud observer streams can authenticate with the local broker's observer token when a local and cloud session coexist, because this session-scoped manager mints by projectId and getSessionForProject prefers the local session. Consider minting through the current sessionKey (and scoping the token cache accordingly) so the stream's token/workspace matches the broker session it is attached to.</violation>
</file>
Reply with feedback, questions, or to request a fix.
Re-trigger cubic
| } | ||
|
|
||
| for (const entry of update.threadReplies) { | ||
| const parent = existingMessages.find((message) => message.id === entry.parentId) |
There was a problem hiding this comment.
P1: Thread replies can be lost when multiple update.threadReplies entries target the same parent (or when parent arrives in update.messages in the same batch). The loop resolves from existingMessages only, so it doesn't accumulate against already-prepared incoming parent state before reconcileMessages merges.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At src/renderer/src/lib/observer-chat.ts, line 61:
<comment>Thread replies can be lost when multiple `update.threadReplies` entries target the same parent (or when parent arrives in `update.messages` in the same batch). The loop resolves from `existingMessages` only, so it doesn't accumulate against already-prepared `incoming` parent state before `reconcileMessages` merges.</comment>
<file context>
@@ -0,0 +1,77 @@
+ }
+
+ for (const entry of update.threadReplies) {
+ const parent = existingMessages.find((message) => message.id === entry.parentId)
+ // A reply whose parent we have never merged is skipped — the store only
+ // carries thread replies attached to a full parent record. The polling
</file context>
| } | ||
|
|
||
| private async runCycle(): Promise<void> { | ||
| if (this.state !== 'running') return |
There was a problem hiding this comment.
P2: A stale socket callback can still emit events after reconnect because the handler only checks state, not socket identity. Adding a this.socket === socket guard makes reconnect behavior match the repo’s stale-listener hardening pattern.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At src/main/observer-stream.ts, line 375:
<comment>A stale socket callback can still emit events after reconnect because the handler only checks `state`, not socket identity. Adding a `this.socket === socket` guard makes reconnect behavior match the repo’s stale-listener hardening pattern.</comment>
<file context>
@@ -0,0 +1,630 @@
+ }
+
+ private async runCycle(): Promise<void> {
+ if (this.state !== 'running') return
+ try {
+ const token = await this.options.mintToken({ forceFresh: this.nextMintForceFresh })
</file context>
|
|
||
| socket.on('close', ((code: unknown) => { | ||
| if (settledOpen) { | ||
| if (this.socket === socket) this.socket = null |
There was a problem hiding this comment.
P2: Auth-expired live sockets can reconnect with the same stale token because post-open close codes are treated as a normal close. Handling auth close codes in the settled-open branch and marking nextMintForceFresh would keep re-auth behavior consistent with the pre-open auth-failure path.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At src/main/observer-stream.ts, line 488:
<comment>Auth-expired live sockets can reconnect with the same stale token because post-open close codes are treated as a normal close. Handling auth close codes in the settled-open branch and marking `nextMintForceFresh` would keep re-auth behavior consistent with the pre-open auth-failure path.</comment>
<file context>
@@ -0,0 +1,630 @@
+
+ socket.on('close', ((code: unknown) => {
+ if (settledOpen) {
+ if (this.socket === socket) this.socket = null
+ resolve()
+ } else {
</file context>
| mintToken: async ({ forceFresh }) => { | ||
| if (forceFresh) this.observerTokens.delete(projectId) | ||
| try { | ||
| return (await this.mintObserverToken(projectId)).token |
There was a problem hiding this comment.
P2: Cloud observer streams can authenticate with the local broker's observer token when a local and cloud session coexist, because this session-scoped manager mints by projectId and getSessionForProject prefers the local session. Consider minting through the current sessionKey (and scoping the token cache accordingly) so the stream's token/workspace matches the broker session it is attached to.
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 1867:
<comment>Cloud observer streams can authenticate with the local broker's observer token when a local and cloud session coexist, because this session-scoped manager mints by projectId and getSessionForProject prefers the local session. Consider minting through the current sessionKey (and scoping the token cache accordingly) so the stream's token/workspace matches the broker session it is attached to.</comment>
<file context>
@@ -1830,6 +1843,69 @@ export class BrokerManager {
+ mintToken: async ({ forceFresh }) => {
+ if (forceFresh) this.observerTokens.delete(projectId)
+ try {
+ return (await this.mintObserverToken(projectId)).token
+ } catch (err) {
+ // mintObserverTokenUncached maps a broker 404 (route predates
</file context>
|
|
||
| const limit = Math.min(this.options.backfillPageLimit ?? DEFAULT_BACKFILL_PAGE_LIMIT, MAX_BACKFILL_PAGE_LIMIT) | ||
| let since = this.cursor | ||
| for (;;) { |
There was a problem hiding this comment.
P2: The backfill pagination loop doesn't check whether the manager has been stopped between page fetches. If stop() is called while backfill is in progress (e.g., a session is dropped), this loop will continue issuing network requests and processing events until pagination naturally terminates. Adding a if (this.state !== 'running') break at the top of the loop (and after the first-run fetch) would avoid wasted work.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At src/main/observer-stream.ts, line 544:
<comment>The backfill pagination loop doesn't check whether the manager has been stopped between page fetches. If `stop()` is called while backfill is in progress (e.g., a session is dropped), this loop will continue issuing network requests and processing events until pagination naturally terminates. Adding a `if (this.state !== 'running') break` at the top of the loop (and after the first-run fetch) would avoid wasted work.</comment>
<file context>
@@ -0,0 +1,630 @@
+
+ const limit = Math.min(this.options.backfillPageLimit ?? DEFAULT_BACKFILL_PAGE_LIMIT, MAX_BACKFILL_PAGE_LIMIT)
+ let since = this.cursor
+ for (;;) {
+ const page = await this.fetchEventsPage(token, since, limit)
+ this.processEvents(page.events)
</file context>
2 participants
Summary
Wires pear's Slack-like UI to the new observer plane (durable workspace event log — AgentWorkforce/relaycast#233), replacing dependence on broker
relay_inboundside effects + 3s active-room polling as the primary live path. Gated behindPEAR_OBSERVER_STREAM=1(default off) with silent fallback to the existing untouched polling on any missing server capability, so it's safe to merge ahead of the engine/relay releases.How it works (
src/main/observer-stream.ts, newObserverStreamManager):POST /api/observer-token(reuses pear's existingmintObserverTokencaller/cache), opens the live/v1/wsobserver WebSocket and buffers frames, REST-backfillsGET /v1/workspace/eventsfrom a persisted per-workspace cursor (paginated ≤500), merges byseq(dedupe, ascending), then goes live — advancing and debounce-persisting the cursor (observerStreamCursorsin the existing projects store).latest_seqinstead of replaying full history (REST reconcile owns room hydration)./v1/workspace/events404 (old engine), or WS auth failure — each degrades silently to today's polling.Store integration: observer events are translated into
BrokerReconciledChatMessagebatches and merged via the samereconcileMessagespath the 3s polling feeds — observer payloads carry canonical relaycast message ids, so id-dedupe converges regardless of which plane delivers first (synthesizingrelay_inboundevents would duplicate against REST-reconciled records and mis-time backfilled history). DMtoparticipants are resolved from the conversation's known messages;thread.replymerges by cloning the in-store parent with the appended reply.message.reacted/message.readare left as an explicit TODO — the renderer store has no incremental merge path for them yet.Glue in
src/main/broker.ts(start/stop per session), plus a newobserver:chat-updateIPC channel (ObserverChatUpdatetype through preload/ipc-mock).Test Plan
Screenshots
N/A
🤖 Generated with Claude Code
https://claude.ai/code/session_01VASvErz2MxkQMCsSzTgsbe
Generated by Claude Code