diff --git a/public/flags.json b/public/flags.json
index f8832b6..bebb588 100644
--- a/public/flags.json
+++ b/public/flags.json
@@ -1,5 +1,6 @@
 {
   "flags": {
-    "realtime-oracle": false
+    "realtime-oracle": false,
+    "realtime-pool-stats": false
   }
 }
diff --git a/src/hooks/usePoolStats.ts b/src/hooks/usePoolStats.ts
index e69de29..25ee62a 100644
--- a/src/hooks/usePoolStats.ts
+++ b/src/hooks/usePoolStats.ts
@@ -0,0 +1,111 @@
+"use client";
+
+import { useEffect, useState } from "react";
+import { useQuery, useQueryClient } from "@tanstack/react-query";
+import { ApiUnreachableError } from "@/lib/api/client";
+import type { DataSource } from "@/lib/api/dataSource";
+import { fetchPoolStats, type PoolStats } from "@/lib/api/pool";
+import { FIXTURE_POOL_STATS } from "@/lib/fixtures/poolStats";
+import { useFeatureFlag } from "@/lib/flags/useFeatureFlag";
+import { realtimeClient } from "@/lib/realtime/client";
+import { readPoolStats } from "@/lib/stellar/rpc";
+
+interface PoolStatsState {
+  data: PoolStats | null;
+  loading: boolean;
+  error: string | null;
+  isFixture: boolean;
+  dataSource: DataSource | null;
+  /** True when the live stream is enabled but disconnected, so `data` is the REST snapshot. */
+  isStale: boolean;
+}
+
+interface PoolStatsResult {
+  data: PoolStats | null;
+  dataSource: DataSource | null;
+  error: string | null;
+}
+
+const NUMERIC_FIELDS: (keyof PoolStats)[] = ["utilizationBps", "apyBps", "sharePrice", "maxUtilizationBps"];
+const STRING_FIELDS: (keyof PoolStats)[] = ["totalUsdc", "totalShares", "lockedUsdc", "premiumAccrued", "availableUsdc"];
+/** A pushed TVL differing from REST by more than this ratio is treated as a bad frame. */
+const MAX_TVL_DRIFT = 0.5;
+
+/** Pushed frames are full snapshots; reject anything partial rather than half-applying it. */
+function isFullSnapshot(p: unknown): p is PoolStats {
+  const o = p as Record<string, unknown> | null;
+  return (
+    !!o &&
+    NUMERIC_FIELDS.every((k) => typeof o[k] === "number" && Number.isFinite(o[k] as number)) &&
+    STRING_FIELDS.every((k) => typeof o[k] === "string" && /^\d+$/.test(o[k] as string))
+  );
+}
+
+function agreesWithRest(pushed: PoolStats, rest: PoolStats | null): boolean {
+  const base = Number(rest?.totalUsdc ?? 0);
+  if (!rest || base === 0) return true;
+  return Math.abs(Number(pushed.totalUsdc) - base) / base <= MAX_TVL_DRIFT;
+}
+
+/**
+ * Pool-wide stats: API → on-chain read → bundled fixture, optionally kept live
+ * by the `pool.stats` realtime topic (behind the `realtime-pool-stats` flag).
+ */
+export function usePoolStats(): PoolStatsState {
+  const live = useFeatureFlag("realtime-pool-stats");
+  const queryClient = useQueryClient();
+  const [pushed, setPushed] = useState<PoolStats | null>(null);
+  const [connected, setConnected] = useState(realtimeClient.getStatus() === "open");
+
+  const { data: result, isLoading } = useQuery<PoolStatsResult>({
+    queryKey: ["poolStats"],
+    queryFn: async ({ signal }) => {
+      try {
+        return { data: await fetchPoolStats(signal), dataSource: "api", error: null };
+      } catch (err) {
+        if (signal?.aborted) throw err;
+        if (!(err instanceof ApiUnreachableError)) {
+          return { data: null, dataSource: null, error: err instanceof Error ? err.message : "Failed to load pool stats" };
+        }
+        try {
+          return { data: await readPoolStats(signal), dataSource: "chain", error: null };
+        } catch {
+          return { data: FIXTURE_POOL_STATS, dataSource: "fixture", error: null };
+        }
+      }
+    },
+  });
+
+  const rest = result?.data ?? null;
+
+  useEffect(() => {
+    if (!live) return;
+    setConnected(realtimeClient.getStatus() === "open");
+    const offStatus = realtimeClient.onStatus((s) => setConnected(s === "open"));
+    const off = realtimeClient.subscribe("pool.stats", (payload) => {
+      if (!isFullSnapshot(payload)) return;
+      const current = queryClient.getQueryData<PoolStatsResult>(["poolStats"])?.data ?? null;
+      if (agreesWithRest(payload, current)) setPushed(payload);
+      else {
+        setPushed(null);
+        void queryClient.invalidateQueries({ queryKey: ["poolStats"] });
+      }
+    });
+    return () => {
+      off();
+      offStatus();
+    };
+  }, [live, queryClient]);
+
+  // On disconnect fall back to the REST snapshot.
+  const useLive = live && connected && pushed !== null;
+
+  return {
+    data: useLive ? pushed : rest,
+    loading: isLoading,
+    error: result?.error ?? null,
+    isFixture: result?.dataSource === "fixture",
+    dataSource: result?.dataSource ?? null,
+    isStale: live && !connected,
+  };
+}
diff --git a/src/lib/flags/flagsConfig.ts b/src/lib/flags/flagsConfig.ts
index 2ecb795..f553be3 100644
--- a/src/lib/flags/flagsConfig.ts
+++ b/src/lib/flags/flagsConfig.ts
@@ -5,6 +5,7 @@ export interface FlagsDocument {
 /** Safe default if the remote flags source is unreachable — new/risky features default off. */
 export const DEFAULT_FLAGS: Record<string, boolean> = {
   "realtime-oracle": false,
+  "realtime-pool-stats": false,
 };
 
 export const FLAGS_ENDPOINT = "/flags.json";
diff --git a/src/lib/realtime/client.ts b/src/lib/realtime/client.ts
index 8be8d3c..ce52877 100644
--- a/src/lib/realtime/client.ts
+++ b/src/lib/realtime/client.ts
@@ -1,59 +1,155 @@
 import { API_BASE_URL } from "@/lib/api/client";
+import { Coalescer, TopicSequencer, type Envelope } from "./merge";
+import { SseTransport } from "./sse";
+import { TabLeader } from "./tabs";
+import { WebSocketTransport, type Transport, type TransportKind } from "./transport";
 
 type Handler = (payload: unknown) => void;
+type Status = "idle" | "connecting" | "open" | "closed";
 
 const MAX_BACKOFF_MS = 30_000;
 const BASE_BACKOFF_MS = 1_000;
+const GAP_TIMEOUT_MS = 2_000;
 
 /**
- * A small reconnecting WebSocket wrapper exposing a subscribe-by-topic API.
+ * A small reconnecting realtime wrapper exposing a subscribe-by-topic API.
  * The backend contract this assumes (documented, not yet implemented
- * upstream): connect to `${wsUrl}`, receive JSON messages shaped
- * `{ topic: string, payload: unknown }`, one topic per push-based update
- * type (e.g. "oracle_update"). Callers must treat this connection as
- * best-effort — always keep an independent REST fetch as the source of
- * truth for initial state.
+ * upstream): JSON messages shaped
+ * `{ topic: string, seq?: number, kind?: "replace" | "append", payload: unknown }`.
+ * WebSocket is preferred; if it never opens, the client falls back to SSE
+ * (outbound `subscribe`/`resync` messages are POSTed since SSE is read-only).
+ * Only one browser tab holds the connection; others receive its messages over
+ * BroadcastChannel. Callers must treat this connection as best-effort — always
+ * keep an independent REST fetch as the source of truth for initial state.
  */
 class RealtimeClient {
-  private socket: WebSocket | null = null;
+  private transport: Transport | null = null;
   private handlers = new Map<string, Set<Handler>>();
+  private sequencers = new Map<string, TopicSequencer>();
+  private gapTimers = new Map<string, ReturnType<typeof setTimeout>>();
+  private coalescer = new Coalescer((topic, payload) => this.handlers.get(topic)?.forEach((h) => h(payload)));
+  private statusListeners = new Set<(status: Status) => void>();
+  private status: Status = "idle";
   private attempt = 0;
+  private everOpened = false;
+  private useSse = false;
   private closedByUser = false;
+  private started = false;
   private reconnectTimer: ReturnType<typeof setTimeout> | null = null;
+  private leader: TabLeader | null = null;
+  private subscribed = new Set<string>();
+
+  get activeTransport(): TransportKind | null {
+    return this.transport?.kind ?? null;
+  }
+
+  getStatus(): Status {
+    return this.status;
+  }
+
+  onStatus(listener: (status: Status) => void): () => void {
+    this.statusListeners.add(listener);
+    return () => this.statusListeners.delete(listener);
+  }
+
+  private setStatus(status: Status): void {
+    if (this.status === status) return;
+    this.status = status;
+    this.statusListeners.forEach((l) => l(status));
+  }
 
   connect(): void {
-    if (typeof WebSocket === "undefined" || this.socket) return;
+    if (typeof WebSocket === "undefined" || this.started) return;
+    this.started = true;
     this.closedByUser = false;
-    this.open();
+    this.leader = new TabLeader(
+      (isLeader) => (isLeader ? this.open() : this.closeTransport()),
+      (raw) => this.ingest(raw),
+      () => this.syncTopics()
+    );
+    this.leader.start();
+  }
+
+  private baseUrl(): string {
+    return API_BASE_URL.replace(/^http/, "ws").replace(/\/api\/v1$/, "/ws");
   }
 
   private open(): void {
-    const wsUrl = API_BASE_URL.replace(/^http/, "ws").replace(/\/api\/v1$/, "/ws");
+    if (this.transport || !this.leader?.isLeader) return;
+    const wsUrl = this.baseUrl();
+    const httpUrl = API_BASE_URL.replace(/\/api\/v1$/, "");
+    this.transport = this.useSse
+      ? new SseTransport(`${httpUrl}/events`, `${httpUrl}/events/subscribe`)
+      : new WebSocketTransport(wsUrl);
+    this.setStatus("connecting");
+    const transport = this.transport;
+    transport.connect({
+      onOpen: () => {
+        this.attempt = 0;
+        this.subscribed.clear();
+        this.everOpened = true;
+        this.setStatus("open");
+        this.syncTopics();
+      },
+      onMessage: (raw) => {
+        this.leader?.broadcast(raw);
+        this.ingest(raw);
+      },
+      onClose: () => {
+        if (this.transport !== transport) return;
+        this.transport = null;
+        this.setStatus("closed");
+        // WebSocket never established (blocked proxy): switch to SSE.
+        if (!this.everOpened && !this.useSse && typeof EventSource !== "undefined") this.useSse = true;
+        if (!this.closedByUser) this.scheduleReconnect();
+      },
+    });
+  }
+
+  private closeTransport(): void {
+    const t = this.transport;
+    this.transport = null;
+    t?.close();
+  }
+
+  private ingest(raw: string): void {
+    let env: Envelope;
     try {
-      this.socket = new WebSocket(wsUrl);
+      env = JSON.parse(raw) as Envelope;
     } catch {
-      this.scheduleReconnect();
-      return;
+      return; // Malformed message — ignore, don't crash the connection.
     }
+    if (typeof env?.topic !== "string") return;
 
-    this.socket.addEventListener("open", () => {
-      this.attempt = 0;
-    });
-    this.socket.addEventListener("message", (event) => {
-      try {
-        const { topic, payload } = JSON.parse(event.data as string) as { topic: string; payload: unknown };
-        this.handlers.get(topic)?.forEach((handler) => handler(payload));
-      } catch {
-        // Malformed message — ignore, don't crash the connection.
-      }
-    });
-    this.socket.addEventListener("close", () => {
-      this.socket = null;
-      if (!this.closedByUser) this.scheduleReconnect();
-    });
-    this.socket.addEventListener("error", () => {
-      this.socket?.close();
-    });
+    let seqr = this.sequencers.get(env.topic);
+    if (!seqr) {
+      seqr = new TopicSequencer((fromSeq, toSeq) => this.transport?.send({ type: "resync", topic: env.topic, fromSeq, toSeq }));
+      this.sequencers.set(env.topic, seqr);
+    }
+    const ready = seqr.push(env);
+    ready.forEach((e) => this.coalescer.add(e));
+
+    // A hole is outstanding if nothing was released for an out-of-order arrival: resolve it after a timeout.
+    clearTimeout(this.gapTimers.get(env.topic));
+    if (env.seq !== undefined && !ready.length) {
+      this.gapTimers.set(
+        env.topic,
+        setTimeout(() => {
+          seqr!.flush().forEach((e) => this.coalescer.add(e));
+        }, GAP_TIMEOUT_MS)
+      );
+    }
+  }
+
+  /** Topics needed by this tab or any other tab; the leader subscribes to the union. */
+  private syncTopics(): void {
+    const own = [...this.handlers.keys()].filter((t) => this.handlers.get(t)!.size > 0);
+    this.leader?.announceTopics(own);
+    if (!this.leader?.isLeader) return;
+    const all = new Set([...own, ...this.leader.remoteTopics()]);
+    all.forEach((topic) => !this.subscribed.has(topic) && this.transport?.send({ type: "subscribe", topic }));
+    this.subscribed.forEach((topic) => !all.has(topic) && this.transport?.send({ type: "unsubscribe", topic }));
+    this.subscribed = all;
   }
 
   private scheduleReconnect(): void {
@@ -67,20 +163,29 @@ class RealtimeClient {
   }
 
   subscribe(topic: string, handler: Handler): () => void {
+    const isNew = !this.handlers.get(topic)?.size;
     if (!this.handlers.has(topic)) this.handlers.set(topic, new Set());
     this.handlers.get(topic)!.add(handler);
     this.connect();
+    if (isNew) this.syncTopics();
 
     return () => {
       this.handlers.get(topic)?.delete(handler);
+      if (!this.handlers.get(topic)?.size) this.syncTopics();
     };
   }
 
   disconnect(): void {
     this.closedByUser = true;
+    this.started = false;
     if (this.reconnectTimer) clearTimeout(this.reconnectTimer);
-    this.socket?.close();
-    this.socket = null;
+    this.reconnectTimer = null;
+    this.gapTimers.forEach(clearTimeout);
+    this.gapTimers.clear();
+    this.leader?.stop();
+    this.leader = null;
+    this.closeTransport();
+    this.setStatus("closed");
   }
 }
 
diff --git a/src/lib/realtime/merge.ts b/src/lib/realtime/merge.ts
new file mode 100644
index 0000000..8918269
--- /dev/null
+++ b/src/lib/realtime/merge.ts
@@ -0,0 +1,100 @@
+export type EventKind = "replace" | "append";
+
+/** Wire envelope. `seq` is per-topic and monotonic; `kind` defaults to "replace". */
+export interface Envelope {
+  topic: string;
+  seq?: number;
+  kind?: EventKind;
+  payload: unknown;
+}
+
+export const REORDER_WINDOW = 32;
+export const MAX_APPEND_BUFFER = 5_000;
+
+/**
+ * Per-topic ordering: delivers envelopes in `seq` order, buffers out-of-order
+ * arrivals up to `window`, drops duplicates, and reports a gap (via `onGap`)
+ * once the window overflows or `flush()` is called with a hole outstanding.
+ */
+export class TopicSequencer {
+  private last: number | null = null;
+  private buffer = new Map<number, Envelope>();
+
+  constructor(
+    private readonly onGap: (fromSeq: number, toSeq: number) => void,
+    private readonly window = REORDER_WINDOW
+  ) {}
+
+  push(env: Envelope): Envelope[] {
+    const seq = env.seq;
+    if (seq === undefined) return [env];
+    if (this.last === null) this.last = seq - 1;
+    if (seq <= this.last) return []; // duplicate / already delivered
+    this.buffer.set(seq, env);
+    const out = this.drain();
+    if (this.buffer.size > this.window) out.push(...this.skipGap());
+    return out;
+  }
+
+  /** Force-resolve an outstanding hole (e.g. after a timeout): flags the gap and releases what is buffered. */
+  flush(): Envelope[] {
+    return this.buffer.size ? this.skipGap() : [];
+  }
+
+  private drain(): Envelope[] {
+    const out: Envelope[] = [];
+    for (let next = this.last! + 1, e = this.buffer.get(next); e; next += 1, e = this.buffer.get(next)) {
+      this.buffer.delete(next);
+      this.last = next;
+      out.push(e);
+    }
+    return out;
+  }
+
+  private skipGap(): Envelope[] {
+    const lowest = Math.min(...this.buffer.keys());
+    this.onGap(this.last! + 1, lowest - 1);
+    this.last = lowest - 1;
+    return this.drain();
+  }
+}
+
+/**
+ * Bounds how often consumers are notified. "replace" events keep only the
+ * latest per topic (drop-oldest); "append" events are always queued in order
+ * and never dropped.
+ */
+export class Coalescer {
+  private replace = new Map<string, unknown>();
+  private append = new Map<string, unknown[]>();
+  private timer: ReturnType<typeof setTimeout> | null = null;
+
+  constructor(
+    private readonly deliver: (topic: string, payload: unknown) => void,
+    private readonly intervalMs = 50
+  ) {}
+
+  add(env: Envelope): void {
+    if (env.kind === "append") {
+      const q = this.append.get(env.topic) ?? [];
+      q.push(env.payload);
+      this.append.set(env.topic, q);
+      // Backpressure: a slow consumer gets an early flush rather than losing events.
+      if (q.length >= MAX_APPEND_BUFFER) return this.flush();
+    } else {
+      this.replace.set(env.topic, env.payload);
+    }
+    this.timer ??= setTimeout(() => this.flush(), this.intervalMs);
+  }
+
+  flush(): void {
+    if (this.timer) clearTimeout(this.timer);
+    this.timer = null;
+    const appends = this.append;
+    const replaces = this.replace;
+    this.append = new Map();
+    this.replace = new Map();
+    appends.forEach((q, topic) => q.forEach((p) => this.deliver(topic, p)));
+    replaces.forEach((p, topic) => this.deliver(topic, p));
+  }
+}
diff --git a/src/lib/realtime/realtime.test.ts b/src/lib/realtime/realtime.test.ts
new file mode 100644
index 0000000..e5f9ffd
--- /dev/null
+++ b/src/lib/realtime/realtime.test.ts
@@ -0,0 +1,79 @@
+import { describe, it, expect, vi, afterEach } from "vitest";
+import { Coalescer, TopicSequencer, type Envelope } from "./merge";
+import { SseTransport } from "./sse";
+
+const env = (seq: number, kind: Envelope["kind"] = "replace"): Envelope => ({ topic: "t", seq, kind, payload: seq });
+
+describe("TopicSequencer", () => {
+  it("reorders within the window and drops duplicates", () => {
+    const gap = vi.fn();
+    const s = new TopicSequencer(gap, 8);
+    expect(s.push(env(1)).map((e) => e.seq)).toEqual([1]);
+    expect(s.push(env(3))).toEqual([]);
+    expect(s.push(env(2)).map((e) => e.seq)).toEqual([2, 3]);
+    expect(s.push(env(2))).toEqual([]);
+    expect(gap).not.toHaveBeenCalled();
+  });
+
+  it("flags a gap when the window overflows", () => {
+    const gap = vi.fn();
+    const s = new TopicSequencer(gap, 2);
+    s.push(env(1));
+    s.push(env(3));
+    s.push(env(4));
+    const out = s.push(env(5));
+    expect(gap).toHaveBeenCalledWith(2, 2);
+    expect(out.map((e) => e.seq)).toEqual([3, 4, 5]);
+  });
+
+  it("flush() resolves an outstanding hole", () => {
+    const gap = vi.fn();
+    const s = new TopicSequencer(gap);
+    s.push(env(1));
+    s.push(env(4));
+    expect(s.flush().map((e) => e.seq)).toEqual([4]);
+    expect(gap).toHaveBeenCalledWith(2, 3);
+  });
+});
+
+describe("Coalescer", () => {
+  afterEach(() => vi.useRealTimers());
+
+  it("keeps latest replace event but never drops append events", () => {
+    vi.useFakeTimers();
+    const deliver = vi.fn();
+    const c = new Coalescer(deliver, 50);
+    for (let i = 0; i < 5_000; i++) c.add({ topic: "oracle", kind: "replace", payload: i });
+    for (let i = 0; i < 1_000; i++) c.add({ topic: "claims", kind: "append", payload: i });
+    vi.advanceTimersByTime(60);
+    const oracle = deliver.mock.calls.filter(([t]) => t === "oracle");
+    const claims = deliver.mock.calls.filter(([t]) => t === "claims");
+    expect(oracle).toEqual([["oracle", 4_999]]);
+    expect(claims).toHaveLength(1_000);
+  });
+});
+
+describe("SseTransport", () => {
+  afterEach(() => vi.unstubAllGlobals());
+
+  it("passes the token in the query and closes on error so the client owns retry", () => {
+    const close = vi.fn();
+    const instances: { url: string; onerror?: () => void }[] = [];
+    vi.stubGlobal(
+      "EventSource",
+      class {
+        onerror?: () => void;
+        close = close;
+        constructor(public url: string) {
+          instances.push(this);
+        }
+      }
+    );
+    const onClose = vi.fn();
+    new SseTransport("http://x/events", "http://x/sub", () => "tok").connect({ onOpen: vi.fn(), onMessage: vi.fn(), onClose });
+    expect(instances[0].url).toBe("http://x/events?token=tok");
+    instances[0].onerror?.();
+    expect(close).toHaveBeenCalled();
+    expect(onClose).toHaveBeenCalled();
+  });
+});
diff --git a/src/lib/realtime/sse.ts b/src/lib/realtime/sse.ts
new file mode 100644
index 0000000..fcf9ef4
--- /dev/null
+++ b/src/lib/realtime/sse.ts
@@ -0,0 +1,52 @@
+import type { Transport, TransportHandlers } from "./transport";
+
+/**
+ * Server-Sent Events transport. SSE is unidirectional, so outbound control
+ * messages (subscribe / resync) go through a separate REST POST to
+ * `${postUrl}`; the stream itself is `${streamUrl}`. Auth is a `token` query
+ * param because EventSource cannot set headers.
+ */
+export class SseTransport implements Transport {
+  readonly kind = "sse" as const;
+  private source: EventSource | null = null;
+
+  constructor(
+    private readonly streamUrl: string,
+    private readonly postUrl: string,
+    private readonly getToken: () => string | null = () => null
+  ) {}
+
+  connect({ onOpen, onMessage, onClose }: TransportHandlers): void {
+    if (typeof EventSource === "undefined") {
+      onClose();
+      return;
+    }
+    const token = this.getToken();
+    const url = token ? `${this.streamUrl}${this.streamUrl.includes("?") ? "&" : "?"}token=${encodeURIComponent(token)}` : this.streamUrl;
+    this.source = new EventSource(url);
+    this.source.onopen = onOpen;
+    this.source.onmessage = (e) => onMessage(e.data as string);
+    // EventSource auto-reconnects natively; close it so the client's own backoff is the only retry logic.
+    this.source.onerror = () => {
+      this.source?.close();
+      this.source = null;
+      onClose();
+    };
+  }
+
+  send(message: object): void {
+    const token = this.getToken();
+    void fetch(this.postUrl, {
+      method: "POST",
+      headers: { "Content-Type": "application/json", ...(token ? { Authorization: `Bearer ${token}` } : {}) },
+      body: JSON.stringify(message),
+    }).catch(() => {
+      // Best-effort: the next reconnect re-sends subscriptions.
+    });
+  }
+
+  close(): void {
+    this.source?.close();
+    this.source = null;
+  }
+}
diff --git a/src/lib/realtime/tabs.ts b/src/lib/realtime/tabs.ts
new file mode 100644
index 0000000..0cfed8d
--- /dev/null
+++ b/src/lib/realtime/tabs.ts
@@ -0,0 +1,95 @@
+const CHANNEL = "refract-realtime";
+const HEARTBEAT_MS = 2_000;
+const LEADER_TIMEOUT_MS = 6_000;
+
+type TabMessage =
+  | { t: "hb"; id: string }
+  | { t: "bye"; id: string }
+  | { t: "topics"; id: string; topics: string[] }
+  | { t: "data"; raw: string };
+
+/**
+ * Leader election over BroadcastChannel so only one tab holds the socket.
+ * The lowest tab id wins a conflict; a follower promotes itself when the
+ * leader's heartbeat stops (closed or crashed tab). Without BroadcastChannel
+ * every tab is its own leader (the pre-existing behaviour).
+ */
+export class TabLeader {
+  readonly id = Math.random().toString(36).slice(2);
+  isLeader = false;
+  private channel: BroadcastChannel | null = null;
+  private lastHeartbeat = 0;
+  private hbTimer: ReturnType<typeof setInterval> | null = null;
+  private tabTopics = new Map<string, Set<string>>();
+
+  constructor(
+    private readonly onRole: (isLeader: boolean) => void,
+    private readonly onData: (raw: string) => void,
+    private readonly onTopicsChanged: () => void
+  ) {}
+
+  start(): void {
+    if (typeof BroadcastChannel === "undefined") return this.setLeader(true);
+    this.channel = new BroadcastChannel(CHANNEL);
+    this.channel.onmessage = (e: MessageEvent<TabMessage>) => this.receive(e.data);
+    this.hbTimer = setInterval(() => this.tick(), HEARTBEAT_MS);
+    // First tab claims leadership after a short listen window.
+    setTimeout(() => this.tick(), 300);
+    if (typeof window !== "undefined") window.addEventListener("pagehide", () => this.stop());
+  }
+
+  private tick(): void {
+    if (this.isLeader) this.post({ t: "hb", id: this.id });
+    else if (Date.now() - this.lastHeartbeat > (this.lastHeartbeat === 0 ? 0 : LEADER_TIMEOUT_MS)) this.setLeader(true);
+  }
+
+  private receive(msg: TabMessage): void {
+    if (msg.t === "hb") {
+      this.lastHeartbeat = Date.now();
+      if (this.isLeader && msg.id < this.id) this.setLeader(false);
+    } else if (msg.t === "bye") {
+      this.tabTopics.delete(msg.id);
+      if (!this.isLeader) this.lastHeartbeat = 0; // promote promptly if the leader left
+      this.onTopicsChanged();
+    } else if (msg.t === "topics") {
+      this.tabTopics.set(msg.id, new Set(msg.topics));
+      this.onTopicsChanged();
+    } else if (!this.isLeader) {
+      this.onData(msg.raw);
+    }
+  }
+
+  private setLeader(value: boolean): void {
+    if (this.isLeader === value) return;
+    this.isLeader = value;
+    if (value) this.post({ t: "hb", id: this.id });
+    this.onRole(value);
+  }
+
+  announceTopics(topics: string[]): void {
+    this.post({ t: "topics", id: this.id, topics });
+  }
+
+  /** Union of topics other tabs need — combined with this tab's own by the client (ref-counted across tabs). */
+  remoteTopics(): Set<string> {
+    const all = new Set<string>();
+    this.tabTopics.forEach((s) => s.forEach((t) => all.add(t)));
+    return all;
+  }
+
+  broadcast(raw: string): void {
+    if (this.isLeader) this.post({ t: "data", raw });
+  }
+
+  private post(msg: TabMessage): void {
+    this.channel?.postMessage(msg);
+  }
+
+  stop(): void {
+    this.post({ t: "bye", id: this.id });
+    if (this.hbTimer) clearInterval(this.hbTimer);
+    this.channel?.close();
+    this.channel = null;
+    this.isLeader = false;
+  }
+}
diff --git a/src/lib/realtime/transport.ts b/src/lib/realtime/transport.ts
new file mode 100644
index 0000000..7313ee7
--- /dev/null
+++ b/src/lib/realtime/transport.ts
@@ -0,0 +1,45 @@
+export type TransportKind = "websocket" | "sse";
+
+export interface TransportHandlers {
+  onOpen: () => void;
+  onMessage: (data: string) => void;
+  onClose: () => void;
+}
+
+/** Minimal contract both transports satisfy so hooks stay transport-agnostic. */
+export interface Transport {
+  readonly kind: TransportKind;
+  connect(handlers: TransportHandlers): void;
+  /** Outbound control message (subscribe / resync). */
+  send(message: object): void;
+  close(): void;
+}
+
+export class WebSocketTransport implements Transport {
+  readonly kind = "websocket" as const;
+  private socket: WebSocket | null = null;
+
+  constructor(private readonly url: string) {}
+
+  connect({ onOpen, onMessage, onClose }: TransportHandlers): void {
+    try {
+      this.socket = new WebSocket(this.url);
+    } catch {
+      onClose();
+      return;
+    }
+    this.socket.addEventListener("open", onOpen);
+    this.socket.addEventListener("message", (e) => onMessage(e.data as string));
+    this.socket.addEventListener("close", onClose);
+    this.socket.addEventListener("error", () => this.socket?.close());
+  }
+
+  send(message: object): void {
+    if (this.socket?.readyState === WebSocket.OPEN) this.socket.send(JSON.stringify(message));
+  }
+
+  close(): void {
+    this.socket?.close();
+    this.socket = null;
+  }
+}