diff --git a/CHANGELOG.md b/CHANGELOG.md
index 22ebd2f1d..e91dfaba6 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,7 +5,11 @@ All notable changes to Agent Relay will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## [Unreleased - Patch]
+## [Unreleased - Minor]
+
+### Added
+
+- `@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
 
@@ -13,6 +17,8 @@ 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 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-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/sdk` messaging events map the canonical `message.reacted` WebSocket event onto `reactionAdded`/`reactionRemoved`; previously only the non-canonical `reaction.added`/`reaction.removed` names were handled, so reaction listeners never fired against current Relaycast engines.
 
 ## [10.4.0] - 2026-07-15
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<RelayCast> {
diff --git a/crates/broker/src/runtime/api.rs b/crates/broker/src/runtime/api.rs
index 32d2767c0..8c5346718 100644
--- a/crates/broker/src/runtime/api.rs
+++ b/crates/broker/src/runtime/api.rs
@@ -27,6 +27,9 @@ pub(crate) fn default_observer_token_scopes() -> Vec<ObserverScope> {
         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 b148ae3e5..b6388e19a 100644
--- a/crates/broker/src/runtime/tests.rs
+++ b/crates/broker/src/runtime/tests.rs
@@ -3414,14 +3414,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/package-lock.json b/package-lock.json
index c34f7c4d9..a3879f596 100644
--- a/package-lock.json
+++ b/package-lock.json
@@ -1,12 +1,12 @@
 {
   "name": "@agent-relay/monorepo",
-  "version": "10.3.0",
+  "version": "10.4.0",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "@agent-relay/monorepo",
-      "version": "10.3.0",
+      "version": "10.4.0",
       "license": "Apache-2.0",
       "workspaces": [
         "packages/*"
@@ -10249,44 +10249,44 @@
     },
     "packages/brand": {
       "name": "@agent-relay/brand",
-      "version": "10.3.0"
+      "version": "10.4.0"
     },
     "packages/broker-darwin-arm64": {
       "name": "@agent-relay/broker-darwin-arm64",
-      "version": "10.3.0",
+      "version": "10.4.0",
       "license": "MIT"
     },
     "packages/broker-darwin-x64": {
       "name": "@agent-relay/broker-darwin-x64",
-      "version": "10.3.0",
+      "version": "10.4.0",
       "license": "MIT"
     },
     "packages/broker-linux-arm64": {
       "name": "@agent-relay/broker-linux-arm64",
-      "version": "10.3.0",
+      "version": "10.4.0",
       "license": "MIT"
     },
     "packages/broker-linux-x64": {
       "name": "@agent-relay/broker-linux-x64",
-      "version": "10.3.0",
+      "version": "10.4.0",
       "license": "MIT"
     },
     "packages/broker-win32-x64": {
       "name": "@agent-relay/broker-win32-x64",
-      "version": "10.3.0",
+      "version": "10.4.0",
       "license": "MIT"
     },
     "packages/cli": {
       "name": "agent-relay",
-      "version": "10.3.0",
+      "version": "10.4.0",
       "license": "Apache-2.0",
       "dependencies": {
-        "@agent-relay/cloud": "10.3.0",
-        "@agent-relay/config": "10.3.0",
-        "@agent-relay/fleet": "10.3.0",
-        "@agent-relay/harness-driver": "10.3.0",
-        "@agent-relay/sdk": "10.3.0",
-        "@agent-relay/utils": "10.3.0",
+        "@agent-relay/cloud": "10.4.0",
+        "@agent-relay/config": "10.4.0",
+        "@agent-relay/fleet": "10.4.0",
+        "@agent-relay/harness-driver": "10.4.0",
+        "@agent-relay/sdk": "10.4.0",
+        "@agent-relay/utils": "10.4.0",
         "@modelcontextprotocol/sdk": "^1.0.0",
         "@relayfile/client": "^0.10.21",
         "@relayflows/cli": "^1.0.1",
@@ -10314,9 +10314,9 @@
     },
     "packages/cloud": {
       "name": "@agent-relay/cloud",
-      "version": "10.3.0",
+      "version": "10.4.0",
       "dependencies": {
-        "@agent-relay/config": "10.3.0",
+        "@agent-relay/config": "10.4.0",
         "@aws-sdk/client-s3": "3.1020.0",
         "ignore": "^7.0.5",
         "tar": "^7.5.10"
@@ -10332,7 +10332,7 @@
     },
     "packages/config": {
       "name": "@agent-relay/config",
-      "version": "10.3.0",
+      "version": "10.4.0",
       "dependencies": {
         "zod": "^3.23.8",
         "zod-to-json-schema": "^3.23.1"
@@ -10345,20 +10345,20 @@
     },
     "packages/evals": {
       "name": "@agent-relay/evals",
-      "version": "10.3.0",
+      "version": "10.4.0",
       "license": "Apache-2.0",
       "dependencies": {
-        "@agent-relay/harness-driver": "10.3.0",
-        "@agent-relay/integration-prompts": "10.3.0"
+        "@agent-relay/harness-driver": "10.4.0",
+        "@agent-relay/integration-prompts": "10.4.0"
       }
     },
     "packages/fleet": {
       "name": "@agent-relay/fleet",
-      "version": "10.3.0",
+      "version": "10.4.0",
       "license": "Apache-2.0",
       "dependencies": {
-        "@agent-relay/harness-driver": "10.3.0",
-        "@agent-relay/harnesses": "10.3.0",
+        "@agent-relay/harness-driver": "10.4.0",
+        "@agent-relay/harnesses": "10.4.0",
         "@relaycast/sdk": "^6.0.0",
         "ws": "^8.18.3",
         "zod": "^3.23.8"
@@ -10369,40 +10369,40 @@
     },
     "packages/harness-driver": {
       "name": "@agent-relay/harness-driver",
-      "version": "10.3.0",
+      "version": "10.4.0",
       "license": "Apache-2.0",
       "dependencies": {
-        "@agent-relay/sdk": "10.3.0",
+        "@agent-relay/sdk": "10.4.0",
         "ws": "^8.18.3",
         "zod": "^3.23.8"
       },
       "optionalDependencies": {
-        "@agent-relay/broker-darwin-arm64": "10.3.0",
-        "@agent-relay/broker-darwin-x64": "10.3.0",
-        "@agent-relay/broker-linux-arm64": "10.3.0",
-        "@agent-relay/broker-linux-x64": "10.3.0",
-        "@agent-relay/broker-win32-x64": "10.3.0"
+        "@agent-relay/broker-darwin-arm64": "10.4.0",
+        "@agent-relay/broker-darwin-x64": "10.4.0",
+        "@agent-relay/broker-linux-arm64": "10.4.0",
+        "@agent-relay/broker-linux-x64": "10.4.0",
+        "@agent-relay/broker-win32-x64": "10.4.0"
       }
     },
     "packages/harnesses": {
       "name": "@agent-relay/harnesses",
-      "version": "10.3.0",
+      "version": "10.4.0",
       "license": "Apache-2.0",
       "dependencies": {
-        "@agent-relay/harness-driver": "10.3.0",
-        "@agent-relay/sdk": "10.3.0"
+        "@agent-relay/harness-driver": "10.4.0",
+        "@agent-relay/sdk": "10.4.0"
       }
     },
     "packages/integration-prompts": {
       "name": "@agent-relay/integration-prompts",
-      "version": "10.3.0",
+      "version": "10.4.0",
       "license": "Apache-2.0"
     },
     "packages/policy": {
       "name": "@agent-relay/policy",
-      "version": "10.3.0",
+      "version": "10.4.0",
       "dependencies": {
-        "@agent-relay/config": "10.3.0"
+        "@agent-relay/config": "10.4.0"
       },
       "devDependencies": {
         "@types/node": "^22.19.3",
@@ -10411,7 +10411,7 @@
     },
     "packages/sdk": {
       "name": "@agent-relay/sdk",
-      "version": "10.3.0",
+      "version": "10.4.0",
       "dependencies": {
         "@relaycast/sdk": "^6.0.0",
         "@relaycast/types": "^6.0.0",
@@ -10432,9 +10432,9 @@
     },
     "packages/utils": {
       "name": "@agent-relay/utils",
-      "version": "10.3.0",
+      "version": "10.4.0",
       "dependencies": {
-        "@agent-relay/config": "10.3.0",
+        "@agent-relay/config": "10.4.0",
         "compare-versions": "^6.1.1"
       },
       "devDependencies": {
diff --git a/packages/sdk/package.json b/packages/sdk/package.json
index 0d7594798..518a73553 100644
--- a/packages/sdk/package.json
+++ b/packages/sdk/package.json
@@ -51,7 +51,7 @@
     "build": "node -e \"require('node:fs').rmSync('dist',{recursive:true,force:true})\" && npx tsc -p tsconfig.build.json",
     "build:full": "npm run build",
     "check": "tsc -p tsconfig.json --noEmit",
-    "test": "vitest run src/__tests__/agent-relay.test.ts src/__tests__/facade.test.ts src/__tests__/listeners.test.ts src/__tests__/integrations.test.ts src/__tests__/messaging.test.ts src/__tests__/delivery-actions.test.ts src/__tests__/relaycast-errors.test.ts src/__tests__/register-action-relay.test.ts src/__tests__/thin-client.test.ts src/__tests__/typed-action-handle.test.ts src/__tests__/webhooks.test.ts",
+    "test": "vitest run src/__tests__/agent-relay.test.ts src/__tests__/facade.test.ts src/__tests__/listeners.test.ts src/__tests__/integrations.test.ts src/__tests__/messaging.test.ts src/__tests__/delivery-actions.test.ts src/__tests__/relaycast-errors.test.ts src/__tests__/register-action-relay.test.ts src/__tests__/thin-client.test.ts src/__tests__/typed-action-handle.test.ts src/__tests__/webhooks.test.ts src/__tests__/event-fanin.test.ts src/__tests__/observer-source.test.ts",
     "test:types": "vitest run --typecheck.only --typecheck.tsconfig tsconfig.typetest.json",
     "prepack": "npm run build"
   },
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..367bea2a7
--- /dev/null
+++ b/packages/sdk/src/__tests__/event-fanin.test.ts
@@ -0,0 +1,311 @@
+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<string, Set<(event: RelayMessagingEvent) => 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 new Set([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<typeof vi.fn>;
+    disconnect: ReturnType<typeof vi.fn>;
+    subscribe: ReturnType<typeof vi.fn>;
+  };
+  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('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();
+    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<typeof createFakeEventsSurface>['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<string, ReturnType<typeof createFakeEventsSurface>>();
+    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..136430d19 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 new Set([type, 'any'])) {
+      for (const handler of handlers.get(key) ?? []) handler(event);
+    }
   };
   return { on, emit };
 }
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..53b392ad4
--- /dev/null
+++ b/packages/sdk/src/__tests__/observer-source.test.ts
@@ -0,0 +1,600 @@
+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<typeof import('@relaycast/sdk')>();
+  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<typeof vi.fn>;
+    disconnect: ReturnType<typeof vi.fn>;
+  } = {
+    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<string, unknown> {
+  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<string, unknown> {
+  return {
+    seq,
+    type: 'message.created',
+    channel_id: 'c1',
+    payload: liveFrame(messageId),
+    created_at: '2026-07-02T00:00:00Z',
+  };
+}
+
+function jsonResponse(events: Record<string, unknown>[], latestSeq: number, nextSince?: number) {
+  return {
+    ok: true,
+    status: 200,
+    json: async () => ({
+      ok: true,
+      data: {
+        events,
+        latest_seq: latestSeq,
+        ...(nextSince !== undefined ? { next_since: nextSince } : {}),
+      },
+    }),
+  } 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<string, unknown>[], 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<void> {
+  // 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<typeof createObserverEventSource>) {
+  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<RelayMessagingEvent, { type: 'messageCreated' }>).message.messageId);
+}
+
+afterEach(() => {
+  relaycastMocks.relayCast.mockReset();
+  vi.unstubAllGlobals();
+});
+
+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;
+    const gate = new Promise<void>((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<typeof vi.fn>).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<typeof vi.fn>).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<typeof vi.fn>).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('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<void>((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<Response>((_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')]);
+
+    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<typeof vi.fn>).mock.calls.map((call) =>
+      String(call[0])
+    );
+    expect(calls[calls.length - 1]).toContain('since=2');
+  });
+});
+
+describe('AgentRelay observer mode', () => {
+  function createObserverRelay(overrides: Record<string, unknown> = {}) {
+    // 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('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.
+    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();
+      options: unknown;
+      constructor(url: string, options?: unknown) {
+        this.url = url;
+        this.options = options;
+        sockets.push(this);
+      }
+    }
+    vi.stubGlobal('WebSocket', FakeWebSocket);
+    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();
+
+    expect(sockets).toHaveLength(1);
+    // 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)) });
+
+    expect(received).toHaveLength(2);
+  });
+});
diff --git a/packages/sdk/src/agent-relay.ts b/packages/sdk/src/agent-relay.ts
index c8ac50a35..254839cdd 100644
--- a/packages/sdk/src/agent-relay.ts
+++ b/packages/sdk/src/agent-relay.ts
@@ -7,8 +7,11 @@ import {
   type RelaycastTelemetryOptions,
 } from './relaycast-telemetry.js';
 import {
+  createEventFanIn,
+  createObserverEventSource,
   RelaycastMessagingClient,
   type RelayAgentRegistration,
+  type RelayEventFanIn,
   type RelayMessaging,
   type RelaycastMessagingOptions,
 } from './messaging/index.js';
