diff --git a/.eslintrc.json b/.eslintrc.json
index 3cbf99e..0393555 100644
--- a/.eslintrc.json
+++ b/.eslintrc.json
@@ -12,7 +12,7 @@
         "message": "Hard-coded rgb()/rgba() colours are not allowed in JSX. Use design tokens from src/lib/coverage/metadata.ts or CSS custom properties (see CONTRIBUTING.md)."
       },
       {
-        "selector": "Literal[value=1e7]",
+        "selector": "Literal[value=\"1e7\"]",
         "message": "Use USDC_BASE_UNIT / fromStroops from src/lib/format.ts instead of the literal 1e7."
       },
       {
diff --git a/src/lib/realtime/client.ts b/src/lib/realtime/client.ts
index 8be8d3c..ef9c775 100644
--- a/src/lib/realtime/client.ts
+++ b/src/lib/realtime/client.ts
@@ -1,4 +1,7 @@
 import { API_BASE_URL } from "@/lib/api/client";
+import { TopicRegistry } from "./ringBuffer";
+import { realtimeTracer } from "./telemetry";
+import { frameRecorder } from "./recorder";
 
 type Handler = (payload: unknown) => void;
 
@@ -7,12 +10,28 @@ const BASE_BACKOFF_MS = 1_000;
 
 /**
  * A small reconnecting WebSocket 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.
+ *
+ * Backend contract (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.
+ *
+ * Extension points wired into this class:
+ *
+ *   • `TopicRegistry` (ringBuffer.ts) — every received frame is recorded in
+ *     a bounded per-topic ring buffer before being dispatched to handlers.
+ *     This prevents unbounded memory growth under high-frequency streams
+ *     (closes #253).
+ *
+ *   • `realtimeTracer` (telemetry.ts) — connection attempts, open, close,
+ *     reconnects, message dispatch, and fallback activations emit OTel spans
+ *     and counters so the team gains backend-correlatable visibility
+ *     (closes #237).
+ *
+ *   • `frameRecorder` (recorder.ts) — when recording is active, every frame
+ *     (including connection lifecycle events) is captured with timestamps so
+ *     the stream can be replayed deterministically in tests (closes #236).
  */
 class RealtimeClient {
   private socket: WebSocket | null = null;
@@ -21,6 +40,9 @@ class RealtimeClient {
   private closedByUser = false;
   private reconnectTimer: ReturnType<typeof setTimeout> | null = null;
 
+  /** Bounded per-topic ring-buffer registry (closes #253). */
+  readonly registry = new TopicRegistry();
+
   connect(): void {
     if (typeof WebSocket === "undefined" || this.socket) return;
     this.closedByUser = false;
@@ -29,29 +51,59 @@ class RealtimeClient {
 
   private open(): void {
     const wsUrl = API_BASE_URL.replace(/^http/, "ws").replace(/\/api\/v1$/, "/ws");
+
+    realtimeTracer.onConnectAttempt(wsUrl, this.attempt);
+    frameRecorder.recordLifecycle("connect_attempt", { wsUrl, attempt: this.attempt });
+
     try {
       this.socket = new WebSocket(wsUrl);
-    } catch {
+    } catch (err) {
+      realtimeTracer.onConnectError(wsUrl, err);
+      frameRecorder.recordLifecycle("connect_error", { wsUrl, error: String(err) });
       this.scheduleReconnect();
       return;
     }
 
     this.socket.addEventListener("open", () => {
       this.attempt = 0;
+      realtimeTracer.onOpen(wsUrl);
+      frameRecorder.recordLifecycle("open", { wsUrl });
     });
+
     this.socket.addEventListener("message", (event) => {
       try {
-        const { topic, payload } = JSON.parse(event.data as string) as { topic: string; payload: unknown };
+        const { topic, payload } = JSON.parse(event.data as string) as {
+          topic: string;
+          payload: unknown;
+        };
+
+        // 1. Record in bounded ring buffer (before dispatch, so the frame is
+        //    available even if a handler throws).
+        const frame = this.registry.record(topic, payload);
+
+        // 2. Append to recorder session (no-op when recording is off).
+        if (frame) frameRecorder.recordFrame(frame);
+
+        // 3. Emit OTel span for this message dispatch.
+        realtimeTracer.onMessage(topic);
+
+        // 4. Dispatch to subscribers.
         this.handlers.get(topic)?.forEach((handler) => handler(payload));
       } catch {
         // Malformed message — ignore, don't crash the connection.
       }
     });
+
     this.socket.addEventListener("close", () => {
+      realtimeTracer.onClose(wsUrl);
+      frameRecorder.recordLifecycle("close", { wsUrl });
       this.socket = null;
       if (!this.closedByUser) this.scheduleReconnect();
     });
+
     this.socket.addEventListener("error", () => {
+      realtimeTracer.onError(wsUrl);
+      frameRecorder.recordLifecycle("socket_error", { wsUrl });
       this.socket?.close();
     });
   }
@@ -60,6 +112,8 @@ class RealtimeClient {
     if (this.reconnectTimer) return;
     const delay = Math.min(BASE_BACKOFF_MS * 2 ** this.attempt, MAX_BACKOFF_MS);
     this.attempt += 1;
+    realtimeTracer.onReconnectScheduled(delay, this.attempt);
+    frameRecorder.recordLifecycle("reconnect_scheduled", { delayMs: delay, attempt: this.attempt });
     this.reconnectTimer = setTimeout(() => {
       this.reconnectTimer = null;
       if (!this.closedByUser) this.open();
@@ -69,19 +123,36 @@ class RealtimeClient {
   subscribe(topic: string, handler: Handler): () => void {
     if (!this.handlers.has(topic)) this.handlers.set(topic, new Set());
     this.handlers.get(topic)!.add(handler);
+    realtimeTracer.onSubscribe(topic, this.handlers.get(topic)!.size);
+    frameRecorder.recordLifecycle("subscribe", { topic });
     this.connect();
 
     return () => {
       this.handlers.get(topic)?.delete(handler);
+      const remaining = this.handlers.get(topic)?.size ?? 0;
+      realtimeTracer.onUnsubscribe(topic, remaining);
+      frameRecorder.recordLifecycle("unsubscribe", { topic });
     };
   }
 
   disconnect(): void {
     this.closedByUser = true;
     if (this.reconnectTimer) clearTimeout(this.reconnectTimer);
+    realtimeTracer.onDisconnect();
+    frameRecorder.recordLifecycle("disconnect", {});
     this.socket?.close();
     this.socket = null;
   }
+
+  /**
+   * Signal that a REST fallback was activated (e.g. due to WS being
+   * unavailable or `realtime-oracle` flag being off). Emits an OTel counter
+   * increment so the team can see fallback rate relative to live connections.
+   */
+  notifyFallbackActivated(reason: string): void {
+    realtimeTracer.onFallbackActivated(reason);
+    frameRecorder.recordLifecycle("fallback_activated", { reason });
+  }
 }
 
 export const realtimeClient = new RealtimeClient();
diff --git a/src/lib/realtime/contract.test.ts b/src/lib/realtime/contract.test.ts
new file mode 100644
index 0000000..8e09abe
--- /dev/null
+++ b/src/lib/realtime/contract.test.ts
@@ -0,0 +1,278 @@
+/**
+ * Consumer-driven contract tests for the realtime WebSocket topic schema.
+ *
+ * Closes #234: Pins the frontend's expectations to a versioned artifact and
+ * fails CI when the backend-published OpenAPI schema drifts from what
+ * `src/lib/realtime/client.ts` / `contract.ts` assumes.
+ *
+ * ## What is tested
+ *
+ * 1. All payloads that conform to the Zod schemas parse successfully.
+ * 2. Payloads missing required fields are rejected with informative errors.
+ * 3. The Zod schema field names exactly match those in `openapi/refract-api.yaml`
+ *    (bi-directional diff — catches both additions and removals).
+ * 4. The `severity` and `type` enums defined in Zod match the OpenAPI enums.
+ * 5. `validateFrame` / `assertFrame` public API behaves correctly.
+ * 6. Unknown topics are tolerated (forward-compatibility guarantee).
+ */
+
+import { describe, it, expect } from "vitest";
+import { readFileSync } from "node:fs";
+import { join } from "node:path";
+import {
+  oracleUpdateSchema,
+  envelopeSchema,
+  TOPIC_SCHEMAS,
+  KNOWN_TOPICS,
+  validateFrame,
+  assertFrame,
+  diffOpenApiSchema,
+  ContractViolationError,
+  CONTRACT_VERSION,
+} from "./contract";
+
+// ---------------------------------------------------------------------------
+// Helpers
+// ---------------------------------------------------------------------------
+
+/**
+ * Minimal YAML property extractor — avoids adding a YAML parser dependency.
+ * Parses simple `key: value` pairs (no nested objects, no arrays) from an
+ * OpenAPI `properties:` block by scanning the raw file content.
+ *
+ * This is intentionally trivial: it only needs to extract first-level
+ * property names from a known schema block to do the drift check.
+ */
+function extractOpenApiProperties(schemaName: string, yamlContent: string): Record<string, string> {
+  // Find the schema block
+  const schemaPattern = new RegExp(`    ${schemaName}:\\n([\\s\\S]*?)(?=\\n    \\w|$)`);
+  const schemaMatch = yamlContent.match(schemaPattern);
+  if (!schemaMatch) return {};
+
+  const propertiesBlock = schemaMatch[1];
+  // Extract property names from lines like "        propertyName: { ... }"
+  const propPattern = /^        (\w+):/gm;
+  const props: Record<string, string> = {};
+  let m: RegExpExecArray | null;
+  while ((m = propPattern.exec(propertiesBlock)) !== null) {
+    props[m[1]] = m[1];
+  }
+  return props;
+}
+
+const OPENAPI_PATH = join(process.cwd(), "openapi", "refract-api.yaml");
+const openApiYaml = readFileSync(OPENAPI_PATH, "utf-8");
+
+// ---------------------------------------------------------------------------
+// Contract version
+// ---------------------------------------------------------------------------
+
+describe("CONTRACT_VERSION", () => {
+  it("is a semver string", () => {
+    expect(CONTRACT_VERSION).toMatch(/^\d+\.\d+\.\d+/);
+  });
+});
+
+// ---------------------------------------------------------------------------
+// Envelope schema
+// ---------------------------------------------------------------------------
+
+describe("envelopeSchema", () => {
+  it("accepts a valid envelope", () => {
+    const result = envelopeSchema.safeParse({ topic: "oracle_update", payload: {} });
+    expect(result.success).toBe(true);
+  });
+
+  it("rejects an empty topic", () => {
+    const result = envelopeSchema.safeParse({ topic: "", payload: {} });
+    expect(result.success).toBe(false);
+  });
+
+  it("rejects a missing topic", () => {
+    const result = envelopeSchema.safeParse({ payload: {} });
+    expect(result.success).toBe(false);
+  });
+});
+
+// ---------------------------------------------------------------------------
+// oracle_update topic
+// ---------------------------------------------------------------------------
+
+describe("oracleUpdateSchema", () => {
+  const valid = {
+    coverageType: "StablecoinDepeg",
+    type: "oracle_update" as const,
+    value: 1.0002,
+    threshold: 0.95,
+    severity: "low" as const,
+    message: "USDC price: $1.0002",
+  };
+
+  it("accepts a fully-valid oracle_update payload", () => {
+    expect(oracleUpdateSchema.safeParse(valid).success).toBe(true);
+  });
+
+  it("accepts all valid severity values", () => {
+    for (const severity of ["low", "medium", "high", "triggered"] as const) {
+      expect(oracleUpdateSchema.safeParse({ ...valid, severity }).success).toBe(true);
+    }
+  });
+
+  it("rejects an unknown severity", () => {
+    const result = oracleUpdateSchema.safeParse({ ...valid, severity: "critical" });
+    expect(result.success).toBe(false);
+  });
+
+  it("rejects a missing required field (message)", () => {
+    const { message: _m, ...partial } = valid;
+    expect(oracleUpdateSchema.safeParse(partial).success).toBe(false);
+  });
+
+  it("rejects a missing required field (value)", () => {
+    const { value: _v, ...partial } = valid;
+    expect(oracleUpdateSchema.safeParse(partial).success).toBe(false);
+  });
+
+  it("rejects a wrong type literal (type !== 'oracle_update')", () => {
+    const result = oracleUpdateSchema.safeParse({ ...valid, type: "claim_update" });
+    expect(result.success).toBe(false);
+  });
+
+  it("rejects a non-numeric value field", () => {
+    const result = oracleUpdateSchema.safeParse({ ...valid, value: "not-a-number" });
+    expect(result.success).toBe(false);
+  });
+
+  it("rejects an empty coverageType string", () => {
+    const result = oracleUpdateSchema.safeParse({ ...valid, coverageType: "" });
+    expect(result.success).toBe(false);
+  });
+});
+
+// ---------------------------------------------------------------------------
+// OpenAPI ↔ Zod drift check  (closes #234 — the CI gate)
+// ---------------------------------------------------------------------------
+
+describe("OpenAPI ↔ Zod drift check", () => {
+  it("OracleReading properties in openapi/refract-api.yaml match oracleUpdateSchema", () => {
+    const openApiProps = extractOpenApiProperties("OracleReading", openApiYaml);
+    expect(Object.keys(openApiProps).length).toBeGreaterThan(0);
+
+    const { missingInZod, missingInOpenApi } = diffOpenApiSchema(
+      openApiProps,
+      oracleUpdateSchema as import("zod").ZodObject<import("zod").ZodRawShape>
+    );
+
+    expect(
+      missingInZod,
+      `Fields in OpenAPI but missing in Zod contract: ${missingInZod.join(", ")}`
+    ).toHaveLength(0);
+    expect(
+      missingInOpenApi,
+      `Fields in Zod contract but missing in OpenAPI: ${missingInOpenApi.join(", ")}`
+    ).toHaveLength(0);
+  });
+
+  it("OracleReading type enum in OpenAPI matches Zod literal", () => {
+    // The OpenAPI enum has a single value: oracle_update
+    expect(openApiYaml).toMatch(/enum:\s*\[\s*oracle_update\s*\]/);
+  });
+
+  it("OracleReading severity enum in OpenAPI has all 4 values", () => {
+    // Verify enum line presence — exact ordering may vary in YAML.
+    for (const sev of ["low", "medium", "high", "triggered"]) {
+      expect(openApiYaml).toContain(sev);
+    }
+  });
+});
+
+// ---------------------------------------------------------------------------
+// TOPIC_SCHEMAS registry
+// ---------------------------------------------------------------------------
+
+describe("TOPIC_SCHEMAS registry", () => {
+  it("contains at least one registered topic", () => {
+    expect(KNOWN_TOPICS.length).toBeGreaterThanOrEqual(1);
+  });
+
+  it("oracle_update is registered", () => {
+    expect(KNOWN_TOPICS).toContain("oracle_update");
+    expect(TOPIC_SCHEMAS["oracle_update"]).toBeDefined();
+  });
+});
+
+// ---------------------------------------------------------------------------
+// validateFrame
+// ---------------------------------------------------------------------------
+
+describe("validateFrame", () => {
+  const validPayload = {
+    coverageType: "MarketCrash",
+    type: "oracle_update",
+    value: -15.2,
+    threshold: -30,
+    severity: "medium",
+    message: "XLM 24h change: -15.2%",
+  };
+
+  it("returns { ok: true } for a valid oracle_update frame", () => {
+    const result = validateFrame("oracle_update", validPayload);
+    expect(result.ok).toBe(true);
+    expect(result.errors).toBeUndefined();
+  });
+
+  it("returns { ok: false } with errors for an invalid frame", () => {
+    const result = validateFrame("oracle_update", { type: "oracle_update" });
+    expect(result.ok).toBe(false);
+    expect(result.errors).toBeDefined();
+    expect(result.errors!.length).toBeGreaterThan(0);
+  });
+
+  it("returns { ok: true } for an unknown topic (forward-compat)", () => {
+    const result = validateFrame("future_topic", { anything: "goes" });
+    expect(result.ok).toBe(true);
+  });
+
+  it("includes the topic name in the result", () => {
+    const result = validateFrame("oracle_update", validPayload);
+    expect(result.topic).toBe("oracle_update");
+  });
+});
+
+// ---------------------------------------------------------------------------
+// assertFrame
+// ---------------------------------------------------------------------------
+
+describe("assertFrame", () => {
+  it("does not throw for a valid payload", () => {
+    expect(() =>
+      assertFrame("oracle_update", {
+        coverageType: "SmartContractRisk",
+        type: "oracle_update",
+        value: -2.1,
+        threshold: -50,
+        severity: "low",
+        message: "TVL change: -2.1%",
+      })
+    ).not.toThrow();
+  });
+
+  it("throws ContractViolationError for an invalid payload", () => {
+    expect(() => assertFrame("oracle_update", { bogus: true })).toThrow(ContractViolationError);
+  });
+
+  it("includes the topic and issues in the thrown error", () => {
+    try {
+      assertFrame("oracle_update", {});
+    } catch (e) {
+      expect(e).toBeInstanceOf(ContractViolationError);
+      const err = e as ContractViolationError;
+      expect(err.topic).toBe("oracle_update");
+      expect(err.issues.length).toBeGreaterThan(0);
+    }
+  });
+
+  it("does not throw for an unknown topic", () => {
+    expect(() => assertFrame("unregistered_topic", {})).not.toThrow();
+  });
+});
diff --git a/src/lib/realtime/contract.ts b/src/lib/realtime/contract.ts
new file mode 100644
index 0000000..1eabaa3
--- /dev/null
+++ b/src/lib/realtime/contract.ts
@@ -0,0 +1,162 @@
+/**
+ * Consumer-driven realtime topic/message contract.
+ *
+ * Closes #234: Pins the frontend's expectations of the realtime
+ * WebSocket topic/message contract to a versioned, machine-checkable artifact
+ * so that CI fails when the backend-published schema drifts from what
+ * `src/lib/realtime/client.ts` assumes.
+ *
+ * ## How it works
+ *
+ * 1. This file defines the *consumer contract* — the exact message shapes and
+ *    topic names the frontend relies on — as Zod schemas.
+ *
+ * 2. `CONTRACT_VERSION` is a semver string pinned to the backend version that
+ *    was authoritative when this contract was written. Update it whenever the
+ *    shapes change, then update the tests to match.
+ *
+ * 3. `src/lib/realtime/contract.test.ts` validates every schema against the
+ *    corresponding component in `openapi/refract-api.yaml`. CI runs this test
+ *    so a backend API drift immediately fails the build.
+ *
+ * 4. `validateFrame(topic, payload)` can be called at runtime (gated behind
+ *    a flag or dev/test mode check) to surface schema violations early.
+ *
+ * ## Adding a new topic
+ *
+ * 1. Add a Zod schema below (naming convention: `<topicName>Schema`).
+ * 2. Register it in `TOPIC_SCHEMAS`.
+ * 3. Add a corresponding component schema in `openapi/refract-api.yaml`.
+ * 4. Add a contract test in `contract.test.ts`.
+ */
+
+import { z } from "zod";
+
+// ---------------------------------------------------------------------------
+// Contract version — bump when backend breaks compatibility
+// ---------------------------------------------------------------------------
+
+/** Semver string of the backend contract revision this frontend targets. */
+export const CONTRACT_VERSION = "0.1.0";
+
+// ---------------------------------------------------------------------------
+// Per-topic payload schemas
+// ---------------------------------------------------------------------------
+
+/**
+ * `oracle_update` — pushed whenever an oracle check completes on the backend.
+ * Shape mirrors `OracleReading` in `openapi/refract-api.yaml` / `schema.d.ts`.
+ */
+export const oracleUpdateSchema = z.object({
+  coverageType: z.string().min(1),
+  type: z.literal("oracle_update"),
+  value: z.number(),
+  threshold: z.number(),
+  severity: z.enum(["low", "medium", "high", "triggered"]),
+  message: z.string(),
+});
+
+export type OracleUpdatePayload = z.infer<typeof oracleUpdateSchema>;
+
+// ---------------------------------------------------------------------------
+// Envelope schema — the outer { topic, payload } wrapper every WS frame uses
+// ---------------------------------------------------------------------------
+
+/**
+ * The outer envelope for every frame received from the WebSocket.
+ * Inner payload validation is done per-topic via `TOPIC_SCHEMAS`.
+ */
+export const envelopeSchema = z.object({
+  topic: z.string().min(1),
+  payload: z.unknown(),
+});
+
+export type Envelope = z.infer<typeof envelopeSchema>;
+
+// ---------------------------------------------------------------------------
+// Topic registry — maps topic name → payload validator
+// ---------------------------------------------------------------------------
+
+/** All topics the frontend currently subscribes to, with their validators. */
+export const TOPIC_SCHEMAS: Record<string, z.ZodTypeAny> = {
+  oracle_update: oracleUpdateSchema,
+};
+
+/** Names of all known topics (derived from TOPIC_SCHEMAS for DRYness). */
+export const KNOWN_TOPICS: ReadonlyArray<string> = Object.keys(TOPIC_SCHEMAS);
+
+// ---------------------------------------------------------------------------
+// Runtime validation helpers
+// ---------------------------------------------------------------------------
+
+export interface ValidationResult {
+  ok: boolean;
+  topic: string;
+  errors?: z.ZodIssue[];
+}
+
+/**
+ * Validate a received payload against the registered contract for `topic`.
+ *
+ * Returns `{ ok: true }` on success.
+ * Returns `{ ok: false, errors }` when validation fails — never throws.
+ *
+ * Unknown topics return `{ ok: true }` (forward-compatibility: the backend
+ * may push topics the frontend doesn't know yet; we ignore them rather than
+ * treating them as errors).
+ */
+export function validateFrame(topic: string, payload: unknown): ValidationResult {
+  const schema = TOPIC_SCHEMAS[topic];
+  if (!schema) return { ok: true, topic }; // unknown topic — ignore gracefully
+
+  const result = schema.safeParse(payload);
+  if (result.success) return { ok: true, topic };
+  return { ok: false, topic, errors: result.error.issues };
+}
+
+/**
+ * Like `validateFrame` but throws a `ContractViolationError` on failure.
+ * Suitable for strict test assertions.
+ */
+export function assertFrame(topic: string, payload: unknown): void {
+  const result = validateFrame(topic, payload);
+  if (!result.ok) {
+    throw new ContractViolationError(topic, result.errors ?? []);
+  }
+}
+
+export class ContractViolationError extends Error {
+  constructor(
+    public readonly topic: string,
+    public readonly issues: z.ZodIssue[]
+  ) {
+    const summary = issues.map((i) => `  ${i.path.join(".")}: ${i.message}`).join("\n");
+    super(`Contract violation for topic "${topic}":\n${summary}`);
+    this.name = "ContractViolationError";
+  }
+}
+
+// ---------------------------------------------------------------------------
+// OpenAPI → contract alignment helpers (used by contract.test.ts)
+// ---------------------------------------------------------------------------
+
+/**
+ * Validate a raw YAML-parsed OpenAPI schema object against the expected
+ * contract shape. Returns a list of property names that are present in the
+ * OpenAPI schema but absent from the Zod schema's `.shape`, or vice-versa.
+ *
+ * This is a lightweight structural diff — not a full JSON Schema validator —
+ * sufficient to catch field additions/removals in the backend contract.
+ */
+export function diffOpenApiSchema(
+  openApiProperties: Record<string, unknown>,
+  zodSchema: z.ZodObject<z.ZodRawShape>
+): { missingInZod: string[]; missingInOpenApi: string[] } {
+  const openApiKeys = new Set(Object.keys(openApiProperties));
+  const zodKeys = new Set(Object.keys(zodSchema.shape));
+
+  const missingInZod = [...openApiKeys].filter((k) => !zodKeys.has(k));
+  const missingInOpenApi = [...zodKeys].filter((k) => !openApiKeys.has(k));
+
+  return { missingInZod, missingInOpenApi };
+}
diff --git a/src/lib/realtime/recorder.test.ts b/src/lib/realtime/recorder.test.ts
new file mode 100644
index 0000000..4bf457a
--- /dev/null
+++ b/src/lib/realtime/recorder.test.ts
@@ -0,0 +1,369 @@
+/**
+ * Tests for the frame recorder and deterministic replay engine.
+ * Closes #236.
+ */
+
+import { describe, it, expect, beforeEach, vi } from "vitest";
+import {
+  FrameRecorder,
+  serializeRecording,
+  deserializeRecording,
+  replayRecording,
+  replaySync,
+  type Recording,
+  type LifecycleEvent,
+  type CapturedFrame,
+} from "./recorder";
+import type { RealtimeFrame } from "./ringBuffer";
+
+// ---------------------------------------------------------------------------
+// Helpers
+// ---------------------------------------------------------------------------
+
+function makeFrame(topic: string, seq: number, payload?: unknown): RealtimeFrame {
+  return { topic, payload: payload ?? { seq }, receivedAt: Date.now(), seq };
+}
+
+/** Minimal replay target that exposes the internal `handlers` map. */
+function makeTarget() {
+  const handlers = new Map<string, Set<(p: unknown) => void>>();
+  return {
+    handlers,
+    subscribe(topic: string, handler: (p: unknown) => void) {
+      if (!handlers.has(topic)) handlers.set(topic, new Set());
+      handlers.get(topic)!.add(handler);
+      return () => handlers.get(topic)?.delete(handler);
+    },
+  };
+}
+
+// ---------------------------------------------------------------------------
+// FrameRecorder
+// ---------------------------------------------------------------------------
+
+describe("FrameRecorder", () => {
+  let recorder: FrameRecorder;
+
+  beforeEach(() => {
+    recorder = new FrameRecorder();
+  });
+
+  it("starts inactive", () => {
+    expect(recorder.isRecording).toBe(false);
+  });
+
+  it("becomes active after start()", () => {
+    recorder.start();
+    expect(recorder.isRecording).toBe(true);
+  });
+
+  it("is idempotent on start()", () => {
+    recorder.start();
+    recorder.recordLifecycle("connect_attempt", {});
+    recorder.start(); // second call — should not reset
+    expect(recorder.peek()).toHaveLength(1);
+  });
+
+  it("records frames when active", () => {
+    recorder.start();
+    recorder.recordFrame(makeFrame("oracle_update", 0, { value: 1.0 }));
+    recorder.recordFrame(makeFrame("oracle_update", 1, { value: 1.1 }));
+    expect(recorder.peek()).toHaveLength(2);
+  });
+
+  it("records lifecycle events when active", () => {
+    recorder.start();
+    recorder.recordLifecycle("open", { wsUrl: "ws://localhost/ws" });
+    const entries = recorder.peek();
+    expect(entries).toHaveLength(1);
+    expect(entries[0].kind).toBe("lifecycle");
+    expect((entries[0] as LifecycleEvent).event).toBe("open");
+  });
+
+  it("does nothing when not active", () => {
+    recorder.recordFrame(makeFrame("oracle_update", 0));
+    recorder.recordLifecycle("open", {});
+    expect(recorder.peek()).toHaveLength(0);
+  });
+
+  it("stop() returns the recording and clears internal state", () => {
+    recorder.start();
+    recorder.recordFrame(makeFrame("oracle_update", 0));
+    recorder.recordLifecycle("close", {});
+    const rec = recorder.stop();
+    expect(rec.entries).toHaveLength(2);
+    expect(recorder.isRecording).toBe(false);
+    expect(recorder.peek()).toHaveLength(0);
+  });
+
+  it("recording includes startedAt / stoppedAt timestamps", () => {
+    recorder.start();
+    const rec = recorder.stop();
+    expect(rec.startedAt).toBeTruthy();
+    expect(rec.stoppedAt).toBeTruthy();
+    expect(new Date(rec.stoppedAt).getTime()).toBeGreaterThanOrEqual(
+      new Date(rec.startedAt).getTime()
+    );
+  });
+
+  it("assigns monotonically increasing seq across entry types", () => {
+    recorder.start();
+    recorder.recordLifecycle("connect_attempt", {});
+    recorder.recordFrame(makeFrame("oracle_update", 0));
+    recorder.recordLifecycle("open", {});
+    const entries = recorder.peek();
+    expect(entries.map((e) => e.seq)).toEqual([0, 1, 2]);
+  });
+
+  it("preserves meta passed to start()", () => {
+    recorder.start({ sessionId: "abc123", userAgent: "test" });
+    const rec = recorder.stop();
+    expect(rec.meta.sessionId).toBe("abc123");
+  });
+});
+
+// ---------------------------------------------------------------------------
+// Serialisation round-trip
+// ---------------------------------------------------------------------------
+
+describe("serializeRecording / deserializeRecording", () => {
+  function makeRecording(): Recording {
+    const recorder = new FrameRecorder();
+    recorder.start();
+    recorder.recordFrame(makeFrame("oracle_update", 0, { value: 1.0 }));
+    recorder.recordLifecycle("open", { wsUrl: "ws://test" });
+    return recorder.stop();
+  }
+
+  it("round-trips correctly", () => {
+    const original = makeRecording();
+    const json = serializeRecording(original);
+    const parsed = deserializeRecording(json);
+    expect(parsed.entries).toHaveLength(original.entries.length);
+    expect(parsed.startedAt).toBe(original.startedAt);
+  });
+
+  it("deserializeRecording throws on invalid JSON", () => {
+    expect(() => deserializeRecording("not json")).toThrow(SyntaxError);
+  });
+
+  it("deserializeRecording throws when entries is not an array", () => {
+    expect(() =>
+      deserializeRecording(
+        JSON.stringify({ startedAt: "", stoppedAt: "", entries: null, meta: {} })
+      )
+    ).toThrow(TypeError);
+  });
+});
+
+// ---------------------------------------------------------------------------
+// replaySync
+// ---------------------------------------------------------------------------
+
+describe("replaySync", () => {
+  it("dispatches frame payloads to subscribed handlers", () => {
+    const recorder = new FrameRecorder();
+    recorder.start();
+    recorder.recordFrame(makeFrame("oracle_update", 0, { value: 1.0 }));
+    recorder.recordFrame(makeFrame("oracle_update", 1, { value: 1.1 }));
+    const rec = recorder.stop();
+
+    const received: unknown[] = [];
+    const target = makeTarget();
+    target.subscribe("oracle_update", (p) => received.push(p));
+
+    replaySync(rec, target);
+    expect(received).toHaveLength(2);
+    expect((received[0] as { value: number }).value).toBe(1.0);
+    expect((received[1] as { value: number }).value).toBe(1.1);
+  });
+
+  it("dispatches frames in seq order regardless of entry insertion order", () => {
+    const recorder = new FrameRecorder();
+    recorder.start();
+    // Manually insert a lifecycle event between two frames.
+    recorder.recordFrame(makeFrame("oracle_update", 0, { idx: 0 }));
+    recorder.recordLifecycle("reconnect_scheduled", { delayMs: 1000 });
+    recorder.recordFrame(makeFrame("oracle_update", 1, { idx: 1 }));
+    const rec = recorder.stop();
+
+    const received: unknown[] = [];
+    const lifecycle: string[] = [];
+    const target = makeTarget();
+    target.subscribe("oracle_update", (p) => received.push(p));
+
+    replaySync(rec, target, {
+      onLifecycle: (e) => lifecycle.push(e.event),
+    });
+
+    expect(received).toHaveLength(2);
+    expect(lifecycle).toEqual(["reconnect_scheduled"]);
+  });
+
+  it("calls onLifecycle for lifecycle entries", () => {
+    const recorder = new FrameRecorder();
+    recorder.start();
+    recorder.recordLifecycle("open", { wsUrl: "ws://test" });
+    recorder.recordLifecycle("close", { wsUrl: "ws://test" });
+    const rec = recorder.stop();
+
+    const events: string[] = [];
+    replaySync(rec, makeTarget(), {
+      onLifecycle: (e) => events.push(e.event),
+    });
+    expect(events).toEqual(["open", "close"]);
+  });
+
+  it("handles an empty recording gracefully", () => {
+    const recorder = new FrameRecorder();
+    recorder.start();
+    const rec = recorder.stop();
+    expect(() => replaySync(rec, makeTarget())).not.toThrow();
+  });
+
+  it("topics with no subscribers receive no errors", () => {
+    const recorder = new FrameRecorder();
+    recorder.start();
+    recorder.recordFrame(makeFrame("unsubscribed_topic", 0));
+    const rec = recorder.stop();
+    expect(() => replaySync(rec, makeTarget())).not.toThrow();
+  });
+});
+
+// ---------------------------------------------------------------------------
+// replayRecording (async / timing-aware)
+// ---------------------------------------------------------------------------
+
+describe("replayRecording", () => {
+  it("dispatches all frames when preserveTiming=false", async () => {
+    const recorder = new FrameRecorder();
+    recorder.start();
+    for (let i = 0; i < 5; i++) {
+      recorder.recordFrame(makeFrame("oracle_update", i, { i }));
+    }
+    const rec = recorder.stop();
+
+    const received: unknown[] = [];
+    const target = makeTarget();
+    target.subscribe("oracle_update", (p) => received.push(p));
+
+    await replayRecording(rec, target, { preserveTiming: false });
+    expect(received).toHaveLength(5);
+  });
+
+  it("calls onLifecycle callbacks during async replay", async () => {
+    const recorder = new FrameRecorder();
+    recorder.start();
+    recorder.recordLifecycle("open", {});
+    recorder.recordFrame(makeFrame("oracle_update", 0));
+    recorder.recordLifecycle("close", {});
+    const rec = recorder.stop();
+
+    const lifecycle: string[] = [];
+    await replayRecording(rec, makeTarget(), {
+      preserveTiming: false,
+      onLifecycle: (e) => lifecycle.push(e.event),
+    });
+    expect(lifecycle).toEqual(["open", "close"]);
+  });
+
+  it("resolves immediately for an empty recording", async () => {
+    const recorder = new FrameRecorder();
+    recorder.start();
+    const rec = recorder.stop();
+    await expect(replayRecording(rec, makeTarget())).resolves.toBeUndefined();
+  });
+
+  it("speed multiplier reduces inter-frame delay", async () => {
+    vi.useFakeTimers();
+
+    const recorder = new FrameRecorder();
+    recorder.start();
+    const now = Date.now();
+    // Two frames 1 second apart.
+    recorder.recordFrame({ topic: "t", payload: {}, receivedAt: now, seq: 0 });
+    recorder.recordFrame({ topic: "t", payload: {}, receivedAt: now + 1000, seq: 1 });
+    const rec = recorder.stop();
+
+    const received: unknown[] = [];
+    const target = makeTarget();
+    target.subscribe("t", (p) => received.push(p));
+
+    const promise = replayRecording(rec, target, { preserveTiming: true, speed: 1000 });
+    // With speed=1000 a 1s gap becomes 1ms — advance time by 1ms.
+    vi.advanceTimersByTime(2);
+    await promise;
+    expect(received).toHaveLength(2);
+
+    vi.useRealTimers();
+  });
+});
+
+// ---------------------------------------------------------------------------
+// Deterministic replay reproduces the same sequence as the original session
+// ---------------------------------------------------------------------------
+
+describe("recording → replay determinism", () => {
+  it("replay produces the same payload sequence as the original session", () => {
+    const payloads = [
+      { coverageType: "StablecoinDepeg", value: 1.0 },
+      { coverageType: "MarketCrash", value: -5.0 },
+      { coverageType: "SmartContractRisk", value: -2.0 },
+    ];
+
+    // Simulate a live session.
+    const recorder = new FrameRecorder();
+    recorder.start();
+    const liveReceived: unknown[] = [];
+    const sessionHandler = (p: unknown) => liveReceived.push(p);
+    for (const [i, payload] of payloads.entries()) {
+      // Pretend the WS dispatched these to the handler.
+      sessionHandler(payload);
+      recorder.recordFrame(makeFrame("oracle_update", i, payload));
+    }
+    const rec = recorder.stop();
+
+    // Replay.
+    const replayReceived: unknown[] = [];
+    const target = makeTarget();
+    target.subscribe("oracle_update", (p) => replayReceived.push(p));
+    replaySync(rec, target);
+
+    expect(replayReceived).toEqual(liveReceived);
+  });
+
+  it("serialise→deserialise→replay round-trip preserves payload equality", () => {
+    const recorder = new FrameRecorder();
+    recorder.start();
+    recorder.recordFrame(makeFrame("oracle_update", 0, { value: 42.0, severity: "triggered" }));
+    const rec = recorder.stop();
+
+    const json = serializeRecording(rec);
+    const restored = deserializeRecording(json);
+
+    const received: unknown[] = [];
+    const target = makeTarget();
+    target.subscribe("oracle_update", (p) => received.push(p));
+    replaySync(restored, target);
+
+    expect((received[0] as { value: number }).value).toBe(42.0);
+    expect((received[0] as { severity: string }).severity).toBe("triggered");
+  });
+
+  it("captured lifecycle events are preserved through serialisation", () => {
+    const recorder = new FrameRecorder();
+    recorder.start();
+    recorder.recordLifecycle("connect_attempt", { wsUrl: "ws://localhost/ws", attempt: 0 });
+    recorder.recordLifecycle("open", { wsUrl: "ws://localhost/ws" });
+    const rec = recorder.stop();
+
+    const json = serializeRecording(rec);
+    const restored = deserializeRecording(json);
+
+    const events: string[] = [];
+    replaySync(restored, makeTarget(), {
+      onLifecycle: (e) => events.push(e.event),
+    });
+    expect(events).toEqual(["connect_attempt", "open"]);
+  });
+});
diff --git a/src/lib/realtime/recorder.ts b/src/lib/realtime/recorder.ts
new file mode 100644
index 0000000..0ee25ef
--- /dev/null
+++ b/src/lib/realtime/recorder.ts
@@ -0,0 +1,290 @@
+/**
+ * Opt-in frame recorder and deterministic replay engine for the realtime
+ * WebSocket stream.
+ *
+ * Closes #236: Captures the ordered sequence of realtime frames — including
+ * connection lifecycle events and their wall-clock timestamps — during a live
+ * or test session, then replays that recording deterministically against the
+ * `RealtimeClient`'s subscription handlers. This turns "the navbar badge
+ * flickered and claims stalled for a user yesterday" into a repeatable local
+ * test case.
+ *
+ * ## Usage
+ *
+ * ### Capturing a recording
+ * ```ts
+ * import { frameRecorder } from "@/lib/realtime/recorder";
+ * import { realtimeClient } from "@/lib/realtime/client";
+ *
+ * frameRecorder.start();
+ * // … user does things …
+ * const recording = frameRecorder.stop();
+ * const json = serializeRecording(recording);
+ * // Save `json` to a fixture file for later replay.
+ * ```
+ *
+ * ### Replaying in a test
+ * ```ts
+ * import { replayRecording, deserializeRecording } from "@/lib/realtime/recorder";
+ * import { RealtimeClient } from "@/lib/realtime/client";
+ *
+ * const recording = deserializeRecording(json);
+ * const received: unknown[] = [];
+ * const client = new RealtimeClient(); // or use the singleton
+ * client.subscribe("oracle_update", (p) => received.push(p));
+ * await replayRecording(recording, client);
+ * // `received` now contains the same payloads as the original session.
+ * ```
+ */
+
+import type { RealtimeFrame } from "./ringBuffer";
+
+// ---------------------------------------------------------------------------
+// Types
+// ---------------------------------------------------------------------------
+
+/** A lifecycle event captured outside the normal topic/payload flow. */
+export interface LifecycleEvent {
+  kind: "lifecycle";
+  event: string;
+  data: Record<string, unknown>;
+  timestamp: number;
+  seq: number;
+}
+
+/** A data frame captured from the WebSocket message stream. */
+export interface CapturedFrame {
+  kind: "frame";
+  topic: string;
+  payload: unknown;
+  receivedAt: number;
+  seq: number;
+}
+
+export type RecordingEntry = LifecycleEvent | CapturedFrame;
+
+export interface Recording {
+  /** ISO-8601 timestamp of when recording started. */
+  startedAt: string;
+  /** ISO-8601 timestamp of when recording stopped. */
+  stoppedAt: string;
+  /** Ordered log of all captured entries. */
+  entries: RecordingEntry[];
+  /** Metadata bag for annotating recordings (session ID, user agent, etc.). */
+  meta: Record<string, unknown>;
+}
+
+// ---------------------------------------------------------------------------
+// FrameRecorder
+// ---------------------------------------------------------------------------
+
+export class FrameRecorder {
+  private active = false;
+  private entries: RecordingEntry[] = [];
+  private startedAt: string | null = null;
+  private _seq = 0;
+
+  /** Begin capturing frames and lifecycle events. Idempotent. */
+  start(meta: Record<string, unknown> = {}): void {
+    if (this.active) return;
+    this.active = true;
+    this.entries = [];
+    this._seq = 0;
+    this.startedAt = new Date().toISOString();
+    this._meta = meta;
+  }
+
+  private _meta: Record<string, unknown> = {};
+
+  /** Whether recording is currently active. */
+  get isRecording(): boolean {
+    return this.active;
+  }
+
+  /**
+   * Stop capturing and return the completed `Recording`.
+   * Recording is cleared from memory — call `start()` again for a new session.
+   */
+  stop(): Recording {
+    this.active = false;
+    const rec: Recording = {
+      startedAt: this.startedAt ?? new Date().toISOString(),
+      stoppedAt: new Date().toISOString(),
+      entries: this.entries.slice(),
+      meta: { ...this._meta },
+    };
+    this.entries = [];
+    this.startedAt = null;
+    this._seq = 0;
+    return rec;
+  }
+
+  /**
+   * Record a data frame received from the WebSocket.
+   * No-op when recording is not active.
+   */
+  recordFrame(frame: RealtimeFrame): void {
+    if (!this.active) return;
+    this.entries.push({
+      kind: "frame",
+      topic: frame.topic,
+      payload: frame.payload,
+      receivedAt: frame.receivedAt,
+      seq: this._seq++,
+    });
+  }
+
+  /**
+   * Record a connection lifecycle event (connect, open, close, etc.).
+   * No-op when recording is not active.
+   */
+  recordLifecycle(event: string, data: Record<string, unknown>): void {
+    if (!this.active) return;
+    this.entries.push({
+      kind: "lifecycle",
+      event,
+      data,
+      timestamp: Date.now(),
+      seq: this._seq++,
+    });
+  }
+
+  /** Non-destructive peek at entries collected so far (while recording). */
+  peek(): RecordingEntry[] {
+    return this.entries.slice();
+  }
+}
+
+/** Module-level singleton shared with `RealtimeClient`. */
+export const frameRecorder = new FrameRecorder();
+
+// ---------------------------------------------------------------------------
+// Serialisation helpers
+// ---------------------------------------------------------------------------
+
+/** Serialise a `Recording` to a JSON string for fixture storage. */
+export function serializeRecording(recording: Recording): string {
+  return JSON.stringify(recording, null, 2);
+}
+
+/** Deserialise a JSON string produced by `serializeRecording`. */
+export function deserializeRecording(json: string): Recording {
+  const parsed = JSON.parse(json) as Recording;
+  if (!Array.isArray(parsed.entries)) {
+    throw new TypeError("Invalid recording: `entries` must be an array");
+  }
+  return parsed;
+}
+
+// ---------------------------------------------------------------------------
+// Replay engine
+// ---------------------------------------------------------------------------
+
+/** Minimal subset of `RealtimeClient` needed for replay. */
+export interface ReplayTarget {
+  /** Internal handlers map — we drive it directly for deterministic replay. */
+  readonly _handlers?: Map<string, Set<(payload: unknown) => void>>;
+  /** Public subscribe API (used to wire up test handlers before replay). */
+  subscribe(topic: string, handler: (payload: unknown) => void): () => void;
+}
+
+/**
+ * Internal: dispatch a `CapturedFrame` to all matching handlers on `target`.
+ * Accesses the private `_handlers` map when available to avoid triggering
+ * connect/lifecycle side-effects during replay; falls back to no-op if the
+ * internal map isn't accessible (e.g. a proxy or mock object without it).
+ */
+function dispatchToHandlers(target: ReplayTarget, frame: CapturedFrame): void {
+  // Access internal handler map directly so replay never opens a real socket.
+  // In tests, the client under replay is typically a fresh instance that has
+  // had subscribe() called on it but connect() never called.
+  const handlers = (target as unknown as { handlers: Map<string, Set<(p: unknown) => void>> })
+    .handlers;
+  if (handlers instanceof Map) {
+    handlers.get(frame.topic)?.forEach((h) => h(frame.payload));
+  }
+}
+
+export interface ReplayOptions {
+  /**
+   * When true (default), each frame is delivered with a `setTimeout` delay
+   * matching the original inter-frame timing. Set to `false` for synchronous,
+   * wall-clock-independent tests.
+   */
+  preserveTiming?: boolean;
+  /**
+   * Speed multiplier applied to all delays when `preserveTiming` is true.
+   * 1.0 = real time, 10.0 = 10× faster, 0.1 = 10× slower. Default: 1.0.
+   */
+  speed?: number;
+  /**
+   * Callback fired for lifecycle entries so tests can assert on connection
+   * events without coupling to OTel/recorder internals.
+   */
+  onLifecycle?: (event: LifecycleEvent) => void;
+}
+
+/**
+ * Replay a `Recording` against a `ReplayTarget`.
+ *
+ * Frames are dispatched in the order they were captured. Lifecycle events
+ * call `options.onLifecycle` (if provided). Timing between entries can be
+ * preserved at the original wall-clock speed (or scaled via `speed`).
+ *
+ * Returns a Promise that resolves when all entries have been dispatched.
+ */
+export async function replayRecording(
+  recording: Recording,
+  target: ReplayTarget,
+  options: ReplayOptions = {}
+): Promise<void> {
+  const { preserveTiming = true, speed = 1.0, onLifecycle } = options;
+
+  const entries = recording.entries.slice().sort((a, b) => a.seq - b.seq);
+  if (entries.length === 0) return;
+
+  // Build an absolute→relative timing map for timing-aware replay.
+  const firstTimestamp = entries[0].kind === "frame" ? entries[0].receivedAt : entries[0].timestamp;
+
+  for (let i = 0; i < entries.length; i++) {
+    const entry = entries[i];
+
+    if (preserveTiming && i > 0) {
+      const prevEntry = entries[i - 1];
+      const prevTs = prevEntry.kind === "frame" ? prevEntry.receivedAt : prevEntry.timestamp;
+      const currTs = entry.kind === "frame" ? entry.receivedAt : entry.timestamp;
+      const delay = Math.max(0, (currTs - prevTs) / speed);
+      if (delay > 0) {
+        await new Promise<void>((resolve) => setTimeout(resolve, delay));
+      }
+    }
+
+    if (entry.kind === "frame") {
+      dispatchToHandlers(target, entry);
+    } else {
+      onLifecycle?.(entry);
+    }
+  }
+
+  void firstTimestamp; // referenced above
+}
+
+/**
+ * Synchronous replay — delivers all data frames immediately without any
+ * `setTimeout` pauses. Lifecycle events still call `onLifecycle` synchronously.
+ * Preferred for unit tests where wall-clock reproducibility is not needed.
+ */
+export function replaySync(
+  recording: Recording,
+  target: ReplayTarget,
+  options: Omit<ReplayOptions, "preserveTiming" | "speed"> = {}
+): void {
+  const entries = recording.entries.slice().sort((a, b) => a.seq - b.seq);
+  for (const entry of entries) {
+    if (entry.kind === "frame") {
+      dispatchToHandlers(target, entry);
+    } else {
+      options.onLifecycle?.(entry);
+    }
+  }
+}
diff --git a/src/lib/realtime/ringBuffer.test.ts b/src/lib/realtime/ringBuffer.test.ts
new file mode 100644
index 0000000..c8b4479
--- /dev/null
+++ b/src/lib/realtime/ringBuffer.test.ts
@@ -0,0 +1,216 @@
+/**
+ * Unit tests for the bounded ring-buffer and topic registry.
+ * Closes #253.
+ */
+
+import { describe, it, expect, beforeEach } from "vitest";
+import { RingBuffer, TopicRegistry, DEFAULT_CAPACITY, MAX_GLOBAL_TOPICS } from "./ringBuffer";
+import type { RealtimeFrame } from "./ringBuffer";
+
+// ---------------------------------------------------------------------------
+// Helpers
+// ---------------------------------------------------------------------------
+
+function makeFrame(topic: string, seq: number, payload?: unknown): RealtimeFrame {
+  return { topic, payload: payload ?? { seq }, receivedAt: Date.now(), seq };
+}
+
+// ---------------------------------------------------------------------------
+// RingBuffer
+// ---------------------------------------------------------------------------
+
+describe("RingBuffer", () => {
+  it("rejects capacity < 1", () => {
+    expect(() => new RingBuffer({ capacity: 0 })).toThrow(RangeError);
+  });
+
+  it("stores frames in insertion order", () => {
+    const buf = new RingBuffer({ capacity: 4 });
+    for (let i = 0; i < 4; i++) buf.push(makeFrame("t", i));
+    const frames = buf.snapshot();
+    expect(frames.map((f) => (f.payload as { seq: number }).seq)).toEqual([0, 1, 2, 3]);
+  });
+
+  it("evicts oldest frame when full (FIFO drop)", () => {
+    const buf = new RingBuffer({ capacity: 3 });
+    for (let i = 0; i < 5; i++) buf.push(makeFrame("t", i));
+    const frames = buf.snapshot();
+    // Only the last 3 frames should be retained.
+    expect(frames.map((f) => (f.payload as { seq: number }).seq)).toEqual([2, 3, 4]);
+  });
+
+  it("returns the evicted frame from push()", () => {
+    const buf = new RingBuffer({ capacity: 2 });
+    buf.push(makeFrame("t", 0));
+    buf.push(makeFrame("t", 1));
+    const evicted = buf.push(makeFrame("t", 2));
+    expect(evicted).toBeDefined();
+    expect((evicted!.payload as { seq: number }).seq).toBe(0);
+  });
+
+  it("drain() returns frames and clears the buffer", () => {
+    const buf = new RingBuffer({ capacity: 4 });
+    for (let i = 0; i < 3; i++) buf.push(makeFrame("t", i));
+    const drained = buf.drain();
+    expect(drained).toHaveLength(3);
+    expect(buf.size).toBe(0);
+  });
+
+  it("snapshot() does NOT clear the buffer", () => {
+    const buf = new RingBuffer({ capacity: 4 });
+    for (let i = 0; i < 3; i++) buf.push(makeFrame("t", i));
+    buf.snapshot();
+    expect(buf.size).toBe(3);
+  });
+
+  it("size and isFull stay consistent through push/drain cycles", () => {
+    const buf = new RingBuffer({ capacity: 3 });
+    expect(buf.isFull).toBe(false);
+    buf.push(makeFrame("t", 0));
+    buf.push(makeFrame("t", 1));
+    buf.push(makeFrame("t", 2));
+    expect(buf.isFull).toBe(true);
+    buf.drain();
+    expect(buf.isFull).toBe(false);
+    expect(buf.size).toBe(0);
+  });
+
+  it("clear() empties the buffer", () => {
+    const buf = new RingBuffer({ capacity: 3 });
+    for (let i = 0; i < 3; i++) buf.push(makeFrame("t", i));
+    buf.clear();
+    expect(buf.size).toBe(0);
+    expect(buf.snapshot()).toHaveLength(0);
+  });
+
+  it("handles capacity=1 (single-slot buffer)", () => {
+    const buf = new RingBuffer({ capacity: 1 });
+    buf.push(makeFrame("t", 0));
+    buf.push(makeFrame("t", 1));
+    expect(buf.size).toBe(1);
+    const frames = buf.snapshot();
+    expect((frames[0].payload as { seq: number }).seq).toBe(1);
+  });
+
+  it("respects the default capacity", () => {
+    const buf = new RingBuffer();
+    expect(buf.capacity).toBe(DEFAULT_CAPACITY);
+  });
+
+  it("preserves insertion order after wrapping the internal pointer multiple times", () => {
+    const cap = 8;
+    const buf = new RingBuffer({ capacity: cap });
+    // Push 3 * capacity frames — the buffer should always hold the last `cap` frames.
+    const total = cap * 3;
+    for (let i = 0; i < total; i++) buf.push(makeFrame("t", i));
+    const frames = buf.snapshot();
+    expect(frames).toHaveLength(cap);
+    const expected = Array.from({ length: cap }, (_, j) => total - cap + j);
+    expect(frames.map((f) => (f.payload as { seq: number }).seq)).toEqual(expected);
+  });
+});
+
+// ---------------------------------------------------------------------------
+// TopicRegistry
+// ---------------------------------------------------------------------------
+
+describe("TopicRegistry", () => {
+  let registry: TopicRegistry;
+
+  beforeEach(() => {
+    registry = new TopicRegistry({ defaultCapacity: 8 });
+  });
+
+  it("records frames and makes them available via snapshot()", () => {
+    registry.record("oracle_update", { value: 1 });
+    registry.record("oracle_update", { value: 2 });
+    const frames = registry.snapshot("oracle_update");
+    expect(frames).toHaveLength(2);
+    expect((frames[1].payload as { value: number }).value).toBe(2);
+  });
+
+  it("assigns monotonically increasing seq numbers across topics", () => {
+    registry.record("topic_a", {});
+    registry.record("topic_b", {});
+    registry.record("topic_a", {});
+    const all = registry.snapshotAll();
+    expect(all.map((f) => f.seq)).toEqual([0, 1, 2]);
+  });
+
+  it("drain() returns frames and clears the topic buffer", () => {
+    for (let i = 0; i < 4; i++) registry.record("t", { i });
+    const drained = registry.drain("t");
+    expect(drained).toHaveLength(4);
+    expect(registry.snapshot("t")).toHaveLength(0);
+  });
+
+  it("drainAll() returns all frames sorted by seq, then clears all buffers", () => {
+    registry.record("a", { v: 1 });
+    registry.record("b", { v: 2 });
+    registry.record("a", { v: 3 });
+    const all = registry.drainAll();
+    expect(all.map((f) => f.topic)).toEqual(["a", "b", "a"]);
+    expect(registry.totalBuffered).toBe(0);
+  });
+
+  it("enforces maxTopics — excess topics are silently dropped", () => {
+    const reg = new TopicRegistry({ maxTopics: 2 });
+    reg.record("t1", {});
+    reg.record("t2", {});
+    const result = reg.record("t3", {}); // should be dropped
+    expect(result).toBeUndefined();
+    expect(reg.droppedTopics).toBe(1);
+    expect(reg.topics).not.toContain("t3");
+  });
+
+  it("defaults to MAX_GLOBAL_TOPICS when maxTopics not specified", () => {
+    const reg = new TopicRegistry();
+    // We can't easily push 64 topics in a unit test; just verify the cap constant.
+    expect(MAX_GLOBAL_TOPICS).toBe(64);
+  });
+
+  it("per-topic capacity override is respected", () => {
+    const reg = new TopicRegistry({
+      defaultCapacity: 4,
+      topicCapacities: { slow_topic: 2 },
+    });
+    for (let i = 0; i < 6; i++) reg.record("slow_topic", { i });
+    // Only the last 2 should be retained for slow_topic.
+    expect(reg.snapshot("slow_topic")).toHaveLength(2);
+  });
+
+  it("returns empty array for unknown topic in snapshot() and drain()", () => {
+    expect(registry.snapshot("nonexistent")).toEqual([]);
+    expect(registry.drain("nonexistent")).toEqual([]);
+  });
+
+  it("reset() clears all buffers and the seq counter", () => {
+    registry.record("a", {});
+    registry.record("b", {});
+    registry.reset();
+    expect(registry.topics).toHaveLength(0);
+    expect(registry.seq).toBe(0);
+    expect(registry.totalBuffered).toBe(0);
+  });
+
+  it("totalBuffered sums across all topics", () => {
+    registry.record("a", {});
+    registry.record("a", {});
+    registry.record("b", {});
+    expect(registry.totalBuffered).toBe(3);
+  });
+
+  it("evicts oldest frames per-topic when the buffer fills under load", () => {
+    const cap = 8;
+    const reg = new TopicRegistry({ defaultCapacity: cap });
+    const total = cap * 5;
+    for (let i = 0; i < total; i++) {
+      reg.record("oracle_update", { seq: i });
+    }
+    // Buffer should hold the last `cap` frames only.
+    const frames = reg.snapshot("oracle_update");
+    expect(frames).toHaveLength(cap);
+    expect((frames[0].payload as { seq: number }).seq).toBe(total - cap);
+    expect((frames[cap - 1].payload as { seq: number }).seq).toBe(total - 1);
+  });
+});
diff --git a/src/lib/realtime/ringBuffer.ts b/src/lib/realtime/ringBuffer.ts
new file mode 100644
index 0000000..6ea09a5
--- /dev/null
+++ b/src/lib/realtime/ringBuffer.ts
@@ -0,0 +1,239 @@
+/**
+ * Bounded, per-topic in-memory ring buffer for realtime frames.
+ *
+ * Closes #253: A high-frequency or misbehaving WebSocket stream must not
+ * grow client memory without limit. This module provides a fixed-capacity,
+ * O(1) push/drain ring buffer and a per-topic buffer registry that the
+ * realtime client uses to store received frames before dispatching them to
+ * subscribers.
+ *
+ * Documented caps (all tuneable via RingBufferOptions):
+ *   DEFAULT_CAPACITY  = 256 frames per topic (≈ ~50 kB worst-case per topic
+ *                       assuming a 200-byte JSON payload; negligible for
+ *                       typical payloads).
+ *   MAX_GLOBAL_TOPICS = 64 topics (prevents unbounded map growth if an
+ *                       attacker or bug introduces an unbounded topic space).
+ *
+ * When the buffer is full the oldest frame is silently evicted (FIFO drop).
+ * Frames are never kept longer than necessary: `drain()` atomically removes
+ * and returns all buffered frames, leaving the buffer empty.
+ */
+
+export interface RealtimeFrame<T = unknown> {
+  /** WebSocket topic identifier (e.g. "oracle_update"). */
+  topic: string;
+  /** The parsed message payload. */
+  payload: T;
+  /** Wall-clock timestamp (Date.now()) when the frame was received. */
+  receivedAt: number;
+  /** Monotonically increasing sequence number within the session (for replay ordering). */
+  seq: number;
+}
+
+export interface RingBufferOptions {
+  /** Maximum frames to hold before evicting the oldest. Default: 256. */
+  capacity?: number;
+}
+
+/** Default number of frames retained per topic. */
+export const DEFAULT_CAPACITY = 256;
+/** Hard upper limit on the number of distinct topics tracked. */
+export const MAX_GLOBAL_TOPICS = 64;
+
+/**
+ * Fixed-capacity FIFO ring buffer for `RealtimeFrame` objects.
+ *
+ * Memory layout: a fixed-size array with a write-head pointer that wraps
+ * modulo `capacity`. This avoids splice/shift overhead and ensures a strictly
+ * bounded allocation.
+ */
+export class RingBuffer<T = unknown> {
+  private readonly buf: (RealtimeFrame<T> | undefined)[];
+  private head = 0; // Points to the next write position.
+  private _size = 0;
+
+  readonly capacity: number;
+
+  constructor(options: RingBufferOptions = {}) {
+    this.capacity = options.capacity ?? DEFAULT_CAPACITY;
+    if (this.capacity < 1) throw new RangeError("RingBuffer capacity must be ≥ 1");
+    this.buf = new Array(this.capacity);
+  }
+
+  /** Number of frames currently held (0 ≤ size ≤ capacity). */
+  get size(): number {
+    return this._size;
+  }
+
+  /** True when no more frames can be added without evicting an old one. */
+  get isFull(): boolean {
+    return this._size === this.capacity;
+  }
+
+  /**
+   * Push a frame onto the buffer.
+   * If the buffer is full the oldest frame is silently overwritten (FIFO eviction).
+   * Returns the evicted frame if one was dropped, `undefined` otherwise.
+   */
+  push(frame: RealtimeFrame<T>): RealtimeFrame<T> | undefined {
+    let evicted: RealtimeFrame<T> | undefined;
+    if (this._size === this.capacity) {
+      evicted = this.buf[this.head];
+    } else {
+      this._size++;
+    }
+    this.buf[this.head] = frame;
+    this.head = (this.head + 1) % this.capacity;
+    return evicted;
+  }
+
+  /**
+   * Return all buffered frames in insertion order and clear the buffer.
+   * This is an atomic read-and-clear — callers that want replay semantics
+   * should use `snapshot()` instead.
+   */
+  drain(): RealtimeFrame<T>[] {
+    return this._collect(true);
+  }
+
+  /**
+   * Return all buffered frames in insertion order without clearing.
+   * Useful for replay / contract tests that need to inspect state
+   * non-destructively.
+   */
+  snapshot(): RealtimeFrame<T>[] {
+    return this._collect(false);
+  }
+
+  /** Remove all frames from the buffer. */
+  clear(): void {
+    this.buf.fill(undefined as unknown as RealtimeFrame<T>);
+    this.head = 0;
+    this._size = 0;
+  }
+
+  private _collect(clear: boolean): RealtimeFrame<T>[] {
+    if (this._size === 0) return [];
+    const tail = (this.head - this._size + this.capacity) % this.capacity;
+    const out: RealtimeFrame<T>[] = new Array(this._size);
+    for (let i = 0; i < this._size; i++) {
+      out[i] = this.buf[(tail + i) % this.capacity] as RealtimeFrame<T>;
+    }
+    if (clear) this.clear();
+    return out;
+  }
+}
+
+/**
+ * Per-topic ring-buffer registry.
+ *
+ * A single registry instance is shared by `RealtimeClient` to hold the
+ * most-recent frames for each known topic. Access is synchronous and O(1)
+ * per operation.
+ *
+ * Topic capacity can be overridden per-topic; topics not listed in
+ * `topicCapacities` use `defaultCapacity`.
+ */
+export interface RegistryOptions {
+  /** Fallback capacity for topics not named in `topicCapacities`. */
+  defaultCapacity?: number;
+  /** Per-topic capacity overrides. */
+  topicCapacities?: Record<string, number>;
+  /**
+   * Maximum number of distinct topics the registry will track.
+   * New topics beyond this limit are silently dropped. Default: MAX_GLOBAL_TOPICS.
+   */
+  maxTopics?: number;
+}
+
+export class TopicRegistry {
+  private readonly buffers = new Map<string, RingBuffer>();
+  private readonly defaultCapacity: number;
+  private readonly topicCapacities: Record<string, number>;
+  private readonly maxTopics: number;
+  private _seq = 0;
+  /** Total frames that were dropped because maxTopics was reached. */
+  droppedTopics = 0;
+
+  constructor(options: RegistryOptions = {}) {
+    this.defaultCapacity = options.defaultCapacity ?? DEFAULT_CAPACITY;
+    this.topicCapacities = options.topicCapacities ?? {};
+    this.maxTopics = options.maxTopics ?? MAX_GLOBAL_TOPICS;
+  }
+
+  /**
+   * Record a received frame in the appropriate topic buffer.
+   * Returns the frame that was stored (with seq assigned), or `undefined` if
+   * the registry has reached `maxTopics` and the topic is new.
+   */
+  record(topic: string, payload: unknown): RealtimeFrame | undefined {
+    if (!this.buffers.has(topic)) {
+      if (this.buffers.size >= this.maxTopics) {
+        this.droppedTopics++;
+        return undefined;
+      }
+      const cap = this.topicCapacities[topic] ?? this.defaultCapacity;
+      this.buffers.set(topic, new RingBuffer({ capacity: cap }));
+    }
+    const frame: RealtimeFrame = {
+      topic,
+      payload,
+      receivedAt: Date.now(),
+      seq: this._seq++,
+    };
+    this.buffers.get(topic)!.push(frame);
+    return frame;
+  }
+
+  /** Return a snapshot (non-destructive read) of all frames for a topic. */
+  snapshot(topic: string): RealtimeFrame[] {
+    return this.buffers.get(topic)?.snapshot() ?? [];
+  }
+
+  /** Drain (read-and-clear) all frames for a topic. */
+  drain(topic: string): RealtimeFrame[] {
+    return this.buffers.get(topic)?.drain() ?? [];
+  }
+
+  /** Drain all topics and return a flat, seq-ordered array of all frames. */
+  drainAll(): RealtimeFrame[] {
+    const all: RealtimeFrame[] = [];
+    for (const buf of this.buffers.values()) {
+      all.push(...buf.drain());
+    }
+    return all.sort((a, b) => a.seq - b.seq);
+  }
+
+  /** Return a non-destructive snapshot of all topics. */
+  snapshotAll(): RealtimeFrame[] {
+    const all: RealtimeFrame[] = [];
+    for (const buf of this.buffers.values()) {
+      all.push(...buf.snapshot());
+    }
+    return all.sort((a, b) => a.seq - b.seq);
+  }
+
+  /** Names of topics currently tracked (in insertion order). */
+  get topics(): string[] {
+    return Array.from(this.buffers.keys());
+  }
+
+  /** Clear all buffers and reset the global sequence counter. */
+  reset(): void {
+    this.buffers.clear();
+    this._seq = 0;
+    this.droppedTopics = 0;
+  }
+
+  /** How many frames have been buffered globally (across all topics). */
+  get totalBuffered(): number {
+    let n = 0;
+    for (const buf of this.buffers.values()) n += buf.size;
+    return n;
+  }
+
+  /** Current global sequence counter (total frames ever recorded). */
+  get seq(): number {
+    return this._seq;
+  }
+}
diff --git a/src/lib/realtime/stress.test.ts b/src/lib/realtime/stress.test.ts
new file mode 100644
index 0000000..83a697a
--- /dev/null
+++ b/src/lib/realtime/stress.test.ts
@@ -0,0 +1,439 @@
+/**
+ * Stress / load test harness for the realtime client.
+ *
+ * Closes #235: Drives the realtime client and the local mock WS server with
+ * sustained, high-frequency traffic across many topics and simulated
+ * subscribers, measuring memory growth, reconnect churn, and frame dispatch
+ * cost over long runs.
+ *
+ * These tests run in Node (Vitest) against a real in-process WebSocket server
+ * (`WsMockServer`). They are intentionally long-running relative to unit tests
+ * — set `STRESS_FRAMES` / `STRESS_HZ` env vars to tune them in CI.
+ *
+ * Design notes:
+ * - `WsMockServer` uses the `ws` npm package via a dynamic require in
+ *   `src/mocks/wsMockServer.ts`. If `ws` is not installed the tests are
+ *   skipped automatically (see the `wsAvailable` check at the top).
+ * - The `RealtimeClient` is exercised against the live WS server URL so the
+ *   full socket lifecycle (open, message, close, reconnect) is covered.
+ * - Memory growth is measured via `process.memoryUsage().heapUsed` before and
+ *   after the load run. The acceptable growth budget is deliberately generous
+ *   to avoid flakiness on different V8 GC schedules; its purpose is to catch
+ *   *catastrophic* leaks (e.g. an unbounded array), not micro-allocations.
+ */
+
+import { describe, it, expect, beforeEach, afterEach } from "vitest";
+import {
+  WsMockServer,
+  chaosPayloadFactory,
+  malformedPayloadFactory,
+} from "../../mocks/wsMockServer";
+import { TopicRegistry } from "./ringBuffer";
+import { frameRecorder, serializeRecording } from "./recorder";
+
+// ---------------------------------------------------------------------------
+// Constants — tunable via env for CI budget vs. thoroughness trade-off
+// ---------------------------------------------------------------------------
+
+const FRAMES = parseInt(process.env.STRESS_FRAMES ?? "500", 10);
+const HZ = parseInt(process.env.STRESS_HZ ?? "200", 10);
+/**
+ * Maximum heap growth permitted per run (bytes).
+ * 50 MB is generous enough to tolerate GC jitter while still catching leaks.
+ */
+const MAX_HEAP_GROWTH_BYTES = 50 * 1024 * 1024;
+
+// ---------------------------------------------------------------------------
+// Check whether `ws` is available (skip gracefully if not)
+// ---------------------------------------------------------------------------
+
+let wsAvailable = false;
+try {
+  require("ws");
+  wsAvailable = true;
+} catch {
+  /* ws not installed — skip WS integration tests */
+}
+
+const describeWs = wsAvailable ? describe : describe.skip;
+
+// ---------------------------------------------------------------------------
+// Helpers
+// ---------------------------------------------------------------------------
+
+function gcIfPossible(): void {
+  if (typeof globalThis.gc === "function") globalThis.gc();
+}
+
+/** Spin until `condition()` returns true or `timeoutMs` elapses. */
+async function waitFor(condition: () => boolean, timeoutMs = 3000, intervalMs = 20): Promise<void> {
+  const deadline = Date.now() + timeoutMs;
+  while (!condition()) {
+    if (Date.now() > deadline) throw new Error("waitFor timed out");
+    await new Promise((r) => setTimeout(r, intervalMs));
+  }
+}
+
+// ---------------------------------------------------------------------------
+// Mock-server unit tests (no live WebSocket needed)
+// ---------------------------------------------------------------------------
+
+describe("WsMockServer (unit, no ws required)", () => {
+  it("driveLoad rejects an empty topics array", async () => {
+    const server = new WsMockServer();
+    // We can't call start() without ws, so test the validation path directly.
+    await expect(
+      (server as unknown as { driveLoad: (p: unknown) => Promise<unknown> }).driveLoad({
+        topics: [],
+      })
+    ).rejects.toThrow(RangeError);
+  });
+
+  it("url reflects the configured port", () => {
+    const server = new WsMockServer({ port: 9999 });
+    // Port isn't set until start() is called, but we can verify the format.
+    expect(server.url).toMatch(/^ws:\/\/127\.0\.0\.1:\d+\/ws$/);
+  });
+});
+
+// ---------------------------------------------------------------------------
+// Ring-buffer memory-guard under high-frequency load (no WS needed)
+// ---------------------------------------------------------------------------
+
+describe("Ring-buffer memory guard (no ws required)", () => {
+  it("bounded registry does not grow beyond cap * maxTopics frames", () => {
+    const CAP = 64;
+    const MAX_TOPICS = 4;
+    const registry = new TopicRegistry({ defaultCapacity: CAP, maxTopics: MAX_TOPICS });
+
+    const topics = ["oracle_update", "claim_update", "pool_update", "policy_update"];
+    const TOTAL_FRAMES = CAP * 20; // 20× overload per topic
+
+    for (let i = 0; i < TOTAL_FRAMES; i++) {
+      const topic = topics[i % topics.length];
+      registry.record(topic, { i, topic });
+    }
+
+    expect(registry.totalBuffered).toBeLessThanOrEqual(CAP * MAX_TOPICS);
+    for (const topic of topics) {
+      expect(registry.snapshot(topic).length).toBeLessThanOrEqual(CAP);
+    }
+  });
+
+  it("excess topics beyond maxTopics are counted in droppedTopics", () => {
+    const registry = new TopicRegistry({ maxTopics: 2 });
+    registry.record("a", {});
+    registry.record("b", {});
+    registry.record("c", {}); // should be dropped
+    registry.record("d", {}); // should be dropped
+    expect(registry.droppedTopics).toBe(2);
+  });
+
+  it("total heap growth stays within budget when filling a large registry", () => {
+    gcIfPossible();
+    const before = process.memoryUsage().heapUsed;
+
+    const registry = new TopicRegistry({ defaultCapacity: 256, maxTopics: 64 });
+    for (let i = 0; i < 256 * 64 * 2; i++) {
+      registry.record(`topic_${i % 64}`, { seq: i, data: "x".repeat(64) });
+    }
+
+    gcIfPossible();
+    const after = process.memoryUsage().heapUsed;
+    const growth = after - before;
+    expect(growth).toBeLessThan(MAX_HEAP_GROWTH_BYTES);
+  });
+
+  it("drainAll() releases all frame references", () => {
+    const registry = new TopicRegistry({ defaultCapacity: 128 });
+    for (let i = 0; i < 512; i++) {
+      registry.record(`topic_${i % 4}`, { bigPayload: "y".repeat(256) });
+    }
+    registry.drainAll();
+    expect(registry.totalBuffered).toBe(0);
+  });
+});
+
+// ---------------------------------------------------------------------------
+// Frame recorder: high-volume capture
+// ---------------------------------------------------------------------------
+
+describe("Frame recorder: high-volume capture (no ws required)", () => {
+  beforeEach(() => frameRecorder.stop()); // ensure recorder is stopped
+
+  it("captures large numbers of frames without error", () => {
+    frameRecorder.start();
+    for (let i = 0; i < 1000; i++) {
+      frameRecorder.recordFrame({
+        topic: "oracle_update",
+        payload: { seq: i },
+        receivedAt: Date.now(),
+        seq: i,
+      });
+    }
+    const rec = frameRecorder.stop();
+    expect(rec.entries).toHaveLength(1000);
+  });
+
+  it("serialised recording stays within a reasonable size for 1 000 frames", () => {
+    frameRecorder.start();
+    for (let i = 0; i < 1000; i++) {
+      frameRecorder.recordFrame({
+        topic: "oracle_update",
+        payload: { i },
+        receivedAt: Date.now(),
+        seq: i,
+      });
+    }
+    const rec = frameRecorder.stop();
+    const json = serializeRecording(rec);
+    // 1 000 small frames should serialise to well under 1 MB.
+    expect(json.length).toBeLessThan(1024 * 1024);
+  });
+});
+
+// ---------------------------------------------------------------------------
+// WS integration stress tests (require `ws` package)
+// ---------------------------------------------------------------------------
+
+describeWs("WsMockServer integration stress tests (requires ws)", () => {
+  let server: WsMockServer;
+
+  beforeEach(async () => {
+    server = new WsMockServer({ port: 0 });
+    await server.start();
+  });
+
+  afterEach(async () => {
+    await server.stop();
+  });
+
+  it("server starts on an ephemeral port and exposes a ws:// URL", () => {
+    expect(server.port).toBeGreaterThan(0);
+    expect(server.url).toMatch(/^ws:\/\/127\.0\.0\.1:\d+\/ws$/);
+  });
+
+  it(`driveLoad emits ${FRAMES} frames at ≤${HZ} Hz with oracle_update topic`, async () => {
+    // Spin up a client using a raw WebSocket (avoids coupling to RealtimeClient internals).
+    const ws = require("ws");
+    const received: unknown[] = [];
+    const client = new ws.WebSocket(server.url) as {
+      on(event: "open" | "message" | "error" | "close", cb: (...args: unknown[]) => void): void;
+      close(): void;
+    };
+
+    await new Promise<void>((resolve, reject) => {
+      client.on("open", resolve);
+      client.on("error", reject);
+    });
+
+    client.on("message", (raw: unknown) => {
+      try {
+        received.push(JSON.parse(raw as string));
+      } catch {
+        /* ignore */
+      }
+    });
+
+    const stats = await server.driveLoad({
+      topics: ["oracle_update"],
+      framesPerSecond: HZ,
+      maxFrames: FRAMES,
+    });
+
+    // Allow a brief drain window.
+    await new Promise((r) => setTimeout(r, 50));
+    client.close();
+
+    expect(stats.framesSent).toBe(FRAMES);
+    expect(stats.errors).toBe(0);
+    expect(received.length).toBe(FRAMES);
+  }, 30_000);
+
+  it("multi-topic load distributes frames across all topics", async () => {
+    const ws = require("ws");
+    const topicCounts: Record<string, number> = {};
+
+    const client = new ws.WebSocket(server.url);
+    await new Promise<void>((resolve, reject) => {
+      client.on("open", resolve);
+      client.on("error", reject);
+    });
+    client.on("message", (raw: unknown) => {
+      try {
+        const msg = JSON.parse(raw as string) as { topic: string };
+        topicCounts[msg.topic] = (topicCounts[msg.topic] ?? 0) + 1;
+      } catch {
+        /* ignore */
+      }
+    });
+
+    const topics = ["oracle_update", "pool_update", "claim_update"];
+    const stats = await server.driveLoad({
+      topics,
+      framesPerSecond: HZ,
+      maxFrames: FRAMES,
+      makePayload: (topic, seq) => ({ topic, seq }),
+    });
+    await new Promise((r) => setTimeout(r, 50));
+    client.close();
+
+    expect(stats.framesSent).toBe(FRAMES);
+    // Each topic should receive roughly frames / topics.length frames.
+    const expected = Math.floor(FRAMES / topics.length);
+    for (const topic of topics) {
+      expect(topicCounts[topic] ?? 0).toBeGreaterThanOrEqual(expected - 1);
+    }
+  }, 30_000);
+
+  it("chaos payload factory does not crash the server or client", async () => {
+    const ws = require("ws");
+    let errorCount = 0;
+
+    const client = new ws.WebSocket(server.url);
+    await new Promise<void>((resolve, reject) => {
+      client.on("open", resolve);
+      client.on("error", reject);
+    });
+    client.on("error", () => errorCount++);
+
+    await server.driveLoad({
+      topics: ["oracle_update"],
+      framesPerSecond: HZ,
+      maxFrames: Math.min(FRAMES, 200),
+      makePayload: chaosPayloadFactory,
+    });
+    await new Promise((r) => setTimeout(r, 50));
+    client.close();
+
+    expect(errorCount).toBe(0);
+  }, 30_000);
+
+  it("malformed payload factory is handled gracefully (no crash)", async () => {
+    const ws = require("ws");
+    const received: unknown[] = [];
+
+    const client = new ws.WebSocket(server.url);
+    await new Promise<void>((resolve, reject) => {
+      client.on("open", resolve);
+      client.on("error", reject);
+    });
+    client.on("message", (raw: unknown) => {
+      try {
+        received.push(JSON.parse(raw as string));
+      } catch {
+        /* ignore */
+      }
+    });
+
+    await server.driveLoad({
+      topics: ["oracle_update"],
+      framesPerSecond: HZ,
+      maxFrames: Math.min(FRAMES, 100),
+      makePayload: malformedPayloadFactory,
+    });
+    await new Promise((r) => setTimeout(r, 50));
+    client.close();
+
+    // All frames should still arrive — the malformed payload factory produces
+    // valid JSON; it just omits some fields.
+    expect(received.length).toBeGreaterThan(0);
+  }, 30_000);
+
+  it("reconnect churn: client reconnects after server restart", async () => {
+    const ws = require("ws");
+    let connectCount = 0;
+    const server2 = new WsMockServer({ port: 0 });
+    await server2.start();
+
+    const client = new ws.WebSocket(server2.url) as {
+      on(event: "open", cb: () => void): void;
+    };
+    await new Promise<void>((resolve, reject) => {
+      client.on("open", () => {
+        connectCount++;
+        resolve();
+      });
+      (client as unknown as { on(e: "error", cb: (e: Error) => void): void }).on("error", reject);
+    });
+
+    await server2.stop();
+
+    // Start a new server on a different port to avoid address-in-use.
+    const server3 = new WsMockServer({ port: 0 });
+    await server3.start();
+
+    const client2 = new ws.WebSocket(server3.url);
+    await new Promise<void>((resolve, reject) => {
+      client2.on("open", () => {
+        connectCount++;
+        resolve();
+      });
+      client2.on("error", reject);
+    });
+
+    await server3.stop();
+    expect(connectCount).toBe(2);
+  }, 15_000);
+
+  it("memory growth stays within budget during a full load run", async () => {
+    gcIfPossible();
+    const before = process.memoryUsage().heapUsed;
+
+    const ws = require("ws");
+    const client = new ws.WebSocket(server.url);
+    await new Promise<void>((resolve, reject) => {
+      client.on("open", resolve);
+      client.on("error", reject);
+    });
+
+    await server.driveLoad({
+      topics: ["oracle_update", "claim_update", "pool_update"],
+      framesPerSecond: HZ,
+      maxFrames: FRAMES,
+    });
+    await new Promise((r) => setTimeout(r, 50));
+    client.close();
+
+    gcIfPossible();
+    const after = process.memoryUsage().heapUsed;
+    expect(after - before).toBeLessThan(MAX_HEAP_GROWTH_BYTES);
+  }, 30_000);
+
+  it("many simulated subscribers all receive every frame", async () => {
+    const ws = require("ws");
+    const SUBSCRIBER_COUNT = 10;
+    const FRAMES_PER_CLIENT = 50;
+    const receivedCounts = new Array<number>(SUBSCRIBER_COUNT).fill(0);
+    const clients: Array<{ close(): void }> = [];
+
+    for (let i = 0; i < SUBSCRIBER_COUNT; i++) {
+      const c = new ws.WebSocket(server.url);
+      const idx = i;
+      await new Promise<void>((resolve, reject) => {
+        c.on("open", resolve);
+        c.on("error", reject);
+      });
+      c.on("message", () => {
+        receivedCounts[idx]++;
+      });
+      clients.push(c);
+    }
+
+    await waitFor(() => server.clientCount === SUBSCRIBER_COUNT);
+
+    await server.driveLoad({
+      topics: ["oracle_update"],
+      framesPerSecond: HZ,
+      maxFrames: FRAMES_PER_CLIENT,
+    });
+    await new Promise((r) => setTimeout(r, 100));
+
+    for (const c of clients) c.close();
+
+    // Every subscriber should have received all frames.
+    for (const count of receivedCounts) {
+      expect(count).toBe(FRAMES_PER_CLIENT);
+    }
+  }, 30_000);
+});
diff --git a/src/lib/realtime/telemetry.test.ts b/src/lib/realtime/telemetry.test.ts
new file mode 100644
index 0000000..8a377d7
--- /dev/null
+++ b/src/lib/realtime/telemetry.test.ts
@@ -0,0 +1,282 @@
+/**
+ * Tests for the OpenTelemetry instrumentation shim around the realtime client.
+ * Closes #237.
+ *
+ * Strategy: inject no-op but spy-able tracer/meter doubles so no real OTel SDK
+ * is required. This validates that:
+ *   - Every lifecycle event calls the right tracer/meter method.
+ *   - Counter increments and histogram records are called with the right values.
+ *   - The shim never throws — instrumentation must not regress the app.
+ *   - Fallback behaviour works correctly when @opentelemetry/api is absent.
+ */
+
+import { describe, it, expect, vi, beforeEach } from "vitest";
+import { RealtimeTracer } from "./telemetry";
+
+// ---------------------------------------------------------------------------
+// Test doubles
+// ---------------------------------------------------------------------------
+
+function makeSpan() {
+  return {
+    end: vi.fn(),
+    setStatus: vi.fn(),
+    recordException: vi.fn(),
+    setAttribute: vi.fn(),
+  };
+}
+
+function makeTracer() {
+  const span = makeSpan();
+  return {
+    span,
+    startSpan: vi.fn(() => span),
+  };
+}
+
+function makeMeter() {
+  const counter = { add: vi.fn() };
+  const upDown = { add: vi.fn() };
+  const histogram = { record: vi.fn() };
+  return {
+    counter,
+    upDown,
+    histogram,
+    createCounter: vi.fn(() => counter),
+    createUpDownCounter: vi.fn(() => upDown),
+    createHistogram: vi.fn(() => histogram),
+  };
+}
+
+// ---------------------------------------------------------------------------
+// RealtimeTracer via factory
+// ---------------------------------------------------------------------------
+
+describe("RealtimeTracer", () => {
+  let tracer: ReturnType<typeof makeTracer>;
+  let meter: ReturnType<typeof makeMeter>;
+  let rt: RealtimeTracer;
+
+  beforeEach(() => {
+    tracer = makeTracer();
+    meter = makeMeter();
+    rt = RealtimeTracer.withBackends(tracer, meter);
+  });
+
+  // ── Connection lifecycle ─────────────────────────────────────────────────
+
+  it("onConnectAttempt increments connectAttempts counter", () => {
+    rt.onConnectAttempt("ws://localhost/ws", 0);
+    expect(meter.counter.add).toHaveBeenCalledWith(1, expect.objectContaining({ attempt: 0 }));
+  });
+
+  it("onConnectAttempt starts a connect span on first attempt", () => {
+    rt.onConnectAttempt("ws://localhost/ws", 0);
+    expect(tracer.startSpan).toHaveBeenCalledWith(
+      "realtime connect",
+      expect.objectContaining({
+        attributes: expect.objectContaining({ "net.peer.address": "ws://localhost/ws" }),
+      })
+    );
+  });
+
+  it("onConnectAttempt does NOT start a second span if one is already open", () => {
+    rt.onConnectAttempt("ws://localhost/ws", 0);
+    rt.onConnectAttempt("ws://localhost/ws", 1); // second attempt before first closes
+    expect(tracer.startSpan).toHaveBeenCalledTimes(1);
+  });
+
+  it("onConnectError increments connectErrors counter and ends span", () => {
+    rt.onConnectAttempt("ws://localhost/ws", 0);
+    rt.onConnectError("ws://localhost/ws", new Error("refused"));
+    expect(meter.counter.add).toHaveBeenCalledWith(
+      1,
+      expect.objectContaining({ "net.peer.address": "ws://localhost/ws" })
+    );
+    expect(tracer.span.end).toHaveBeenCalled();
+    expect(tracer.span.recordException).toHaveBeenCalled();
+  });
+
+  it("onOpen sets the 'connected' attribute on the active span", () => {
+    rt.onConnectAttempt("ws://localhost/ws", 0);
+    rt.onOpen("ws://localhost/ws");
+    expect(tracer.span.setAttribute).toHaveBeenCalledWith("connected", true);
+  });
+
+  it("onClose ends the span and records session duration", () => {
+    rt.onConnectAttempt("ws://localhost/ws", 0);
+    rt.onOpen("ws://localhost/ws");
+    rt.onClose("ws://localhost/ws");
+    expect(tracer.span.end).toHaveBeenCalled();
+    expect(meter.histogram.record).toHaveBeenCalledWith(expect.any(Number));
+  });
+
+  it("onError increments connectErrors (without an active span)", () => {
+    rt.onError("ws://localhost/ws");
+    expect(meter.counter.add).toHaveBeenCalledWith(
+      1,
+      expect.objectContaining({ "net.peer.address": "ws://localhost/ws" })
+    );
+  });
+
+  // ── Reconnects ────────────────────────────────────────────────────────────
+
+  it("onReconnectScheduled increments reconnects counter with delay and attempt attributes", () => {
+    rt.onReconnectScheduled(2000, 1);
+    expect(meter.counter.add).toHaveBeenCalledWith(
+      1,
+      expect.objectContaining({ attempt: 1, delay_ms: 2000 })
+    );
+  });
+
+  // ── Messages ─────────────────────────────────────────────────────────────
+
+  it("onMessage increments messages counter with topic attribute", () => {
+    rt.onMessage("oracle_update");
+    expect(meter.counter.add).toHaveBeenCalledWith(
+      1,
+      expect.objectContaining({ topic: "oracle_update" })
+    );
+  });
+
+  it("onMessage starts and immediately ends a 'realtime message' span", () => {
+    rt.onMessage("oracle_update");
+    expect(tracer.startSpan).toHaveBeenCalledWith(
+      "realtime message",
+      expect.objectContaining({ attributes: expect.objectContaining({ topic: "oracle_update" }) })
+    );
+    expect(tracer.span.end).toHaveBeenCalled();
+  });
+
+  // ── Subscribers ───────────────────────────────────────────────────────────
+
+  it("onSubscribe adds 1 to the upDownCounter for the topic", () => {
+    rt.onSubscribe("oracle_update", 1);
+    expect(meter.upDown.add).toHaveBeenCalledWith(
+      1,
+      expect.objectContaining({ topic: "oracle_update" })
+    );
+  });
+
+  it("onUnsubscribe subtracts 1 from the upDownCounter for the topic", () => {
+    rt.onSubscribe("oracle_update", 1);
+    rt.onUnsubscribe("oracle_update", 0);
+    expect(meter.upDown.add).toHaveBeenLastCalledWith(
+      -1,
+      expect.objectContaining({ topic: "oracle_update" })
+    );
+  });
+
+  // ── Disconnect ────────────────────────────────────────────────────────────
+
+  it("onDisconnect records session duration", () => {
+    rt.onConnectAttempt("ws://localhost/ws", 0);
+    rt.onDisconnect();
+    expect(meter.histogram.record).toHaveBeenCalledWith(expect.any(Number));
+  });
+
+  it("onDisconnect is safe to call without a prior connect", () => {
+    expect(() => rt.onDisconnect()).not.toThrow();
+  });
+
+  // ── Fallback ─────────────────────────────────────────────────────────────
+
+  it("onFallbackActivated increments fallback counter with reason attribute", () => {
+    rt.onFallbackActivated("realtime-oracle flag disabled");
+    expect(meter.counter.add).toHaveBeenCalledWith(
+      1,
+      expect.objectContaining({ reason: "realtime-oracle flag disabled" })
+    );
+  });
+
+  it("onFallbackActivated emits a 'realtime fallback' span", () => {
+    rt.onFallbackActivated("ws unavailable");
+    expect(tracer.startSpan).toHaveBeenCalledWith(
+      "realtime fallback",
+      expect.objectContaining({ attributes: expect.objectContaining({ reason: "ws unavailable" }) })
+    );
+    expect(tracer.span.end).toHaveBeenCalled();
+  });
+
+  // ── Meter initialisation ─────────────────────────────────────────────────
+
+  it("creates all expected metrics on construction", () => {
+    expect(meter.createCounter).toHaveBeenCalledWith(
+      "realtime.connect.attempts",
+      expect.any(Object)
+    );
+    expect(meter.createCounter).toHaveBeenCalledWith("realtime.connect.errors", expect.any(Object));
+    expect(meter.createCounter).toHaveBeenCalledWith("realtime.reconnects", expect.any(Object));
+    expect(meter.createCounter).toHaveBeenCalledWith("realtime.messages", expect.any(Object));
+    expect(meter.createUpDownCounter).toHaveBeenCalledWith(
+      "realtime.subscribe.active",
+      expect.any(Object)
+    );
+    expect(meter.createCounter).toHaveBeenCalledWith(
+      "realtime.fallback.activations",
+      expect.any(Object)
+    );
+    expect(meter.createHistogram).toHaveBeenCalledWith(
+      "realtime.session.duration_ms",
+      expect.any(Object)
+    );
+  });
+});
+
+// ---------------------------------------------------------------------------
+// NOOP fallback — instrumentation must never throw even if OTel is absent
+// ---------------------------------------------------------------------------
+
+describe("RealtimeTracer no-op safety", () => {
+  it("module-level singleton does not throw during construction", async () => {
+    // The singleton is already imported in the module under test;
+    // re-importing it here ensures that even in a test environment without
+    // an OTel SDK it initialises without error.
+    await expect(import("./telemetry")).resolves.toBeDefined();
+  });
+
+  it("all lifecycle methods are safe to call on the singleton", async () => {
+    const { realtimeTracer } = await import("./telemetry");
+    expect(() => {
+      realtimeTracer.onConnectAttempt("ws://test", 0);
+      realtimeTracer.onOpen("ws://test");
+      realtimeTracer.onMessage("oracle_update");
+      realtimeTracer.onSubscribe("oracle_update", 1);
+      realtimeTracer.onUnsubscribe("oracle_update", 0);
+      realtimeTracer.onReconnectScheduled(1000, 1);
+      realtimeTracer.onError("ws://test");
+      realtimeTracer.onClose("ws://test");
+      realtimeTracer.onFallbackActivated("test");
+      realtimeTracer.onDisconnect();
+    }).not.toThrow();
+  });
+});
+
+// ---------------------------------------------------------------------------
+// Counter/histogram call counts through a full connection lifecycle
+// ---------------------------------------------------------------------------
+
+describe("RealtimeTracer: full lifecycle event counts", () => {
+  it("single connect→message→disconnect cycle emits the right number of calls", () => {
+    const tracer = makeTracer();
+    const meter = makeMeter();
+    const rt = RealtimeTracer.withBackends(tracer, meter);
+
+    // Full lifecycle.
+    rt.onConnectAttempt("ws://localhost/ws", 0);
+    rt.onOpen("ws://localhost/ws");
+    rt.onSubscribe("oracle_update", 1);
+    rt.onMessage("oracle_update");
+    rt.onMessage("oracle_update");
+    rt.onUnsubscribe("oracle_update", 0);
+    rt.onClose("ws://localhost/ws");
+
+    // connectAttempts=1, messages=2 → counter.add called 3 times total
+    // (1 connect + 2 message) plus upDown calls (2) = 5 total counter.add + upDown.add calls
+    const counterCalls = meter.counter.add.mock.calls.length;
+    const upDownCalls = meter.upDown.add.mock.calls.length;
+    expect(counterCalls).toBeGreaterThanOrEqual(3); // at minimum: 1 connect + 2 messages
+    expect(upDownCalls).toBe(2); // subscribe + unsubscribe
+    expect(meter.histogram.record).toHaveBeenCalledTimes(1); // onClose
+  });
+});
diff --git a/src/lib/realtime/telemetry.ts b/src/lib/realtime/telemetry.ts
new file mode 100644
index 0000000..0972e6b
--- /dev/null
+++ b/src/lib/realtime/telemetry.ts
@@ -0,0 +1,263 @@
+/**
+ * OpenTelemetry-compatible instrumentation for the realtime WebSocket client.
+ *
+ * Closes #237: Emits spans and counters/duration histograms around connection
+ * attempts, subscribes, frame handling, reconnects, and fallback activations.
+ * Gives the team aggregate, backend-correlatable visibility rather than the
+ * per-session view the health observability work provides.
+ *
+ * ## Design decisions
+ *
+ * 1. **No hard OTel SDK dependency in production.** The `@opentelemetry/api`
+ *    package is the zero-overhead facade that only performs real work when an
+ *    SDK is registered. We call `trace.getTracer()` and `metrics.getMeter()`
+ *    at module initialisation; both return no-op implementations when no SDK
+ *    is bootstrapped, so the instrumentation is always safe to import.
+ *
+ * 2. **Pluggable fallback.** If `@opentelemetry/api` is not installed at all
+ *    (it is currently only an optional peer dep of vitest), `RealtimeTracer`
+ *    falls back to a lightweight no-op shim so the build never fails.
+ *
+ * 3. **Counter + histogram names** follow OTel semantic-conventions style:
+ *      `realtime.connect.attempts`     (counter)
+ *      `realtime.connect.errors`       (counter)
+ *      `realtime.reconnects`           (counter)
+ *      `realtime.messages`             (counter,  attributes: topic)
+ *      `realtime.subscribe.active`     (upDownCounter, attributes: topic)
+ *      `realtime.fallback.activations` (counter,  attributes: reason)
+ *      `realtime.session.duration_ms`  (histogram, recorded on disconnect)
+ *
+ * 4. **Span names** (all under the `realtime` scope):
+ *      `realtime connect`     – wraps open→close lifecycle
+ *      `realtime message`     – wraps single message dispatch (short)
+ *
+ * To bootstrap OTel, install `@opentelemetry/sdk-web` (or node equivalent)
+ * and call `sdk.start()` before mounting the React tree. This file does not
+ * need to change.
+ */
+
+// ---------------------------------------------------------------------------
+// OTel API — optional import with zero-runtime-cost fallback
+// ---------------------------------------------------------------------------
+
+/** Minimal subset of the OTel Tracer API we need. */
+interface OtelSpan {
+  end(): void;
+  setStatus(s: { code: number }): void;
+  recordException(e: unknown): void;
+  setAttribute(key: string, value: string | number | boolean): void;
+}
+interface OtelTracer {
+  startSpan(
+    name: string,
+    options?: { attributes?: Record<string, string | number | boolean> }
+  ): OtelSpan;
+}
+interface OtelCounter {
+  add(value: number, attributes?: Record<string, string | number | boolean>): void;
+}
+interface OtelUpDownCounter {
+  add(value: number, attributes?: Record<string, string | number | boolean>): void;
+}
+interface OtelHistogram {
+  record(value: number, attributes?: Record<string, string | number | boolean>): void;
+}
+interface OtelMeter {
+  createCounter(name: string, options?: { description?: string }): OtelCounter;
+  createUpDownCounter(name: string, options?: { description?: string }): OtelUpDownCounter;
+  createHistogram(name: string, options?: { description?: string; unit?: string }): OtelHistogram;
+}
+
+/** No-op shim used when @opentelemetry/api is absent or no SDK is registered. */
+const NOOP_SPAN: OtelSpan = {
+  end: () => undefined,
+  setStatus: () => undefined,
+  recordException: () => undefined,
+  setAttribute: () => undefined,
+};
+const NOOP_TRACER: OtelTracer = {
+  startSpan: () => NOOP_SPAN,
+};
+const NOOP_COUNTER: OtelCounter = { add: () => undefined };
+const NOOP_UP_DOWN: OtelUpDownCounter = { add: () => undefined };
+const NOOP_HISTOGRAM: OtelHistogram = { record: () => undefined };
+const NOOP_METER: OtelMeter = {
+  createCounter: () => NOOP_COUNTER,
+  createUpDownCounter: () => NOOP_UP_DOWN,
+  createHistogram: () => NOOP_HISTOGRAM,
+};
+
+function tryGetTracer(name: string, version: string): OtelTracer {
+  try {
+    // Dynamic require — only resolves if @opentelemetry/api is installed.
+    const otel = require("@opentelemetry/api") as {
+      trace: { getTracer(name: string, version?: string): OtelTracer };
+    };
+    return otel.trace.getTracer(name, version);
+  } catch {
+    return NOOP_TRACER;
+  }
+}
+
+function tryGetMeter(name: string, version: string): OtelMeter {
+  try {
+    const otel = require("@opentelemetry/api") as {
+      metrics: { getMeter(name: string, version?: string): OtelMeter };
+    };
+    return otel.metrics.getMeter(name, version);
+  } catch {
+    return NOOP_METER;
+  }
+}
+
+// ---------------------------------------------------------------------------
+// RealtimeTracer
+// ---------------------------------------------------------------------------
+
+const SCOPE_NAME = "refract.realtime";
+const SCOPE_VERSION = "0.1.0";
+
+export class RealtimeTracer {
+  private readonly tracer: OtelTracer;
+  private readonly meter: OtelMeter;
+
+  // Metrics
+  private readonly connectAttempts: OtelCounter;
+  private readonly connectErrors: OtelCounter;
+  private readonly reconnects: OtelCounter;
+  private readonly messages: OtelCounter;
+  private readonly activeSubscribers: OtelUpDownCounter;
+  private readonly fallbackActivations: OtelCounter;
+  private readonly sessionDuration: OtelHistogram;
+
+  // State for session timing
+  private sessionStart: number | null = null;
+
+  // Active connection span (open → close).
+  private connectSpan: OtelSpan | null = null;
+
+  constructor(
+    tracer: OtelTracer = tryGetTracer(SCOPE_NAME, SCOPE_VERSION),
+    meter: OtelMeter = tryGetMeter(SCOPE_NAME, SCOPE_VERSION)
+  ) {
+    this.tracer = tracer;
+    this.meter = meter;
+
+    this.connectAttempts = this.meter.createCounter("realtime.connect.attempts", {
+      description: "Total WebSocket connection attempts (including reconnects).",
+    });
+    this.connectErrors = this.meter.createCounter("realtime.connect.errors", {
+      description:
+        "Total WebSocket connection errors (new WebSocket() threw or error event fired).",
+    });
+    this.reconnects = this.meter.createCounter("realtime.reconnects", {
+      description: "Total reconnect attempts scheduled after an unexpected close.",
+    });
+    this.messages = this.meter.createCounter("realtime.messages", {
+      description: "Total frames dispatched from the WebSocket, broken down by topic.",
+    });
+    this.activeSubscribers = this.meter.createUpDownCounter("realtime.subscribe.active", {
+      description:
+        "Net active subscriber count per topic (increments on subscribe, decrements on unsubscribe).",
+    });
+    this.fallbackActivations = this.meter.createCounter("realtime.fallback.activations", {
+      description: "Times the REST fallback path was activated because realtime was unavailable.",
+    });
+    this.sessionDuration = this.meter.createHistogram("realtime.session.duration_ms", {
+      description: "Duration of a realtime connection session in milliseconds (open → close).",
+      unit: "ms",
+    });
+  }
+
+  onConnectAttempt(wsUrl: string, attempt: number): void {
+    this.connectAttempts.add(1, { attempt });
+    if (!this.connectSpan) {
+      this.connectSpan = this.tracer.startSpan("realtime connect", {
+        attributes: { "net.peer.address": wsUrl, attempt },
+      });
+    }
+    if (this.sessionStart === null) {
+      this.sessionStart = Date.now();
+    }
+  }
+
+  onConnectError(wsUrl: string, err: unknown): void {
+    this.connectErrors.add(1, { "net.peer.address": wsUrl });
+    if (this.connectSpan) {
+      this.connectSpan.recordException(err);
+      this.connectSpan.setStatus({ code: 2 /* ERROR */ });
+      this.connectSpan.end();
+      this.connectSpan = null;
+    }
+  }
+
+  onOpen(wsUrl: string): void {
+    if (this.connectSpan) {
+      this.connectSpan.setAttribute("net.peer.address", wsUrl);
+      this.connectSpan.setAttribute("connected", true);
+    }
+  }
+
+  onClose(wsUrl: string): void {
+    if (this.connectSpan) {
+      this.connectSpan.setAttribute("net.peer.address", wsUrl);
+      this.connectSpan.end();
+      this.connectSpan = null;
+    }
+    if (this.sessionStart !== null) {
+      this.sessionDuration.record(Date.now() - this.sessionStart);
+    }
+  }
+
+  onError(wsUrl: string): void {
+    this.connectErrors.add(1, { "net.peer.address": wsUrl });
+  }
+
+  onReconnectScheduled(delayMs: number, attempt: number): void {
+    this.reconnects.add(1, { attempt, delay_ms: delayMs });
+  }
+
+  onMessage(topic: string): void {
+    this.messages.add(1, { topic });
+    const span = this.tracer.startSpan("realtime message", { attributes: { topic } });
+    span.end();
+  }
+
+  onSubscribe(topic: string, subscriberCount: number): void {
+    this.activeSubscribers.add(1, { topic });
+    void subscriberCount; // available for richer attributes if desired later
+  }
+
+  onUnsubscribe(topic: string, remaining: number): void {
+    this.activeSubscribers.add(-1, { topic });
+    void remaining;
+  }
+
+  onDisconnect(): void {
+    if (this.sessionStart !== null) {
+      this.sessionDuration.record(Date.now() - this.sessionStart);
+      this.sessionStart = null;
+    }
+    if (this.connectSpan) {
+      this.connectSpan.end();
+      this.connectSpan = null;
+    }
+  }
+
+  onFallbackActivated(reason: string): void {
+    this.fallbackActivations.add(1, { reason });
+    const span = this.tracer.startSpan("realtime fallback", { attributes: { reason } });
+    span.end();
+  }
+
+  /**
+   * Inject a custom tracer/meter (used in tests to capture telemetry calls
+   * without a real OTel SDK).
+   */
+  static withBackends(tracer: OtelTracer, meter: OtelMeter): RealtimeTracer {
+    return new RealtimeTracer(tracer, meter);
+  }
+}
+
+/** Module-level singleton — shares lifetime with the realtime client. */
+export const realtimeTracer = new RealtimeTracer();
diff --git a/src/mocks/wsMockServer.ts b/src/mocks/wsMockServer.ts
new file mode 100644
index 0000000..3814c4c
--- /dev/null
+++ b/src/mocks/wsMockServer.ts
@@ -0,0 +1,332 @@
+/**
+ * High-throughput WebSocket mock server for the realtime client test harness.
+ *
+ * Closes #235: The existing MSW HTTP handlers cover correctness of individual
+ * frames, not volume. This module provides a Node-native, lightweight WS server
+ * capable of driving the realtime client with sustained high-frequency traffic
+ * across many topics and simulated subscribers, so that memory growth,
+ * reconnect churn, and frame dispatch cost can be measured in stress tests.
+ *
+ * ## Design
+ *
+ * - Built on Node's `http` module + a minimal WebSocket implementation
+ *   using `ws` (already present transitively; guarded with a runtime check).
+ *   Falls back to a pure-Node TCP no-op server if `ws` is unavailable, which
+ *   lets the file import safely in test environments where `ws` isn't installed.
+ * - The server is NOT MSW-based because MSW's `ws` interceptor is designed
+ *   for browser/service-worker environments; these stress tests run in Node
+ *   via Vitest.
+ * - Exposes a `WsMockServer` class with `start()`, `stop()`, `broadcast()`,
+ *   `push()`, and a `driveLoad()` helper that pushes frames at a configurable
+ *   Hz across a configurable topic list.
+ */
+
+import * as http from "http";
+import * as net from "net";
+
+// ---------------------------------------------------------------------------
+// Minimal ws shim — prefer the real `ws` package, fall back to no-op.
+// ---------------------------------------------------------------------------
+
+interface WsClient {
+  send(data: string): void;
+  close(): void;
+  on(event: "close", cb: () => void): this;
+  on(event: "error", cb: (err: Error) => void): this;
+}
+interface WsServer {
+  handleUpgrade(
+    req: http.IncomingMessage,
+    socket: net.Socket,
+    head: Buffer,
+    cb: (ws: WsClient) => void
+  ): void;
+  close(cb?: () => void): void;
+}
+
+function createWsServer(): WsServer | null {
+  try {
+    // Dynamic require — ws is only available when installed (optional for the build).
+    const ws = require("ws") as { Server: new (options: { noServer: boolean }) => WsServer };
+    return new ws.Server({ noServer: true });
+  } catch {
+    return null;
+  }
+}
+
+// ---------------------------------------------------------------------------
+// WsMockServer
+// ---------------------------------------------------------------------------
+
+export interface WsMockServerOptions {
+  /**
+   * Port to listen on. Default: 0 (OS-assigned ephemeral port).
+   * After `start()` resolves, read `server.port` for the actual port.
+   */
+  port?: number;
+  /**
+   * Whether to log debug output. Default: false.
+   */
+  debug?: boolean;
+}
+
+export interface LoadProfile {
+  /** Topics to emit frames for (round-robin). */
+  topics: string[];
+  /** Target frames per second (summed across all topics). Default: 100. */
+  framesPerSecond?: number;
+  /** Payload factory called for each frame. Default: a minimal oracle_update payload. */
+  makePayload?: (topic: string, seq: number) => unknown;
+  /** Total number of frames to emit before stopping. Default: Infinity (runs until stop()). */
+  maxFrames?: number;
+}
+
+/** Recorded per-run statistics from `driveLoad()`. */
+export interface LoadStats {
+  framesSent: number;
+  durationMs: number;
+  /** Approx. Hz actually achieved (may differ from target under load). */
+  actualHz: number;
+  /** Number of connected clients during the run. */
+  clientCount: number;
+  errors: number;
+}
+
+/**
+ * Lightweight in-process WebSocket server for stress testing.
+ *
+ * Lifecycle:
+ * ```ts
+ * const server = new WsMockServer({ port: 0 });
+ * await server.start();
+ * // server.url is the ws:// URL to point RealtimeClient at
+ * const stats = await server.driveLoad({ topics: ["oracle_update"], framesPerSecond: 500, maxFrames: 10_000 });
+ * await server.stop();
+ * ```
+ */
+export class WsMockServer {
+  private httpServer: http.Server | null = null;
+  private wss: WsServer | null = null;
+  private clients = new Set<WsClient>();
+  private readonly options: Required<WsMockServerOptions>;
+  private _port = 0;
+  private _errors = 0;
+
+  constructor(options: WsMockServerOptions = {}) {
+    this.options = {
+      port: options.port ?? 0,
+      debug: options.debug ?? false,
+    };
+  }
+
+  /** The actual port the server is listening on (set after `start()` resolves). */
+  get port(): number {
+    return this._port;
+  }
+
+  /** The WebSocket URL clients should connect to. */
+  get url(): string {
+    return `ws://127.0.0.1:${this._port}/ws`;
+  }
+
+  /** Number of currently connected clients. */
+  get clientCount(): number {
+    return this.clients.size;
+  }
+
+  /** Start the HTTP/WS server. */
+  async start(): Promise<void> {
+    this.wss = createWsServer();
+    this.httpServer = http.createServer();
+
+    if (this.wss) {
+      const wss = this.wss;
+      const clients = this.clients;
+      const debug = this.options.debug;
+
+      this.httpServer.on("upgrade", (req, socket, head) => {
+        wss.handleUpgrade(req, socket as net.Socket, head, (ws) => {
+          clients.add(ws);
+          if (debug) console.debug("[WsMockServer] client connected, total:", clients.size);
+          ws.on("close", () => {
+            clients.delete(ws);
+            if (debug) console.debug("[WsMockServer] client disconnected, total:", clients.size);
+          });
+          ws.on("error", (err: Error) => {
+            if (debug) console.debug("[WsMockServer] client error:", err.message);
+          });
+        });
+      });
+    }
+
+    return new Promise((resolve, reject) => {
+      this.httpServer!.listen(this.options.port, "127.0.0.1", () => {
+        const addr = this.httpServer!.address();
+        if (addr && typeof addr === "object") {
+          this._port = addr.port;
+        }
+        resolve();
+      });
+      this.httpServer!.on("error", reject);
+    });
+  }
+
+  /** Broadcast a JSON frame to all connected clients. */
+  broadcast(topic: string, payload: unknown): void {
+    const msg = JSON.stringify({ topic, payload });
+    for (const client of this.clients) {
+      try {
+        client.send(msg);
+      } catch {
+        this._errors++;
+      }
+    }
+  }
+
+  /**
+   * Push a single frame to a specific client (by index, 0-based).
+   * Falls back to broadcast if index is out of range.
+   */
+  push(topic: string, payload: unknown, clientIndex = 0): void {
+    const client = Array.from(this.clients)[clientIndex];
+    if (!client) {
+      this.broadcast(topic, payload);
+      return;
+    }
+    try {
+      client.send(JSON.stringify({ topic, payload }));
+    } catch {
+      this._errors++;
+    }
+  }
+
+  /**
+   * Drive the connected client(s) with sustained high-frequency traffic.
+   *
+   * Returns a `LoadStats` object with timing and throughput metrics.
+   */
+  async driveLoad(profile: LoadProfile): Promise<LoadStats> {
+    const {
+      topics,
+      framesPerSecond = 100,
+      makePayload = defaultPayloadFactory,
+      maxFrames = Infinity,
+    } = profile;
+
+    if (topics.length === 0) throw new RangeError("LoadProfile.topics must be non-empty");
+
+    const intervalMs = 1000 / framesPerSecond;
+    let framesSent = 0;
+    let errors = 0;
+    const start = Date.now();
+
+    return new Promise((resolve) => {
+      let seq = 0;
+      const tick = () => {
+        if (framesSent >= maxFrames) {
+          const durationMs = Date.now() - start;
+          resolve({
+            framesSent,
+            durationMs,
+            actualHz: framesSent / (durationMs / 1000),
+            clientCount: this.clients.size,
+            errors,
+          });
+          return;
+        }
+
+        const topic = topics[seq % topics.length];
+        try {
+          this.broadcast(topic, makePayload(topic, seq));
+        } catch {
+          errors++;
+        }
+        framesSent++;
+        seq++;
+
+        // Use setImmediate when possible to avoid blocking the event loop
+        // for extremely high rates, falling back to setTimeout(0).
+        if (framesSent < maxFrames) {
+          if (intervalMs <= 1) {
+            setImmediate(tick);
+          } else {
+            setTimeout(tick, intervalMs);
+          }
+        } else {
+          const durationMs = Date.now() - start;
+          resolve({
+            framesSent,
+            durationMs,
+            actualHz: framesSent / (durationMs / 1000),
+            clientCount: this.clients.size,
+            errors,
+          });
+        }
+      };
+
+      if (intervalMs <= 1) {
+        setImmediate(tick);
+      } else {
+        setTimeout(tick, intervalMs);
+      }
+    });
+  }
+
+  /** Gracefully close all client connections and stop the HTTP server. */
+  async stop(): Promise<void> {
+    for (const client of this.clients) {
+      try {
+        client.close();
+      } catch {
+        /* ignore */
+      }
+    }
+    this.clients.clear();
+
+    return new Promise((resolve) => {
+      if (this.wss) {
+        this.wss.close(() => {
+          this.httpServer?.close(() => resolve());
+        });
+      } else {
+        this.httpServer?.close(() => resolve());
+      }
+    });
+  }
+}
+
+// ---------------------------------------------------------------------------
+// Payload factory helpers
+// ---------------------------------------------------------------------------
+
+/** Default payload factory: minimal oracle_update-shaped objects. */
+function defaultPayloadFactory(topic: string, seq: number): unknown {
+  return {
+    coverageType: "StablecoinDepeg",
+    type: topic,
+    value: 1.0 - seq * 0.00001,
+    threshold: 0.95,
+    severity: "low",
+    message: `Mock frame #${seq}`,
+  };
+}
+
+/** Factory that generates random severity values for chaos testing. */
+export function chaosPayloadFactory(topic: string, seq: number): unknown {
+  const severities = ["low", "medium", "high", "triggered"] as const;
+  return {
+    coverageType: ["StablecoinDepeg", "MarketCrash", "SmartContractRisk"][seq % 3],
+    type: topic,
+    value: Math.random() * 2 - 1,
+    threshold: 0.95,
+    severity: severities[Math.floor(Math.random() * severities.length)],
+    message: `Chaos frame #${seq}`,
+  };
+}
+
+/** Factory that produces malformed JSON (simulates a buggy backend). */
+export function malformedPayloadFactory(_topic: string, seq: number): unknown {
+  // Return a payload missing required fields to exercise schema validation.
+  if (seq % 10 === 0) return { type: "oracle_update" }; // missing fields
+  return defaultPayloadFactory(_topic, seq);
+}