From 543d5ddf8a121a60a75c00da3ac5106b3e26d124 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 2 Jul 2026 01:05:51 +0000 Subject: [PATCH 01/12] fix(sdk): stream relay.addListener events through registered agent clients MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Claude-Session: https://claude.ai/code/session_01VASvErz2MxkQMCsSzTgsbe --- CHANGELOG.md | 2 +- .../sdk/src/__tests__/event-fanin.test.ts | 261 +++++++++++++++ packages/sdk/src/__tests__/listeners.test.ts | 6 +- packages/sdk/src/agent-relay.ts | 16 +- packages/sdk/src/listeners.ts | 13 +- packages/sdk/src/messaging/event-fanin.ts | 313 ++++++++++++++++++ packages/sdk/src/messaging/index.ts | 5 + 7 files changed, 609 insertions(+), 7 deletions(-) create mode 100644 packages/sdk/src/__tests__/event-fanin.test.ts create mode 100644 packages/sdk/src/messaging/event-fanin.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 69fefddfb..fe5ede8c1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,7 +26,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### 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. - `agent-relay integration subscribe` now resolves provider-native `--resource` values through relayfile before binding, so Slack channel names, GitHub repos, Linear team keys, and Telegram chats bind to matching relayfile VFS globs while explicit `/`-prefixed globs still work. - `agent-relay integration subscribe` is now idempotent and supports multiple resources/channels per provider. Each inbound webhook is scoped to its `(provider, resource)` binding (not one-per-provider), so subscribing a second Slack channel — or two sources into the same relay channel — no longer collides on the unique `(workspace, webhook name)` index or clobbers the other binding's webhook. Re-subscribing creates the replacement webhook/subscription before retiring the old one, so a transient failure can't leave you with no working binding; a failed cleanup now warns instead of being silently swallowed. The relay channel id is normalized (`#general` → `general`) consistently across the webhook, subscription filter, relayfile bind, and writeback-secret lookup, and `listBindings` now maps relayfile's `pathGlob` field so unsubscribe/replace match correctly. - `agent-relay-broker` bootstrap `node.register` no longer advertises a generic `"spawn"` capability. Because the engine does not treat bare `"spawn"` as a placement capability (only `spawn:*`), it materialized a `spawn` action pinned to whichever node bootstrapped first, which then hijacked capability-based spawn placement for the whole workspace — every `spawn` invoke was dispatched to that node, ignoring `cli`/`target_node`/least-loaded routing. The pre-sidecar descriptor now carries no capabilities; real `spawn:*`/action capabilities arrive on the sidecar's `node.register`. diff --git a/packages/sdk/src/__tests__/event-fanin.test.ts b/packages/sdk/src/__tests__/event-fanin.test.ts new file mode 100644 index 000000000..2bba03dc5 --- /dev/null +++ b/packages/sdk/src/__tests__/event-fanin.test.ts @@ -0,0 +1,261 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +import { AgentRelay, ActionRegistry } from '../index.js'; +import { createEventFanIn } from '../messaging/event-fanin.js'; +import type { + RelayMessaging, + RelayMessagingEvent, + RelayMessagingEventsSurface, +} from '../messaging/index.js'; + +/** + * Fake events surface mirroring RelaycastMessagingClient's emit semantics: + * every event fans out to its own type key AND to 'any'. + */ +function createFakeEventsSurface() { + const handlers = new Map void>>(); + const on = vi.fn((type: string, handler: (event: RelayMessagingEvent) => void) => { + const set = handlers.get(type) ?? new Set(); + set.add(handler); + handlers.set(type, set); + return () => set.delete(handler); + }); + const emit = (event: RelayMessagingEvent) => { + for (const key of [event.type, 'any']) { + for (const handler of handlers.get(key) ?? []) handler(event); + } + }; + const surface = { + connect: vi.fn(), + disconnect: vi.fn(async () => {}), + subscribe: vi.fn(), + unsubscribe: vi.fn(), + on, + } as unknown as RelayMessagingEventsSurface & { + connect: ReturnType; + disconnect: ReturnType; + subscribe: ReturnType; + }; + return { surface, emit }; +} + +function messageCreated(messageId: string, text = 'hi', channel = 'general'): RelayMessagingEvent { + return { + type: 'messageCreated', + channel, + message: { messageId, text } as never, + } as RelayMessagingEvent; +} + +describe('createEventFanIn', () => { + it('forwards events from a source added before connect', () => { + const fanIn = createEventFanIn(undefined); + const { surface, emit } = createFakeEventsSurface(); + fanIn.addSource(surface); + + const received: RelayMessagingEvent[] = []; + fanIn.on('any', (event) => { + received.push(event); + }); + + fanIn.connect(); + expect(surface.connect).toHaveBeenCalledTimes(1); + + emit(messageCreated('m1')); + expect(received).toHaveLength(1); + expect(received[0].type).toBe('messageCreated'); + }); + + it('connects sources added after connect() was requested', () => { + const fanIn = createEventFanIn(undefined, { noSourceWarningMs: 0 }); + const received: RelayMessagingEvent[] = []; + fanIn.on('messageCreated', (event) => { + received.push(event); + }); + fanIn.connect(); + + const { surface, emit } = createFakeEventsSurface(); + fanIn.addSource(surface); + expect(surface.connect).toHaveBeenCalledTimes(1); + + emit(messageCreated('m1')); + expect(received).toHaveLength(1); + }); + + it('dedupes the same message arriving from two sources, keeps distinct ones', () => { + const fanIn = createEventFanIn(undefined); + const a = createFakeEventsSurface(); + const b = createFakeEventsSurface(); + fanIn.addSource(a.surface); + fanIn.addSource(b.surface); + fanIn.connect(); + + const received: RelayMessagingEvent[] = []; + fanIn.on('any', (event) => { + received.push(event); + }); + + a.emit(messageCreated('m1')); + b.emit(messageCreated('m1')); + expect(received).toHaveLength(1); + + a.emit(messageCreated('m2')); + expect(received).toHaveLength(2); + }); + + it('never dedupes per-source transport events', () => { + const fanIn = createEventFanIn(undefined); + const a = createFakeEventsSurface(); + const b = createFakeEventsSurface(); + fanIn.addSource(a.surface); + fanIn.addSource(b.surface); + fanIn.connect(); + + const received: RelayMessagingEvent[] = []; + fanIn.on('any', (event) => { + received.push(event); + }); + + a.emit({ type: 'error' } as RelayMessagingEvent); + b.emit({ type: 'error' } as RelayMessagingEvent); + expect(received).toHaveLength(2); + }); + + it('uses the workspace fallback only until the first agent source connects', () => { + const fallback = createFakeEventsSurface(); + const fanIn = createEventFanIn(fallback.surface, { noSourceWarningMs: 0 }); + + const received: RelayMessagingEvent[] = []; + fanIn.on('any', (event) => { + received.push(event); + }); + + fanIn.connect(); + expect(fallback.surface.connect).toHaveBeenCalledTimes(1); + fallback.emit(messageCreated('m1')); + expect(received).toHaveLength(1); + + const agent = createFakeEventsSurface(); + fanIn.addSource(agent.surface); + expect(agent.surface.connect).toHaveBeenCalledTimes(1); + expect(fallback.surface.disconnect).toHaveBeenCalledTimes(1); + + // Detached fallback no longer forwards. + fallback.emit(messageCreated('m2')); + expect(received).toHaveLength(1); + agent.emit(messageCreated('m3')); + expect(received).toHaveLength(2); + }); + + it('replays desired channel subscriptions onto late sources', () => { + const fanIn = createEventFanIn(undefined, { noSourceWarningMs: 0 }); + fanIn.subscribe(['general', 'ops']); + fanIn.connect(); + + const { surface } = createFakeEventsSurface(); + fanIn.addSource(surface); + expect(surface.subscribe).toHaveBeenCalledWith(['general', 'ops']); + }); + + describe('no-source warning', () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + afterEach(() => { + vi.useRealTimers(); + }); + + it('reports when connect() has no agent source within the window', () => { + const onError = vi.fn(); + const fanIn = createEventFanIn(undefined, { onError, noSourceWarningMs: 1000 }); + fanIn.connect(); + + vi.advanceTimersByTime(1001); + expect(onError).toHaveBeenCalledTimes(1); + expect(String(onError.mock.calls[0][0])).toContain('no registered agent'); + }); + + it('stays silent when an agent source arrives in time', () => { + const onError = vi.fn(); + const fanIn = createEventFanIn(undefined, { onError, noSourceWarningMs: 1000 }); + fanIn.connect(); + fanIn.addSource(createFakeEventsSurface().surface); + + vi.advanceTimersByTime(1001); + expect(onError).not.toHaveBeenCalled(); + }); + }); +}); + +describe('AgentRelay listener fan-in', () => { + function createFakeMessaging(events: ReturnType['surface']) { + return { + messages: {}, + events, + workspace: {}, + agents: { + register: vi.fn(async (input: { name: string }) => ({ + id: `id-${input.name}`, + name: input.name, + token: `tok-${input.name}`, + })), + }, + } as unknown as RelayMessaging; + } + + function createHarness() { + const workspaceEvents = createFakeEventsSurface(); + const agentEvents = new Map>(); + const messaging = createFakeMessaging(workspaceEvents.surface); + const relay = new AgentRelay({ + messaging, + actions: new ActionRegistry(), + createAgentMessaging: (token) => { + const bus = createFakeEventsSurface(); + agentEvents.set(token, bus); + return createFakeMessaging(bus.surface); + }, + }); + return { relay, workspaceEvents, agentEvents }; + } + + it('relay.addListener receives channel messages via an agent registered first', async () => { + const { relay, agentEvents } = createHarness(); + await relay.workspace.register({ name: 'listener' }); + + const handler = vi.fn(); + relay.addListener('message.created', handler); + + const bus = agentEvents.get('tok-listener')!; + expect(bus.surface.connect).toHaveBeenCalled(); + bus.emit(messageCreated('m1', 'hello', 'general')); + + expect(handler).toHaveBeenCalledTimes(1); + expect(handler.mock.calls[0][0].envelope.channel).toEqual({ name: 'general' }); + }); + + it('relay.addListener attached before register streams once the agent arrives', async () => { + const { relay, agentEvents } = createHarness(); + const handler = vi.fn(); + relay.addListener('message.created', handler); + + await relay.workspace.register({ name: 'late' }); + const bus = agentEvents.get('tok-late')!; + expect(bus.surface.connect).toHaveBeenCalled(); + + bus.emit(messageCreated('m1')); + expect(handler).toHaveBeenCalledTimes(1); + }); + + it('a message fanned to two registered agents surfaces once', async () => { + const { relay, agentEvents } = createHarness(); + await relay.workspace.register(['one', 'two']); + + const handler = vi.fn(); + relay.addListener('message.created', handler); + + agentEvents.get('tok-one')!.emit(messageCreated('m1')); + agentEvents.get('tok-two')!.emit(messageCreated('m1')); + expect(handler).toHaveBeenCalledTimes(1); + }); +}); diff --git a/packages/sdk/src/__tests__/listeners.test.ts b/packages/sdk/src/__tests__/listeners.test.ts index 93553fbb3..0fd62118d 100644 --- a/packages/sdk/src/__tests__/listeners.test.ts +++ b/packages/sdk/src/__tests__/listeners.test.ts @@ -12,7 +12,11 @@ function createEventBus() { return () => set.delete(handler); }; const emit = (type: string, event: unknown) => { - for (const handler of handlers.get(type) ?? []) handler(event); + // Mirror RelaycastMessagingClient's emit contract: every event fans out + // to its own type key AND to 'any' (the events fan-in listens on 'any'). + for (const key of [type, 'any']) { + for (const handler of handlers.get(key) ?? []) handler(event); + } }; return { on, emit }; } diff --git a/packages/sdk/src/agent-relay.ts b/packages/sdk/src/agent-relay.ts index c8ac50a35..c16ebb273 100644 --- a/packages/sdk/src/agent-relay.ts +++ b/packages/sdk/src/agent-relay.ts @@ -7,8 +7,10 @@ import { type RelaycastTelemetryOptions, } from './relaycast-telemetry.js'; import { + createEventFanIn, RelaycastMessagingClient, type RelayAgentRegistration, + type RelayEventFanIn, type RelayMessaging, type RelaycastMessagingOptions, } from './messaging/index.js'; @@ -104,6 +106,7 @@ export class AgentRelay implements AgentRelayAgent { private readonly messagingOptions: RelaycastMessagingOptions; private readonly clientsByToken = new Map(); private readonly createAgentMessaging: (token: string) => RelayMessaging; + private readonly eventFanIn: RelayEventFanIn; private enrichedMessages?: EnrichedMessages; private workspaceFacade?: RelayWorkspace; private hub?: ListenerHub; @@ -119,6 +122,14 @@ export class AgentRelay implements AgentRelayAgent { this.createAgentMessaging = createAgentMessaging ?? ((token) => new RelaycastMessagingClient({ ...this.messagingOptions, agentToken: token })); + // Relaycast v5 streams events over each agent's node transport; the + // workspace-key stream cannot receive them. The fan-in makes + // `relay.addListener(...)` stream through every registered agent client + // (added in `messagingForToken`), keeping the workspace client only as a + // pre-registration fallback. + this.eventFanIn = createEventFanIn(this.messaging.events, { + onError: (error) => this.reportError(error, { source: 'listener', selector: 'events.connect' }), + }); if (onError) { this.errorHooks.add(onError); } @@ -179,7 +190,7 @@ export class AgentRelay implements AgentRelayAgent { private get listenerHub(): ListenerHub { if (!this.hub) { - this.hub = createListenerHub(this.messaging.events, this.actions, { + this.hub = createListenerHub(this.eventFanIn, this.actions, { onError: (error, context) => this.reportError(error, context), }); } @@ -325,6 +336,9 @@ export class AgentRelay implements AgentRelayAgent { if (!client) { client = this.createAgentMessaging(token); this.clientsByToken.set(token, client); + // Every registered agent's connection feeds the workspace-level + // listener hub; the fan-in connects it lazily once a listener exists. + this.eventFanIn.addSource(client.events); } return client; } diff --git a/packages/sdk/src/listeners.ts b/packages/sdk/src/listeners.ts index 6314a1eab..707a18671 100644 --- a/packages/sdk/src/listeners.ts +++ b/packages/sdk/src/listeners.ts @@ -648,12 +648,17 @@ export function createListenerHub( ): (() => void) => { // Open the event stream — `events.on(...)` only registers handlers; the // socket is opened by `events.connect()` (idempotent). Agent-scoped clients - // stream through their own connection; workspace-key clients stream all - // workspace-visible events through the workspace stream. + // stream through their own connection; the workspace-level hub streams + // through registered agent clients via the events fan-in. A connect + // failure must be surfaced, not swallowed: a listener attached to a + // stream that never opens receives nothing, silently. try { context.events.connect(); - } catch { - // No stream available (no agent token and no workspace stream). + } catch (error) { + makeReporter(context, { + source: 'listener', + selector: typeof selector === 'string' ? selector : 'predicate', + })(error); } if (typeof selector !== 'string') { return selector.subscribe(context, handler as ListenerHandler); diff --git a/packages/sdk/src/messaging/event-fanin.ts b/packages/sdk/src/messaging/event-fanin.ts new file mode 100644 index 000000000..9ddc3ca86 --- /dev/null +++ b/packages/sdk/src/messaging/event-fanin.ts @@ -0,0 +1,313 @@ +import type { + RelayMessagingEvent, + RelayMessagingEventMap, + RelayMessagingEventsSurface, +} from './types.js'; + +/** + * Options for {@link createEventFanIn}. + */ +export interface EventFanInOptions { + /** + * Window in which identical events arriving from different sources are + * collapsed into one emission. Cross-source duplicates of the same server + * event arrive within milliseconds of each other; legitimate repeats + * (re-reactions, presence flaps) are separated by far more than this. + */ + dedupeWindowMs?: number; + /** Maximum number of tracked dedupe keys before the oldest are evicted. */ + dedupeCapacity?: number; + /** + * How long after `connect()` to wait for a registered-agent source before + * reporting that the stream has nothing real to connect to. + */ + noSourceWarningMs?: number; + /** Receives source connect/subscribe failures and the no-source warning. */ + onError?: (error: unknown) => void; + /** Clock override for tests. */ + now?: () => number; +} + +/** + * A {@link RelayMessagingEventsSurface} that fans in events from every + * registered agent's messaging client. + * + * Relaycast v5 delivers events over each agent's node transport + * (`/v1/node/ws`); the legacy workspace stream (`/v1/ws`) rejects workspace + * keys, so a workspace-scoped client on its own can never receive channel + * messages. The fan-in makes `relay.addListener(...)` work by streaming + * through the registered agents instead: each agent client added via + * `addSource` is connected (once `connect()` has been requested) and its + * events are forwarded, deduplicated across sources so a message delivered to + * several locally-registered agents surfaces once. + */ +export interface RelayEventFanIn extends RelayMessagingEventsSurface { + /** Add a registered agent's events surface. Idempotent per surface. */ + addSource(source: RelayMessagingEventsSurface): void; + /** Whether any registered-agent sources have been added. */ + hasAgentSources(): boolean; +} + +const DEFAULT_DEDUPE_WINDOW_MS = 30_000; +const DEFAULT_DEDUPE_CAPACITY = 2048; +const DEFAULT_NO_SOURCE_WARNING_MS = 10_000; + +/** + * Derive a cross-source identity for an event, or `null` for events that must + * never be deduplicated (per-connection transport state, unknown frames). + */ +function dedupeKey(event: RelayMessagingEvent): string | null { + switch (event.type) { + case 'messageCreated': + case 'messageUpdated': + case 'threadReply': + return event.message?.messageId ? `${event.type}:${event.message.messageId}` : null; + case 'dmReceived': + case 'groupDmReceived': + return event.message?.messageId + ? `${event.type}:${event.conversationId}:${event.message.messageId}` + : null; + case 'messageRead': + return `${event.type}:${event.messageId}:${event.agentName}`; + case 'reactionAdded': + case 'reactionRemoved': + return `${event.type}:${event.messageId}:${event.emoji}:${event.agentName}`; + case 'agentOnline': + case 'agentOffline': + return `${event.type}:${event.agent?.name}`; + case 'agentSpawnRequested': + case 'agentReleaseRequested': + return `${event.type}:${event.agent?.name}`; + case 'channelCreated': + case 'channelUpdated': + case 'channelArchived': + return `${event.type}:${event.channel?.name}`; + case 'memberJoined': + case 'memberLeft': + case 'channelMuted': + case 'channelUnmuted': + return `${event.type}:${event.channel}:${event.agentName}`; + case 'actionInvoked': + return `${event.type}:${event.invocationId}`; + default: + // connected / disconnected / error / reconnecting / + // permanentlyDisconnected / unknown: per-source transport state. + return null; + } +} + +/** + * Create an events fan-in. + * + * @param fallback - The workspace-scoped client's events surface, used only + * while no agent sources exist (it may still work against servers that + * accept the workspace stream). It is detached as soon as the first agent + * source connects. + */ +export function createEventFanIn( + fallback: RelayMessagingEventsSurface | undefined, + options: EventFanInOptions = {} +): RelayEventFanIn { + const dedupeWindowMs = options.dedupeWindowMs ?? DEFAULT_DEDUPE_WINDOW_MS; + const dedupeCapacity = options.dedupeCapacity ?? DEFAULT_DEDUPE_CAPACITY; + const noSourceWarningMs = options.noSourceWarningMs ?? DEFAULT_NO_SOURCE_WARNING_MS; + const now = options.now ?? (() => Date.now()); + + const handlers = new Map void | Promise>>(); + const sources: RelayMessagingEventsSurface[] = []; + const seenSources = new Set(); + const desiredChannels = new Set(); + /** Dedupe keys → last-seen timestamp, insertion-ordered for eviction. */ + const seenEvents = new Map(); + + let connectRequested = false; + let fallbackForwarding: (() => void) | undefined; + let fallbackConnected = false; + let noSourceTimer: ReturnType | undefined; + + const report = (error: unknown): void => { + if (options.onError) { + try { + options.onError(error); + } catch { + // Error hooks must not throw into the event source. + } + return; + } + console.warn('[agent-relay] event stream:', error); + }; + + const emitLocal = (event: RelayMessagingEvent): void => { + for (const key of [event.type, 'any'] as const) { + for (const handler of handlers.get(key) ?? []) { + try { + void Promise.resolve(handler(event)).catch(report); + } catch (error) { + report(error); + } + } + } + }; + + const forward = (event: RelayMessagingEvent): void => { + const key = dedupeKey(event); + if (key) { + const at = now(); + const prev = seenEvents.get(key); + if (prev !== undefined && at - prev < dedupeWindowMs) return; + seenEvents.delete(key); + seenEvents.set(key, at); + while (seenEvents.size > dedupeCapacity) { + const oldest = seenEvents.keys().next().value; + if (oldest === undefined) break; + seenEvents.delete(oldest); + } + } + emitLocal(event); + }; + + const clearNoSourceTimer = (): void => { + if (noSourceTimer !== undefined) { + clearTimeout(noSourceTimer); + noSourceTimer = undefined; + } + }; + + const scheduleNoSourceWarning = (): void => { + if (noSourceTimer !== undefined || noSourceWarningMs <= 0) return; + noSourceTimer = setTimeout(() => { + noSourceTimer = undefined; + if (!connectRequested || sources.length > 0) return; + report( + new Error( + 'Listening for relay events, but no registered agent is connected. ' + + 'Relaycast delivers events over each agent\'s node transport, and the ' + + 'workspace-key stream cannot receive them. Register an agent first ' + + '(`relay.workspace.register(...)` / `workspace.reconnect(...)`) so the ' + + 'listener has a live connection to stream through.' + ) + ); + }, noSourceWarningMs); + (noSourceTimer as { unref?: () => void }).unref?.(); + }; + + const connectSource = (source: RelayMessagingEventsSurface): void => { + if (typeof source.connect === 'function') { + try { + source.connect(); + } catch (error) { + report(error); + } + } + if (desiredChannels.size > 0 && typeof source.subscribe === 'function') { + try { + source.subscribe([...desiredChannels]); + } catch (error) { + report(error); + } + } + }; + + const attachFallback = (): void => { + // Injected fakes may carry a partial surface; only a stream that can be + // observed is worth attaching. + if (!fallback || fallbackForwarding || typeof fallback.on !== 'function') return; + fallbackForwarding = fallback.on('any', forward); + 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 + // still arrive later, so surface the failure without giving up. + report(error); + } + } + }; + + const detachFallback = (): void => { + if (!fallbackForwarding) return; + fallbackForwarding(); + fallbackForwarding = undefined; + 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 + // `createAgentMessaging`) contributes nothing; skip it entirely. + if (!source || typeof source.on !== 'function') return; + 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: () => { + connectRequested = true; + if (sources.length > 0) { + for (const source of sources) connectSource(source); + return; + } + attachFallback(); + scheduleNoSourceWarning(); + }, + + disconnect: async () => { + connectRequested = false; + clearNoSourceTimer(); + detachFallback(); + 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) { + try { + 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: ( + event: K, + handler: (...args: RelayMessagingEventMap[K]) => void | Promise + ): (() => void) => { + const set = handlers.get(event) ?? new Set(); + set.add(handler as (event: RelayMessagingEvent) => void | Promise); + handlers.set(event, set); + return () => { + set.delete(handler as (event: RelayMessagingEvent) => void | Promise); + }; + }, + }; +} diff --git a/packages/sdk/src/messaging/index.ts b/packages/sdk/src/messaging/index.ts index 85a6d0b0e..8ff3842c6 100644 --- a/packages/sdk/src/messaging/index.ts +++ b/packages/sdk/src/messaging/index.ts @@ -1,5 +1,10 @@ export * from './types.js'; export * from './normalize.js'; +export { + createEventFanIn, + type EventFanInOptions, + type RelayEventFanIn, +} from './event-fanin.js'; export { RelayPlacementError, RelaycastMessagingClient, From 21281ebce12fd0aef552a2e47345d208304dd58d Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Thu, 2 Jul 2026 03:11:01 +0000 Subject: [PATCH 02/12] style: auto-format with Prettier --- packages/sdk/src/__tests__/event-fanin.test.ts | 6 +----- packages/sdk/src/messaging/event-fanin.ts | 8 ++------ packages/sdk/src/messaging/index.ts | 6 +----- 3 files changed, 4 insertions(+), 16 deletions(-) diff --git a/packages/sdk/src/__tests__/event-fanin.test.ts b/packages/sdk/src/__tests__/event-fanin.test.ts index 2bba03dc5..c9b65e5fe 100644 --- a/packages/sdk/src/__tests__/event-fanin.test.ts +++ b/packages/sdk/src/__tests__/event-fanin.test.ts @@ -2,11 +2,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { AgentRelay, ActionRegistry } from '../index.js'; import { createEventFanIn } from '../messaging/event-fanin.js'; -import type { - RelayMessaging, - RelayMessagingEvent, - RelayMessagingEventsSurface, -} from '../messaging/index.js'; +import type { RelayMessaging, RelayMessagingEvent, RelayMessagingEventsSurface } from '../messaging/index.js'; /** * Fake events surface mirroring RelaycastMessagingClient's emit semantics: diff --git a/packages/sdk/src/messaging/event-fanin.ts b/packages/sdk/src/messaging/event-fanin.ts index 9ddc3ca86..bb92a4998 100644 --- a/packages/sdk/src/messaging/event-fanin.ts +++ b/packages/sdk/src/messaging/event-fanin.ts @@ -1,8 +1,4 @@ -import type { - RelayMessagingEvent, - RelayMessagingEventMap, - RelayMessagingEventsSurface, -} from './types.js'; +import type { RelayMessagingEvent, RelayMessagingEventMap, RelayMessagingEventsSurface } from './types.js'; /** * Options for {@link createEventFanIn}. @@ -181,7 +177,7 @@ export function createEventFanIn( report( new Error( 'Listening for relay events, but no registered agent is connected. ' + - 'Relaycast delivers events over each agent\'s node transport, and the ' + + "Relaycast delivers events over each agent's node transport, and the " + 'workspace-key stream cannot receive them. Register an agent first ' + '(`relay.workspace.register(...)` / `workspace.reconnect(...)`) so the ' + 'listener has a live connection to stream through.' diff --git a/packages/sdk/src/messaging/index.ts b/packages/sdk/src/messaging/index.ts index 8ff3842c6..45e3c5dd0 100644 --- a/packages/sdk/src/messaging/index.ts +++ b/packages/sdk/src/messaging/index.ts @@ -1,10 +1,6 @@ export * from './types.js'; export * from './normalize.js'; -export { - createEventFanIn, - type EventFanInOptions, - type RelayEventFanIn, -} from './event-fanin.js'; +export { createEventFanIn, type EventFanInOptions, type RelayEventFanIn } from './event-fanin.js'; export { RelayPlacementError, RelaycastMessagingClient, From d71fc621d4593d454901f21c80bf9559659362db Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 2 Jul 2026 03:23:24 +0000 Subject: [PATCH 03/12] =?UTF-8?q?fix(sdk):=20address=20fan-in=20review=20?= =?UTF-8?q?=E2=80=94=20source-aware=20dedupe,=20disconnect=20cleanup?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Claude-Session: https://claude.ai/code/session_01VASvErz2MxkQMCsSzTgsbe --- CHANGELOG.md | 2 +- .../sdk/src/__tests__/event-fanin.test.ts | 56 +++++++++++- packages/sdk/src/__tests__/listeners.test.ts | 2 +- packages/sdk/src/messaging/event-fanin.ts | 85 +++++++++++++++---- 4 files changed, 124 insertions(+), 21 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index fe5ede8c1..c3854e2f4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,7 +26,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed -- `@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 receives channel messages, DMs, and thread replies by streaming through registered agent clients (`workspace.register`/`reconnect`) over the node transport, deduplicating events delivered to multiple locally-registered agents; previously listeners silently received nothing. Listener connect failures now surface through `onError` instead of being swallowed, and a listener with no registered agent warns after 10s. - `agent-relay integration subscribe` now resolves provider-native `--resource` values through relayfile before binding, so Slack channel names, GitHub repos, Linear team keys, and Telegram chats bind to matching relayfile VFS globs while explicit `/`-prefixed globs still work. - `agent-relay integration subscribe` is now idempotent and supports multiple resources/channels per provider. Each inbound webhook is scoped to its `(provider, resource)` binding (not one-per-provider), so subscribing a second Slack channel — or two sources into the same relay channel — no longer collides on the unique `(workspace, webhook name)` index or clobbers the other binding's webhook. Re-subscribing creates the replacement webhook/subscription before retiring the old one, so a transient failure can't leave you with no working binding; a failed cleanup now warns instead of being silently swallowed. The relay channel id is normalized (`#general` → `general`) consistently across the webhook, subscription filter, relayfile bind, and writeback-secret lookup, and `listBindings` now maps relayfile's `pathGlob` field so unsubscribe/replace match correctly. - `agent-relay-broker` bootstrap `node.register` no longer advertises a generic `"spawn"` capability. Because the engine does not treat bare `"spawn"` as a placement capability (only `spawn:*`), it materialized a `spawn` action pinned to whichever node bootstrapped first, which then hijacked capability-based spawn placement for the whole workspace — every `spawn` invoke was dispatched to that node, ignoring `cli`/`target_node`/least-loaded routing. The pre-sidecar descriptor now carries no capabilities; real `spawn:*`/action capabilities arrive on the sidecar's `node.register`. diff --git a/packages/sdk/src/__tests__/event-fanin.test.ts b/packages/sdk/src/__tests__/event-fanin.test.ts index c9b65e5fe..3736c554b 100644 --- a/packages/sdk/src/__tests__/event-fanin.test.ts +++ b/packages/sdk/src/__tests__/event-fanin.test.ts @@ -17,7 +17,7 @@ function createFakeEventsSurface() { return () => set.delete(handler); }); const emit = (event: RelayMessagingEvent) => { - for (const key of [event.type, 'any']) { + for (const key of new Set([event.type, 'any'])) { for (const handler of handlers.get(key) ?? []) handler(event); } }; @@ -99,6 +99,60 @@ describe('createEventFanIn', () => { expect(received).toHaveLength(2); }); + it('passes genuine repeats from the same source while collapsing cross-source copies', () => { + const fanIn = createEventFanIn(undefined); + const a = createFakeEventsSurface(); + const b = createFakeEventsSurface(); + fanIn.addSource(a.surface); + fanIn.addSource(b.surface); + fanIn.connect(); + + const received: RelayMessagingEvent[] = []; + fanIn.on('any', (event) => { + received.push(event); + }); + + const updated = (): RelayMessagingEvent => + ({ type: 'messageUpdated', channel: 'general', message: { messageId: 'm1', text: 'edit' } } as never); + + // First edit: A delivers, B's copy collapses. + a.emit(updated()); + b.emit(updated()); + expect(received).toHaveLength(1); + + // Second edit within the window: A already delivered the previous + // occurrence, so this is a new occurrence — it must NOT be dropped. + a.emit(updated()); + expect(received).toHaveLength(2); + // ...and B's copy of the second edit collapses again. + b.emit(updated()); + expect(received).toHaveLength(2); + }); + + it('disconnect() stops forwarding; connect() resumes it', async () => { + const fanIn = createEventFanIn(undefined); + const { surface, emit } = createFakeEventsSurface(); + fanIn.addSource(surface); + fanIn.connect(); + + const received: RelayMessagingEvent[] = []; + fanIn.on('any', (event) => { + received.push(event); + }); + + emit(messageCreated('m1')); + expect(received).toHaveLength(1); + + await fanIn.disconnect(); + expect(surface.disconnect).toHaveBeenCalledTimes(1); + emit(messageCreated('m2')); + expect(received).toHaveLength(1); + + fanIn.connect(); + emit(messageCreated('m3')); + expect(received).toHaveLength(2); + }); + it('never dedupes per-source transport events', () => { const fanIn = createEventFanIn(undefined); const a = createFakeEventsSurface(); diff --git a/packages/sdk/src/__tests__/listeners.test.ts b/packages/sdk/src/__tests__/listeners.test.ts index 0fd62118d..136430d19 100644 --- a/packages/sdk/src/__tests__/listeners.test.ts +++ b/packages/sdk/src/__tests__/listeners.test.ts @@ -14,7 +14,7 @@ function createEventBus() { const emit = (type: string, event: unknown) => { // Mirror RelaycastMessagingClient's emit contract: every event fans out // to its own type key AND to 'any' (the events fan-in listens on 'any'). - for (const key of [type, 'any']) { + for (const key of new Set([type, 'any'])) { for (const handler of handlers.get(key) ?? []) handler(event); } }; diff --git a/packages/sdk/src/messaging/event-fanin.ts b/packages/sdk/src/messaging/event-fanin.ts index bb92a4998..7fe3ef067 100644 --- a/packages/sdk/src/messaging/event-fanin.ts +++ b/packages/sdk/src/messaging/event-fanin.ts @@ -5,10 +5,11 @@ import type { RelayMessagingEvent, RelayMessagingEventMap, RelayMessagingEventsS */ export interface EventFanInOptions { /** - * Window in which identical events arriving from different sources are - * collapsed into one emission. Cross-source duplicates of the same server - * event arrive within milliseconds of each other; legitimate repeats - * (re-reactions, presence flaps) are separated by far more than this. + * Window in which the same occurrence arriving from *different* sources is + * collapsed into one emission. Cross-source duplicates of one server event + * arrive within milliseconds of each other; a repeat from a source that + * already delivered the previous occurrence is a new occurrence and always + * passes through. */ dedupeWindowMs?: number; /** Maximum number of tracked dedupe keys before the oldest are evicted. */ @@ -92,6 +93,12 @@ function dedupeKey(event: RelayMessagingEvent): string | null { } } +/** One observed occurrence: when it was first seen and which sources delivered it. */ +interface OccurrenceRecord { + at: number; + sources: Set; +} + /** * Create an events fan-in. * @@ -112,9 +119,11 @@ export function createEventFanIn( const handlers = new Map void | Promise>>(); const sources: RelayMessagingEventsSurface[] = []; const seenSources = new Set(); + /** Active `on('any', ...)` forwarding per source, so disconnect can detach. */ + const sourceForwarding = new Map void>(); const desiredChannels = new Set(); - /** Dedupe keys → last-seen timestamp, insertion-ordered for eviction. */ - const seenEvents = new Map(); + /** Dedupe keys → current occurrence, insertion-ordered for eviction. */ + const seenEvents = new Map(); let connectRequested = false; let fallbackForwarding: (() => void) | undefined; @@ -145,14 +154,26 @@ export function createEventFanIn( } }; - const forward = (event: RelayMessagingEvent): void => { + /** + * Forward an event from one source, collapsing cross-source duplicates. + * + * The engine emits one delivery per locally-registered recipient, so N + * sources produce N copies of one occurrence — only the first is emitted. + * A copy from a source that already delivered the previous occurrence is a + * genuine repeat (message edit, re-reaction, presence flap) and starts a + * new occurrence, so same-source repeats are never dropped. + */ + const forward = (source: RelayMessagingEventsSurface, event: RelayMessagingEvent): void => { const key = dedupeKey(event); if (key) { const at = now(); - const prev = seenEvents.get(key); - if (prev !== undefined && at - prev < dedupeWindowMs) return; + const record = seenEvents.get(key); + if (record && at - record.at < dedupeWindowMs && !record.sources.has(source)) { + record.sources.add(source); + return; + } seenEvents.delete(key); - seenEvents.set(key, at); + seenEvents.set(key, { at, sources: new Set([source]) }); while (seenEvents.size > dedupeCapacity) { const oldest = seenEvents.keys().next().value; if (oldest === undefined) break; @@ -187,6 +208,16 @@ export function createEventFanIn( (noSourceTimer as { unref?: () => void }).unref?.(); }; + const attachSourceForwarding = (source: RelayMessagingEventsSurface): void => { + if (sourceForwarding.has(source)) return; + sourceForwarding.set(source, source.on('any', (event) => forward(source, event))); + }; + + const detachAllForwarding = (): void => { + for (const off of sourceForwarding.values()) off(); + sourceForwarding.clear(); + }; + const connectSource = (source: RelayMessagingEventsSurface): void => { if (typeof source.connect === 'function') { try { @@ -208,7 +239,7 @@ export function createEventFanIn( // Injected fakes may carry a partial surface; only a stream that can be // observed is worth attaching. if (!fallback || fallbackForwarding || typeof fallback.on !== 'function') return; - fallbackForwarding = fallback.on('any', forward); + fallbackForwarding = fallback.on('any', (event) => forward(fallback, event)); if (typeof fallback.connect === 'function') { try { fallback.connect(); @@ -227,10 +258,12 @@ export function createEventFanIn( fallbackForwarding = undefined; if (fallbackConnected) { fallbackConnected = false; - try { - void fallback?.disconnect().catch(() => {}); - } catch { - // Fallback surfaces without a disconnect are simply left as-is. + if (typeof fallback?.disconnect === 'function') { + try { + void fallback.disconnect().catch(() => {}); + } catch { + // Fallback surfaces whose disconnect misbehaves are left as-is. + } } } }; @@ -243,7 +276,7 @@ export function createEventFanIn( if (seenSources.has(source)) return; seenSources.add(source); sources.push(source); - source.on('any', forward); + attachSourceForwarding(source); if (connectRequested) { connectSource(source); clearNoSourceTimer(); @@ -258,7 +291,11 @@ export function createEventFanIn( connect: () => { connectRequested = true; if (sources.length > 0) { - for (const source of sources) connectSource(source); + for (const source of sources) { + // Re-attach forwarding dropped by a prior disconnect(). + attachSourceForwarding(source); + connectSource(source); + } return; } attachFallback(); @@ -269,12 +306,23 @@ export function createEventFanIn( connectRequested = false; clearNoSourceTimer(); detachFallback(); - await Promise.allSettled(sources.map((source) => Promise.resolve().then(() => source.disconnect()))); + // Stop forwarding first so a source another owner reconnects later + // cannot feed listeners that were explicitly disconnected. + detachAllForwarding(); + await Promise.allSettled( + sources.map((source) => + Promise.resolve().then(() => { + if (typeof source.disconnect === 'function') return source.disconnect(); + return undefined; + }) + ) + ); }, subscribe: (channels) => { for (const channel of channels) desiredChannels.add(channel); for (const source of sources) { + if (typeof source.subscribe !== 'function') continue; try { source.subscribe(channels); } catch (error) { @@ -286,6 +334,7 @@ export function createEventFanIn( unsubscribe: (channels) => { for (const channel of channels) desiredChannels.delete(channel); for (const source of sources) { + if (typeof source.unsubscribe !== 'function') continue; try { source.unsubscribe(channels); } catch (error) { From f1a5c92b22698481cb083dea4d5f475e9756429b Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 2 Jul 2026 03:24:18 +0000 Subject: [PATCH 04/12] style: format fan-in files with Prettier Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01VASvErz2MxkQMCsSzTgsbe --- packages/sdk/src/__tests__/event-fanin.test.ts | 2 +- packages/sdk/src/messaging/event-fanin.ts | 5 ++++- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/packages/sdk/src/__tests__/event-fanin.test.ts b/packages/sdk/src/__tests__/event-fanin.test.ts index 3736c554b..367bea2a7 100644 --- a/packages/sdk/src/__tests__/event-fanin.test.ts +++ b/packages/sdk/src/__tests__/event-fanin.test.ts @@ -113,7 +113,7 @@ describe('createEventFanIn', () => { }); const updated = (): RelayMessagingEvent => - ({ type: 'messageUpdated', channel: 'general', message: { messageId: 'm1', text: 'edit' } } as never); + ({ type: 'messageUpdated', channel: 'general', message: { messageId: 'm1', text: 'edit' } }) as never; // First edit: A delivers, B's copy collapses. a.emit(updated()); diff --git a/packages/sdk/src/messaging/event-fanin.ts b/packages/sdk/src/messaging/event-fanin.ts index 7fe3ef067..3ea87050d 100644 --- a/packages/sdk/src/messaging/event-fanin.ts +++ b/packages/sdk/src/messaging/event-fanin.ts @@ -210,7 +210,10 @@ export function createEventFanIn( const attachSourceForwarding = (source: RelayMessagingEventsSurface): void => { if (sourceForwarding.has(source)) return; - sourceForwarding.set(source, source.on('any', (event) => forward(source, event))); + sourceForwarding.set( + source, + source.on('any', (event) => forward(source, event)) + ); }; const detachAllForwarding = (): void => { From 7327f215ab0d18c7acb97289513ffcf39790bb63 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 2 Jul 2026 03:57:38 +0000 Subject: [PATCH 05/12] =?UTF-8?q?feat(sdk):=20observer=20mode=20=E2=80=94?= =?UTF-8?q?=20durable=20event-log=20backfill=20+=20live=20observer=20strea?= =?UTF-8?q?m?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Claude-Session: https://claude.ai/code/session_01VASvErz2MxkQMCsSzTgsbe --- .../sdk/src/__tests__/observer-source.test.ts | 375 ++++++++++++++++++ packages/sdk/src/agent-relay.ts | 96 ++++- packages/sdk/src/messaging/index.ts | 5 + packages/sdk/src/messaging/observer-source.ts | 294 ++++++++++++++ 4 files changed, 759 insertions(+), 11 deletions(-) create mode 100644 packages/sdk/src/__tests__/observer-source.test.ts create mode 100644 packages/sdk/src/messaging/observer-source.ts diff --git a/packages/sdk/src/__tests__/observer-source.test.ts b/packages/sdk/src/__tests__/observer-source.test.ts new file mode 100644 index 000000000..8fd0a5af5 --- /dev/null +++ b/packages/sdk/src/__tests__/observer-source.test.ts @@ -0,0 +1,375 @@ +import { afterEach, describe, expect, it, vi } from 'vitest'; + +const relaycastMocks = vi.hoisted(() => { + const relayCast = vi.fn(); + return { relayCast }; +}); + +vi.mock('@relaycast/sdk', async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, RelayCast: relaycastMocks.relayCast }; +}); + +import { AgentRelay } from '../index.js'; +import { createObserverEventSource, type ObserverLiveStream } from '../messaging/observer-source.js'; +import type { RelayMessaging, RelayMessagingEvent } from '../messaging/index.js'; + +function createFakeLiveStream() { + const handlers = new Set<(event: unknown) => void>(); + const stream: ObserverLiveStream & { + connect: ReturnType; + disconnect: ReturnType; + } = { + connect: vi.fn(), + disconnect: vi.fn(), + on: { + any: (handler: (event: unknown) => void) => { + handlers.add(handler); + return () => handlers.delete(handler); + }, + }, + }; + return { + stream, + emit: (event: unknown) => { + for (const handler of [...handlers]) handler(event); + }, + }; +} + +/** Raw server frame as the live observer WS delivers it. */ +function liveFrame(messageId: string, seq?: number): Record { + return { + type: 'message.created', + channel: 'general', + message: { id: messageId, text: `text-${messageId}` }, + ...(seq !== undefined ? { seq } : {}), + }; +} + +/** Durable event-log row as GET /v1/workspace/events returns it. */ +function logRow(seq: number, messageId: string): Record { + return { + seq, + type: 'message.created', + channel_id: 'c1', + payload: liveFrame(messageId), + created_at: '2026-07-02T00:00:00Z', + }; +} + +function jsonResponse(events: Record[], latestSeq: number) { + return { + ok: true, + status: 200, + json: async () => ({ ok: true, data: { events, latest_seq: latestSeq } }), + } as Response; +} + +function notFoundResponse() { + return { ok: false, status: 404, json: async () => ({ ok: false }) } as Response; +} + +/** Serve pages of the given log rows keyed off the `since` query parameter. */ +function createBackfillFetch(rows: Record[], latestSeq?: number) { + const latest = latestSeq ?? (rows.length > 0 ? (rows[rows.length - 1].seq as number) : 0); + return vi.fn(async (input: RequestInfo | URL) => { + const url = new URL(String(input)); + const since = Number(url.searchParams.get('since') ?? '0'); + const limit = Number(url.searchParams.get('limit') ?? '500'); + const page = rows.filter((row) => (row.seq as number) > since).slice(0, limit); + return jsonResponse(page, latest); + }) as unknown as typeof fetch; +} + +async function settle(): Promise { + // Let the async backfill loop (fetch + json awaits) run to completion. + for (let i = 0; i < 10; i += 1) await Promise.resolve(); + await new Promise((resolve) => setTimeout(resolve, 0)); +} + +function collect(source: ReturnType) { + const received: RelayMessagingEvent[] = []; + source.on('any', (event) => { + received.push(event); + }); + return received; +} + +function messageIds(events: RelayMessagingEvent[]): string[] { + return events + .filter((event) => event.type === 'messageCreated') + .map((event) => (event as Extract).message.messageId); +} + +afterEach(() => { + relaycastMocks.relayCast.mockReset(); + vi.unstubAllGlobals(); +}); + +describe('createObserverEventSource', () => { + it('backfills from the log, then merges buffered live frames deduped and ordered by seq', async () => { + const live = createFakeLiveStream(); + let releaseBackfill!: () => void; + const gate = new Promise((resolve) => { + releaseBackfill = resolve; + }); + const rows = [logRow(1, 'm1'), logRow(2, 'm2'), logRow(3, 'm3')]; + const fetchImpl = vi.fn(async (input: RequestInfo | URL) => { + await gate; + const url = new URL(String(input)); + const since = Number(url.searchParams.get('since') ?? '0'); + return jsonResponse( + rows.filter((row) => (row.seq as number) > since), + 3 + ); + }) as unknown as typeof fetch; + + const cursors: number[] = []; + const source = createObserverEventSource({ + observerToken: 'ot_live_test', + baseUrl: 'https://api.example.test', + createLiveStream: () => live.stream, + fetch: fetchImpl, + onCursor: (seq) => cursors.push(seq), + }); + const received = collect(source); + + source.connect(); + expect(live.stream.connect).toHaveBeenCalledTimes(1); + + // Live frames arrive while the backfill is in flight: seq 4/3 buffered + // out of order, seq 3 is also covered by the backfill. + live.emit(liveFrame('m4', 4)); + live.emit(liveFrame('m3', 3)); + expect(received).toHaveLength(0); + + releaseBackfill(); + await settle(); + + expect(messageIds(received)).toEqual(['m1', 'm2', 'm3', 'm4']); + expect(cursors).toEqual([1, 2, 3, 4]); + }); + + it('paginates the backfill until latest_seq', async () => { + const live = createFakeLiveStream(); + const rows = [logRow(1, 'm1'), logRow(2, 'm2'), logRow(3, 'm3'), logRow(4, 'm4')]; + const fetchImpl = createBackfillFetch(rows); + + const source = createObserverEventSource({ + observerToken: 'ot_live_test', + baseUrl: 'https://api.example.test', + createLiveStream: () => live.stream, + fetch: fetchImpl, + backfillPageSize: 2, + }); + const received = collect(source); + + source.connect(); + await settle(); + + const calls = (fetchImpl as unknown as ReturnType).mock.calls.map((call) => + String(call[0]) + ); + expect(calls).toEqual([ + 'https://api.example.test/v1/workspace/events?since=0&limit=2', + 'https://api.example.test/v1/workspace/events?since=2&limit=2', + ]); + expect(messageIds(received)).toEqual(['m1', 'm2', 'm3', 'm4']); + }); + + it('resumes from sinceSeq and skips already-seen events', async () => { + const live = createFakeLiveStream(); + const rows = [logRow(3, 'm3'), logRow(4, 'm4')]; + const fetchImpl = createBackfillFetch(rows, 4); + + const source = createObserverEventSource({ + observerToken: 'ot_live_test', + baseUrl: 'https://api.example.test', + sinceSeq: 2, + createLiveStream: () => live.stream, + fetch: fetchImpl, + }); + const received = collect(source); + + source.connect(); + await settle(); + + const firstUrl = String((fetchImpl as unknown as ReturnType).mock.calls[0][0]); + expect(firstUrl).toContain('since=2'); + expect(messageIds(received)).toEqual(['m3', 'm4']); + }); + + it('dedupes live frames at or below the cursor and passes seq-less frames through', async () => { + const live = createFakeLiveStream(); + const fetchImpl = createBackfillFetch([logRow(1, 'm1'), logRow(2, 'm2')]); + + const source = createObserverEventSource({ + observerToken: 'ot_live_test', + baseUrl: 'https://api.example.test', + createLiveStream: () => live.stream, + fetch: fetchImpl, + }); + const received = collect(source); + + source.connect(); + await settle(); + + live.emit(liveFrame('m2', 2)); // duplicate of a backfilled event + live.emit(liveFrame('m3', 3)); + live.emit(liveFrame('m3', 3)); // duplicate live redelivery + live.emit(liveFrame('m-live-only')); // log append failed: no seq, live-only + live.emit({ type: 'open' }); // transport frames have no seq + + expect(messageIds(received)).toEqual(['m1', 'm2', 'm3', 'm-live-only']); + expect(received.some((event) => event.type === 'connected')).toBe(true); + }); + + it('sends the observer token as a bearer on backfill requests', async () => { + const live = createFakeLiveStream(); + const fetchImpl = createBackfillFetch([]); + + createObserverEventSource({ + observerToken: 'ot_live_secret', + baseUrl: 'https://api.example.test', + createLiveStream: () => live.stream, + fetch: fetchImpl, + }).connect(); + await settle(); + + const init = (fetchImpl as unknown as ReturnType).mock.calls[0][1] as RequestInit; + expect(init.headers).toEqual({ Authorization: 'Bearer ot_live_secret' }); + }); + + it('degrades to live-only when the backfill endpoint 404s', async () => { + const live = createFakeLiveStream(); + const fetchImpl = vi.fn(async () => notFoundResponse()) as unknown as typeof fetch; + const onError = vi.fn(); + const cursors: number[] = []; + + const source = createObserverEventSource({ + observerToken: 'ot_live_test', + baseUrl: 'https://api.example.test', + createLiveStream: () => live.stream, + fetch: fetchImpl, + onError, + onCursor: (seq) => cursors.push(seq), + }); + const received = collect(source); + + source.connect(); + live.emit(liveFrame('m1', 1)); // buffered until the 404 resolves + await settle(); + live.emit(liveFrame('m2', 2)); + + expect(messageIds(received)).toEqual(['m1', 'm2']); + // A missing endpoint is expected on older engines, not an error. + expect(onError).not.toHaveBeenCalled(); + // The cursor still tracks live seq so callers can persist it. + expect(cursors).toEqual([1, 2]); + }); + + it('reports backfill failures and still delivers the live stream', async () => { + const live = createFakeLiveStream(); + const fetchImpl = vi.fn(async () => { + throw new Error('network down'); + }) as unknown as typeof fetch; + const onError = vi.fn(); + + const source = createObserverEventSource({ + observerToken: 'ot_live_test', + baseUrl: 'https://api.example.test', + createLiveStream: () => live.stream, + fetch: fetchImpl, + onError, + }); + const received = collect(source); + + source.connect(); + await settle(); + live.emit(liveFrame('m1', 1)); + + expect(onError).toHaveBeenCalledTimes(1); + expect(messageIds(received)).toEqual(['m1']); + }); + + it('disconnect stops the live stream; reconnect backfills from the cursor', async () => { + const live = createFakeLiveStream(); + const fetchImpl = createBackfillFetch([logRow(1, 'm1'), logRow(2, 'm2')]); + + const source = createObserverEventSource({ + observerToken: 'ot_live_test', + baseUrl: 'https://api.example.test', + createLiveStream: () => live.stream, + fetch: fetchImpl, + }); + const received = collect(source); + + source.connect(); + await settle(); + await source.disconnect(); + expect(live.stream.disconnect).toHaveBeenCalledTimes(1); + + live.emit(liveFrame('m-late', 3)); // detached: must not emit + expect(messageIds(received)).toEqual(['m1', 'm2']); + + source.connect(); + await settle(); + const calls = (fetchImpl as unknown as ReturnType).mock.calls.map((call) => + String(call[0]) + ); + expect(calls[calls.length - 1]).toContain('since=2'); + }); +}); + +describe('AgentRelay observer mode', () => { + function createObserverRelay(overrides: Record = {}) { + // A partial messaging fake: observer mode never uses the workspace + // client's event stream and register/reconnect throw at the facade. + const messaging = { + workspace: { info: vi.fn(async () => ({})), fleetNodes: {} }, + agents: {}, + events: undefined, + } as unknown as RelayMessaging; + return new AgentRelay({ observerToken: 'ot_live_test', messaging, ...overrides }); + } + + it('workspace.register() throws a read-only error', async () => { + const relay = createObserverRelay(); + await expect(async () => relay.workspace.register('Reviewer')).rejects.toThrow( + /observer tokens are read-only; use a workspace key to register agents/ + ); + }); + + it('workspace.reconnect() throws a read-only error', async () => { + const relay = createObserverRelay(); + await expect(relay.workspace.reconnect({ apiToken: 'rat_test' })).rejects.toThrow( + /observer tokens are read-only; use a workspace key to register agents/ + ); + }); + + it('streams observer events through relay.addListener', async () => { + const live = createFakeLiveStream(); + relaycastMocks.relayCast.mockImplementation(function () { + return live.stream; + }); + vi.stubGlobal('fetch', createBackfillFetch([logRow(1, 'm1')])); + + const relay = new AgentRelay({ + observerToken: 'ot_live_test', + baseUrl: 'https://api.example.test', + }); + + const received: unknown[] = []; + relay.addListener('message.created', (event) => { + received.push(event); + }); + await settle(); + live.emit(liveFrame('m2', 2)); + + expect(relaycastMocks.relayCast).toHaveBeenCalledWith( + expect.objectContaining({ apiKey: 'ot_live_test', baseUrl: 'https://api.example.test' }) + ); + expect(received).toHaveLength(2); + }); +}); diff --git a/packages/sdk/src/agent-relay.ts b/packages/sdk/src/agent-relay.ts index c16ebb273..254839cdd 100644 --- a/packages/sdk/src/agent-relay.ts +++ b/packages/sdk/src/agent-relay.ts @@ -8,6 +8,7 @@ import { } from './relaycast-telemetry.js'; import { createEventFanIn, + createObserverEventSource, RelaycastMessagingClient, type RelayAgentRegistration, type RelayEventFanIn, @@ -50,6 +51,27 @@ export interface AgentRelayOptions extends RelaycastMessagingOptions { actions?: AgentRelayActions; /** Factory for agent-token-scoped messaging clients. Defaults to a Relaycast client. */ createAgentMessaging?: (token: string) => RelayMessaging; + /** + * Read-only observer token (`ot_live_...`) with `stream:read` scope. + * When set, `relay.addListener(...)` streams from the workspace observer + * plane — a durable per-workspace event log plus the live `/v1/ws` observer + * stream — instead of fanning in through registered agent clients. Observer + * tokens are read-only: `workspace.register()` and `workspace.reconnect()` + * throw in observer mode. + */ + observerToken?: string; + /** + * Observer-mode resume cursor: skip durable-log events at or below this + * per-workspace sequence number. Pair with {@link AgentRelayOptions.onCursor} + * to resume a persisted stream without gaps or duplicates. + */ + sinceSeq?: number; + /** + * Observer-mode cursor hook, called with each advanced sequence number. + * Persisting the cursor is the caller's job; feed the last value back as + * `sinceSeq` on the next start. + */ + onCursor?: (seq: number) => void; /** * Receives listener and action handler errors with a context identifying * the listener selector or action name. When unset, errors are logged as @@ -102,6 +124,7 @@ export class AgentRelay implements AgentRelayAgent { readonly messaging: RelayMessaging; private readonly actions: AgentRelayActions; readonly workspaceKey?: string; + private readonly observerToken?: string; private readonly messagingOptions: RelaycastMessagingOptions; private readonly clientsByToken = new Map(); @@ -113,23 +136,56 @@ export class AgentRelay implements AgentRelayAgent { private readonly errorHooks = new Set(); constructor(options: AgentRelayOptions = {}) { - const { messaging, actions, workspaceKey, createAgentMessaging, onError, ...messagingOptions } = options; + const { + messaging, + actions, + workspaceKey, + createAgentMessaging, + onError, + observerToken, + sinceSeq, + onCursor, + ...messagingOptions + } = options; const resolvedWorkspaceKey = workspaceKey ?? messagingOptions.apiKey; this.workspaceKey = resolvedWorkspaceKey; - this.messagingOptions = { ...messagingOptions, workspaceKey: resolvedWorkspaceKey }; + this.observerToken = observerToken; + // Observer-only construction: fall back to the observer token as the REST + // credential so the client can be built; writes are rejected server-side. + this.messagingOptions = { ...messagingOptions, workspaceKey: resolvedWorkspaceKey ?? observerToken }; this.messaging = messaging ?? new RelaycastMessagingClient(this.messagingOptions); this.actions = actions ?? new ActionRegistry(); this.createAgentMessaging = createAgentMessaging ?? ((token) => new RelaycastMessagingClient({ ...this.messagingOptions, agentToken: token })); - // Relaycast v5 streams events over each agent's node transport; the - // workspace-key stream cannot receive them. The fan-in makes - // `relay.addListener(...)` stream through every registered agent client - // (added in `messagingForToken`), keeping the workspace client only as a - // pre-registration fallback. - this.eventFanIn = createEventFanIn(this.messaging.events, { - onError: (error) => this.reportError(error, { source: 'listener', selector: 'events.connect' }), - }); + if (observerToken) { + // Observer mode: the listener hub streams from the workspace observer + // plane (durable event log + live observer stream). The observer source + // is the sole source — agent registration is disabled — so the fan-in + // never falls back to the workspace client and never warns about + // missing agent sources. + this.eventFanIn = createEventFanIn(undefined, { + onError: (error) => this.reportError(error, { source: 'listener', selector: 'events.connect' }), + }); + this.eventFanIn.addSource( + createObserverEventSource({ + observerToken, + baseUrl: messagingOptions.baseUrl, + sinceSeq, + onCursor, + onError: (error) => this.reportError(error, { source: 'listener', selector: 'events.connect' }), + }) + ); + } else { + // Relaycast v5 streams events over each agent's node transport; the + // workspace-key stream cannot receive them. The fan-in makes + // `relay.addListener(...)` stream through every registered agent client + // (added in `messagingForToken`), keeping the workspace client only as a + // pre-registration fallback. + this.eventFanIn = createEventFanIn(this.messaging.events, { + onError: (error) => this.reportError(error, { source: 'listener', selector: 'events.connect' }), + }); + } if (onError) { this.errorHooks.add(onError); } @@ -223,10 +279,24 @@ export class AgentRelay implements AgentRelayAgent { get workspace(): RelayWorkspace { if (!this.workspaceFacade) { - this.workspaceFacade = createWorkspaceFacade(this.messaging, { + const facade = createWorkspaceFacade(this.messaging, { buildAgentClient: (registration) => this.buildAgentClient(registration), reconnectAgent: (apiToken) => this.reconnectAgent(apiToken), }); + // Observer mode is read-only: registering (or rehydrating) participant + // agents requires a workspace key, and mixing participant sources into + // the observer stream would double-deliver events. + this.workspaceFacade = this.observerToken + ? { + ...facade, + register: (async () => { + throw new Error(observerReadOnlyMessage('register()')); + }) as RelayWorkspace['register'], + reconnect: async () => { + throw new Error(observerReadOnlyMessage('reconnect()')); + }, + } + : facade; } return this.workspaceFacade; } @@ -351,6 +421,10 @@ export class AgentRelay implements AgentRelayAgent { } } +function observerReadOnlyMessage(method: string): string { + return `${method} is not available in observer mode: observer tokens are read-only; use a workspace key to register agents.`; +} + function extractWorkspaceKey(payload: Record): string | undefined { const data = payload.data && typeof payload.data === 'object' ? (payload.data as Record) : {}; diff --git a/packages/sdk/src/messaging/index.ts b/packages/sdk/src/messaging/index.ts index 45e3c5dd0..df5e72f6a 100644 --- a/packages/sdk/src/messaging/index.ts +++ b/packages/sdk/src/messaging/index.ts @@ -1,6 +1,11 @@ export * from './types.js'; export * from './normalize.js'; export { createEventFanIn, type EventFanInOptions, type RelayEventFanIn } from './event-fanin.js'; +export { + createObserverEventSource, + type ObserverEventSourceOptions, + type ObserverLiveStream, +} from './observer-source.js'; export { RelayPlacementError, RelaycastMessagingClient, diff --git a/packages/sdk/src/messaging/observer-source.ts b/packages/sdk/src/messaging/observer-source.ts new file mode 100644 index 000000000..07431a4e2 --- /dev/null +++ b/packages/sdk/src/messaging/observer-source.ts @@ -0,0 +1,294 @@ +import { RelayCast } from '@relaycast/sdk'; + +import { normalizeMessagingEvent } from './normalize.js'; +import type { RelayMessagingEvent, RelayMessagingEventMap, RelayMessagingEventsSurface } from './types.js'; + +/** + * The slice of the live observer stream the source depends on. A `RelayCast` + * client constructed with an observer token satisfies it: `connect()` opens + * `/v1/ws?token=` and `on.any(...)` delivers the raw server + * event frames. + */ +export interface ObserverLiveStream { + connect(): void; + disconnect(): void; + on: { any(handler: (event: unknown) => void): () => void }; +} + +/** + * Options for {@link createObserverEventSource}. + */ +export interface ObserverEventSourceOptions { + /** Read-only observer token (`ot_live_...`) with `stream:read` scope. */ + observerToken: string; + /** Relaycast base URL. Defaults to the hosted gateway. */ + baseUrl?: string; + /** + * Resume the durable event log after this per-workspace sequence number + * (exclusive). Defaults to `0` (backfill from the start of the log). + */ + sinceSeq?: number; + /** + * Receives every advanced cursor value. Persisting the cursor is the + * caller's job: store the last value and pass it back as `sinceSeq` to + * resume without gaps or duplicates. + */ + onCursor?: (seq: number) => void; + /** Receives live-stream and backfill failures. Defaults to console warnings. */ + onError?: (error: unknown) => void; + /** Live stream factory override for tests. Defaults to a `RelayCast` client. */ + createLiveStream?: () => ObserverLiveStream; + /** Fetch override for tests. Defaults to the global `fetch`. */ + fetch?: typeof globalThis.fetch; + /** REST backfill page size. The server caps pages at 500 events. */ + backfillPageSize?: number; +} + +const DEFAULT_BASE_URL = 'https://cast.agentrelay.com'; +const MAX_BACKFILL_PAGE_SIZE = 500; + +/** One raw event frame from the durable workspace event log. */ +interface BackfillEventRow { + seq: number; + payload: unknown; +} + +type UnknownRecord = Record; + +function isRecord(value: unknown): value is UnknownRecord { + return value !== null && typeof value === 'object' && !Array.isArray(value); +} + +/** + * Read the per-workspace monotonic sequence number stamped on a live frame. + * Frames without a `seq` were never appended to the durable log (server-side + * append failure) and are treated as live-only. + */ +function readSeq(raw: unknown): number | undefined { + if (!isRecord(raw)) return undefined; + const value = raw.seq; + return typeof value === 'number' && Number.isFinite(value) ? value : undefined; +} + +function parseBackfillEvents(payload: unknown): { events: BackfillEventRow[]; latestSeq: number } { + const record = isRecord(payload) ? payload : {}; + const data = isRecord(record.data) ? record.data : {}; + const rows = Array.isArray(data.events) ? data.events : []; + const events: BackfillEventRow[] = []; + for (const row of rows) { + if (!isRecord(row)) continue; + const seq = typeof row.seq === 'number' && Number.isFinite(row.seq) ? row.seq : undefined; + if (seq === undefined) continue; + events.push({ seq, payload: row.payload }); + } + const latestSeq = + typeof data.latest_seq === 'number' && Number.isFinite(data.latest_seq) ? data.latest_seq : 0; + return { events, latestSeq }; +} + +/** + * Create a {@link RelayMessagingEventsSurface} backed by the workspace + * observer plane: the durable per-workspace event log plus the live + * observer WebSocket stream. + * + * On `connect()` the source: + * + * 1. opens the live stream and buffers incoming frames, + * 2. REST-backfills `GET /v1/workspace/events` from the in-memory cursor + * (starting at `sinceSeq`, default `0`), paginating until `latest_seq`, + * 3. emits the backfilled events, then the buffered/live frames — deduped + * and ordered by `seq`. Frames without a `seq` pass straight through. + * + * When the backfill endpoint is missing (404 on older engines) or fails, + * the source degrades to live-only streaming. Raw frames flow through + * {@link normalizeMessagingEvent}, so listeners receive the same public event + * shapes as every other source. + * + * @param options - Observer token, cursor, and injectable transports. + * @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(/\/+$/, ''); + const pageSize = Math.min( + Math.max(options.backfillPageSize ?? MAX_BACKFILL_PAGE_SIZE, 1), + MAX_BACKFILL_PAGE_SIZE + ); + const fetchImpl = options.fetch ?? globalThis.fetch; + const createLiveStream = + options.createLiveStream ?? + (() => new RelayCast({ apiKey: options.observerToken, baseUrl }) as unknown as ObserverLiveStream); + + const handlers = new Map void | Promise>>(); + + /** Highest durable-log sequence emitted so far; events at or below it are duplicates. */ + let cursor = options.sinceSeq ?? 0; + let live: ObserverLiveStream | undefined; + let offLive: (() => void) | undefined; + /** Raw live frames received while the backfill is still running, in arrival order. */ + let pending: unknown[] = []; + let backfillDone = false; + /** Invalidates in-flight backfills when the source is disconnected. */ + let epoch = 0; + + const report = (error: unknown): void => { + if (options.onError) { + try { + options.onError(error); + } catch { + // Error hooks must not throw into the event source. + } + return; + } + console.warn('[agent-relay] observer stream:', error); + }; + + const emit = (event: RelayMessagingEvent): void => { + for (const key of [event.type, 'any'] as const) { + for (const handler of handlers.get(key) ?? []) { + try { + void Promise.resolve(handler(event)).catch(report); + } catch (error) { + report(error); + } + } + } + }; + + const advanceCursor = (seq: number): void => { + cursor = seq; + if (options.onCursor) { + try { + options.onCursor(seq); + } catch { + // Cursor hooks must not throw into the event source. + } + } + }; + + /** Emit a raw frame, deduping seq-stamped frames against the cursor. */ + const deliver = (raw: unknown): void => { + const seq = readSeq(raw); + if (seq !== undefined) { + if (seq <= cursor) return; + advanceCursor(seq); + } + emit(normalizeMessagingEvent(raw)); + }; + + const handleLiveFrame = (raw: unknown): void => { + if (!backfillDone) { + pending.push(raw); + return; + } + deliver(raw); + }; + + /** + * Flush frames buffered during the backfill. Seq-stamped frames are ordered + * by `seq` among themselves (stable sort — frames without a `seq` keep their + * arrival position) and deduped against the cursor; the rest pass through. + */ + const flushPending = (): void => { + backfillDone = true; + const buffered = pending; + pending = []; + 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 backfillPage = async ( + 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}` }, + }); + // Older engines have no durable event log; observer mode then works + // live-only and the cursor advances from seq-stamped live frames. + if (response.status === 404) return undefined; + if (!response.ok) { + throw new Error(`observer backfill failed: HTTP ${response.status} for GET /v1/workspace/events`); + } + return parseBackfillEvents(await response.json()); + }; + + const runBackfill = async (): Promise => { + const startedEpoch = epoch; + try { + for (;;) { + const page = await backfillPage(cursor); + if (epoch !== startedEpoch) return; + if (!page || page.events.length === 0) break; + for (const event of page.events) { + if (event.seq <= cursor) continue; + advanceCursor(event.seq); + emit(normalizeMessagingEvent(event.payload)); + } + if (cursor >= page.latestSeq) break; + } + } catch (error) { + if (epoch !== startedEpoch) return; + // Backfill is best-effort: degrade to live-only rather than losing the + // stream entirely. + report(error); + } + if (epoch !== startedEpoch) return; + flushPending(); + }; + + return { + connect: (): void => { + if (live) return; + backfillDone = false; + pending = []; + 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. + } + void runBackfill(); + }, + + disconnect: async (): Promise => { + epoch += 1; + backfillDone = false; + pending = []; + offLive?.(); + offLive = undefined; + const stream = live; + live = undefined; + if (stream) { + try { + stream.disconnect(); + } catch (error) { + report(error); + } + } + }, + + // The observer stream is workspace-wide; there is no per-channel + // subscription surface on the observer socket. + subscribe: (): void => {}, + unsubscribe: (): void => {}, + + on: ( + event: K, + handler: (...args: RelayMessagingEventMap[K]) => void | Promise + ): (() => void) => { + const set = handlers.get(event) ?? new Set(); + set.add(handler as (event: RelayMessagingEvent) => void | Promise); + handlers.set(event, set); + return () => { + set.delete(handler as (event: RelayMessagingEvent) => void | Promise); + }; + }, + }; +} From fbecb7015584eb38afa6c00cf37f0130a70789ba Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 2 Jul 2026 03:57:51 +0000 Subject: [PATCH 06/12] fix(broker): mute joined channels for the broker-self agent MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Claude-Session: https://claude.ai/code/session_01VASvErz2MxkQMCsSzTgsbe --- crates/broker/src/relaycast/ws.rs | 49 +++++++++++++++++++++++-------- 1 file changed, 37 insertions(+), 12 deletions(-) diff --git a/crates/broker/src/relaycast/ws.rs b/crates/broker/src/relaycast/ws.rs index 7c10a4268..d0ddc787a 100644 --- a/crates/broker/src/relaycast/ws.rs +++ b/crates/broker/src/relaycast/ws.rs @@ -343,12 +343,15 @@ impl RelaycastHttpClient { metadata: None, }; match agent_client.ensure_joined_channel(request).await { - Ok(outcome) => tracing::info!( - channel = %outcome.name, - created = outcome.created, - joined = outcome.joined, - "ensured default channel membership" - ), + Ok(outcome) => { + tracing::info!( + channel = %outcome.name, + created = outcome.created, + joined = outcome.joined, + "ensured default channel membership" + ); + mute_self_channel(&agent_client, &outcome.name).await; + } Err(error) => { tracing::warn!(channel = %name, error = %error, "failed to ensure default channel membership"); } @@ -385,12 +388,15 @@ impl RelaycastHttpClient { metadata: None, }; match agent_client.ensure_joined_channel(request).await { - Ok(outcome) => tracing::info!( - channel = %outcome.name, - created = outcome.created, - joined = outcome.joined, - "ensured extra channel membership" - ), + Ok(outcome) => { + tracing::info!( + channel = %outcome.name, + created = outcome.created, + joined = outcome.joined, + "ensured extra channel membership" + ); + mute_self_channel(&agent_client, &outcome.name).await; + } Err(error) => { tracing::warn!(channel = %name, error = %error, "failed to ensure extra channel membership"); } @@ -573,6 +579,25 @@ impl RelaycastHttpClient { } } +/// Mute a channel for the broker-self agent, best-effort. +/// +/// The broker-self identity lives on an implicit direct node that never +/// connects, so every channel message fanned out to it writes a delivery row +/// that queues forever and churns through TTL expiry. The engine's channel +/// delivery fan-out skips muted members (mentions still deliver), so muting +/// the broker-self membership stops those dead-letter rows at the source. +/// Failures only log a warning — muting is an optimization and must never +/// fail startup. +async fn mute_self_channel(agent_client: &AgentClient, channel: &str) { + if let Err(error) = agent_client.mute_channel(channel).await { + tracing::warn!( + channel = %channel, + error = %error, + "failed to mute channel for broker-self agent; channel deliveries will queue for its offline node" + ); + } +} + /// Build a `RelayCast` workspace client from an API key and optional base URL. /// When `base_url` is `None`, the SDK applies its own default. fn build_relay_client(api_key: &str, base_url: Option<&str>) -> Option { From b0959ea2fc83c74f3d172e8d44dc538fb82d4cf4 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 2 Jul 2026 03:57:51 +0000 Subject: [PATCH 07/12] docs: observer-plane spec + changelog entries for observer mode and broker 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 Claude-Session: https://claude.ai/code/session_01VASvErz2MxkQMCsSzTgsbe --- CHANGELOG.md | 2 + specs/observer-plane.md | 201 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 203 insertions(+) create mode 100644 specs/observer-plane.md diff --git a/CHANGELOG.md b/CHANGELOG.md index c3854e2f4..6681fef1d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `agent-relay skills add` installs the `/orchestrate` skill (from `agentrelay.com/skill.md`) into your coding harnesses. An interactive TUI asks whether to install for the current project or globally and which harnesses to target (Claude Code, Codex, Cursor, Gemini, OpenCode); `--global`/`--local`, `--harness `, and `--all` flags drive it non-interactively. - `agent-relay up --verbose` now prints step-by-step startup progress (port resolution, broker process spawn, handshake retries, fleet sidecar, node-delivery wait, agent spawns) and streams the broker's own startup-phase logs and stderr live, instead of only surfacing a terse error if startup fails. +- `@agent-relay/sdk` observer mode: `new AgentRelay({ observerToken })` streams `relay.addListener(...)` read-only from the workspace observer plane — the durable event log is REST-backfilled and merged with the live stream, deduped and ordered by `seq`, with `sinceSeq`/`onCursor` options to persist and resume the cursor across restarts. Degrades to live-only against engines without the backfill endpoint; `workspace.register()`/`reconnect()` throw in observer mode (observer tokens are read-only). ### Changed @@ -43,6 +44,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `agent-relay-broker` node-only delivery: `seq:0` fan-out frames (reactions, read receipts, and `action.completed`/`action.failed`/`action.denied` results) are no longer dropped — action results are injected into the calling agent's PTY; reaction/read receipts are acked (PTY surfacing deferred). Inbound `deliver`/`action.invoke` frames tolerate unknown future engine fields instead of being dropped without an ack (which caused infinite redelivery), and the broker's per-agent delivery-dedup memory is now bounded. - `agent-relay-broker` node-control no longer logs a spurious `agent.register reply did not match a pending registration` warning. The engine replies to every node-control request (`node.register`, `inventory.sync`) with a fresh snowflake id; those non-agent replies are now ignored at debug instead of warned, and `agent.register` replies correlate by request id with a robust fallback to matching the pending registration by agent name. - `agent-relay-broker` node `release` action reports a faithful `action.result`: a genuinely unknown worker returns an error while an already-exited worker still reports success. +- `agent-relay-broker` now mutes the default/extra channels it joins for its own broker-self agent, so channel messages stop writing delivery rows to that identity's permanently-offline implicit node (they previously queued until TTL expiry on every message). Muting is best-effort and never fails startup. - `agent-relay integration webhook|subscription` commands work reliably in local broker workflows even when shell auth is stale or missing. - The Bun-compiled `agent-relay` standalone binary now bundles workspace packages from their compiled JS instead of their `.d.ts`, so `local up` starts the implicit Fleet local node instead of failing with `Fleet local node skipped: … is not a function`. The `tsconfig` `paths` that mapped `@agent-relay/*` to declaration files (no runtime exports) were redundant with the npm workspace symlinks and have been removed. - `agent-relay` and `@agent-relay/sdk` require `@relaycast/sdk` `^4.1.2`, whose matching `@relaycast/types` package is now published, so publish installs resolve cleanly without pinning. diff --git a/specs/observer-plane.md b/specs/observer-plane.md new file mode 100644 index 000000000..8574bf45f --- /dev/null +++ b/specs/observer-plane.md @@ -0,0 +1,201 @@ +# Observer Plane — Durable Workspace Event Log and Cursor Protocol + +**Status**: Draft +**Date**: 2026-07-02 +**Author**: Design session (Will + Claude) + +--- + +## 1. Problem + +Relaycast v5 made participant delivery durable — every agent receives through +its node's `deliveries` mailbox with replay on reconnect — but everything that +_watches_ a workspace still rides fire-and-forget transports. That split +produces one recurring class of bug: **any consumer that is not a delivery +recipient has no way to observe the workspace reliably.** + +Concrete instances: + +- **The workspace stream loses events on reconnect.** The `/v1/ws` observer + stream is fan-out only: a dropped socket loses every event emitted while the + client was away, and there is no way to ask for the gap. In-memory resync + rings help with short blips but are bounded and per-connection, not durable. +- **Pear falls back to polling reconciliation.** Because the stream can't be + trusted across reconnects, UIs like Pear re-fetch rosters and message + histories on a timer and diff them against local state — expensive, laggy, + and still racy between polls. +- **The broker's `relay_inbound` only sees local recipients' deliveries.** + A broker observes workspace traffic through the delivery frames of the + agents it hosts. Channel messages whose recipients are all remote (or whose + only local "recipient" is an identity that never drains) are invisible to it. +- **Dashboard-identity deliveries queue forever.** Watch-only identities (the + broker-self agent, dashboard/console identities) get registered as agents so + something fans events at them — but they live on permanently-offline + implicit direct nodes, so every channel message writes a delivery row that + sits queued until TTL expiry. The mailbox becomes a dead-letter queue and + the sweeps churn. + +All four are the same defect: observation is being emulated with participant +machinery (agent identities + deliveries) or with a lossy stream, because +there is no first-class read path. + +## 2. The frame: two planes + +The fix is to make watching a workspace a separate plane with its own +contract, instead of a degenerate form of participating. + +| Plane | Who | Credential | Transport | Durability | +| --------------- | ---------------------------- | --------------------------------------------- | ------------------------------------ | ----------------------------------- | +| **Participant** | agents | agent token (via node registration) | `/v1/node/ws` deliver frames | per-recipient `deliveries` mailbox | +| **Observer** | UIs, SDK listeners, auditors | observer token (`ot_live_...`, `stream:read`) | `/v1/ws` live stream + REST backfill | per-workspace append-only event log | + +**Participant plane — unchanged.** v5 node deliveries stay exactly as they +are: agents register through a node, sends are stateless REST, inbound is a +durable per-recipient mailbox drained over the node socket with replay on +reconnect. Nothing in this spec touches message delivery semantics. + +**Observer plane — new.** A durable, append-only, per-workspace event log +plus a client cursor protocol. Observers never hold agent identities, never +receive deliveries, and never appear in the roster. Reading the workspace +cannot create server-side per-reader state beyond the shared log. + +## 3. Observer plane contract + +### 3.1 The event log + +The engine appends every workspace-visible server event (the same +client-shaped JSON the WS delivers: `message.created`, `thread.reply`, +`message.reacted`, membership/presence/channel events, ...) to a +per-workspace log table: + +| column | meaning | +| ------------ | ------------------------------------------------------------ | +| `seq` | per-workspace monotonic sequence number (assigned on append) | +| `type` | dotted server event type | +| `channel_id` | channel the event belongs to, when applicable | +| `payload` | the client-shaped event JSON, verbatim | +| `created_at` | append time | + +`seq` semantics: + +- Monotonic and unique **per workspace**; it totally orders the log. +- Stamped onto the corresponding live WS frame as a top-level `seq` field, so + the stream and the log speak the same coordinate system. +- A live frame **may lack `seq`** when the server-side log append failed. Such + frames are live-only: observers pass them through but cannot dedupe or + replay them. Append failure must never block event fan-out. + +The log is retention-bounded (age and/or row cap per workspace); a cursor +older than retention backfills from the oldest retained event. + +### 3.2 Backfill: `GET /v1/workspace/events` + +``` +GET {baseUrl}/v1/workspace/events?since=&limit= (n <= 500) +Authorization: Bearer +``` + +Response: + +```json +{ + "ok": true, + "data": { + "events": [ + { "seq": 41, "type": "message.created", "channel_id": "c1", "payload": { ... }, "created_at": "..." } + ], + "latest_seq": 57 + } +} +``` + +- `events` are ordered by `seq` ascending, strictly greater than `since`. +- `payload` is the same client-shaped event JSON the WS delivers, so one + normalization path serves both legs. +- Auth: observer tokens with `stream:read`. Workspace keys and agent tokens + are rejected, mirroring the live `/v1/ws` endpoint. +- Older engines without the log 404 this route; clients must degrade to + live-only streaming. + +### 3.3 Live stream + +Unchanged endpoint, upgraded frames: `WS /v1/ws?token=` accepts +observer tokens with `stream:read` (rejects workspace keys and agent tokens) +and delivers the standard server event shapes, now carrying `seq` when the +log append succeeded. + +### 3.4 Client cursor protocol + +The client owns exactly one piece of state: the highest `seq` it has emitted +(the **cursor**). Persisting it is the caller's job. The connect sequence: + +1. **Connect live and buffer.** Open `/v1/ws` with the observer token; hold + incoming frames in memory without emitting. +2. **Backfill from the cursor.** Page `GET /v1/workspace/events?since=` + until `latest_seq` is reached, emitting each event and advancing the + cursor. +3. **Merge and go live.** Flush the buffer — drop frames with `seq <=` cursor + (already emitted via backfill), emit the rest ordered by `seq`, pass + seq-less frames through — then emit live frames directly, deduping against + the cursor. + +Properties: no event in the log is dropped across reconnects (the buffer +covers the backfill window; the backfill covers the disconnected window), +no seq-stamped event is emitted twice, and a persisted cursor resumes a +stream across process restarts. A 404 backfill degrades to live-only; the +cursor still advances from live `seq` so persistence keeps working. + +## 4. Per-component changes + +- **Engine (relaycast)** — owns the plane: the event log table + append hook + in event fan-out, `seq` stamping on `/v1/ws` frames, the + `GET /v1/workspace/events` route, observer-token auth on both, and log + retention sweeps. (Being built in parallel; §3 is the contract.) +- **Cloud (relaycast-cloud)** — copies the new D1 migrations from the engine, + routes the new REST path through the existing engine app, and keeps the + workspace-stream KV gating in lockstep on any new publish path. +- **Relay SDK (`@agent-relay/sdk`)** — observer mode: + `new AgentRelay({ observerToken, baseUrl, sinceSeq, onCursor })` streams + `relay.addListener(...)` from a new observer event source + (`packages/sdk/src/messaging/observer-source.ts`) implementing §3.4 — + live leg via a `RelayCast` client on the observer token, backfill via REST, + both normalized through the existing `normalizeMessagingEvent` path. The + source is the fan-in's sole source in observer mode; `workspace.register()` + / `reconnect()` throw (observer tokens are read-only). Cursor persistence + stays with the caller via `sinceSeq`/`onCursor`. +- **Broker (`agent-relay-broker`)** — self-mute: after joining default/extra + channels, the broker mutes them for the broker-self agent + (`crates/broker/src/relaycast/ws.rs`). The engine's channel fan-out skips + muted members, so channel messages stop writing delivery rows to + broker-self's permanently-offline implicit direct node. Best-effort; when + the broker needs full workspace visibility it should hold an observer + cursor instead of a phantom recipient. +- **Pear** — replace polling reconciliation with an observer-token stream + + persisted cursor: hydrate from REST once, then apply §3.4. Polling remains + only as a fallback against pre-log engines (the 404 path). + +## 5. Migration order + +1. **Engine**: event log + `seq` stamping + backfill route ship behind the + existing observer-token auth. Pure addition; old clients ignore `seq`. +2. **Cloud**: bump the engine, copy migrations, deploy. The route 404s until + this lands — which clients already tolerate. +3. **Relay SDK**: observer mode (this repo). Works live-only against + pre-log engines, gapless once 1–2 are deployed. +4. **Broker**: self-mute lands independently (it needs only the + long-shipped channel mute endpoint) and stops the dead-letter churn + immediately. +5. **Pear**: move to the cursor protocol once 1–3 are stable, then retire + polling reconciliation. +6. **Cleanup** (later): drop watch-only agent identities (dashboard/console + registrations) in favor of observer tokens; consider broker workspace + visibility via an observer cursor. + +## 6. Open questions + +- Retention policy defaults (age vs row cap) and whether `latest_seq` should + expose the oldest retained seq so clients can detect truncated backfills. +- Whether channel-scoped observer tokens should filter both the stream and + the backfill (today the plane is workspace-wide). +- Whether the broker should adopt an observer cursor for `relay_inbound` + (fixing remote-recipient blindness) before or after Pear migrates. From 93aad7d4288fc14ca82ebb86b83b68a092c79576 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 2 Jul 2026 04:31:49 +0000 Subject: [PATCH 08/12] fix(sdk,broker): raw observer WebSocket live leg + reaction normalization MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Claude-Session: https://claude.ai/code/session_01VASvErz2MxkQMCsSzTgsbe --- crates/broker/src/runtime/api.rs | 3 + crates/broker/src/runtime/tests.rs | 5 +- .../sdk/src/__tests__/observer-source.test.ts | 29 ++-- packages/sdk/src/messaging/normalize.ts | 11 ++ packages/sdk/src/messaging/observer-source.ts | 134 ++++++++++++++++-- 5 files changed, 162 insertions(+), 20 deletions(-) diff --git a/crates/broker/src/runtime/api.rs b/crates/broker/src/runtime/api.rs index 9529f0bf8..1596d505e 100644 --- a/crates/broker/src/runtime/api.rs +++ b/crates/broker/src/runtime/api.rs @@ -20,6 +20,9 @@ pub(crate) fn default_observer_token_scopes() -> Vec { ObserverScope::ChannelsRead, ObserverScope::ActivityRead, ObserverScope::AgentsRead, + // Reactions surface on the observer stream as `message.reacted`; + // without this scope the live stream filters them out for UIs. + ObserverScope::ReactionsRead, ] } diff --git a/crates/broker/src/runtime/tests.rs b/crates/broker/src/runtime/tests.rs index 2b6fc4273..1dec2b422 100644 --- a/crates/broker/src/runtime/tests.rs +++ b/crates/broker/src/runtime/tests.rs @@ -2793,14 +2793,15 @@ fn default_observer_token_scopes_are_read_only_and_exclude_unneeded_scopes() { ObserverScope::ChannelsRead, ObserverScope::ActivityRead, ObserverScope::AgentsRead, + ObserverScope::ReactionsRead, ] .into_iter() .collect(); assert_eq!( scopes.len(), - 7, - "expected exactly 7 default observer token scopes, got {scopes:?}" + 8, + "expected exactly 8 default observer token scopes, got {scopes:?}" ); assert_eq!( actual, expected, diff --git a/packages/sdk/src/__tests__/observer-source.test.ts b/packages/sdk/src/__tests__/observer-source.test.ts index 8fd0a5af5..e640a5d86 100644 --- a/packages/sdk/src/__tests__/observer-source.test.ts +++ b/packages/sdk/src/__tests__/observer-source.test.ts @@ -349,10 +349,22 @@ describe('AgentRelay observer mode', () => { }); it('streams observer events through relay.addListener', async () => { - const live = createFakeLiveStream(); - relaycastMocks.relayCast.mockImplementation(function () { - return live.stream; - }); + // The default live leg is a raw WebSocket (it must see the top-level + // `seq`, which higher-level clients strip); stub the global constructor. + const sockets: FakeWebSocket[] = []; + class FakeWebSocket { + url: string; + onopen: (() => void) | null = null; + onmessage: ((message: { data: string }) => void) | null = null; + onclose: (() => void) | null = null; + onerror: (() => void) | null = null; + close = vi.fn(); + constructor(url: string) { + this.url = url; + sockets.push(this); + } + } + vi.stubGlobal('WebSocket', FakeWebSocket); vi.stubGlobal('fetch', createBackfillFetch([logRow(1, 'm1')])); const relay = new AgentRelay({ @@ -365,11 +377,12 @@ describe('AgentRelay observer mode', () => { received.push(event); }); await settle(); - live.emit(liveFrame('m2', 2)); - expect(relaycastMocks.relayCast).toHaveBeenCalledWith( - expect.objectContaining({ apiKey: 'ot_live_test', baseUrl: 'https://api.example.test' }) - ); + expect(sockets).toHaveLength(1); + expect(sockets[0].url).toBe('wss://api.example.test/v1/ws?token=ot_live_test'); + sockets[0].onopen?.(); + sockets[0].onmessage?.({ data: JSON.stringify(liveFrame('m2', 2)) }); + expect(received).toHaveLength(2); }); }); diff --git a/packages/sdk/src/messaging/normalize.ts b/packages/sdk/src/messaging/normalize.ts index 90b1c5f08..a0f0987ba 100644 --- a/packages/sdk/src/messaging/normalize.ts +++ b/packages/sdk/src/messaging/normalize.ts @@ -763,6 +763,17 @@ export function normalizeMessagingEvent(input: unknown): RelayMessagingEvent { emoji: readString(record, 'emoji') ?? '', agentName: readString(record, 'agentName', 'agent_name') ?? '', }; + // The engine's raw workspace-stream frames (and the durable event log) + // carry reactions as one `message.reacted` type with an `action` field; + // higher-level clients split it into reaction.added/removed before we see + // it, but raw observer frames arrive in the server shape. + case 'message.reacted': + return { + type: readString(record, 'action') === 'removed' ? 'reactionRemoved' : 'reactionAdded', + messageId: readString(record, 'messageId', 'message_id') ?? '', + emoji: readString(record, 'emoji') ?? '', + agentName: readString(record, 'agentName', 'agent_name') ?? '', + }; case 'action.invoked': return { type: 'actionInvoked', diff --git a/packages/sdk/src/messaging/observer-source.ts b/packages/sdk/src/messaging/observer-source.ts index 07431a4e2..1a0fc0bde 100644 --- a/packages/sdk/src/messaging/observer-source.ts +++ b/packages/sdk/src/messaging/observer-source.ts @@ -1,18 +1,122 @@ -import { RelayCast } from '@relaycast/sdk'; - import { normalizeMessagingEvent } from './normalize.js'; import type { RelayMessagingEvent, RelayMessagingEventMap, RelayMessagingEventsSurface } from './types.js'; /** - * The slice of the live observer stream the source depends on. A `RelayCast` - * client constructed with an observer token satisfies it: `connect()` opens - * `/v1/ws?token=` and `on.any(...)` delivers the raw server - * event frames. + * The slice of the live observer stream the source depends on. The default + * implementation is a raw WebSocket to `/v1/ws?token=` that + * delivers each JSON frame untouched — the frames must arrive raw because the + * durable-log cursor rides on their top-level `seq` field, which higher-level + * clients strip during schema parsing. */ export interface ObserverLiveStream { connect(): void; disconnect(): void; - on: { any(handler: (event: unknown) => void): () => void }; + on: { + any(handler: (event: unknown) => void): () => void; + /** Fires on every socket (re)open; used to re-backfill after a reconnect. */ + open?(handler: () => void): () => void; + }; +} + +/** + * Raw observer WebSocket with capped-backoff auto-reconnect. Uses the global + * `WebSocket` (Node >= 21 and all browsers). Frames are parsed as JSON and + * handed to `on.any` handlers verbatim, preserving the top-level `seq`. + */ +function createRawObserverStream( + baseUrl: string, + token: string, + report: (error: unknown) => void +): ObserverLiveStream { + const anyHandlers = new Set<(event: unknown) => void>(); + const openHandlers = new Set<() => void>(); + let socket: WebSocket | undefined; + let closed = false; + let attempts = 0; + let timer: ReturnType | undefined; + + const wsUrl = `${baseUrl.replace(/^http/, 'ws')}/v1/ws?token=${encodeURIComponent(token)}`; + + const scheduleReconnect = (): void => { + if (closed || timer !== undefined) return; + attempts += 1; + const delay = Math.min(30_000, 1_000 * 2 ** Math.min(attempts - 1, 5)); + timer = setTimeout(() => { + timer = undefined; + open(); + }, delay); + (timer as { unref?: () => void }).unref?.(); + }; + + const open = (): void => { + if (closed || socket) return; + const WebSocketImpl = (globalThis as { WebSocket?: typeof WebSocket }).WebSocket; + if (!WebSocketImpl) { + report( + new Error( + 'No global WebSocket implementation available for the observer stream (Node >= 21 or a browser is required).' + ) + ); + return; + } + let ws: WebSocket; + try { + ws = new WebSocketImpl(wsUrl); + } catch (error) { + report(error); + scheduleReconnect(); + return; + } + socket = ws; + ws.onopen = () => { + attempts = 0; + for (const handler of openHandlers) handler(); + }; + ws.onmessage = (message: MessageEvent) => { + let frame: unknown; + try { + frame = JSON.parse(String(message.data)); + } catch { + return; // Non-JSON frames carry nothing for us. + } + for (const handler of anyHandlers) handler(frame); + }; + ws.onclose = () => { + if (socket === ws) socket = undefined; + scheduleReconnect(); + }; + ws.onerror = () => { + // The close handler owns reconnection. + }; + }; + + return { + connect: open, + disconnect: () => { + closed = true; + if (timer !== undefined) { + clearTimeout(timer); + timer = undefined; + } + const ws = socket; + socket = undefined; + try { + ws?.close(); + } catch { + // Already closed. + } + }, + on: { + any: (handler) => { + anyHandlers.add(handler); + return () => anyHandlers.delete(handler); + }, + open: (handler) => { + openHandlers.add(handler); + return () => openHandlers.delete(handler); + }, + }, + }; } /** @@ -36,7 +140,7 @@ export interface ObserverEventSourceOptions { onCursor?: (seq: number) => void; /** Receives live-stream and backfill failures. Defaults to console warnings. */ onError?: (error: unknown) => void; - /** Live stream factory override for tests. Defaults to a `RelayCast` client. */ + /** Live stream factory override for tests. Defaults to a raw observer WebSocket. */ createLiveStream?: () => ObserverLiveStream; /** Fetch override for tests. Defaults to the global `fetch`. */ fetch?: typeof globalThis.fetch; @@ -115,8 +219,7 @@ export function createObserverEventSource(options: ObserverEventSourceOptions): ); const fetchImpl = options.fetch ?? globalThis.fetch; const createLiveStream = - options.createLiveStream ?? - (() => new RelayCast({ apiKey: options.observerToken, baseUrl }) as unknown as ObserverLiveStream); + options.createLiveStream ?? (() => createRawObserverStream(baseUrl, options.observerToken, report)); const handlers = new Map void | Promise>>(); @@ -248,6 +351,17 @@ export function createObserverEventSource(options: ObserverEventSourceOptions): try { live = createLiveStream(); offLive = live.on.any(handleLiveFrame); + // Frames missed while the socket was down are only in the durable + // 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; + void runBackfill(); + } + hadOpen = true; + }); live.connect(); } catch (error) { report(error); From 180cb8ba028cc3e6db857e8e85be6e34ee686fca Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 2 Jul 2026 04:45:02 +0000 Subject: [PATCH 09/12] fix(sdk): keep the observer token out of the stream URL, enforce wss MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Claude-Session: https://claude.ai/code/session_01VASvErz2MxkQMCsSzTgsbe --- .../sdk/src/__tests__/observer-source.test.ts | 9 ++- packages/sdk/src/messaging/observer-source.ts | 67 +++++++++++++++++-- 2 files changed, 68 insertions(+), 8 deletions(-) diff --git a/packages/sdk/src/__tests__/observer-source.test.ts b/packages/sdk/src/__tests__/observer-source.test.ts index e640a5d86..10580c347 100644 --- a/packages/sdk/src/__tests__/observer-source.test.ts +++ b/packages/sdk/src/__tests__/observer-source.test.ts @@ -359,8 +359,10 @@ describe('AgentRelay observer mode', () => { onclose: (() => void) | null = null; onerror: (() => void) | null = null; close = vi.fn(); - constructor(url: string) { + options: unknown; + constructor(url: string, options?: unknown) { this.url = url; + this.options = options; sockets.push(this); } } @@ -379,7 +381,10 @@ describe('AgentRelay observer mode', () => { await settle(); expect(sockets).toHaveLength(1); - expect(sockets[0].url).toBe('wss://api.example.test/v1/ws?token=ot_live_test'); + // Header auth keeps the token out of the URL; the query downgrade only + // fires on runtimes whose WebSocket rejects constructor options. + expect(sockets[0].url).toBe('wss://api.example.test/v1/ws'); + expect(sockets[0].options).toEqual({ headers: { authorization: 'Bearer ot_live_test' } }); sockets[0].onopen?.(); sockets[0].onmessage?.({ data: JSON.stringify(liveFrame('m2', 2)) }); diff --git a/packages/sdk/src/messaging/observer-source.ts b/packages/sdk/src/messaging/observer-source.ts index 1a0fc0bde..a1d5de006 100644 --- a/packages/sdk/src/messaging/observer-source.ts +++ b/packages/sdk/src/messaging/observer-source.ts @@ -3,7 +3,7 @@ import type { RelayMessagingEvent, RelayMessagingEventMap, RelayMessagingEventsS /** * The slice of the live observer stream the source depends on. The default - * implementation is a raw WebSocket to `/v1/ws?token=` that + * implementation is a raw WebSocket to `/v1/ws` (bearer-authenticated) that * delivers each JSON frame untouched — the frames must arrive raw because the * durable-log cursor rides on their top-level `seq` field, which higher-level * clients strip during schema parsing. @@ -18,10 +18,35 @@ export interface ObserverLiveStream { }; } +/** + * Build the observer stream URL. The scheme is always `wss:` except for + * loopback hosts (local self-hosted engines), and the URL never carries the + * token — authentication travels in the `Authorization` header where the + * runtime supports it, with an explicit query-token downgrade only when it + * does not (see {@link createRawObserverStream}). + */ +function observerWsUrl(baseUrl: string, opts: { includeToken?: string } = {}): string { + const url = new URL(baseUrl); + const loopback = url.hostname === 'localhost' || url.hostname === '127.0.0.1' || url.hostname === '[::1]'; + url.protocol = url.protocol === 'http:' && loopback ? 'ws:' : 'wss:'; + url.pathname = `${url.pathname.replace(/\/+$/, '')}/v1/ws`; + if (opts.includeToken !== undefined) { + url.searchParams.set('token', opts.includeToken); + } + return url.toString(); +} + /** * Raw observer WebSocket with capped-backoff auto-reconnect. Uses the global * `WebSocket` (Node >= 21 and all browsers). Frames are parsed as JSON and * handed to `on.any` handlers verbatim, preserving the top-level `seq`. + * + * Authentication: the token is sent as an `Authorization: Bearer` header via + * the Node (undici) constructor options extension, keeping it out of the URL. + * Runtimes whose `WebSocket` rejects or ignores constructor options — browsers, + * per the WHATWG signature — are detected (constructor throw, or a close + * before the first open on the header attempt) and downgraded once to the + * server's `?token=` query convention. */ function createRawObserverStream( baseUrl: string, @@ -34,13 +59,15 @@ function createRawObserverStream( let closed = false; let attempts = 0; let timer: ReturnType | undefined; + /** Whether header auth has been observed working (any successful open). */ + let everOpened = false; + /** Downgrade flag: the runtime cannot send headers, use the query token. */ + let useQueryToken = false; - const wsUrl = `${baseUrl.replace(/^http/, 'ws')}/v1/ws?token=${encodeURIComponent(token)}`; - - const scheduleReconnect = (): void => { + const scheduleReconnect = (delayOverrideMs?: number): void => { if (closed || timer !== undefined) return; attempts += 1; - const delay = Math.min(30_000, 1_000 * 2 ** Math.min(attempts - 1, 5)); + const delay = delayOverrideMs ?? Math.min(30_000, 1_000 * 2 ** Math.min(attempts - 1, 5)); timer = setTimeout(() => { timer = undefined; open(); @@ -48,6 +75,22 @@ function createRawObserverStream( (timer as { unref?: () => void }).unref?.(); }; + const construct = (WebSocketImpl: typeof WebSocket): WebSocket => { + if (useQueryToken) { + return new WebSocketImpl(observerWsUrl(baseUrl, { includeToken: token })); + } + try { + // Node's undici WebSocket accepts { headers } as a non-standard + // extension; browsers throw on a non-protocols second argument. + return new (WebSocketImpl as new (url: string, options: unknown) => WebSocket)(observerWsUrl(baseUrl), { + headers: { authorization: `Bearer ${token}` }, + }); + } catch { + useQueryToken = true; + return new WebSocketImpl(observerWsUrl(baseUrl, { includeToken: token })); + } + }; + const open = (): void => { if (closed || socket) return; const WebSocketImpl = (globalThis as { WebSocket?: typeof WebSocket }).WebSocket; @@ -61,15 +104,18 @@ function createRawObserverStream( } let ws: WebSocket; try { - ws = new WebSocketImpl(wsUrl); + ws = construct(WebSocketImpl); } catch (error) { report(error); scheduleReconnect(); return; } + let openedHere = false; socket = ws; ws.onopen = () => { attempts = 0; + everOpened = true; + openedHere = true; for (const handler of openHandlers) handler(); }; ws.onmessage = (message: MessageEvent) => { @@ -83,6 +129,15 @@ function createRawObserverStream( }; ws.onclose = () => { if (socket === ws) socket = undefined; + // A close before the first successful open on a header-auth attempt + // 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); + return; + } scheduleReconnect(); }; ws.onerror = () => { From fd1f4499b18d7695153b37a56956c3931a4055c8 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 2 Jul 2026 05:34:50 +0000 Subject: [PATCH 10/12] feat(sdk): advance the observer backfill cursor via next_since MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Claude-Session: https://claude.ai/code/session_01VASvErz2MxkQMCsSzTgsbe --- .../sdk/src/__tests__/observer-source.test.ts | 40 ++++++++++++++++++- packages/sdk/src/messaging/observer-source.ts | 29 ++++++++++++-- specs/observer-plane.md | 12 +++++- 3 files changed, 73 insertions(+), 8 deletions(-) diff --git a/packages/sdk/src/__tests__/observer-source.test.ts b/packages/sdk/src/__tests__/observer-source.test.ts index 10580c347..6a46a0737 100644 --- a/packages/sdk/src/__tests__/observer-source.test.ts +++ b/packages/sdk/src/__tests__/observer-source.test.ts @@ -58,11 +58,18 @@ function logRow(seq: number, messageId: string): Record { }; } -function jsonResponse(events: Record[], latestSeq: number) { +function jsonResponse(events: Record[], latestSeq: number, nextSince?: number) { return { ok: true, status: 200, - json: async () => ({ ok: true, data: { events, latest_seq: latestSeq } }), + json: async () => ({ + ok: true, + data: { + events, + latest_seq: latestSeq, + ...(nextSince !== undefined ? { next_since: nextSince } : {}), + }, + }), } as Response; } @@ -108,6 +115,35 @@ afterEach(() => { }); describe('createObserverEventSource', () => { + it('advances the cursor via next_since when a scoped page is fully filtered', async () => { + // Server consumed rows 1-3 (hidden for this token) and reports + // next_since=3 with an empty page; the visible row 4 arrives on the next + // page. Without next_since the loop would stall on zero progress. + const fetchImpl = vi.fn(async (input: RequestInfo | URL) => { + const url = new URL(String(input)); + const since = Number(url.searchParams.get('since') ?? '0'); + if (since < 3) return jsonResponse([], 4, 3); + if (since < 4) return jsonResponse([logRow(4, 'visible')], 4, 4); + return jsonResponse([], 4, since); + }) as unknown as typeof fetch; + + const cursors: number[] = []; + const source = createObserverEventSource({ + observerToken: 'ot_live_test', + baseUrl: 'https://api.example.test', + fetch: fetchImpl, + createLiveStream: () => createFakeLiveStream().stream, + onCursor: (seq) => cursors.push(seq), + }); + const received = collect(source); + source.connect(); + await settle(); + + expect(messageIds(received)).toEqual(['visible']); + // The cursor advanced through the hidden window (3) and the visible row (4). + expect(cursors).toEqual([3, 4]); + }); + it('backfills from the log, then merges buffered live frames deduped and ordered by seq', async () => { const live = createFakeLiveStream(); let releaseBackfill!: () => void; diff --git a/packages/sdk/src/messaging/observer-source.ts b/packages/sdk/src/messaging/observer-source.ts index a1d5de006..7abfc8760 100644 --- a/packages/sdk/src/messaging/observer-source.ts +++ b/packages/sdk/src/messaging/observer-source.ts @@ -229,7 +229,11 @@ function readSeq(raw: unknown): number | undefined { return typeof value === 'number' && Number.isFinite(value) ? value : undefined; } -function parseBackfillEvents(payload: unknown): { events: BackfillEventRow[]; latestSeq: number } { +function parseBackfillEvents(payload: unknown): { + events: BackfillEventRow[]; + latestSeq: number; + nextSince: number | undefined; +} { const record = isRecord(payload) ? payload : {}; const data = isRecord(record.data) ? record.data : {}; const rows = Array.isArray(data.events) ? data.events : []; @@ -242,7 +246,11 @@ function parseBackfillEvents(payload: unknown): { events: BackfillEventRow[]; la } const latestSeq = typeof data.latest_seq === 'number' && Number.isFinite(data.latest_seq) ? data.latest_seq : 0; - return { events, latestSeq }; + // Scoped-token resume cursor: the seq of the last row the server's scan + // consumed (visible or filtered). Absent on older engines. + const nextSince = + typeof data.next_since === 'number' && Number.isFinite(data.next_since) ? data.next_since : undefined; + return { events, latestSeq, nextSince }; } /** @@ -360,7 +368,9 @@ export function createObserverEventSource(options: ObserverEventSourceOptions): const backfillPage = async ( since: number - ): Promise<{ events: BackfillEventRow[]; latestSeq: number } | undefined> => { + ): Promise< + { events: BackfillEventRow[]; latestSeq: number; nextSince: number | undefined } | undefined + > => { const url = `${baseUrl}/v1/workspace/events?since=${since}&limit=${pageSize}`; const response = await fetchImpl(url, { headers: { Authorization: `Bearer ${options.observerToken}` }, @@ -378,15 +388,26 @@ export function createObserverEventSource(options: ObserverEventSourceOptions): const startedEpoch = epoch; try { for (;;) { + const before = cursor; const page = await backfillPage(cursor); if (epoch !== startedEpoch) return; - if (!page || page.events.length === 0) break; + if (!page) break; for (const event of page.events) { if (event.seq <= cursor) continue; advanceCursor(event.seq); emit(normalizeMessagingEvent(event.payload)); } + // For scoped observer tokens a page can be empty while hidden rows + // were consumed server-side; `next_since` advances the cursor past + // them (events hidden from this token will never be delivered live + // either, so skipping their seqs is safe). + if (page.nextSince !== undefined && page.nextSince > cursor) { + advanceCursor(page.nextSince); + } 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) { if (epoch !== startedEpoch) return; diff --git a/specs/observer-plane.md b/specs/observer-plane.md index 8574bf45f..b8447d4ff 100644 --- a/specs/observer-plane.md +++ b/specs/observer-plane.md @@ -104,12 +104,19 @@ Response: "events": [ { "seq": 41, "type": "message.created", "channel_id": "c1", "payload": { ... }, "created_at": "..." } ], - "latest_seq": 57 + "latest_seq": 57, + "next_since": 45 } } ``` - `events` are ordered by `seq` ascending, strictly greater than `since`. +- `next_since` is the resume cursor: the seq of the last row the server's + scan consumed — visible or hidden by the observer token's scopes/filters. + Scoped tokens can have entire windows filtered out server-side; advancing + by `next_since` (rather than the last visible event) means an all-hidden + page never stalls pagination. Hidden events are never delivered to that + token on the live stream either, so skipping their seqs is safe. - `payload` is the same client-shaped event JSON the WS delivers, so one normalization path serves both legs. - Auth: observer tokens with `stream:read`. Workspace keys and agent tokens @@ -133,7 +140,8 @@ The client owns exactly one piece of state: the highest `seq` it has emitted incoming frames in memory without emitting. 2. **Backfill from the cursor.** Page `GET /v1/workspace/events?since=` until `latest_seq` is reached, emitting each event and advancing the - cursor. + cursor — including past fully-filtered windows via `next_since` (absent on + older engines; stop on a page that makes no progress). 3. **Merge and go live.** Flush the buffer — drop frames with `seq <=` cursor (already emitted via backfill), emit the rest ordered by `seq`, pass seq-less frames through — then emit live frames directly, deduping against From 1f96aa0717b54ad0fdd64d2d4c38fd749d147b00 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 2 Jul 2026 06:05:50 +0000 Subject: [PATCH 11/12] chore(sdk,cli): bump @relaycast/sdk to ^5.1.0 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Claude-Session: https://claude.ai/code/session_01VASvErz2MxkQMCsSzTgsbe --- CHANGELOG.md | 2 +- package-lock.json | 114 +++++++++++++++++++------------------- packages/cli/package.json | 2 +- packages/sdk/package.json | 2 +- 4 files changed, 60 insertions(+), 60 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6681fef1d..bb36c7900 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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. ### Removed diff --git a/package-lock.json b/package-lock.json index 384af8447..fef745468 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "@agent-relay/monorepo", - "version": "9.1.7", + "version": "9.2.1", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "@agent-relay/monorepo", - "version": "9.1.7", + "version": "9.2.1", "license": "Apache-2.0", "workspaces": [ "packages/*" @@ -9940,46 +9940,46 @@ }, "packages/brand": { "name": "@agent-relay/brand", - "version": "9.1.7" + "version": "9.2.1" }, "packages/broker-darwin-arm64": { "name": "@agent-relay/broker-darwin-arm64", - "version": "9.1.7", + "version": "9.2.1", "license": "MIT" }, "packages/broker-darwin-x64": { "name": "@agent-relay/broker-darwin-x64", - "version": "9.1.7", + "version": "9.2.1", "license": "MIT" }, "packages/broker-linux-arm64": { "name": "@agent-relay/broker-linux-arm64", - "version": "9.1.7", + "version": "9.2.1", "license": "MIT" }, "packages/broker-linux-x64": { "name": "@agent-relay/broker-linux-x64", - "version": "9.1.7", + "version": "9.2.1", "license": "MIT" }, "packages/broker-win32-x64": { "name": "@agent-relay/broker-win32-x64", - "version": "9.1.7", + "version": "9.2.1", "license": "MIT" }, "packages/cli": { "name": "agent-relay", - "version": "9.1.7", + "version": "9.2.1", "license": "Apache-2.0", "dependencies": { - "@agent-relay/cloud": "9.1.7", - "@agent-relay/config": "9.1.7", - "@agent-relay/fleet": "9.1.7", - "@agent-relay/harness-driver": "9.1.7", - "@agent-relay/sdk": "9.1.7", - "@agent-relay/utils": "9.1.7", + "@agent-relay/cloud": "9.2.1", + "@agent-relay/config": "9.2.1", + "@agent-relay/fleet": "9.2.1", + "@agent-relay/harness-driver": "9.2.1", + "@agent-relay/sdk": "9.2.1", + "@agent-relay/utils": "9.2.1", "@modelcontextprotocol/sdk": "^1.0.0", - "@relaycast/sdk": "^5.0.5", + "@relaycast/sdk": "^5.1.0", "@relayfile/client": "^0.10.19", "@relayflows/cli": "^1.0.1", "@xterm/headless": "^6.0.0", @@ -10002,11 +10002,11 @@ } }, "packages/cli/node_modules/@relaycast/sdk": { - "version": "5.0.5", - "resolved": "https://registry.npmjs.org/@relaycast/sdk/-/sdk-5.0.5.tgz", - "integrity": "sha512-Lgfg7uzKr3ox0b6lNUQ6hfd5ereoID5+QTXsmnGoQ7pjfj9DilAkPQiyDEM7OCUC+SSQiNViLu0CB1DRofpisw==", + "version": "5.1.0", + "resolved": "https://registry.npmjs.org/@relaycast/sdk/-/sdk-5.1.0.tgz", + "integrity": "sha512-UqPDKLOpIxfvBqpYGTJNxe66RSHUM9y8gc0zOfIh9vV7eKPTd/kGhO1QlXmA+KHpeVDy3Tr8b+Ccm6F72qwV9g==", "dependencies": { - "@relaycast/types": "5.0.5", + "@relaycast/types": "5.1.0", "zod": "^4.3.6" } }, @@ -10020,9 +10020,9 @@ } }, "packages/cli/node_modules/@relaycast/types": { - "version": "5.0.5", - "resolved": "https://registry.npmjs.org/@relaycast/types/-/types-5.0.5.tgz", - "integrity": "sha512-He4BR8zyIPsQ4WaVYOFRrOfJq9FlkzCwvB9f3eG6Wj2oXKLD+vxrtVpUG87DmGA+pNgRReW8sgPM4vFoeAo11A==", + "version": "5.1.0", + "resolved": "https://registry.npmjs.org/@relaycast/types/-/types-5.1.0.tgz", + "integrity": "sha512-+G5uX4ye5aEda0I7acYb151q5I6Ss0qvs2NTNu6wBFvXJDhm/LxQGFVCbEJPi28qYbmJbH6ZOVhwKEvks2jDbQ==", "dependencies": { "zod": "^4.3.6" } @@ -10047,9 +10047,9 @@ }, "packages/cloud": { "name": "@agent-relay/cloud", - "version": "9.1.7", + "version": "9.2.1", "dependencies": { - "@agent-relay/config": "9.1.7", + "@agent-relay/config": "9.2.1", "@aws-sdk/client-s3": "3.1020.0", "ignore": "^7.0.5", "tar": "^7.5.10" @@ -10065,7 +10065,7 @@ }, "packages/config": { "name": "@agent-relay/config", - "version": "9.1.7", + "version": "9.2.1", "dependencies": { "zod": "^3.23.8", "zod-to-json-schema": "^3.23.1" @@ -10078,60 +10078,60 @@ }, "packages/evals": { "name": "@agent-relay/evals", - "version": "9.1.7", + "version": "9.2.1", "license": "Apache-2.0", "dependencies": { - "@agent-relay/harness-driver": "9.1.7", - "@agent-relay/integration-prompts": "9.1.7" + "@agent-relay/harness-driver": "9.2.1", + "@agent-relay/integration-prompts": "9.2.1" } }, "packages/fleet": { "name": "@agent-relay/fleet", - "version": "9.1.7", + "version": "9.2.1", "license": "Apache-2.0", "dependencies": { - "@agent-relay/harness-driver": "9.1.7", - "@agent-relay/harnesses": "9.1.7", - "@agent-relay/sdk": "9.1.7", + "@agent-relay/harness-driver": "9.2.1", + "@agent-relay/harnesses": "9.2.1", + "@agent-relay/sdk": "9.2.1", "zod": "^3.23.8" } }, "packages/harness-driver": { "name": "@agent-relay/harness-driver", - "version": "9.1.7", + "version": "9.2.1", "license": "Apache-2.0", "dependencies": { - "@agent-relay/sdk": "9.1.7", + "@agent-relay/sdk": "9.2.1", "ws": "^8.18.3", "zod": "^3.23.8" }, "optionalDependencies": { - "@agent-relay/broker-darwin-arm64": "9.1.7", - "@agent-relay/broker-darwin-x64": "9.1.7", - "@agent-relay/broker-linux-arm64": "9.1.7", - "@agent-relay/broker-linux-x64": "9.1.7", - "@agent-relay/broker-win32-x64": "9.1.7" + "@agent-relay/broker-darwin-arm64": "9.2.1", + "@agent-relay/broker-darwin-x64": "9.2.1", + "@agent-relay/broker-linux-arm64": "9.2.1", + "@agent-relay/broker-linux-x64": "9.2.1", + "@agent-relay/broker-win32-x64": "9.2.1" } }, "packages/harnesses": { "name": "@agent-relay/harnesses", - "version": "9.1.7", + "version": "9.2.1", "license": "Apache-2.0", "dependencies": { - "@agent-relay/harness-driver": "9.1.7", - "@agent-relay/sdk": "9.1.7" + "@agent-relay/harness-driver": "9.2.1", + "@agent-relay/sdk": "9.2.1" } }, "packages/integration-prompts": { "name": "@agent-relay/integration-prompts", - "version": "9.1.7", + "version": "9.2.1", "license": "Apache-2.0" }, "packages/policy": { "name": "@agent-relay/policy", - "version": "9.1.7", + "version": "9.2.1", "dependencies": { - "@agent-relay/config": "9.1.7" + "@agent-relay/config": "9.2.1" }, "devDependencies": { "@types/node": "^22.19.3", @@ -10140,27 +10140,27 @@ }, "packages/sdk": { "name": "@agent-relay/sdk", - "version": "9.1.7", + "version": "9.2.1", "dependencies": { - "@relaycast/sdk": "^5.0.5" + "@relaycast/sdk": "^5.1.0" }, "devDependencies": { "@types/node": "^22.13.10" } }, "packages/sdk/node_modules/@relaycast/sdk": { - "version": "5.0.5", - "resolved": "https://registry.npmjs.org/@relaycast/sdk/-/sdk-5.0.5.tgz", - "integrity": "sha512-Lgfg7uzKr3ox0b6lNUQ6hfd5ereoID5+QTXsmnGoQ7pjfj9DilAkPQiyDEM7OCUC+SSQiNViLu0CB1DRofpisw==", + "version": "5.1.0", + "resolved": "https://registry.npmjs.org/@relaycast/sdk/-/sdk-5.1.0.tgz", + "integrity": "sha512-UqPDKLOpIxfvBqpYGTJNxe66RSHUM9y8gc0zOfIh9vV7eKPTd/kGhO1QlXmA+KHpeVDy3Tr8b+Ccm6F72qwV9g==", "dependencies": { - "@relaycast/types": "5.0.5", + "@relaycast/types": "5.1.0", "zod": "^4.3.6" } }, "packages/sdk/node_modules/@relaycast/types": { - "version": "5.0.5", - "resolved": "https://registry.npmjs.org/@relaycast/types/-/types-5.0.5.tgz", - "integrity": "sha512-He4BR8zyIPsQ4WaVYOFRrOfJq9FlkzCwvB9f3eG6Wj2oXKLD+vxrtVpUG87DmGA+pNgRReW8sgPM4vFoeAo11A==", + "version": "5.1.0", + "resolved": "https://registry.npmjs.org/@relaycast/types/-/types-5.1.0.tgz", + "integrity": "sha512-+G5uX4ye5aEda0I7acYb151q5I6Ss0qvs2NTNu6wBFvXJDhm/LxQGFVCbEJPi28qYbmJbH6ZOVhwKEvks2jDbQ==", "dependencies": { "zod": "^4.3.6" } @@ -10176,9 +10176,9 @@ }, "packages/utils": { "name": "@agent-relay/utils", - "version": "9.1.7", + "version": "9.2.1", "dependencies": { - "@agent-relay/config": "9.1.7", + "@agent-relay/config": "9.2.1", "compare-versions": "^6.1.1" }, "devDependencies": { diff --git a/packages/cli/package.json b/packages/cli/package.json index 8174de3f1..02d00346b 100644 --- a/packages/cli/package.json +++ b/packages/cli/package.json @@ -50,7 +50,7 @@ "@agent-relay/sdk": "9.2.1", "@agent-relay/utils": "9.2.1", "@modelcontextprotocol/sdk": "^1.0.0", - "@relaycast/sdk": "^5.0.5", + "@relaycast/sdk": "^5.1.0", "@relayfile/client": "^0.10.19", "@relayflows/cli": "^1.0.1", "@xterm/headless": "^6.0.0", diff --git a/packages/sdk/package.json b/packages/sdk/package.json index 372018d94..cdc77e297 100644 --- a/packages/sdk/package.json +++ b/packages/sdk/package.json @@ -59,6 +59,6 @@ "@types/node": "^22.13.10" }, "dependencies": { - "@relaycast/sdk": "^5.0.5" + "@relaycast/sdk": "^5.1.0" } } From 7afc1ec4f0a790e6078192537f26bc8555d027aa Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 15 Jul 2026 19:56:54 +0000 Subject: [PATCH 12/12] fix(sdk): address observer-source review findings + CodeQL ReDoS MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Claude-Session: https://claude.ai/code/session_017qAFBAdEa5jmViDPSHqZYK --- .../sdk/src/__tests__/observer-source.test.ts | 171 ++++++++++++++++ packages/sdk/src/listeners.ts | 7 + packages/sdk/src/messaging/observer-source.ts | 182 ++++++++++++++---- 3 files changed, 327 insertions(+), 33 deletions(-) diff --git a/packages/sdk/src/__tests__/observer-source.test.ts b/packages/sdk/src/__tests__/observer-source.test.ts index 6a46a0737..53b392ad4 100644 --- a/packages/sdk/src/__tests__/observer-source.test.ts +++ b/packages/sdk/src/__tests__/observer-source.test.ts @@ -329,6 +329,133 @@ describe('createObserverEventSource', () => { expect(messageIds(received)).toEqual(['m1']); }); + it('orders a seq-less frame between out-of-order seq frames without dropping either', async () => { + // Regression: a non-total sort could let a higher seq advance the cursor + // before a lower seq buffered behind a seq-less frame was delivered. + const live = createFakeLiveStream(); + let releaseBackfill!: () => void; + const gate = new Promise((resolve) => { + releaseBackfill = resolve; + }); + const rows = [logRow(1, 'm1'), logRow(2, 'm2')]; + const fetchImpl = vi.fn(async (input: RequestInfo | URL) => { + await gate; + const url = new URL(String(input)); + const since = Number(url.searchParams.get('since') ?? '0'); + return jsonResponse( + rows.filter((row) => (row.seq as number) > since), + 2 + ); + }) as unknown as typeof fetch; + + const source = createObserverEventSource({ + observerToken: 'ot_live_test', + baseUrl: 'https://api.example.test', + createLiveStream: () => live.stream, + fetch: fetchImpl, + }); + const received = collect(source); + + source.connect(); + // Arrival order: seq 4, then a seq-less frame, then seq 3. + live.emit(liveFrame('m4', 4)); + live.emit(liveFrame('m-live-only')); + live.emit(liveFrame('m3', 3)); + + releaseBackfill(); + await settle(); + + // seq 3 is delivered (not dropped by seq 4 advancing the cursor first), the + // seq-less frame keeps its arrival slot, and everything arrives seq-ordered. + expect(messageIds(received)).toEqual(['m1', 'm2', 'm3', 'm-live-only', 'm4']); + }); + + it('surfaces a warning instead of silently skipping when a scoped backfill makes no progress', async () => { + // Older engine: a fully filtered page returns no visible events, no + // next_since, and latest_seq still ahead — the loop must stop AND warn. + const live = createFakeLiveStream(); + const fetchImpl = vi.fn(async () => jsonResponse([], 5)) as unknown as typeof fetch; + const onError = vi.fn(); + + createObserverEventSource({ + observerToken: 'ot_live_test', + baseUrl: 'https://api.example.test', + createLiveStream: () => live.stream, + fetch: fetchImpl, + onError, + }).connect(); + await settle(); + + expect(onError).toHaveBeenCalledTimes(1); + expect(String(onError.mock.calls[0][0])).toContain('observer backfill stopped early'); + }); + + it('degrades to live-only when a backfill page hangs past the timeout', async () => { + vi.useFakeTimers(); + try { + const live = createFakeLiveStream(); + const onError = vi.fn(); + // Never resolves on its own; rejects when the source aborts the request. + const fetchImpl = vi.fn( + (_input: RequestInfo | URL, init?: RequestInit) => + new Promise((_resolve, reject) => { + init?.signal?.addEventListener('abort', () => reject(new Error('aborted'))); + }) + ) as unknown as typeof fetch; + + const source = createObserverEventSource({ + observerToken: 'ot_live_test', + baseUrl: 'https://api.example.test', + createLiveStream: () => live.stream, + fetch: fetchImpl, + backfillTimeoutMs: 50, + onError, + }); + const received = collect(source); + + source.connect(); + live.emit(liveFrame('m1', 1)); // buffered while the backfill hangs + await vi.advanceTimersByTimeAsync(60); + + expect(onError).toHaveBeenCalledTimes(1); + // Buffer flushed to live-only delivery after the abort. + expect(messageIds(received)).toEqual(['m1']); + } finally { + vi.useRealTimers(); + } + }); + + it('retries the live stream after a connect-time failure', async () => { + // A throw during live-stream setup must clear `live` so the next connect() + // builds a fresh stream instead of returning early. + const good = createFakeLiveStream(); + let attempt = 0; + const onError = vi.fn(); + const source = createObserverEventSource({ + observerToken: 'ot_live_test', + baseUrl: 'https://api.example.test', + fetch: createBackfillFetch([]), + onError, + createLiveStream: () => { + attempt += 1; + if (attempt === 1) throw new Error('live boom'); + return good.stream; + }, + }); + const received = collect(source); + + source.connect(); + await settle(); + expect(onError).toHaveBeenCalledTimes(1); + + source.connect(); // must retry, not no-op + await settle(); + expect(good.stream.connect).toHaveBeenCalledTimes(1); + + good.emit(liveFrame('m1', 1)); + expect(messageIds(received)).toEqual(['m1']); + }); + it('disconnect stops the live stream; reconnect backfills from the cursor', async () => { const live = createFakeLiveStream(); const fetchImpl = createBackfillFetch([logRow(1, 'm1'), logRow(2, 'm2')]); @@ -384,6 +511,50 @@ describe('AgentRelay observer mode', () => { ); }); + it('keeps the token out of the URL on Node and reports abnormal closes', async () => { + // Security regression guard: a pre-open close on Node (which supports header + // auth) must not permanently downgrade to a `?token=` URL, and abnormal + // closes must surface through onError rather than looping silently. + const sockets: Array<{ url: string; options: unknown; onclose: ((e?: unknown) => void) | null }> = []; + class FakeWebSocket { + url: string; + options: unknown; + onopen: (() => void) | null = null; + onmessage: ((message: { data: string }) => void) | null = null; + onclose: ((e?: unknown) => void) | null = null; + onerror: (() => void) | null = null; + close = vi.fn(); + constructor(url: string, options?: unknown) { + this.url = url; + this.options = options; + sockets.push(this); + } + } + const onError = vi.fn(); + const source = createObserverEventSource({ + observerToken: 'ot_live_secret', + baseUrl: 'https://api.example.test', + fetch: createBackfillFetch([]), + webSocketImpl: FakeWebSocket as unknown as typeof WebSocket, + onError, + }); + source.connect(); + await settle(); + + expect(sockets).toHaveLength(1); + expect(sockets[0].url).toBe('wss://api.example.test/v1/ws'); + expect(sockets[0].options).toEqual({ headers: { authorization: 'Bearer ot_live_secret' } }); + + // Socket closes abnormally before ever opening. + sockets[0].onclose?.({ code: 1006, wasClean: false }); + + expect(onError).toHaveBeenCalled(); + expect(String(onError.mock.calls[0][0])).toContain('closed unexpectedly'); + // No token was ever leaked into a URL, on this or any reconnect socket. + expect(sockets.every((socket) => !socket.url.includes('token='))).toBe(true); + await source.disconnect(); + }); + it('streams observer events through relay.addListener', async () => { // The default live leg is a raw WebSocket (it must see the top-level // `seq`, which higher-level clients strip); stub the global constructor. diff --git a/packages/sdk/src/listeners.ts b/packages/sdk/src/listeners.ts index 707a18671..14da0e8f1 100644 --- a/packages/sdk/src/listeners.ts +++ b/packages/sdk/src/listeners.ts @@ -51,6 +51,12 @@ export type RelayErrorHook = (error: unknown, context: RelayErrorContext) => voi /** Default reporting for handler errors when no `onError` hook is registered. */ export function logRelayHandlerError(error: unknown, context: RelayErrorContext): void { const where = context.action ? `action "${context.action}"` : `"${context.selector ?? 'unknown'}"`; + // A named `operation` (e.g. `connect`) is a wiring/transport failure, not a + // user handler throwing — word it so the two are distinguishable at a glance. + if (context.operation) { + console.warn(`[agent-relay] ${context.source} ${context.operation} for ${where} failed:`, error); + return; + } console.warn(`[agent-relay] ${context.source} handler for ${where} threw:`, error); } @@ -657,6 +663,7 @@ export function createListenerHub( } catch (error) { makeReporter(context, { source: 'listener', + operation: 'connect', selector: typeof selector === 'string' ? selector : 'predicate', })(error); } diff --git a/packages/sdk/src/messaging/observer-source.ts b/packages/sdk/src/messaging/observer-source.ts index 7abfc8760..56faac29a 100644 --- a/packages/sdk/src/messaging/observer-source.ts +++ b/packages/sdk/src/messaging/observer-source.ts @@ -18,6 +18,17 @@ export interface ObserverLiveStream { }; } +/** + * Strip trailing `/` characters in linear time. A regex like `/\/+$/` is a + * polynomial-ReDoS hazard on attacker-influenced strings (CodeQL flags it), so + * we walk the tail explicitly instead. + */ +function stripTrailingSlashes(value: string): string { + let end = value.length; + while (end > 0 && value.charCodeAt(end - 1) === 47 /* '/' */) end -= 1; + return value.slice(0, end); +} + /** * Build the observer stream URL. The scheme is always `wss:` except for * loopback hosts (local self-hosted engines), and the URL never carries the @@ -29,32 +40,47 @@ function observerWsUrl(baseUrl: string, opts: { includeToken?: string } = {}): s const url = new URL(baseUrl); const loopback = url.hostname === 'localhost' || url.hostname === '127.0.0.1' || url.hostname === '[::1]'; url.protocol = url.protocol === 'http:' && loopback ? 'ws:' : 'wss:'; - url.pathname = `${url.pathname.replace(/\/+$/, '')}/v1/ws`; + url.pathname = `${stripTrailingSlashes(url.pathname)}/v1/ws`; if (opts.includeToken !== undefined) { url.searchParams.set('token', opts.includeToken); } return url.toString(); } +/** + * Whether the current runtime is Node.js. Node's `undici` WebSocket accepts an + * `Authorization` header via its constructor-options extension, so Node never + * needs to fall back to the token-in-URL convention — keeping the token out of + * request lines and access logs. + */ +function isNodeRuntime(): boolean { + return typeof process !== 'undefined' && !!(process as { versions?: { node?: string } }).versions?.node; +} + /** * Raw observer WebSocket with capped-backoff auto-reconnect. Uses the global - * `WebSocket` (Node >= 21 and all browsers). Frames are parsed as JSON and - * handed to `on.any` handlers verbatim, preserving the top-level `seq`. + * `WebSocket` (Node >= 22, all browsers), or an injected implementation for + * runtimes without one (e.g. Node < 22 with the `ws` package). Frames are + * parsed as JSON and handed to `on.any` handlers verbatim, preserving the + * top-level `seq`. * * Authentication: the token is sent as an `Authorization: Bearer` header via * the Node (undici) constructor options extension, keeping it out of the URL. - * Runtimes whose `WebSocket` rejects or ignores constructor options — browsers, - * per the WHATWG signature — are detected (constructor throw, or a close - * before the first open on the header attempt) and downgraded once to the - * server's `?token=` query convention. + * Only browser-like runtimes, whose `WebSocket` rejects or silently ignores + * constructor options per the WHATWG signature, fall back to the server's + * `?token=` query convention — and only after the constructor throws or two + * consecutive header attempts close before ever opening, so a single transient + * network blip on Node can never trigger a persistent token-in-URL downgrade. */ function createRawObserverStream( baseUrl: string, token: string, - report: (error: unknown) => void + report: (error: unknown) => void, + webSocketImpl?: typeof WebSocket ): ObserverLiveStream { const anyHandlers = new Set<(event: unknown) => void>(); const openHandlers = new Set<() => void>(); + const nodeRuntime = isNodeRuntime(); let socket: WebSocket | undefined; let closed = false; let attempts = 0; @@ -63,6 +89,8 @@ function createRawObserverStream( let everOpened = false; /** Downgrade flag: the runtime cannot send headers, use the query token. */ let useQueryToken = false; + /** Consecutive header-auth attempts that closed before opening (browser probe). */ + let headerCloseStreak = 0; const scheduleReconnect = (delayOverrideMs?: number): void => { if (closed || timer !== undefined) return; @@ -93,11 +121,14 @@ function createRawObserverStream( const open = (): void => { if (closed || socket) return; - const WebSocketImpl = (globalThis as { WebSocket?: typeof WebSocket }).WebSocket; + const WebSocketImpl = webSocketImpl ?? (globalThis as { WebSocket?: typeof WebSocket }).WebSocket; if (!WebSocketImpl) { report( new Error( - 'No global WebSocket implementation available for the observer stream (Node >= 21 or a browser is required).' + 'No WebSocket implementation available for the observer stream. Node exposes a ' + + 'global `WebSocket` only from v22; on earlier supported runtimes (Node >= 20.9) ' + + 'install one as `globalThis.WebSocket` (e.g. the `ws` package) or pass ' + + '`webSocketImpl`/`createLiveStream` to observer mode.' ) ); return; @@ -116,6 +147,7 @@ function createRawObserverStream( attempts = 0; everOpened = true; openedHere = true; + headerCloseStreak = 0; for (const handler of openHandlers) handler(); }; ws.onmessage = (message: MessageEvent) => { @@ -127,21 +159,44 @@ function createRawObserverStream( } for (const handler of anyHandlers) handler(frame); }; - ws.onclose = () => { + ws.onclose = (event?: { code?: number; reason?: string; wasClean?: boolean }) => { if (socket === ws) socket = undefined; - // A close before the first successful open on a header-auth attempt - // means the runtime accepted the options object but ignored the - // headers (auth rejected): downgrade to the query token and retry - // immediately, once. + if (closed) return; // Intentional teardown owns its own cleanup. + // A close before the first successful open on a header-auth attempt can + // mean the runtime accepted the options object but ignored the headers + // (browsers). Only browser-like runtimes downgrade to the query token, + // and only after two such probe closes — a Node runtime keeps header + // auth (and never leaks the token into the URL) even across transient + // failures. if (!useQueryToken && !everOpened && !openedHere) { - useQueryToken = true; - scheduleReconnect(0); - return; + if (!nodeRuntime) { + headerCloseStreak += 1; + if (headerCloseStreak >= 2) { + useQueryToken = true; + scheduleReconnect(0); + return; + } + scheduleReconnect(0); + return; + } + } + // Surface abnormal closures so invalid tokens / server rejections are not + // hidden behind a silent reconnect loop; clean closes (code 1000) stay quiet. + const code = event?.code; + const clean = event?.wasClean === true || code === 1000; + if (!clean) { + report( + new Error( + `observer WebSocket closed unexpectedly${ + code !== undefined ? ` (code ${code}${event?.reason ? `: ${event.reason}` : ''})` : '' + }` + ) + ); } scheduleReconnect(); }; ws.onerror = () => { - // The close handler owns reconnection. + // Abnormal closures are reported in onclose; onerror carries no extra detail. }; }; @@ -197,14 +252,27 @@ export interface ObserverEventSourceOptions { onError?: (error: unknown) => void; /** Live stream factory override for tests. Defaults to a raw observer WebSocket. */ createLiveStream?: () => ObserverLiveStream; + /** + * WebSocket implementation for the default live stream. Defaults to + * `globalThis.WebSocket`. Supply this (or a global polyfill such as the `ws` + * package) on runtimes without a native `WebSocket` (e.g. Node < 22). + */ + webSocketImpl?: typeof WebSocket; /** Fetch override for tests. Defaults to the global `fetch`. */ fetch?: typeof globalThis.fetch; /** REST backfill page size. The server caps pages at 500 events. */ backfillPageSize?: number; + /** + * Per-request timeout for each backfill page fetch, in milliseconds. A stalled + * backfill would otherwise buffer live frames indefinitely; on timeout the + * source degrades to live-only. Defaults to 30s. + */ + backfillTimeoutMs?: number; } const DEFAULT_BASE_URL = 'https://cast.agentrelay.com'; const MAX_BACKFILL_PAGE_SIZE = 500; +const DEFAULT_BACKFILL_TIMEOUT_MS = 30_000; /** One raw event frame from the durable workspace event log. */ interface BackfillEventRow { @@ -275,14 +343,16 @@ function parseBackfillEvents(payload: unknown): { * @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(/\/+$/, ''); + const baseUrl = stripTrailingSlashes(options.baseUrl ?? DEFAULT_BASE_URL); const pageSize = Math.min( Math.max(options.backfillPageSize ?? MAX_BACKFILL_PAGE_SIZE, 1), MAX_BACKFILL_PAGE_SIZE ); const fetchImpl = options.fetch ?? globalThis.fetch; + const backfillTimeoutMs = options.backfillTimeoutMs ?? DEFAULT_BACKFILL_TIMEOUT_MS; const createLiveStream = - options.createLiveStream ?? (() => createRawObserverStream(baseUrl, options.observerToken, report)); + options.createLiveStream ?? + (() => createRawObserverStream(baseUrl, options.observerToken, report, options.webSocketImpl)); const handlers = new Map void | Promise>>(); @@ -290,6 +360,7 @@ export function createObserverEventSource(options: ObserverEventSourceOptions): let cursor = options.sinceSeq ?? 0; let live: ObserverLiveStream | undefined; let offLive: (() => void) | undefined; + let offOpen: (() => void) | undefined; /** Raw live frames received while the backfill is still running, in arrival order. */ let pending: unknown[] = []; let backfillDone = false; @@ -358,12 +429,20 @@ export function createObserverEventSource(options: ObserverEventSourceOptions): backfillDone = true; const buffered = pending; pending = []; - 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); + // Order seq-stamped frames among themselves; seq-less frames keep their + // arrival slot. Sorting the whole array with a comparator that returns 0 + // for any pair involving a seq-less frame is not a total order (a higher + // seq could sort ahead of a lower one across a seq-less gap and advance the + // cursor past it), so we sort the sequenced frames out-of-band and splice + // them back into the seq-less skeleton. + 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); + } }; const backfillPage = async ( @@ -372,9 +451,20 @@ export function createObserverEventSource(options: ObserverEventSourceOptions): { events: BackfillEventRow[]; latestSeq: number; nextSince: number | undefined } | undefined > => { const url = `${baseUrl}/v1/workspace/events?since=${since}&limit=${pageSize}`; - const response = await fetchImpl(url, { - headers: { Authorization: `Bearer ${options.observerToken}` }, - }); + // Bound each page fetch so a hung backfill endpoint can't buffer live frames + // forever; on timeout the caller degrades to live-only. + const controller = new AbortController(); + const timeout = setTimeout(() => controller.abort(), backfillTimeoutMs); + (timeout as { unref?: () => void }).unref?.(); + let response: Response; + try { + response = await fetchImpl(url, { + headers: { Authorization: `Bearer ${options.observerToken}` }, + signal: controller.signal, + }); + } finally { + clearTimeout(timeout); + } // Older engines have no durable event log; observer mode then works // live-only and the cursor advances from seq-stamped live frames. if (response.status === 404) return undefined; @@ -406,8 +496,18 @@ 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; + // only hidden rows): stop rather than loop forever, but surface it — + // historical events after this point may never be backfilled. + if (cursor <= before) { + report( + new Error( + `observer backfill stopped early: engine returned no visible events and no next_since ` + + `cursor at seq ${cursor} (latest ${page.latestSeq}); historical events past this ` + + `point may be skipped. Upgrade the engine for next_since-based scoped backfill.` + ) + ); + break; + } } } catch (error) { if (epoch !== startedEpoch) return; @@ -431,7 +531,7 @@ 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?.(() => { + offOpen = live.on.open?.(() => { if (hadOpen) { backfillDone = false; void runBackfill(); @@ -440,6 +540,20 @@ export function createObserverEventSource(options: ObserverEventSourceOptions): }); live.connect(); } catch (error) { + // A partially initialized stream would make later connect() calls + // return early (live is set) and never retry. Tear it down so a fresh + // stream can be built on the next connect(). + offLive?.(); + offLive = undefined; + offOpen?.(); + offOpen = 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. @@ -453,6 +567,8 @@ export function createObserverEventSource(options: ObserverEventSourceOptions): pending = []; offLive?.(); offLive = undefined; + offOpen?.(); + offOpen = undefined; const stream = live; live = undefined; if (stream) {