diff --git a/packages/sdk/src/__tests__/messaging.test.ts b/packages/sdk/src/__tests__/messaging.test.ts index dfb49755d..c22e998c4 100644 --- a/packages/sdk/src/__tests__/messaging.test.ts +++ b/packages/sdk/src/__tests__/messaging.test.ts @@ -444,3 +444,71 @@ describe('RelaycastMessagingClient workspace stream', () => { expect(workspace.connect).not.toHaveBeenCalled(); }); }); + +describe('workspace stream (realtime fanout)', () => { + function streamWorkspace(enabled: boolean) { + const config = { enabled, defaultEnabled: false, override: enabled }; + return { + ...createWorkspace(), + workspace: { + info: vi.fn(async () => ({ id: 'ws-1', name: 'demo' })), + stream: { + get: vi.fn(async () => config), + set: vi.fn(async (value: boolean) => ({ ...config, enabled: value, override: value })), + inherit: vi.fn(async () => ({ enabled: false, defaultEnabled: false, override: null })), + }, + }, + connect: vi.fn(), + disconnect: vi.fn(), + on: { any: vi.fn(() => () => {}) }, + }; + } + + it('exposes workspace.stream get/set/inherit and normalizes the config', async () => { + const workspace = streamWorkspace(false); + const client = new RelaycastMessagingClient({ relaycast: workspace }); + + expect(await client.workspace.stream.get()).toEqual({ + enabled: false, + defaultEnabled: false, + override: false, + }); + expect(await client.workspace.stream.set(true)).toEqual({ + enabled: true, + defaultEnabled: false, + override: true, + }); + expect(workspace.workspace.stream.set).toHaveBeenCalledWith(true); + expect((await client.workspace.stream.inherit()).override).toBeNull(); + }); + + it('throws a clear error when the relaycast stream API is unavailable', async () => { + const workspace = { ...createWorkspace(), workspace: { info: vi.fn() } }; + const client = new RelaycastMessagingClient({ relaycast: workspace }); + await expect(client.workspace.stream.get()).rejects.toThrow(/workspace stream API/); + }); + + it('warns once when a workspace-key client connects with fanout disabled', async () => { + const workspace = streamWorkspace(false); + const client = new RelaycastMessagingClient({ relaycast: workspace }); + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}); + + client.events.connect(); + + await vi.waitFor(() => expect(warn).toHaveBeenCalledTimes(1)); + expect(warn.mock.calls[0]?.[0]).toContain('relay.workspace.stream.set(true)'); + warn.mockRestore(); + }); + + it('does not warn when fanout is already enabled', async () => { + const workspace = streamWorkspace(true); + const client = new RelaycastMessagingClient({ relaycast: workspace }); + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}); + + client.events.connect(); + + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(warn).not.toHaveBeenCalled(); + warn.mockRestore(); + }); +}); diff --git a/packages/sdk/src/facade.ts b/packages/sdk/src/facade.ts index 8a5770aa0..b492d56ef 100644 --- a/packages/sdk/src/facade.ts +++ b/packages/sdk/src/facade.ts @@ -9,6 +9,7 @@ import type { RelayMessageMode, RelaySendChannelMessageInput, RelayWorkspaceInfo, + RelayWorkspaceStreamConfig, } from './messaging/index.js'; import { actionSchemaToJsonSchema, @@ -175,6 +176,16 @@ export interface RelayWorkspace { ): Promise; reconnect(input: { apiToken: string }): Promise; info(): Promise; + /** + * Workspace-wide realtime fanout — **off by default**. A workspace-key client + * receives realtime events only after `set(true)`; agent-scoped clients don't + * need it. See https://agentrelay.com/docs/events#realtime-fanout + */ + readonly stream: { + get(): Promise; + set(enabled: boolean): Promise; + inherit(): Promise; + }; } export interface NotifyOptions { @@ -324,6 +335,12 @@ export function createWorkspaceFacade(messaging: RelayMessaging, deps?: Workspac return { info: () => messaging.workspace.info(), + // Lazy (like `info`) so partial mocks needn't implement stream. + stream: { + get: () => messaging.workspace.stream.get(), + set: (enabled: boolean) => messaging.workspace.stream.set(enabled), + inherit: () => messaging.workspace.stream.inherit(), + }, register: register as RelayWorkspace['register'], reconnect: async ({ apiToken }) => { if (!deps) { diff --git a/packages/sdk/src/messaging/relaycast.ts b/packages/sdk/src/messaging/relaycast.ts index fca8b079b..3a592b78c 100644 --- a/packages/sdk/src/messaging/relaycast.ts +++ b/packages/sdk/src/messaging/relaycast.ts @@ -44,6 +44,7 @@ import type { RelayRegisterCapabilityInput, RelayWebhook, RelayWorkspaceInfo, + RelayWorkspaceStreamConfig, InboxAckInput, InboxDeferInput, InboxFailInput, @@ -263,6 +264,12 @@ type RelaycastWorkspaceLike = { }; workspace?: { info(): Promise; + // Workspace-wide realtime fanout config (relaycast 2.5+). + stream?: { + get(): Promise; + set(enabled: boolean): Promise; + inherit(): Promise; + }; }; as?: (agentToken: string, options?: AgentClientOptions) => RelaycastAgentLike; // Workspace-scoped realtime stream (relaycast 2.5+): lets a workspace-key @@ -410,6 +417,17 @@ function createRelaycastClient(options: RelaycastMessagingOptions): RelaycastWor ) as unknown as RelaycastWorkspaceLike; } +/** Coerce a relaycast workspace-stream config (camel or snake case) into the relay shape. */ +function normalizeStreamConfig(raw: unknown): RelayWorkspaceStreamConfig { + const record = (raw ?? {}) as Record; + const override = record.override ?? record.override_value ?? null; + return { + enabled: Boolean(record.enabled), + defaultEnabled: Boolean(record.defaultEnabled ?? record.default_enabled), + override: override === null ? null : Boolean(override), + }; +} + export class RelaycastMessagingClient implements RelayMessagingClient { readonly capabilities: RelayMessagingCapabilities = { serverDeliveryState: false, @@ -426,6 +444,7 @@ export class RelaycastMessagingClient implements RelayMessagingClient { Set<(event: RelayMessagingEvent) => void | Promise> >(); private eventUnsubscribe?: () => void; + private workspaceStreamWarned = false; constructor(options: RelaycastMessagingOptions) { this.relaycast = createRelaycastClient(options); @@ -669,6 +688,9 @@ export class RelaycastMessagingClient implements RelayMessagingClient { if (typeof this.relaycast.connect === 'function' && this.relaycast.on) { this.relaycast.connect(); this.eventUnsubscribe = this.relaycast.on.any(forward); + // A workspace-key client only receives events when fanout is enabled; + // warn (once) instead of silently delivering nothing. + this.warnIfWorkspaceStreamDisabled(); return; } // No agent client and no workspace stream available: preserve the @@ -811,8 +833,48 @@ export class RelaycastMessagingClient implements RelayMessagingClient { } return (await this.relaycast.workspace.info()) as RelayWorkspaceInfo; }, + // Workspace-wide realtime fanout. Off by default. See `events.connect`. + stream: { + get: async (): Promise => + normalizeStreamConfig(await this.requireWorkspaceStream('get').get()), + set: async (enabled: boolean): Promise => + normalizeStreamConfig(await this.requireWorkspaceStream('set').set(enabled)), + inherit: async (): Promise => + normalizeStreamConfig(await this.requireWorkspaceStream('inherit').inherit()), + }, }; + private requireWorkspaceStream( + op: string + ): NonNullable['stream']> { + const stream = this.relaycast.workspace?.stream; + if (!stream) { + throw new Error( + `RelaycastMessagingClient.workspace.stream.${op} requires the relaycast workspace stream API (relaycast 2.5+).` + ); + } + return stream; + } + + /** Warn once if a workspace-key client subscribes while fanout is off (avoids silent zero delivery). */ + private warnIfWorkspaceStreamDisabled(): void { + if (this.workspaceStreamWarned) return; + const stream = this.relaycast.workspace?.stream; + if (!stream) return; + void stream + .get() + .then((raw) => { + if (this.workspaceStreamWarned || normalizeStreamConfig(raw).enabled) return; + this.workspaceStreamWarned = true; + console.warn( + '[agent-relay] Listening on a workspace-key client but realtime fanout is off — no events ' + + 'will arrive. Enable once with `relay.workspace.stream.set(true)`. ' + + 'https://agentrelay.com/docs/events#realtime-fanout' + ); + }) + .catch(() => {}); + } + private requireWebhooks(): NonNullable { if (!this.relaycast.webhooks) { throw new Error('RelaycastMessagingClient.integrations.webhooks requires the relaycast webhooks API.'); diff --git a/packages/sdk/src/messaging/types.ts b/packages/sdk/src/messaging/types.ts index 38e653274..31583edba 100644 --- a/packages/sdk/src/messaging/types.ts +++ b/packages/sdk/src/messaging/types.ts @@ -465,6 +465,14 @@ export interface RelayWorkspaceInfo { [key: string]: unknown; } +/** Workspace-wide realtime fanout config. Off by default; enable with `stream.set(true)`. */ +export interface RelayWorkspaceStreamConfig { + enabled: boolean; + defaultEnabled: boolean; + /** Explicit override, or `null` when inheriting the default. */ + override: boolean | null; +} + export type InboxItemState = 'queued' | 'delivered' | 'failed' | 'deferred' | 'read'; export interface InboxItem { @@ -769,6 +777,12 @@ export interface RelayMessagingClient { }; readonly workspace: { info(): Promise; + /** Workspace-wide realtime fanout. Off by default; enable with `set(true)`. */ + readonly stream: { + get(): Promise; + set(enabled: boolean): Promise; + inherit(): Promise; + }; }; } diff --git a/web/content/docs/events.mdx b/web/content/docs/events.mdx index c0ac67558..203dffd76 100644 --- a/web/content/docs/events.mdx +++ b/web/content/docs/events.mdx @@ -41,6 +41,28 @@ relay.addListener(relay.action('spawn-claude').calledBy(engineer), (event) => { `addListener` returns an unsubscribe function. There is exactly one listener entry point — there is no `relay.on`, `relay.notify`, or `relay.actions` namespace. +## Realtime fanout + +Where listeners get their events depends on the client: + +- **Agent-scoped clients** (from `register` / `reconnect` / harness `create`) stream over their own + connection — realtime works once connected and subscribed to the relevant channels. +- **Workspace-key clients** (from `createWorkspace` / `new AgentRelay({ workspaceKey })`) can receive all + workspace-visible events, but **fanout is off by default**. Until enabled, `addListener` is wired up yet + receives nothing (durable reads like `messages.list` / `inbox.get` still work). + +Enable fanout **once per workspace**: + +```ts +await relay.workspace.stream.set(true); // enable +// relay.workspace.stream.get() -> { enabled, defaultEnabled, override } +// relay.workspace.stream.inherit() -> clear the override +``` + +`addListener` on a workspace-key client with fanout off logs a one-time warning rather than failing +silently. Messages are durable regardless — realtime is the fast path, the stored record is the source of +truth. + ## The event object Every event is a discriminated union keyed on `type`. Listening to a specific name narrows the object in diff --git a/web/content/docs/messaging.mdx b/web/content/docs/messaging.mdx index 7364990d4..010c48485 100644 --- a/web/content/docs/messaging.mdx +++ b/web/content/docs/messaging.mdx @@ -163,6 +163,12 @@ const stop = relay.addListener('message.created', ({ message, envelope }) => { stop(); ``` +On a **workspace-key client** (`createWorkspace` / `new AgentRelay({ workspaceKey })`), realtime fanout is +**off by default** — enable it once with `await relay.workspace.stream.set(true)` or your listener receives +nothing over the wire (durable reads still work). Agent-scoped clients from `register`/`reconnect`/harness +`create` stream over their own connection and don't need this. See +[Events → Realtime fanout](/docs/events#realtime-fanout). + Common messaging event names include `message.created`, `message.reacted`, and `message.read`. See [Events](/docs/events) for the full vocabulary and envelope schema. diff --git a/web/content/docs/quickstart.mdx b/web/content/docs/quickstart.mdx index 89de4d843..61f7d5175 100644 --- a/web/content/docs/quickstart.mdx +++ b/web/content/docs/quickstart.mdx @@ -95,6 +95,16 @@ unsubscribe(); Listeners work across message, delivery, action, and harness-provided session events such as status changes and tool calls. See [Event handlers](/docs/event-handlers) and [Events](/docs/events). +To receive realtime events on a workspace-key client across processes, enable workspace-wide fanout once — +it is off by default: + +```ts +await relay.workspace.stream.set(true); +``` + +Agent-scoped clients (from `register`/`reconnect`/harness `create`) stream over their own connection and +don't need this. See [Events → Realtime fanout](/docs/events#realtime-fanout). + ## Register An Action Actions are fire-and-forget typed capabilities. Invoking returns an acknowledgement immediately; the handler