fix(sdk): stream relay.addListener events through registered agent clients#1226
fix(sdk): stream relay.addListener events through registered agent clients#1226willwashburn wants to merge 11 commits into
Conversation
…ients relay.addListener(...) on a workspace-key AgentRelay silently received nothing against relaycast v5. The listener hub connected the workspace client's event stream, which opens the legacy /v1/ws workspace stream — but v5 rejects workspace keys there (observer token required), so the socket 401'd into an endless reconnect loop, no error surfaced, and the engine's queued deliveries for the listening agent were never drained (its implicit direct node never connected). Route the workspace-level listener hub through a new events fan-in instead: every agent client created by workspace.register()/reconnect() is added as a source and connected over the v5 node transport once a listener exists. Events that reach multiple locally-registered agents (one deliver frame per recipient) are deduplicated across sources within a bounded window; per-source transport events pass through. The workspace stream is kept only as a pre-registration fallback and is detached as soon as the first agent source connects. Failures are no longer silent: listener connect errors route through the onError hooks instead of an empty catch, and a listener still waiting with no registered agent reports after a grace period. Verified end-to-end against a self-hosted engine (5.0.11): the README quickstart pattern now receives #general messages in both orders (register-then-listen and listen-then-register), with deliveries acked server-side. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VASvErz2MxkQMCsSzTgsbe
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughAdds observer-mode workspace streaming, a shared listener fan-in, broker self-muting for offline self identities, and matching spec, changelog, and test coverage. ChangesObserver Plane Streaming and Listener Fan-In
Estimated code review effort: 5 (Critical) | ~120 minutes Possibly related issues
Possibly related PRs
Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ 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 an event fan-in mechanism (createEventFanIn) to stream events through registered agent clients over the v5 node transport, fixing relay.addListener(...) for workspace-key clients. The feedback focuses on enhancing the robustness of the fan-in implementation by tracking and cleaning up event subscriptions on disconnect to prevent memory leaks, defensively checking for optional methods on sources to avoid runtime errors, and refactoring the changelog entry to be more concise and impact-first in accordance with the repository style guide.
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.
| await Promise.allSettled(sources.map((source) => Promise.resolve().then(() => source.disconnect()))); | ||
| }, | ||
|
|
||
| subscribe: (channels) => { | ||
| for (const channel of channels) desiredChannels.add(channel); | ||
| for (const source of sources) { |
There was a problem hiding this comment.
During disconnect(), unsubscribe from all active sources to prevent memory leaks and stop forwarding events. Additionally, defensively check if source.disconnect is a function before calling it, as partial fakes or custom event surfaces might not implement it.
| await Promise.allSettled(sources.map((source) => Promise.resolve().then(() => source.disconnect()))); | |
| }, | |
| subscribe: (channels) => { | |
| for (const channel of channels) desiredChannels.add(channel); | |
| for (const source of sources) { | |
| disconnect: async () => { | |
| connectRequested = false; | |
| clearNoSourceTimer(); | |
| detachFallback(); | |
| for (const unsub of unsubscribes.values()) { | |
| unsub(); | |
| } | |
| unsubscribes.clear(); | |
| await Promise.allSettled( | |
| sources.map((source) => | |
| Promise.resolve().then(() => { | |
| if (typeof source.disconnect === 'function') { | |
| return source.disconnect(); | |
| } | |
| }) | |
| ) | |
| ); | |
| }, |
There was a problem hiding this comment.
Fixed in d71fc62. disconnect() now detaches all source forwarding via a tracked sourceForwarding map (so a source reconnected later by another owner can't feed explicitly-disconnected listeners), guards source.disconnect with a typeof check, and connect() re-attaches forwarding for sources dropped by a prior disconnect. Covered by the new "disconnect() stops forwarding; connect() resumes it" test.
Generated by Claude Code
| ### Fixed | ||
|
|
||
| - `HarnessDriverClient.spawn()` now polls the broker's startup handshake for the full `startupTimeoutMs` budget (default 45s) instead of a fixed ~10s, so a slow-but-healthy Relaycast handshake that keeps answering `503` while warming up is no longer misreported as a spawn failure. | ||
| - `@agent-relay/sdk` `relay.addListener(...)` on a workspace-key client now receives channel messages, DMs, and thread replies. It previously listened on the legacy `/v1/ws` workspace stream, which relaycast v5 rejects for workspace keys (observer token required), so the socket 401'd into a silent reconnect loop and listeners got nothing while deliveries queued unread. The workspace-level listener hub now streams through every registered agent client (`workspace.register`/`reconnect`) over the v5 node transport, deduplicating events that reach multiple locally-registered agents; the workspace stream remains only as a pre-registration fallback. Listener `events.connect()` failures are reported through `onError` instead of being swallowed, and a listener left waiting with no registered agent warns after 10s. |
There was a problem hiding this comment.
The changelog entry is too long and contains detailed implementation backstory (e.g., legacy workspace stream details, 401 errors, and reconnect loops). According to the repository style guide, changelog entries should be concise and impact-first, dropping implementation backstory unless necessary to explain the shipped impact.
| - `@agent-relay/sdk` `relay.addListener(...)` on a workspace-key client now receives channel messages, DMs, and thread replies. It previously listened on the legacy `/v1/ws` workspace stream, which relaycast v5 rejects for workspace keys (observer token required), so the socket 401'd into a silent reconnect loop and listeners got nothing while deliveries queued unread. The workspace-level listener hub now streams through every registered agent client (`workspace.register`/`reconnect`) over the v5 node transport, deduplicating events that reach multiple locally-registered agents; the workspace stream remains only as a pre-registration fallback. Listener `events.connect()` failures are reported through `onError` instead of being swallowed, and a listener left waiting with no registered agent warns after 10s. | |
| - @agent-relay/sdk: Fix `relay.addListener(...)` on workspace-key clients to correctly receive channel messages, DMs, and thread replies by streaming through registered agent clients over the v5 node transport. Listener connection failures are now reported via `onError` instead of being swallowed, and a warning is issued if no registered agent is connected after 10 seconds. |
References
- Changelog entries should be concise and impact-first. Prefer one short bullet per user-visible change: name the command, API, schema, or package touched and the practical effect. Drop issue/PR links, internal review notes, implementation backstory, release-only entries, and 'foundation for...' phrasing unless that text clearly explains the shipped impact. (link)
There was a problem hiding this comment.
Fixed in d71fc62 — trimmed to a single impact-first bullet (package, API, practical effect), dropping the 401/reconnect-loop backstory.
Generated by Claude Code
| if (options.onError) { | ||
| try { |
There was a problem hiding this comment.
To prevent memory leaks and avoid forwarding events after disconnect(), we should track the unsubscribe functions returned by source.on('any', forward) and invoke them during disconnection. Let's declare a map to store these unsubscribe functions.
| if (options.onError) { | |
| try { | |
| let fallbackConnected = false; | |
| let noSourceTimer: ReturnType<typeof setTimeout> | undefined; | |
| const unsubscribes = new Map<RelayMessagingEventsSurface, () => void>(); |
There was a problem hiding this comment.
Fixed in d71fc62 — forwarding unsubscribes are tracked in a sourceForwarding map (keyed by surface) and detached on disconnect().
Generated by Claude Code
| if (seenSources.has(source)) return; | ||
| seenSources.add(source); | ||
| sources.push(source); | ||
| source.on('any', forward); | ||
| if (connectRequested) { | ||
| connectSource(source); | ||
| clearNoSourceTimer(); | ||
| // The agent transport is the real event stream; stop the workspace | ||
| // fallback so it does not sit in a doomed reconnect loop. | ||
| detachFallback(); | ||
| } | ||
| }, | ||
|
|
||
| hasAgentSources: () => sources.length > 0, | ||
|
|
||
| connect: () => { |
There was a problem hiding this comment.
Store the unsubscribe function returned by source.on('any', forward) in the unsubscribes map so we can clean it up later.
| if (seenSources.has(source)) return; | |
| seenSources.add(source); | |
| sources.push(source); | |
| source.on('any', forward); | |
| if (connectRequested) { | |
| connectSource(source); | |
| clearNoSourceTimer(); | |
| // The agent transport is the real event stream; stop the workspace | |
| // fallback so it does not sit in a doomed reconnect loop. | |
| detachFallback(); | |
| } | |
| }, | |
| hasAgentSources: () => sources.length > 0, | |
| connect: () => { | |
| addSource: (source) => { | |
| // A source without a subscribable stream (partial fakes injected via | |
| // `createAgentMessaging`) contributes nothing; skip it entirely. | |
| if (!source || typeof source.on !== 'function') return; | |
| if (seenSources.has(source)) return; | |
| seenSources.add(source); | |
| sources.push(source); | |
| const unsub = source.on('any', forward); | |
| unsubscribes.set(source, unsub); | |
| if (connectRequested) { | |
| connectSource(source); | |
| clearNoSourceTimer(); | |
| // The agent transport is the real event stream; stop the workspace | |
| // fallback so it does not sit in a doomed reconnect loop. | |
| detachFallback(); | |
| } | |
| }, |
There was a problem hiding this comment.
Fixed in d71fc62 — addSource wires forwarding through attachSourceForwarding, which stores the unsubscribe in the sourceForwarding map for later detach.
Generated by Claude Code
| return; | ||
| } | ||
| attachFallback(); | ||
| scheduleNoSourceWarning(); | ||
| }, | ||
|
|
||
| disconnect: async () => { | ||
| connectRequested = false; | ||
| clearNoSourceTimer(); |
There was a problem hiding this comment.
When reconnecting, ensure we re-subscribe to any sources that were unsubscribed during disconnect().
connect: () => {
connectRequested = true;
if (sources.length > 0) {
for (const source of sources) {
if (!unsubscribes.has(source)) {
const unsub = source.on('any', forward);
unsubscribes.set(source, unsub);
}
connectSource(source);
}
return;
}
attachFallback();
scheduleNoSourceWarning();
},There was a problem hiding this comment.
Fixed in d71fc62 — connect() calls attachSourceForwarding for every source (idempotent via the map), so forwarding dropped by a prior disconnect() is re-attached. Covered by the new disconnect/reconnect test.
Generated by Claude Code
| if (fallbackConnected) { | ||
| fallbackConnected = false; | ||
| try { | ||
| void fallback?.disconnect().catch(() => {}); | ||
| } catch { | ||
| // Fallback surfaces without a disconnect are simply left as-is. | ||
| } | ||
| } | ||
| }; | ||
|
|
||
| return { | ||
| addSource: (source) => { | ||
| // A source without a subscribable stream (partial fakes injected via |
There was a problem hiding this comment.
If fallback is a partial fake or custom object that does not implement disconnect, or if disconnect returns a non-Promise value, calling fallback?.disconnect().catch(...) can throw a TypeError (e.g., if disconnect is undefined or returns void). Let's defensively check if disconnect is a function and wrap it safely.
const detachFallback = (): void => {
if (!fallbackForwarding) return;
fallbackForwarding();
fallbackForwarding = undefined;
if (fallbackConnected) {
fallbackConnected = false;
try {
if (typeof fallback?.disconnect === 'function') {
void fallback.disconnect().catch(() => {});
}
} catch {
// Fallback surfaces without a disconnect are simply left as-is.
}
}
};There was a problem hiding this comment.
Fixed in d71fc62 — detachFallback now checks typeof fallback?.disconnect === 'function' before calling, with the try/catch retained for a disconnect that returns a non-promise.
Generated by Claude Code
| source.subscribe(channels); | ||
| } catch (error) { | ||
| report(error); | ||
| } | ||
| } | ||
| }, | ||
|
|
||
| unsubscribe: (channels) => { | ||
| for (const channel of channels) desiredChannels.delete(channel); | ||
| for (const source of sources) { | ||
| try { | ||
| source.unsubscribe(channels); | ||
| } catch (error) { | ||
| report(error); | ||
| } | ||
| } | ||
| }, | ||
|
|
||
| on: <K extends keyof RelayMessagingEventMap>( | ||
| event: K, | ||
| handler: (...args: RelayMessagingEventMap[K]) => void | Promise<void> |
There was a problem hiding this comment.
In subscribe and unsubscribe, we should defensively check if source.subscribe or source.unsubscribe are functions before calling them. Otherwise, calling them on partial fakes or custom event surfaces will throw a TypeError and trigger the report error handler, causing noisy warnings or error hook invocations.
subscribe: (channels) => {
for (const channel of channels) desiredChannels.add(channel);
for (const source of sources) {
if (typeof source.subscribe === 'function') {
try {
source.subscribe(channels);
} catch (error) {
report(error);
}
}
}
},
unsubscribe: (channels) => {
for (const channel of channels) desiredChannels.delete(channel);
for (const source of sources) {
if (typeof source.unsubscribe === 'function') {
try {
source.unsubscribe(channels);
} catch (error) {
report(error);
}
}
}
},There was a problem hiding this comment.
Fixed in d71fc62 — subscribe/unsubscribe now skip sources whose surface lacks the method (typeof guard) instead of routing a TypeError through the error hooks.
Generated by Claude Code
![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: 543d5ddf8a
ℹ️ 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".
| if (desiredChannels.size > 0 && typeof source.subscribe === 'function') { | ||
| try { | ||
| source.subscribe([...desiredChannels]); |
There was a problem hiding this comment.
In the documented workspace.register(...); channels.join(...); relay.addListener('message.created', ...) flow, desiredChannels is empty because addListener never records the joined channel, so this code only opens the agent socket and never calls source.subscribe(...). The underlying relaycast agent client sends channel subscription frames only for channels passed to subscribe, so channel message.created/thread.reply events still won't arrive unless callers make an extra undocumented relay.events.subscribe(['channel']) call.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
This doesn't match how v5 delivers channel messages, so no change here. On the node transport, message.created/thread.reply are not gated on WS subscribe frames: the engine creates durable delivery rows from server-side channel membership (buildChannelDeliveryWrite selects from channel_members) and pushes them as deliver frames; agents are auto-joined to #general at registration, and channels.join(...) is the membership operation. The WS-level subscribe op is a legacy /v1/ws concern — the engine's node-control handler doesn't even have a subscribe message type.
Verified E2E against a self-hosted engine 5.0.11: the exact register → addListener('message.created') flow with no events.subscribe call receives #general messages (and the delivery rows show acked). desiredChannels replay in connectSource exists only to preserve the explicit events.subscribe API's behavior across late-connecting sources, not as a delivery prerequisite.
Generated by Claude Code
| const key = dedupeKey(event); | ||
| if (key) { | ||
| const at = now(); | ||
| const prev = seenEvents.get(key); | ||
| if (prev !== undefined && at - prev < dedupeWindowMs) return; |
There was a problem hiding this comment.
This drops any repeated keyed event within the 30s window, even when it comes from the same source. For example, two message.updated events for the same message (or a remove/re-add reaction cycle by the same agent) within 30 seconds will make the second real event return here and never reach relay.addListener, because the dedupe state is global rather than scoped to cross-source duplicates.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Good catch — fixed in d71fc62. Dedupe is now source-aware: each key tracks the set of sources that delivered the current occurrence. A copy from a source already in the set is a genuine repeat (message edit, re-reaction) and starts a new occurrence, so same-source repeats always pass through; only copies from other sources within the window collapse. Covered by the new "passes genuine repeats from the same source while collapsing cross-source copies" test using two message.updated events for the same message.
Generated by Claude Code
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (1)
CHANGELOG.md (1)
29-29: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winTrim the changelog entry for conciseness.
This bullet spans several sentences covering internal mechanics (
/v1/ws, 401 handling, node transport, dedup logic,onError, 10s timer) rather than a short, impact-first summary. As per coding guidelines, "Changelog entries should be concise and impact-first, with one short bullet per user-visible change. Omit issue/PR links, internal notes, and implementation details."📝 Suggested tighter entry
-- `@agent-relay/sdk` `relay.addListener(...)` on a workspace-key client now receives channel messages, DMs, and thread replies. It previously listened on the legacy `/v1/ws` workspace stream, which relaycast v5 rejects for workspace keys (observer token required), so the socket 401'd into a silent reconnect loop and listeners got nothing while deliveries queued unread. The workspace-level listener hub now streams through every registered agent client (`workspace.register`/`reconnect`) over the v5 node transport, deduplicating events that reach multiple locally-registered agents; the workspace stream remains only as a pre-registration fallback. Listener `events.connect()` failures are reported through `onError` instead of being swallowed, and a listener left waiting with no registered agent warns after 10s. +- `@agent-relay/sdk` `relay.addListener(...)` on a workspace-key client now correctly receives channel messages, DMs, and thread replies instead of silently getting nothing; connect failures now surface via `onError`, and an unregistered listener warns after 10s.🤖 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 `@CHANGELOG.md` at line 29, The changelog bullet in the main entry is too long and includes internal implementation details. Shorten the entry in CHANGELOG.md to an impact-first, user-visible summary for `@agent-relay/sdk` relay.addListener behavior, and remove references to /v1/ws, 401 reconnect loops, node transport, deduplication, onError handling, and the 10s warning.Source: Path instructions
🤖 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 `@packages/sdk/src/messaging/event-fanin.ts`:
- Around line 71-85: The dedupe logic in eventFanin/getEventKey is using coarse
entity-based keys for repeatable state events, which can incorrectly collapse
legitimate transitions for the same agent or channel. Update the key generation
in event-fanin.ts so only events with stable unique IDs are deduped, or derive a
canonical payload identity for cases like agentOnline, channelUpdated, and
memberJoined instead of using just event.type plus agent/channel fields. Keep
the fix centered around getEventKey and the seenEvents dedupe path so repeat
events from the same source are not dropped.
- Around line 207-221: The fallback stream in event-fanin is not receiving
channel subscriptions that were registered before sources existed. Update the
subscription flow in the subscribe/unsubscribe logic and attachFallback() so the
active fallback mirrors desiredChannels whenever fallbackForwarding is active,
including replaying existing channels when the fallback attaches and keeping
them in sync on later subscribe/unsubscribe calls. Use the event-fanin helpers
around attachFallback, subscribe, and unsubscribe to ensure the fallback
observes the same channel set as sources.
---
Nitpick comments:
In `@CHANGELOG.md`:
- Line 29: The changelog bullet in the main entry is too long and includes
internal implementation details. Shorten the entry in CHANGELOG.md to an
impact-first, user-visible summary for `@agent-relay/sdk` relay.addListener
behavior, and remove references to /v1/ws, 401 reconnect loops, node transport,
deduplication, onError handling, and the 10s warning.
🪄 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: 55380139-c650-4503-a1e8-b13fc676211c
📒 Files selected for processing (7)
CHANGELOG.mdpackages/sdk/src/__tests__/event-fanin.test.tspackages/sdk/src/__tests__/listeners.test.tspackages/sdk/src/agent-relay.tspackages/sdk/src/listeners.tspackages/sdk/src/messaging/event-fanin.tspackages/sdk/src/messaging/index.ts
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
4 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="packages/sdk/src/listeners.ts">
<violation number="1" location="packages/sdk/src/listeners.ts:658">
P3: The `logRelayHandlerError` message says `"listener" handler for ... threw:` but this error path handles `context.events.connect()` failures, not handler execution errors. The log message is misleading — a connect failure reads as if a user-supplied handler threw. Use an `operation` field or a dedicated message prefix so connect errors are distinguishable from handler errors at a glance.</violation>
</file>
<file name="packages/sdk/src/messaging/event-fanin.ts">
<violation number="1" location="packages/sdk/src/messaging/event-fanin.ts:219">
P2: `attachFallback()` connects the workspace stream but never replays `desiredChannels` onto it, and `subscribe()`/`unsubscribe()` only iterate `sources` (which excludes the fallback). If channel-scoped subscriptions are added while the fallback is the only active stream (before any agent registers), those channels won't be subscribed on the fallback transport, so channel-scoped events may not arrive during the pre-registration window.
Consider replaying `desiredChannels` in `attachFallback()` after `connect()`, and mirroring `subscribe`/`unsubscribe` calls to the fallback while `fallbackForwarding` is active.</violation>
</file>
Reply with feedback, questions, or to request a fix.
Re-trigger cubic
| if (typeof fallback.connect === 'function') { | ||
| try { | ||
| fallback.connect(); | ||
| fallbackConnected = true; |
There was a problem hiding this comment.
P2: attachFallback() connects the workspace stream but never replays desiredChannels onto it, and subscribe()/unsubscribe() only iterate sources (which excludes the fallback). If channel-scoped subscriptions are added while the fallback is the only active stream (before any agent registers), those channels won't be subscribed on the fallback transport, so channel-scoped events may not arrive during the pre-registration window.
Consider replaying desiredChannels in attachFallback() after connect(), and mirroring subscribe/unsubscribe calls to the fallback while fallbackForwarding is active.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At packages/sdk/src/messaging/event-fanin.ts, line 219:
<comment>`attachFallback()` connects the workspace stream but never replays `desiredChannels` onto it, and `subscribe()`/`unsubscribe()` only iterate `sources` (which excludes the fallback). If channel-scoped subscriptions are added while the fallback is the only active stream (before any agent registers), those channels won't be subscribed on the fallback transport, so channel-scoped events may not arrive during the pre-registration window.
Consider replaying `desiredChannels` in `attachFallback()` after `connect()`, and mirroring `subscribe`/`unsubscribe` calls to the fallback while `fallbackForwarding` is active.</comment>
<file context>
@@ -0,0 +1,313 @@
+ if (typeof fallback.connect === 'function') {
+ try {
+ fallback.connect();
+ fallbackConnected = true;
+ } catch (error) {
+ // A workspace-key client may have no stream at all; agent sources can
</file context>
| } catch { | ||
| // No stream available (no agent token and no workspace stream). | ||
| } catch (error) { | ||
| makeReporter(context, { |
There was a problem hiding this comment.
P3: The logRelayHandlerError message says "listener" handler for ... threw: but this error path handles context.events.connect() failures, not handler execution errors. The log message is misleading — a connect failure reads as if a user-supplied handler threw. Use an operation field or a dedicated message prefix so connect errors are distinguishable from handler errors at a glance.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At packages/sdk/src/listeners.ts, line 658:
<comment>The `logRelayHandlerError` message says `"listener" handler for ... threw:` but this error path handles `context.events.connect()` failures, not handler execution errors. The log message is misleading — a connect failure reads as if a user-supplied handler threw. Use an `operation` field or a dedicated message prefix so connect errors are distinguishable from handler errors at a glance.</comment>
<file context>
@@ -648,12 +648,17 @@ export function createListenerHub(
- } catch {
- // No stream available (no agent token and no workspace stream).
+ } catch (error) {
+ makeReporter(context, {
+ source: 'listener',
+ selector: typeof selector === 'string' ? selector : 'predicate',
</file context>
…anup Review follow-ups on the events fan-in: - Dedupe is now source-aware: only cross-source copies of one occurrence collapse. A repeat from a source that already delivered the previous occurrence (message edit, re-reaction within the window, presence flap) starts a new occurrence and always passes through — the previous global key window could drop legitimate same-source repeats. - disconnect() detaches all source forwarding (and connect() re-attaches it), so a source reconnected later by another owner cannot feed listeners that were explicitly disconnected. - Defensive typeof guards on source/fallback disconnect, subscribe, and unsubscribe so partial surfaces injected via createAgentMessaging don't produce noisy TypeErrors through the error hooks. - Trimmed the changelog entry to the impact-first house style. Test fakes' emit now mirrors the real client contract without double-firing when emitting the literal 'any' key. E2E re-verified against a self-hosted engine (both listener orders receive #general). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VASvErz2MxkQMCsSzTgsbe
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VASvErz2MxkQMCsSzTgsbe
… stream
new AgentRelay({ observerToken }) streams relay.addListener(...) from the
workspace observer plane instead of the registered-agent fan-in. A new
observer event source implements the cursor protocol: connect the live
/v1/ws observer stream and buffer frames, REST-backfill
GET /v1/workspace/events from the in-memory cursor (sinceSeq, default 0)
paginating until latest_seq, then emit backfilled events followed by the
buffered/live ones deduped and ordered by seq. Frames without a seq
(server-side log append failed) pass through as live-only. A 404 backfill
(older engines) degrades gracefully to live-only; the cursor still advances
from live seq. Persisting the cursor stays with the caller via
sinceSeq/onCursor.
Both legs flow through normalizeMessagingEvent, so listeners receive the
same public event shapes as every other source. In observer mode the
source is the fan-in's sole source (no workspace-stream fallback, no
no-agent-source warning), and workspace.register()/reconnect() throw a
clear read-only error at the facade boundary.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VASvErz2MxkQMCsSzTgsbe
ensure_default_channels/ensure_extra_channels join the broker-self agent to channels so the broker can post to them, but that membership also made the engine's channel fan-out write a delivery row per message to broker-self's permanently-offline implicit direct node — rows that queued forever and churned through TTL expiry. The engine skips muted members in channel delivery fan-out, so mute each channel for the self agent right after ensure_joined_channel succeeds (POST /v1/channels/:name/mute via the relaycast crate's AgentClient::mute_channel). Best-effort: failures log a warning and never fail startup. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VASvErz2MxkQMCsSzTgsbe
…roker self-mute specs/observer-plane.md documents the two-plane architecture: the bug class motivating it (lossy workspace stream on reconnect, Pear polling reconciliation, broker relay_inbound blind to remote recipients, dashboard-identity deliveries queuing forever), the unchanged v5 participant plane, the observer-plane contract (event log table, seq semantics, GET /v1/workspace/events, client cursor protocol), per-component changes, and the migration order. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VASvErz2MxkQMCsSzTgsbe
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
| * @returns An events surface suitable as an event fan-in source. | ||
| */ | ||
| export function createObserverEventSource(options: ObserverEventSourceOptions): RelayMessagingEventsSurface { | ||
| const baseUrl = (options.baseUrl ?? DEFAULT_BASE_URL).replace(/\/+$/, ''); |
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 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 `@packages/sdk/src/messaging/observer-source.ts`:
- Around line 207-209: The backfill fetch in observer-source’s request flow can
hang indefinitely, leaving backfillDone false and causing pending live frames to
buffer forever. Update the fetchImpl call in the backfill logic to use an
abortable timeout so stalled requests fail promptly, and make sure the existing
catch path in observer-source degrades to live-only delivery when the timeout
aborts.
- Around line 248-256: The live stream setup in observer-source’s connect flow
leaves a partially initialized `live` instance assigned when
`createLiveStream()`, `live.on.any(handleLiveFrame)`, or `live.connect()`
throws, which blocks later retries. Update the `connect()` path to clear or
reset `live` (and any related listener state like `offLive`) inside the catch
block before returning, so subsequent calls can create a fresh stream and retry
initialization.
- Around line 195-200: The buffered frame ordering in observer-source is not
total, so seq-less frames can compare equal to both seq frames and be delivered
out of order before the cursor advances. Update the sorting logic in the
buffered delivery path to order all pending frames deterministically in the
frame-processing flow around readSeq and deliver, ensuring seq-bearing frames
are ordered consistently ahead of advancing the cursor and seq-less frames do
not break ordering. After the sort, verify the cursor only advances after the
correct lowest-seq frames have been delivered.
🪄 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: f310e628-56d4-4199-8923-0242a022e307
📒 Files selected for processing (7)
CHANGELOG.mdcrates/broker/src/relaycast/ws.rspackages/sdk/src/__tests__/observer-source.test.tspackages/sdk/src/agent-relay.tspackages/sdk/src/messaging/index.tspackages/sdk/src/messaging/observer-source.tsspecs/observer-plane.md
✅ Files skipped from review due to trivial changes (1)
- CHANGELOG.md
🚧 Files skipped from review as they are similar to previous changes (1)
- packages/sdk/src/messaging/index.ts
| buffered.sort((a, b) => { | ||
| const seqA = readSeq(a); | ||
| const seqB = readSeq(b); | ||
| return seqA !== undefined && seqB !== undefined ? seqA - seqB : 0; | ||
| }); | ||
| for (const raw of buffered) deliver(raw); |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Make pending-frame ordering total before advancing the cursor.
When a seq-less frame is buffered between out-of-order seq frames, the comparator returns 0 for those comparisons, so [seq:4, open, seq:3] may deliver 4 first; cursor then advances to 4 and drops 3.
Proposed fix
- buffered.sort((a, b) => {
- const seqA = readSeq(a);
- const seqB = readSeq(b);
- return seqA !== undefined && seqB !== undefined ? seqA - seqB : 0;
- });
- for (const raw of buffered) deliver(raw);
+ const sequenced = buffered
+ .map((raw) => ({ raw, seq: readSeq(raw) }))
+ .filter((item): item is { raw: unknown; seq: number } => item.seq !== undefined)
+ .sort((a, b) => a.seq - b.seq);
+ let nextSequenced = 0;
+ for (const raw of buffered) {
+ deliver(readSeq(raw) === undefined ? raw : sequenced[nextSequenced++].raw);
+ }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| buffered.sort((a, b) => { | |
| const seqA = readSeq(a); | |
| const seqB = readSeq(b); | |
| return seqA !== undefined && seqB !== undefined ? seqA - seqB : 0; | |
| }); | |
| for (const raw of buffered) deliver(raw); | |
| const sequenced = buffered | |
| .map((raw) => ({ raw, seq: readSeq(raw) })) | |
| .filter((item): item is { raw: unknown; seq: number } => item.seq !== undefined) | |
| .sort((a, b) => a.seq - b.seq); | |
| let nextSequenced = 0; | |
| for (const raw of buffered) { | |
| deliver(readSeq(raw) === undefined ? raw : sequenced[nextSequenced++].raw); | |
| } |
🤖 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 `@packages/sdk/src/messaging/observer-source.ts` around lines 195 - 200, The
buffered frame ordering in observer-source is not total, so seq-less frames can
compare equal to both seq frames and be delivered out of order before the cursor
advances. Update the sorting logic in the buffered delivery path to order all
pending frames deterministically in the frame-processing flow around readSeq and
deliver, ensuring seq-bearing frames are ordered consistently ahead of advancing
the cursor and seq-less frames do not break ordering. After the sort, verify the
cursor only advances after the correct lowest-seq frames have been delivered.
| const response = await fetchImpl(url, { | ||
| headers: { Authorization: `Bearer ${options.observerToken}` }, | ||
| }); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Bound backfill requests so live frames are not buffered forever.
If fetchImpl hangs, backfillDone never flips and live events accumulate in pending without being delivered. Add an abortable timeout and let the existing catch path degrade to live-only.
🤖 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 `@packages/sdk/src/messaging/observer-source.ts` around lines 207 - 209, The
backfill fetch in observer-source’s request flow can hang indefinitely, leaving
backfillDone false and causing pending live frames to buffer forever. Update the
fetchImpl call in the backfill logic to use an abortable timeout so stalled
requests fail promptly, and make sure the existing catch path in observer-source
degrades to live-only delivery when the timeout aborts.
| try { | ||
| live = createLiveStream(); | ||
| offLive = live.on.any(handleLiveFrame); | ||
| live.connect(); | ||
| } catch (error) { | ||
| report(error); | ||
| // With no live stream, backfilled events are all we can deliver; | ||
| // don't hold them hostage in the buffer. | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Clear partially initialized live streams after connect failures.
If live.on.any(...) or live.connect() throws after live is assigned, future connect() calls return early and the listener cannot retry the live stream.
Proposed fix
} catch (error) {
+ offLive?.();
+ offLive = undefined;
+ const stream = live;
+ live = undefined;
+ try {
+ stream?.disconnect();
+ } catch {
+ // Best-effort cleanup after a failed live-stream initialization.
+ }
report(error);
// With no live stream, backfilled events are all we can deliver;📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| try { | |
| live = createLiveStream(); | |
| offLive = live.on.any(handleLiveFrame); | |
| live.connect(); | |
| } catch (error) { | |
| report(error); | |
| // With no live stream, backfilled events are all we can deliver; | |
| // don't hold them hostage in the buffer. | |
| } | |
| try { | |
| live = createLiveStream(); | |
| offLive = live.on.any(handleLiveFrame); | |
| live.connect(); | |
| } catch (error) { | |
| offLive?.(); | |
| offLive = undefined; | |
| const stream = live; | |
| live = undefined; | |
| try { | |
| stream?.disconnect(); | |
| } catch { | |
| // Best-effort cleanup after a failed live-stream initialization. | |
| } | |
| report(error); | |
| // With no live stream, backfilled events are all we can deliver; | |
| // don't hold them hostage in the buffer. | |
| } |
🤖 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 `@packages/sdk/src/messaging/observer-source.ts` around lines 248 - 256, The
live stream setup in observer-source’s connect flow leaves a partially
initialized `live` instance assigned when `createLiveStream()`,
`live.on.any(handleLiveFrame)`, or `live.connect()` throws, which blocks later
retries. Update the `connect()` path to clear or reset `live` (and any related
listener state like `offLive`) inside the catch block before returning, so
subsequent calls can create a fresh stream and retry initialization.
There was a problem hiding this comment.
3 issues found across 7 files (changes from recent commits).
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="packages/sdk/src/messaging/observer-source.ts">
<violation number="1" location="packages/sdk/src/messaging/observer-source.ts:198">
P1: Buffered live events can be lost when an unsequenced transport frame sits between out-of-order sequenced frames. The current sort comparator does not actually order seq-stamped frames across seq-less frames, so a higher seq can advance `cursor` before a lower seq is delivered.</violation>
<violation number="2" location="packages/sdk/src/messaging/observer-source.ts:207">
P2: The `fetchImpl` call has no timeout or `AbortSignal`. If the backfill endpoint hangs (e.g., network partition, proxy stall), `backfillDone` never flips and live frames accumulate in `pending` without delivery — effectively a silent DoS on the observer stream. Consider adding an `AbortController` with a reasonable timeout (e.g., 30s) so the source can degrade to live-only rather than buffering indefinitely.</violation>
<violation number="3" location="packages/sdk/src/messaging/observer-source.ts:251">
P2: A transient observer WebSocket connect failure can permanently disable live-stream retries for this source instance. Since `live` remains set after `live.connect()` throws, later `connect()` calls return early instead of creating a fresh stream.</violation>
</file>
Tip: Review your code locally with the cubic CLI to iterate faster.
Re-trigger cubic
| buffered.sort((a, b) => { | ||
| const seqA = readSeq(a); | ||
| const seqB = readSeq(b); | ||
| return seqA !== undefined && seqB !== undefined ? seqA - seqB : 0; |
There was a problem hiding this comment.
P1: Buffered live events can be lost when an unsequenced transport frame sits between out-of-order sequenced frames. The current sort comparator does not actually order seq-stamped frames across seq-less frames, so a higher seq can advance cursor before a lower seq is delivered.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At packages/sdk/src/messaging/observer-source.ts, line 198:
<comment>Buffered live events can be lost when an unsequenced transport frame sits between out-of-order sequenced frames. The current sort comparator does not actually order seq-stamped frames across seq-less frames, so a higher seq can advance `cursor` before a lower seq is delivered.</comment>
<file context>
@@ -0,0 +1,294 @@
+ buffered.sort((a, b) => {
+ const seqA = readSeq(a);
+ const seqB = readSeq(b);
+ return seqA !== undefined && seqB !== undefined ? seqA - seqB : 0;
+ });
+ for (const raw of buffered) deliver(raw);
</file context>
| try { | ||
| live = createLiveStream(); | ||
| offLive = live.on.any(handleLiveFrame); | ||
| live.connect(); |
There was a problem hiding this comment.
P2: A transient observer WebSocket connect failure can permanently disable live-stream retries for this source instance. Since live remains set after live.connect() throws, later connect() calls return early instead of creating a fresh stream.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At packages/sdk/src/messaging/observer-source.ts, line 251:
<comment>A transient observer WebSocket connect failure can permanently disable live-stream retries for this source instance. Since `live` remains set after `live.connect()` throws, later `connect()` calls return early instead of creating a fresh stream.</comment>
<file context>
@@ -0,0 +1,294 @@
+ try {
+ live = createLiveStream();
+ offLive = live.on.any(handleLiveFrame);
+ live.connect();
+ } catch (error) {
+ report(error);
</file context>
| since: number | ||
| ): Promise<{ events: BackfillEventRow[]; latestSeq: number } | undefined> => { | ||
| const url = `${baseUrl}/v1/workspace/events?since=${since}&limit=${pageSize}`; | ||
| const response = await fetchImpl(url, { |
There was a problem hiding this comment.
P2: The fetchImpl call has no timeout or AbortSignal. If the backfill endpoint hangs (e.g., network partition, proxy stall), backfillDone never flips and live frames accumulate in pending without delivery — effectively a silent DoS on the observer stream. Consider adding an AbortController with a reasonable timeout (e.g., 30s) so the source can degrade to live-only rather than buffering indefinitely.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At packages/sdk/src/messaging/observer-source.ts, line 207:
<comment>The `fetchImpl` call has no timeout or `AbortSignal`. If the backfill endpoint hangs (e.g., network partition, proxy stall), `backfillDone` never flips and live frames accumulate in `pending` without delivery — effectively a silent DoS on the observer stream. Consider adding an `AbortController` with a reasonable timeout (e.g., 30s) so the source can degrade to live-only rather than buffering indefinitely.</comment>
<file context>
@@ -0,0 +1,294 @@
+ since: number
+ ): Promise<{ events: BackfillEventRow[]; latestSeq: number } | undefined> => {
+ const url = `${baseUrl}/v1/workspace/events?since=${since}&limit=${pageSize}`;
+ const response = await fetchImpl(url, {
+ headers: { Authorization: `Bearer ${options.observerToken}` },
+ });
</file context>
…tion Cross-repo E2E against the new durable event log surfaced three seams in the observer plane: - The observer source's live leg used the RelayCast client, whose schema parsing strips the top-level `seq` from frames — the cursor never advanced from live events, so resume re-delivered everything. The default live stream is now a raw WebSocket to /v1/ws (capped-backoff reconnect, re-backfill from the cursor on every reopen) so frames arrive with `seq` intact. - Raw engine frames carry reactions as one `message.reacted` type with an `action` field (higher-level clients split it before normalize sees it); normalizeMessagingEvent now maps that shape to reactionAdded/reactionRemoved so observer listeners receive reactions. - The broker's default observer-token scopes gain `reactions:read` — the live stream filters message.reacted for tokens without it, so UI tokens minted via POST /api/observer-token silently lost reactions. Verified end-to-end against a self-hosted engine: live observation with cursor advance (messages and reactions), read-only enforcement, and an offline gap recovered exactly via cursor backfill with no duplicates. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VASvErz2MxkQMCsSzTgsbe
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 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 `@packages/sdk/src/messaging/observer-source.ts`:
- Around line 84-90: The WebSocket handlers in observer-source.ts are swallowing
abnormal failures and immediately reconnecting, which can hide
auth/permission/server rejection issues. Update the ws.onclose and ws.onerror
logic in the observer source connection flow so that non-clean close/error
events are forwarded through onError before any reconnect is scheduled, while
preserving the existing reconnect behavior for normal disconnects. Use the
existing socket, scheduleReconnect, and onError handling paths to distinguish
abnormal failures from routine closes.
- Around line 22-24: The default observer source in observer-source.ts assumes a
global WebSocket, but packages/sdk still supports Node 20.9.x where that global
is unavailable. Update the observer source implementation (and its associated
WebSocket setup logic) to either use a non-global fallback path for WebSocket
creation or raise the package engine floor so the supported Node range matches
the runtime requirement. Ensure the change is localized around the
observer-source module and any WebSocket client initialization helpers it uses.
🪄 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: ec42c899-821e-40a1-b2d5-00743f3e1d9a
📒 Files selected for processing (5)
crates/broker/src/runtime/api.rscrates/broker/src/runtime/tests.rspackages/sdk/src/__tests__/observer-source.test.tspackages/sdk/src/messaging/normalize.tspackages/sdk/src/messaging/observer-source.ts
🚧 Files skipped from review as they are similar to previous changes (1)
- packages/sdk/src/tests/observer-source.test.ts
There was a problem hiding this comment.
3 issues found across 5 files (changes from recent commits).
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="packages/sdk/src/messaging/observer-source.ts">
<violation number="1" location="packages/sdk/src/messaging/observer-source.ts:53">
P1: The default observer live stream now requires a global `WebSocket`, which is only available in Node 21+. The repo’s declared Node engine is `>=20.9.0`, so SDK consumers on Node 20 will see a live-stream error instead of receiving observer events. The previous `RelayCast` client provided its own transport and supported the existing engine range. Please either add a WebSocket polyfill (e.g., `ws`) for the default Node transport, allow callers to inject a `WebSocket` implementation, or align the SDK’s engine requirement with this new runtime dependency.</violation>
<violation number="2" location="packages/sdk/src/messaging/observer-source.ts:84">
P2: **Observer WebSocket failures are silently swallowed, undermining the louder-error goal.** The new raw WebSocket live stream only schedules a reconnect on `ws.onclose` and leaves `ws.onerror` as a no-op. That means invalid observer tokens, server-side socket drops, or other connection failures enter an endless silent reconnect loop instead of reaching the `onError` hook, even though the `ObserverEventSourceOptions.onError` docstring says it receives live-stream failures.
Consider routing abnormal WebSocket closures through `report(...)`. The `closed` flag is already set before an intentional `disconnect()`, so you can guard against false positives during teardown. For example, in `ws.onclose` report a non-clean close with the close code (and reason, if present), then let the reconnect logic continue. This keeps the auto-reconnect behavior but gives callers the actionable signal they need to debug authentication or network issues.</violation>
<violation number="3" location="packages/sdk/src/messaging/observer-source.ts:358">
P2: The `on.open` handler registered in `connect()` is never unsubscribed. You store and clean up `offLive` for `on.any`, but the `open` subscription is discarded. If a reused `ObserverLiveStream` instance is supplied (which the test fakes do), each connect/disconnect cycle adds another stale `open` handler that triggers `runBackfill()` on every reconnect, causing duplicate backfill work and potential duplicate event delivery. Capture the `open` unsubscribe and call it in `disconnect()` alongside `offLive`.</violation>
</file>
Tip: Review your code locally with the cubic CLI to iterate faster.
Re-trigger cubic
|
|
||
| const open = (): void => { | ||
| if (closed || socket) return; | ||
| const WebSocketImpl = (globalThis as { WebSocket?: typeof WebSocket }).WebSocket; |
There was a problem hiding this comment.
P1: The default observer live stream now requires a global WebSocket, which is only available in Node 21+. The repo’s declared Node engine is >=20.9.0, so SDK consumers on Node 20 will see a live-stream error instead of receiving observer events. The previous RelayCast client provided its own transport and supported the existing engine range. Please either add a WebSocket polyfill (e.g., ws) for the default Node transport, allow callers to inject a WebSocket implementation, or align the SDK’s engine requirement with this new runtime dependency.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At packages/sdk/src/messaging/observer-source.ts, line 53:
<comment>The default observer live stream now requires a global `WebSocket`, which is only available in Node 21+. The repo’s declared Node engine is `>=20.9.0`, so SDK consumers on Node 20 will see a live-stream error instead of receiving observer events. The previous `RelayCast` client provided its own transport and supported the existing engine range. Please either add a WebSocket polyfill (e.g., `ws`) for the default Node transport, allow callers to inject a `WebSocket` implementation, or align the SDK’s engine requirement with this new runtime dependency.</comment>
<file context>
@@ -1,18 +1,122 @@
+
+ const open = (): void => {
+ if (closed || socket) return;
+ const WebSocketImpl = (globalThis as { WebSocket?: typeof WebSocket }).WebSocket;
+ if (!WebSocketImpl) {
+ report(
</file context>
| // log: on every reopen after the first, buffer live frames again and | ||
| // re-backfill from the cursor to close the gap. | ||
| let hadOpen = false; | ||
| live.on.open?.(() => { |
There was a problem hiding this comment.
P2: The on.open handler registered in connect() is never unsubscribed. You store and clean up offLive for on.any, but the open subscription is discarded. If a reused ObserverLiveStream instance is supplied (which the test fakes do), each connect/disconnect cycle adds another stale open handler that triggers runBackfill() on every reconnect, causing duplicate backfill work and potential duplicate event delivery. Capture the open unsubscribe and call it in disconnect() alongside offLive.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At packages/sdk/src/messaging/observer-source.ts, line 358:
<comment>The `on.open` handler registered in `connect()` is never unsubscribed. You store and clean up `offLive` for `on.any`, but the `open` subscription is discarded. If a reused `ObserverLiveStream` instance is supplied (which the test fakes do), each connect/disconnect cycle adds another stale `open` handler that triggers `runBackfill()` on every reconnect, causing duplicate backfill work and potential duplicate event delivery. Capture the `open` unsubscribe and call it in `disconnect()` alongside `offLive`.</comment>
<file context>
@@ -248,6 +351,17 @@ export function createObserverEventSource(options: ObserverEventSourceOptions):
+ // log: on every reopen after the first, buffer live frames again and
+ // re-backfill from the cursor to close the gap.
+ let hadOpen = false;
+ live.on.open?.(() => {
+ if (hadOpen) {
+ backfillDone = false;
</file context>
| ws.onclose = () => { | ||
| if (socket === ws) socket = undefined; | ||
| scheduleReconnect(); | ||
| }; | ||
| ws.onerror = () => { | ||
| // The close handler owns reconnection. | ||
| }; |
There was a problem hiding this comment.
P2: Observer WebSocket failures are silently swallowed, undermining the louder-error goal. The new raw WebSocket live stream only schedules a reconnect on ws.onclose and leaves ws.onerror as a no-op. That means invalid observer tokens, server-side socket drops, or other connection failures enter an endless silent reconnect loop instead of reaching the onError hook, even though the ObserverEventSourceOptions.onError docstring says it receives live-stream failures.
Consider routing abnormal WebSocket closures through report(...). The closed flag is already set before an intentional disconnect(), so you can guard against false positives during teardown. For example, in ws.onclose report a non-clean close with the close code (and reason, if present), then let the reconnect logic continue. This keeps the auto-reconnect behavior but gives callers the actionable signal they need to debug authentication or network issues.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At packages/sdk/src/messaging/observer-source.ts, line 84:
<comment>**Observer WebSocket failures are silently swallowed, undermining the louder-error goal.** The new raw WebSocket live stream only schedules a reconnect on `ws.onclose` and leaves `ws.onerror` as a no-op. That means invalid observer tokens, server-side socket drops, or other connection failures enter an endless silent reconnect loop instead of reaching the `onError` hook, even though the `ObserverEventSourceOptions.onError` docstring says it receives live-stream failures.
Consider routing abnormal WebSocket closures through `report(...)`. The `closed` flag is already set before an intentional `disconnect()`, so you can guard against false positives during teardown. For example, in `ws.onclose` report a non-clean close with the close code (and reason, if present), then let the reconnect logic continue. This keeps the auto-reconnect behavior but gives callers the actionable signal they need to debug authentication or network issues.</comment>
<file context>
@@ -1,18 +1,122 @@
+ }
+ for (const handler of anyHandlers) handler(frame);
+ };
+ ws.onclose = () => {
+ if (socket === ws) socket = undefined;
+ scheduleReconnect();
</file context>
| ws.onclose = () => { | |
| if (socket === ws) socket = undefined; | |
| scheduleReconnect(); | |
| }; | |
| ws.onerror = () => { | |
| // The close handler owns reconnection. | |
| }; | |
| ws.onclose = (event: CloseEvent) => { | |
| if (socket === ws) socket = undefined; | |
| if (!closed && !event.wasClean) { | |
| report( | |
| new Error( | |
| `observer WebSocket closed unexpectedly (code ${event.code}${event.reason ? `: ${event.reason}` : ''})` | |
| ) | |
| ); | |
| } | |
| scheduleReconnect(); | |
| }; | |
| ws.onerror = () => { | |
| // Abnormal closures are reported above; onerror carries no additional detail. | |
| }; |
Hardens the observer live leg against the CodeQL findings on the raw WebSocket connection: - The token no longer travels in the URL: it is sent as an Authorization: Bearer header via Node's undici WebSocket constructor options (the engine's upgrade path accepts both forms). Runtimes whose WebSocket rejects or ignores constructor options — browsers, per the WHATWG signature — are detected (constructor throw, or close before the first open) and downgraded once to the server's ?token= query convention. - The stream URL scheme is always wss:// except for loopback hosts (local self-hosted engines), instead of blindly mapping http -> ws. E2E re-verified against a self-hosted engine over the header-auth path: live observation, reactions, and cursor backfill after an offline gap all intact. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VASvErz2MxkQMCsSzTgsbe
There was a problem hiding this comment.
1 issue found across 2 files (changes from recent commits).
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="packages/sdk/src/messaging/observer-source.ts">
<violation number="1" location="packages/sdk/src/messaging/observer-source.ts:136">
P1: The WebSocket auth-downgrade heuristic can misclassify transient pre-open failures and permanently switch the observer stream to sending the token in the URL. A close before the first `open` is treated as proof that the runtime ignored the `Authorization` header, but the same symptom is produced by network drops, TLS errors, server restarts, or temporary auth/backend failures. Once `useQueryToken` is set, it is never reset, so a one-off startup hiccup causes a persistent security regression (token in query string) and may cause a persistent failure if the deployment only accepts header auth. Consider narrowing the downgrade signal: rely on the constructor throw for runtimes that do not support options, and/or require a specific close code/reason before concluding that headers are unsupported. If a fallback is still needed, retry header auth at least once before permanently switching to the URL token.</violation>
</file>
Tip: Review your code locally with the cubic CLI to iterate faster.
Re-trigger cubic
| // means the runtime accepted the options object but ignored the | ||
| // headers (auth rejected): downgrade to the query token and retry | ||
| // immediately, once. | ||
| if (!useQueryToken && !everOpened && !openedHere) { |
There was a problem hiding this comment.
P1: The WebSocket auth-downgrade heuristic can misclassify transient pre-open failures and permanently switch the observer stream to sending the token in the URL. A close before the first open is treated as proof that the runtime ignored the Authorization header, but the same symptom is produced by network drops, TLS errors, server restarts, or temporary auth/backend failures. Once useQueryToken is set, it is never reset, so a one-off startup hiccup causes a persistent security regression (token in query string) and may cause a persistent failure if the deployment only accepts header auth. Consider narrowing the downgrade signal: rely on the constructor throw for runtimes that do not support options, and/or require a specific close code/reason before concluding that headers are unsupported. If a fallback is still needed, retry header auth at least once before permanently switching to the URL token.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At packages/sdk/src/messaging/observer-source.ts, line 136:
<comment>The WebSocket auth-downgrade heuristic can misclassify transient pre-open failures and permanently switch the observer stream to sending the token in the URL. A close before the first `open` is treated as proof that the runtime ignored the `Authorization` header, but the same symptom is produced by network drops, TLS errors, server restarts, or temporary auth/backend failures. Once `useQueryToken` is set, it is never reset, so a one-off startup hiccup causes a persistent security regression (token in query string) and may cause a persistent failure if the deployment only accepts header auth. Consider narrowing the downgrade signal: rely on the constructor throw for runtimes that do not support options, and/or require a specific close code/reason before concluding that headers are unsupported. If a fallback is still needed, retry header auth at least once before permanently switching to the URL token.</comment>
<file context>
@@ -83,6 +129,15 @@ function createRawObserverStream(
+ // means the runtime accepted the options object but ignored the
+ // headers (auth rejected): downgrade to the query token and retry
+ // immediately, once.
+ if (!useQueryToken && !everOpened && !openedHere) {
+ useQueryToken = true;
+ scheduleReconnect(0);
</file context>
The engine's GET /v1/workspace/events now returns next_since — the seq of the last row the server's scan consumed, visible or hidden — so scoped observer tokens whose windows are fully filtered server-side still make pagination progress. The backfill loop advances the cursor by it (hidden events are never delivered live to that token either, so skipping their seqs is safe) and stops on any page that makes no progress, preserving behavior against older engines without the field. Spec updated. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VASvErz2MxkQMCsSzTgsbe
There was a problem hiding this comment.
1 issue found across 3 files (changes from recent commits).
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="packages/sdk/src/messaging/observer-source.ts">
<violation number="1" location="packages/sdk/src/messaging/observer-source.ts:410">
P1: The backfill loop can silently skip historical events for scoped observer tokens when the engine returns an empty filtered page without `next_since`. In that case `cursor` does not advance, `latestSeq` is still ahead, and the new `if (cursor <= before) break;` exits without reporting an error or advancing the cursor. Any visible events that occur after the hidden block will never be backfilled. Consider reporting an error through the existing `report(...)` hook in this fallback path, or at least surfacing a warning that the backfill is incomplete, rather than stopping silently. If skipping is intentional for this compatibility mode, a test and a logged reason would make the trade-off explicit and prevent future regressions.</violation>
</file>
Tip: Review your code locally with the cubic CLI to iterate faster.
Re-trigger cubic
| if (cursor >= page.latestSeq) break; | ||
| // No progress this page (older engine without next_since returning | ||
| // only hidden rows): stop rather than loop forever. | ||
| if (cursor <= before) break; |
There was a problem hiding this comment.
P1: The backfill loop can silently skip historical events for scoped observer tokens when the engine returns an empty filtered page without next_since. In that case cursor does not advance, latestSeq is still ahead, and the new if (cursor <= before) break; exits without reporting an error or advancing the cursor. Any visible events that occur after the hidden block will never be backfilled. Consider reporting an error through the existing report(...) hook in this fallback path, or at least surfacing a warning that the backfill is incomplete, rather than stopping silently. If skipping is intentional for this compatibility mode, a test and a logged reason would make the trade-off explicit and prevent future regressions.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At packages/sdk/src/messaging/observer-source.ts, line 410:
<comment>The backfill loop can silently skip historical events for scoped observer tokens when the engine returns an empty filtered page without `next_since`. In that case `cursor` does not advance, `latestSeq` is still ahead, and the new `if (cursor <= before) break;` exits without reporting an error or advancing the cursor. Any visible events that occur after the hidden block will never be backfilled. Consider reporting an error through the existing `report(...)` hook in this fallback path, or at least surfacing a warning that the backfill is incomplete, rather than stopping silently. If skipping is intentional for this compatibility mode, a test and a logged reason would make the trade-off explicit and prevent future regressions.</comment>
<file context>
@@ -378,15 +388,26 @@ export function createObserverEventSource(options: ObserverEventSourceOptions):
if (cursor >= page.latestSeq) break;
+ // No progress this page (older engine without next_since returning
+ // only hidden rows): stop rather than loop forever.
+ if (cursor <= before) break;
}
} catch (error) {
</file context>
| if (cursor <= before) break; | |
| if (cursor <= before) { | |
| report( | |
| new Error( | |
| 'observer backfill stopped: engine returned no visible events and no next_since cursor; historical events after seq ' + | |
| cursor + | |
| ' may be skipped' | |
| ) | |
| ); | |
| break; | |
| } |
Picks up the release carrying the durable workspace event log (relaycast v5.1.0). Verified end-to-end against the PUBLISHED @relaycast/engine 5.1.0 from npm: SDK observer mode receives live messages, reactions, and presence transitions with seq-stamped frames, and recovers an offline gap exactly via cursor backfill — including the presence events that now flow through the log. SDK suite 135/135, build and typecheck clean against 5.1.0. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VASvErz2MxkQMCsSzTgsbe
There was a problem hiding this comment.
1 issue found across 4 files (changes from recent commits).
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="CHANGELOG.md">
<violation number="1" location="CHANGELOG.md:20">
P2: The Unreleased changelog now contains two conflicting `@relaycast/sdk` version claims: the Changed bullet says the SDK was upgraded to `5.1.0`, while a Fixed bullet in the same section still says the packages require `@relaycast/sdk` `^4.1.2`. Since the repo's `package.json` files already depend on `^5.1.0`, the `^4.1.2` entry is stale and will mislead consumers about minimum compatibility. Could you remove or relocate that older bullet so the Unreleased section is internally consistent?</violation>
</file>
Tip: Review your code locally with the cubic CLI to iterate faster.
Re-trigger cubic
| @@ -11,12 +11,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 | |||
|
|
|||
There was a problem hiding this comment.
P2: The Unreleased changelog now contains two conflicting @relaycast/sdk version claims: the Changed bullet says the SDK was upgraded to 5.1.0, while a Fixed bullet in the same section still says the packages require @relaycast/sdk ^4.1.2. Since the repo's package.json files already depend on ^5.1.0, the ^4.1.2 entry is stale and will mislead consumers about minimum compatibility. Could you remove or relocate that older bullet so the Unreleased section is internally consistent?
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At CHANGELOG.md, line 20:
<comment>The Unreleased changelog now contains two conflicting `@relaycast/sdk` version claims: the Changed bullet says the SDK was upgraded to `5.1.0`, while a Fixed bullet in the same section still says the packages require `@relaycast/sdk` `^4.1.2`. Since the repo's `package.json` files already depend on `^5.1.0`, the `^4.1.2` entry is stale and will mislead consumers about minimum compatibility. Could you remove or relocate that older bullet so the Unreleased section is internally consistent?</comment>
<file context>
@@ -17,7 +17,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- `agent-relay integration` commands now talk to relayfile over its local **control-plane unix socket** (`relayfile control-plane serve`) via the published **`@relayfile/client`** package — a typed, version-negotiated client (`/v1/hello` handshake) — instead of shelling out to the `relayfile` CLI and parsing stdout. The daemon is auto-started on first use (or required already-running via `RELAYFILE_REQUIRE_DAEMON=1`); request/response types are generated from relayfile's OpenAPI so contract drift is a build error rather than a runtime surprise. The provider resource is canonicalized to relayfile's stored path-glob before bind/unbind so re-subscribing and unsubscribing match reliably. Requires relayfile ≥ 0.10.17.
- `agent-relay integration subscribe` now points the writeback subscription at the relayfile-cloud ingress and signs it with a per-channel secret fetched from relayfile (`relayfile integration writeback-secret`), instead of a relay-server path that returned 404. The secret is derived server-side and tied to the logged-in account, so there's nothing to provision; `--bridge-url`/`--bridge-secret` still override.
-- relaycast SDKs upgraded to latest: `@relaycast/sdk` 5.0.5 (v4→v5 major), `relaycast` crate 5.0.2, `relaycast-sdk` 0.3.0, Swift relaycast 5.0.5. The v5 `agents.release` now returns an action invocation (like `agents.spawn`); the `remove_agent` MCP tool surfaces that invocation.
+- relaycast SDKs upgraded to latest: `@relaycast/sdk` 5.1.0 (v4→v5 major; 5.1 adds the durable workspace event log consumed by SDK observer mode), `relaycast` crate 5.0.2, `relaycast-sdk` 0.3.0, Swift relaycast 5.0.5. The v5 `agents.release` now returns an action invocation (like `agents.spawn`); the `remove_agent` MCP tool surfaces that invocation.
- The hosted engine base URL default is owned solely by the relaycast SDK. `agent-relay`, `agent-relay-broker`, and the bundled SDKs no longer hardcode a base URL — they pass `RELAYCAST_BASE_URL`/`RELAY_BASE_URL` through for self-hosting and otherwise inherit the SDK default (`cast.agentrelay.com`). The broker reaches the fleet node-control endpoint via the SDK's `node_control_ws_url` helper and only injects `RELAY_BASE_URL` into spawned agents when an override is set.
</file context>
3 participants
Summary
relay.addListener(...)on a workspace-keyAgentRelay— the documented listen pattern in the README quickstart and.claude/rules/sdk.md— silently received zero events against a relaycast v5 engine. This is the root cause of SDK consumers not getting#generalchannel messages.The failure chain:
/v1/wsworkspace stream.catch {}inlisteners.ts, a capability-by-existence guard inrelaycast.ts, and WS auth failures becoming filtered-out pseudo-events./v1/node/wswas never opened), so they satqueuedforever.The fix routes the workspace-level listener hub through a new events fan-in (
packages/sdk/src/messaging/event-fanin.ts):workspace.register()/workspace.reconnect()is added as a source and connected over the v5 node transport once a listener exists — in either order (listen-then-register also works; the source connects lazily when it appears).deliverframe per recipient, so a message reaching several locally-registered agents would fire listeners N times — the fan-in dedupes across sources by event identity within a bounded window. Per-source transport events (error,reconnecting) are never deduped.events.connect()errors route through theonErrorhooks instead of an emptycatch, and a listener still waiting with no registered agent reports an actionable error after a 10s grace period.Also updates the listeners-test event bus fake to mirror the real client's emit contract (events fan out to their type key and
'any'), and adds aFixedentry to[Unreleased]inCHANGELOG.md.Test Plan
packages/sdk/src/__tests__/event-fanin.test.tscovering fan-in forwarding, late-source connect, cross-source dedupe, transport-event passthrough, fallback attach/detach, subscription replay, the no-source warning, andAgentRelayintegration in both orders. Full SDK suite green: 121/121.queuedwith its direct nodeoffline). After the fix, the same script receives the#generalmessage in both register-then-listen and listen-then-register orders, and the engine DB shows deliveriesacked.Screenshots
N/A
🤖 Generated with Claude Code
https://claude.ai/code/session_01VASvErz2MxkQMCsSzTgsbe
Generated by Claude Code