@@ -48,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
@@ -100,25 +124,68 @@ 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<string, RelayMessaging>();
   private readonly createAgentMessaging: (token: string) => RelayMessaging;
+  private readonly eventFanIn: RelayEventFanIn;
   private enrichedMessages?: EnrichedMessages;
   private workspaceFacade?: RelayWorkspace;
   private hub?: ListenerHub;
   private readonly errorHooks = new Set<RelayErrorHook>();
 
   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 }));
+    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);
     }
@@ -179,7 +246,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),
       });
     }
@@ -212,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;
   }
@@ -325,6 +406,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;
   }
@@ -337,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, unknown>): string | undefined {
   const data =
     payload.data && typeof payload.data === 'object' ? (payload.data as Record<string, unknown>) : {};
diff --git a/packages/sdk/src/listeners.ts b/packages/sdk/src/listeners.ts
index 6314a1eab..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);
 }
 
@@ -648,12 +654,18 @@ 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',
+        operation: 'connect',
+        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..3ea87050d
--- /dev/null
+++ b/packages/sdk/src/messaging/event-fanin.ts
@@ -0,0 +1,361 @@
+import type { RelayMessagingEvent, RelayMessagingEventMap, RelayMessagingEventsSurface } from './types.js';
+
+/**
+ * Options for {@link createEventFanIn}.
+ */
+export interface EventFanInOptions {
+  /**
+   * 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. */
+  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;
+  }
+}
+
+/** One observed occurrence: when it was first seen and which sources delivered it. */
+interface OccurrenceRecord {
+  at: number;
+  sources: Set<RelayMessagingEventsSurface>;
+}
+
+/**
+ * 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<string, Set<(event: RelayMessagingEvent) => void | Promise<void>>>();
+  const sources: RelayMessagingEventsSurface[] = [];
+  const seenSources = new Set<RelayMessagingEventsSurface>();
+  /** Active `on('any', ...)` forwarding per source, so disconnect can detach. */
+  const sourceForwarding = new Map<RelayMessagingEventsSurface, () => void>();
+  const desiredChannels = new Set<string>();
+  /** Dedupe keys → current occurrence, insertion-ordered for eviction. */
+  const seenEvents = new Map<string, OccurrenceRecord>();
+
+  let connectRequested = false;
+  let fallbackForwarding: (() => void) | undefined;
+  let fallbackConnected = false;
+  let noSourceTimer: ReturnType<typeof setTimeout> | 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);
+        }
+      }
+    }
+  };
+
+  /**
+   * 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 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, sources: new Set([source]) });
+      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 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 {
+        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', (event) => forward(fallback, event));
+    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;
+      if (typeof fallback?.disconnect === 'function') {
+        try {
+          void fallback.disconnect().catch(() => {});
+        } catch {
+          // Fallback surfaces whose disconnect misbehaves are 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);
+      attachSourceForwarding(source);
+      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) {
+          // Re-attach forwarding dropped by a prior disconnect().
+          attachSourceForwarding(source);
+          connectSource(source);
+        }
+        return;
+      }
+      attachFallback();
+      scheduleNoSourceWarning();
+    },
+
+    disconnect: async () => {
+      connectRequested = false;
+      clearNoSourceTimer();
+      detachFallback();
+      // 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) {
+          report(error);
+        }
+      }
+    },
+
+    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) {
+          report(error);
+        }
+      }
+    },
+
+    on: <K extends keyof RelayMessagingEventMap>(
+      event: K,
+      handler: (...args: RelayMessagingEventMap[K]) => void | Promise<void>
+    ): (() => void) => {
+      const set = handlers.get(event) ?? new Set();
+      set.add(handler as (event: RelayMessagingEvent) => void | Promise<void>);
+      handlers.set(event, set);
+      return () => {
+        set.delete(handler as (event: RelayMessagingEvent) => void | Promise<void>);
+      };
+    },
+  };
+}
diff --git a/packages/sdk/src/messaging/index.ts b/packages/sdk/src/messaging/index.ts
index 4834085a6..b2513de6d 100644
--- a/packages/sdk/src/messaging/index.ts
+++ b/packages/sdk/src/messaging/index.ts
@@ -1,5 +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/normalize.ts b/packages/sdk/src/messaging/normalize.ts
index 8de35e29e..96a3703e1 100644
--- a/packages/sdk/src/messaging/normalize.ts
+++ b/packages/sdk/src/messaging/normalize.ts
@@ -747,7 +747,11 @@ export function normalizeMessagingEvent(input: unknown): RelayMessagingEvent {
         agentName: str(record, 'agent_name') ?? '',
         readAt: opt(str(record, 'read_at')),
       });
-    // Canonical reaction event plus the legacy split pair.
+    // Canonical reaction event plus the legacy split pair. The engine's raw
+    // workspace-stream frames and the durable event log (observer mode) carry
+    // reactions as a single `message.reacted` type with an `action` field;
+    // higher-level clients split it into `reaction.added`/`reaction.removed`
+    // before we see it. All three land here.
     case 'message.reacted':
     case 'reaction.added':
     case 'reaction.removed':
diff --git a/packages/sdk/src/messaging/observer-source.ts b/packages/sdk/src/messaging/observer-source.ts
new file mode 100644
index 000000000..56faac29a
--- /dev/null
+++ b/packages/sdk/src/messaging/observer-source.ts
@@ -0,0 +1,600 @@
+import { normalizeMessagingEvent } from './normalize.js';
+import type { RelayMessagingEvent, RelayMessagingEventMap, RelayMessagingEventsSurface } from './types.js';
+
+/**
+ * The slice of the live observer stream the source depends on. The default
+ * 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.
+ */
+export interface ObserverLiveStream {
+  connect(): void;
+  disconnect(): 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;
+  };
+}
+
+/**
+ * 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
+ * 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 = `${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 >= 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.
+ * 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,
+  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;
+  let timer: ReturnType<typeof setTimeout> | 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;
+  /** Consecutive header-auth attempts that closed before opening (browser probe). */
+  let headerCloseStreak = 0;
+
+  const scheduleReconnect = (delayOverrideMs?: number): void => {
+    if (closed || timer !== undefined) return;
+    attempts += 1;
+    const delay = delayOverrideMs ?? Math.min(30_000, 1_000 * 2 ** Math.min(attempts - 1, 5));
+    timer = setTimeout(() => {
+      timer = undefined;
+      open();
+    }, delay);
+    (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 = webSocketImpl ?? (globalThis as { WebSocket?: typeof WebSocket }).WebSocket;
+    if (!WebSocketImpl) {
+      report(
+        new Error(
+          '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;
+    }
+    let ws: WebSocket;
+    try {
+      ws = construct(WebSocketImpl);
+    } catch (error) {
+      report(error);
+      scheduleReconnect();
+      return;
+    }
+    let openedHere = false;
+    socket = ws;
+    ws.onopen = () => {
+      attempts = 0;
+      everOpened = true;
+      openedHere = true;
+      headerCloseStreak = 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 = (event?: { code?: number; reason?: string; wasClean?: boolean }) => {
+      if (socket === ws) socket = undefined;
+      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) {
+        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 = () => {
+      // Abnormal closures are reported in onclose; onerror carries no extra detail.
+    };
+  };
+
+  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);
+      },
+    },
+  };
+}
+
+/**
+ * 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 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 {
+  seq: number;
+  payload: unknown;
+}
+
+type UnknownRecord = Record<string, unknown>;
+
+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;
+  nextSince: number | undefined;
+} {
+  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;
+  // 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 };
+}
+
+/**
+ * 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 = 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.webSocketImpl));
+
+  const handlers = new Map<string, Set<(event: RelayMessagingEvent) => void | Promise<void>>>();
+
+  /** 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;
+  let offOpen: (() => 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 = [];
+    // 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 (
+    since: number
+  ): Promise<
+    { events: BackfillEventRow[]; latestSeq: number; nextSince: number | undefined } | undefined
+  > => {
+    const url = `${baseUrl}/v1/workspace/events?since=${since}&limit=${pageSize}`;
+    // 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;
+    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<void> => {
+    const startedEpoch = epoch;
+    try {
+      for (;;) {
+        const before = cursor;
+        const page = await backfillPage(cursor);
+        if (epoch !== startedEpoch) return;
+        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, 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;
+      // 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);
+        // 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;
+        offOpen = live.on.open?.(() => {
+          if (hadOpen) {
+            backfillDone = false;
+            void runBackfill();
+          }
+          hadOpen = true;
+        });
+        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.
+      }
+      void runBackfill();
+    },
+
+    disconnect: async (): Promise<void> => {
+      epoch += 1;
+      backfillDone = false;
+      pending = [];
+      offLive?.();
+      offLive = undefined;
+      offOpen?.();
+      offOpen = 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: <K extends keyof RelayMessagingEventMap>(
+      event: K,
+      handler: (...args: RelayMessagingEventMap[K]) => void | Promise<void>
+    ): (() => void) => {
+      const set = handlers.get(event) ?? new Set();
+      set.add(handler as (event: RelayMessagingEvent) => void | Promise<void>);
+      handlers.set(event, set);
+      return () => {
+        set.delete(handler as (event: RelayMessagingEvent) => void | Promise<void>);
+      };
+    },
+  };
+}
diff --git a/specs/observer-plane.md b/specs/observer-plane.md
new file mode 100644
index 000000000..b8447d4ff
--- /dev/null
+++ b/specs/observer-plane.md
@@ -0,0 +1,209 @@
+# 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=<seq>&limit=<n>   (n <= 500)
+Authorization: Bearer <observer token>
+```
+
+Response:
+
+```json
+{
+  "ok": true,
+  "data": {
+    "events": [
+      { "seq": 41, "type": "message.created", "channel_id": "c1", "payload": { ... }, "created_at": "..." }
+    ],
+    "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
+  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=<ot_live_...>` 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=<cursor>`
+   until `latest_seq` is reached, emitting each event and advancing the
+   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
+   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.