fix(sdk): stream relay.addListener events through registered agent clients#1226
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.
1 issue 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/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>
…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}
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
There was a problem hiding this comment.
All reported issues were addressed across 7 files (changes from recent commits).
Tip: Review your code locally with the cubic CLI to iterate faster.
Re-trigger cubic
…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
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.
All reported issues were addressed across 2 files (changes from recent commits).
Tip: Review your code locally with the cubic CLI to iterate faster.
Re-trigger cubic
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.
All reported issues were addressed across 3 files (changes from recent commits).
Tip: Review your code locally with the cubic CLI to iterate faster.
Re-trigger cubic
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.
All reported issues were addressed across 4 files (changes from recent commits).
Tip: Review your code locally with the cubic CLI to iterate faster.
Re-trigger cubic
Resolve conflicts: - packages/sdk/package.json, packages/cli/package.json: take main's dep versions (@relaycast/sdk ^6.0.0; CLI internal deps 10.2.0; @relayfile/client ^0.10.21; CLI no longer declares a direct @relaycast/sdk dep). - CHANGELOG.md: main released the prior [Unreleased] block as 10.2.0; re-add only this PR's genuinely-new entries (observer mode; addListener workspace-key fix; broker self-mute) to the current [Unreleased]. Drop the stale relaycast 5.1.0 bullet (main documents the 6.0 upgrade in released sections). - package-lock.json: regenerated; already consistent with main's v6. Also register the new event-fanin and observer-source test files in the SDK test script so CI runs them. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_017qAFBAdEa5jmViDPSHqZYK
Observer source (packages/sdk/src/messaging/observer-source.ts): - Replace the ReDoS-prone `/\/+$/` trailing-slash regexes with a linear `stripTrailingSlashes` walk (resolves the CodeQL polynomial-regex alert). - Make pending-frame flush a total order: sort seq-stamped frames out-of-band and splice them back into the seq-less skeleton, so a seq-less frame between out-of-order seq frames can no longer advance the cursor past an undelivered lower seq. - Bound each backfill page fetch with an AbortController timeout (`backfillTimeoutMs`, default 30s) so a hung endpoint degrades to live-only instead of buffering forever. - Clear `live`/`offLive`/`offOpen` and tear down a partially initialized live stream on connect failure so later connect() calls retry instead of no-oping. - Capture and unsubscribe the `on.open` handler on disconnect (no stale re-backfill handlers across reconnect cycles). - Never downgrade to a token-in-URL on Node (header auth always works there); only browser-like runtimes fall back, and only after the constructor throws or two pre-open closes — removing the persistent security regression from a single transient blip. Report abnormal WebSocket closes through onError. - Add an injectable `webSocketImpl` and an accurate missing-WebSocket error (Node exposes global WebSocket only from v22). - Warn instead of silently stopping when a scoped backfill makes no progress. Listeners (packages/sdk/src/listeners.ts): - Tag the addListener connect-failure path with operation: 'connect' and word its log distinctly so a transport failure no longer reads as a user handler throwing. Tests: register the new files in the SDK test script (prior commit) and add coverage for total-order flush, the Node no-URL-token/abnormal-close guard, backfill-timeout degradation, connect-failure retry, and the no-progress warning. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_017qAFBAdEa5jmViDPSHqZYK
Rebased onto
|
Main advanced (v10.3.0, v10.4.0 cut; SDK types refactored to derive from @relaycast/types). Resolve: - CHANGELOG.md: releasing v10.3.0 had folded this PR's still-unreleased entries (observer mode, addListener workspace-key fix, broker self-mute) into the published [10.3.0] section. Move them back under [Unreleased], bumped to Minor (observer mode is a new backward-compatible API). Keep main's new Unreleased entries (types refactor, canonical `message.reacted` mapping). - normalize.ts: the auto-merge left two `case 'message.reacted'` labels — main's canonical handler plus this PR's observer-frame handler referencing a now-gone `readString` helper (a compile error). Drop the duplicate; main's grouped message.reacted/reaction.added/reaction.removed case (snake_case wire fields) already covers raw observer frames. - package-lock.json regenerated against v10.4.0 deps (zod ^4, @relaycast/types ^6). SDK typechecks and 151 tests pass; broker compiles against relaycast crate 6.0. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_017qAFBAdEa5jmViDPSHqZYK
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