diff --git a/README.md b/README.md
index a9480af9..e7064a13 100644
--- a/README.md
+++ b/README.md
@@ -130,6 +130,7 @@ await esbuild.build({
 - **Renderer Events** — Capture RUM events from renderer processes via the browser SDK
 - **Custom Duration Vitals** — Measure user-defined durations in the main process
 - **Renderer Profiling** — Collect JS Self-Profiling data from renderer pages and correlate it with RUM
+- **Session Replay** — Record renderer UI sessions in the main process and correlate them with RUM views
 - **User & Account Info** — Attach user and account identity to all RUM events and traces
 - **Operation Monitoring** _(experimental)_ — Track start / succeed / fail steps of critical user-facing workflows
 
@@ -245,6 +246,41 @@ protocol.registerSchemesAsPrivileged([
 ]);
 ```
 
+### Session Replay
+
+The Browser SDK records renderer UI ([Session Replay](https://docs.datadoghq.com/real_user_monitoring/session_replay/browser/)) and streams the records to the main process, which buffers, compresses, and uploads them — correlated with your RUM views.
+
+**1. Enable session replay from the main process.** Set `sessionReplaySampleRate` on the Electron SDK `init()`:
+
+```ts
+await init({
+  clientToken: '<CLIENT_TOKEN>',
+  applicationId: '<APPLICATION_ID>',
+  site: 'datadoghq.com',
+  service: 'my-electron-app',
+  sessionReplaySampleRate: 100, // percentage of sampled sessions that record replay (0–100)
+  defaultPrivacyLevel: 'mask', // 'mask' | 'allow' | 'mask-user-input'
+});
+```
+
+The Electron SDK owns the replay sampling decision and performs the upload. `sessionReplaySampleRate` is applied as a child of `sessionSampleRate`, so with `sessionSampleRate: 0` no session records replay. The `defaultPrivacyLevel` set here is propagated to the renderer recorder through the bridge, so masking is enforced end-to-end.
+
+**2. Enable session replay in the renderer's Browser SDK.** In the pages loaded by the renderer, initialize `@datadog/browser-rum` with session replay turned on:
+
+```ts
+import { datadogRum } from '@datadog/browser-rum';
+
+datadogRum.init({
+  clientToken: '<CLIENT_TOKEN>',
+  applicationId: '<APPLICATION_ID>',
+  site: 'datadoghq.com',
+  service: 'my-electron-app',
+  sessionReplaySampleRate: 100,
+});
+```
+
+The Electron SDK advertises a `records` capability over the bridge, so the Browser SDK streams replay records to the main process instead of uploading them itself — no `startSessionReplayRecording()` call is required. See [Renderer process setup](#renderer-process-setup) for general Browser SDK configuration.
+
 ## API
 
 ### `init(config: InitConfiguration): Promise<boolean>`
@@ -375,18 +411,19 @@ interface FeatureOperationOptions {
 
 ### Configuration Options
 
-| Option                | Type                                     | Required | Default  | Description                                                                                                                               |
-| --------------------- | ---------------------------------------- | -------- | -------- | ----------------------------------------------------------------------------------------------------------------------------------------- |
-| `clientToken`         | `string`                                 | Yes      | —        | Datadog client token                                                                                                                      |
-| `applicationId`       | `string`                                 | Yes      | —        | RUM application ID                                                                                                                        |
-| `site`                | `string`                                 | Yes      | —        | Datadog site (e.g. `datadoghq.com`, `datadoghq.eu`, `us3.datadoghq.com`, `us5.datadoghq.com`, `ap1.datadoghq.com`, `ddog-gov.com`)        |
-| `service`             | `string`                                 | Yes      | —        | Service name                                                                                                                              |
-| `env`                 | `string`                                 | No       | —        | Application environment                                                                                                                   |
-| `version`             | `string`                                 | No       | —        | Application version                                                                                                                       |
-| `sessionSampleRate`   | `number`                                 | No       | `100`    | Percentage of sessions to collect (0–100). `0` collects no sessions; `100` collects all sessions.                                         |
-| `profilingSampleRate` | `number`                                 | No       | `0`      | Percentage of sampled sessions that are profiled (0–100). `0` disables renderer profiling. See [Renderer Profiling](#renderer-profiling). |
-| `telemetrySampleRate` | `number`                                 | No       | `20`     | Telemetry sample rate (0–100)                                                                                                             |
-| `batchSize`           | `'SMALL' \| 'MEDIUM' \| 'LARGE'`         | No       | —        | Batch size for event uploads                                                                                                              |
-| `uploadFrequency`     | `'RARE' \| 'NORMAL' \| 'FREQUENT'`       | No       | —        | Upload frequency for event batches                                                                                                        |
-| `defaultPrivacyLevel` | `'mask' \| 'allow' \| 'mask-user-input'` | No       | `'mask'` | Default privacy level for renderer session replay                                                                                         |
-| `allowedWebViewHosts` | `string[]`                               | No       | `[]`     | Hostnames allowed for the renderer bridge                                                                                                 |
+| Option                    | Type                                     | Required | Default  | Description                                                                                                                                         |
+| ------------------------- | ---------------------------------------- | -------- | -------- | --------------------------------------------------------------------------------------------------------------------------------------------------- |
+| `clientToken`             | `string`                                 | Yes      | —        | Datadog client token                                                                                                                                |
+| `applicationId`           | `string`                                 | Yes      | —        | RUM application ID                                                                                                                                  |
+| `site`                    | `string`                                 | Yes      | —        | Datadog site (e.g. `datadoghq.com`, `datadoghq.eu`, `us3.datadoghq.com`, `us5.datadoghq.com`, `ap1.datadoghq.com`, `ddog-gov.com`)                  |
+| `service`                 | `string`                                 | Yes      | —        | Service name                                                                                                                                        |
+| `env`                     | `string`                                 | No       | —        | Application environment                                                                                                                             |
+| `version`                 | `string`                                 | No       | —        | Application version                                                                                                                                 |
+| `sessionSampleRate`       | `number`                                 | No       | `100`    | Percentage of sessions to collect (0–100). `0` collects no sessions; `100` collects all sessions.                                                   |
+| `sessionReplaySampleRate` | `number`                                 | No       | `0`      | Percentage of sampled sessions that record session replay (0–100). `0` disables renderer session replay. Applied as a child of `sessionSampleRate`. |
+| `profilingSampleRate`     | `number`                                 | No       | `0`      | Percentage of sampled sessions that are profiled (0–100). `0` disables renderer profiling. See [Renderer Profiling](#renderer-profiling).           |
+| `telemetrySampleRate`     | `number`                                 | No       | `20`     | Telemetry sample rate (0–100)                                                                                                                       |
+| `batchSize`               | `'SMALL' \| 'MEDIUM' \| 'LARGE'`         | No       | —        | Batch size for event uploads                                                                                                                        |
+| `uploadFrequency`         | `'RARE' \| 'NORMAL' \| 'FREQUENT'`       | No       | —        | Upload frequency for event batches                                                                                                                  |
+| `defaultPrivacyLevel`     | `'mask' \| 'allow' \| 'mask-user-input'` | No       | `'mask'` | Default privacy level for renderer session replay                                                                                                   |
+| `allowedWebViewHosts`     | `string[]`                               | No       | `[]`     | Hostnames allowed for the renderer bridge                                                                                                           |
diff --git a/e2e/app/src/bridge-window.ts b/e2e/app/src/bridge-window.ts
index 505607b0..2adb51cc 100644
--- a/e2e/app/src/bridge-window.ts
+++ b/e2e/app/src/bridge-window.ts
@@ -5,11 +5,11 @@ datadogRum.init({
   clientToken: 'pub-renderer-token',
   site: 'datadoghq.com',
   service: 'e2e-renderer',
-  sessionSampleRate: 100,
-  profilingSampleRate: 100,
   trackResources: true,
   trackLongTasks: true,
   trackUserInteractions: true,
+  profilingSampleRate: 100,
+  sessionReplaySampleRate: 100,
 });
 
 document.getElementById('status')!.textContent = 'bridge-ready';
diff --git a/e2e/app/src/main-window.html b/e2e/app/src/main-window.html
index e2e05b9e..1f786bb1 100644
--- a/e2e/app/src/main-window.html
+++ b/e2e/app/src/main-window.html
@@ -5,7 +5,7 @@
     <meta name="viewport" content="width=device-width, initial-scale=1.0" />
     <meta
       http-equiv="Content-Security-Policy"
-      content="default-src 'self'; style-src 'unsafe-inline'; script-src 'self'"
+      content="default-src 'self'; style-src 'unsafe-inline'; script-src 'self'; worker-src blob:"
     />
     <title>Main Window</title>
     <style>
diff --git a/e2e/integration/apps/electron-builder-vite/src/renderer/index.ts b/e2e/integration/apps/electron-builder-vite/src/renderer/index.ts
index 4506a181..e2b97692 100644
--- a/e2e/integration/apps/electron-builder-vite/src/renderer/index.ts
+++ b/e2e/integration/apps/electron-builder-vite/src/renderer/index.ts
@@ -5,6 +5,7 @@ datadogRum.init({
   applicationId: 'integration-test-app-id',
   service: 'integration-test-renderer',
   site: 'datadoghq.com',
+  sessionReplaySampleRate: 100,
   trackUserInteractions: false,
   trackResources: false,
   trackLongTasks: false,
diff --git a/e2e/integration/apps/electron-vite-esm/src/renderer/src/main.ts b/e2e/integration/apps/electron-vite-esm/src/renderer/src/main.ts
index 4506a181..e2b97692 100644
--- a/e2e/integration/apps/electron-vite-esm/src/renderer/src/main.ts
+++ b/e2e/integration/apps/electron-vite-esm/src/renderer/src/main.ts
@@ -5,6 +5,7 @@ datadogRum.init({
   applicationId: 'integration-test-app-id',
   service: 'integration-test-renderer',
   site: 'datadoghq.com',
+  sessionReplaySampleRate: 100,
   trackUserInteractions: false,
   trackResources: false,
   trackLongTasks: false,
diff --git a/e2e/integration/apps/electron-vite/src/renderer/src/main.ts b/e2e/integration/apps/electron-vite/src/renderer/src/main.ts
index 4506a181..e2b97692 100644
--- a/e2e/integration/apps/electron-vite/src/renderer/src/main.ts
+++ b/e2e/integration/apps/electron-vite/src/renderer/src/main.ts
@@ -5,6 +5,7 @@ datadogRum.init({
   applicationId: 'integration-test-app-id',
   service: 'integration-test-renderer',
   site: 'datadoghq.com',
+  sessionReplaySampleRate: 100,
   trackUserInteractions: false,
   trackResources: false,
   trackLongTasks: false,
diff --git a/e2e/integration/apps/forge-esbuild-cjs/src/renderer/index.ts b/e2e/integration/apps/forge-esbuild-cjs/src/renderer/index.ts
index 4506a181..e2b97692 100644
--- a/e2e/integration/apps/forge-esbuild-cjs/src/renderer/index.ts
+++ b/e2e/integration/apps/forge-esbuild-cjs/src/renderer/index.ts
@@ -5,6 +5,7 @@ datadogRum.init({
   applicationId: 'integration-test-app-id',
   service: 'integration-test-renderer',
   site: 'datadoghq.com',
+  sessionReplaySampleRate: 100,
   trackUserInteractions: false,
   trackResources: false,
   trackLongTasks: false,
diff --git a/e2e/integration/apps/forge-esbuild-esm/src/renderer/index.ts b/e2e/integration/apps/forge-esbuild-esm/src/renderer/index.ts
index 4506a181..e2b97692 100644
--- a/e2e/integration/apps/forge-esbuild-esm/src/renderer/index.ts
+++ b/e2e/integration/apps/forge-esbuild-esm/src/renderer/index.ts
@@ -5,6 +5,7 @@ datadogRum.init({
   applicationId: 'integration-test-app-id',
   service: 'integration-test-renderer',
   site: 'datadoghq.com',
+  sessionReplaySampleRate: 100,
   trackUserInteractions: false,
   trackResources: false,
   trackLongTasks: false,
diff --git a/e2e/integration/apps/forge-vite/src/renderer/index.ts b/e2e/integration/apps/forge-vite/src/renderer/index.ts
index 4506a181..e2b97692 100644
--- a/e2e/integration/apps/forge-vite/src/renderer/index.ts
+++ b/e2e/integration/apps/forge-vite/src/renderer/index.ts
@@ -5,6 +5,7 @@ datadogRum.init({
   applicationId: 'integration-test-app-id',
   service: 'integration-test-renderer',
   site: 'datadoghq.com',
+  sessionReplaySampleRate: 100,
   trackUserInteractions: false,
   trackResources: false,
   trackLongTasks: false,
diff --git a/e2e/integration/apps/forge-webpack/src/renderer/index.ts b/e2e/integration/apps/forge-webpack/src/renderer/index.ts
index 4506a181..e2b97692 100644
--- a/e2e/integration/apps/forge-webpack/src/renderer/index.ts
+++ b/e2e/integration/apps/forge-webpack/src/renderer/index.ts
@@ -5,6 +5,7 @@ datadogRum.init({
   applicationId: 'integration-test-app-id',
   service: 'integration-test-renderer',
   site: 'datadoghq.com',
+  sessionReplaySampleRate: 100,
   trackUserInteractions: false,
   trackResources: false,
   trackLongTasks: false,
diff --git a/e2e/integration/lib/integrationFixture.ts b/e2e/integration/lib/integrationFixture.ts
index f97d8efd..676c2d61 100644
--- a/e2e/integration/lib/integrationFixture.ts
+++ b/e2e/integration/lib/integrationFixture.ts
@@ -111,6 +111,7 @@ function buildSdkConfig(intake: Intake): InitConfiguration {
     applicationId: 'integration-test-app-id',
     env: 'test',
     version: '1.0.0',
+    sessionReplaySampleRate: 100,
     telemetrySampleRate: 100,
     defaultPrivacyLevel: 'mask',
     allowedWebViewHosts: [],
diff --git a/e2e/lib/bridgeWindowPage.ts b/e2e/lib/bridgeWindowPage.ts
index 2852665b..d468aabd 100644
--- a/e2e/lib/bridgeWindowPage.ts
+++ b/e2e/lib/bridgeWindowPage.ts
@@ -44,6 +44,18 @@ export class BridgeWindowPage {
     );
   }
 
+  async getBridgeCapabilities(): Promise<string[]> {
+    return this.page.evaluate(() => {
+      const bridge = (
+        globalThis as unknown as {
+          DatadogEventBridge: { getCapabilities(): string };
+        }
+      ).DatadogEventBridge;
+
+      return JSON.parse(bridge.getCapabilities()) as string[];
+    });
+  }
+
   async triggerProfilingFlush(): Promise<void> {
     await this.page.close({ runBeforeUnload: true });
     // Wait for IPC propagation: beforeunload → bridge send → main process → write queue
diff --git a/e2e/lib/helpers.ts b/e2e/lib/helpers.ts
index 76832167..21ce4c43 100644
--- a/e2e/lib/helpers.ts
+++ b/e2e/lib/helpers.ts
@@ -120,6 +120,7 @@ async function launchApp(
     env: 'test',
     version: '1.0.0',
     sessionSampleRate: 100,
+    sessionReplaySampleRate: 100,
     profilingSampleRate: 100,
     telemetrySampleRate: 100,
     defaultPrivacyLevel: 'mask',
@@ -135,6 +136,7 @@ async function launchApp(
       site: 'datadoghq.com',
       service: 'e2e-main-window',
       sessionSampleRate: 100,
+      sessionReplaySampleRate: 100,
       trackUserInteractions: true,
       ...rumBrowserSdk,
     });
diff --git a/e2e/lib/intake.ts b/e2e/lib/intake.ts
index 27530b35..abd205ca 100644
--- a/e2e/lib/intake.ts
+++ b/e2e/lib/intake.ts
@@ -1,4 +1,5 @@
 import * as http from 'node:http';
+import zlib from 'node:zlib';
 
 /**
  * Fake Datadog intake used to assert what the SDK sends.
@@ -14,6 +15,19 @@ export interface ReceivedEvent {
   ddforward: string;
 }
 
+export interface ReplaySegment {
+  timestamp: number;
+  /** Parsed JSON from the multipart `event` field — segment metadata + size fields. */
+  metadata: Record<string, unknown>;
+  headers: Record<string, string>;
+  /**
+   * The rrweb records decoded from the compressed `segment` blob, when it was a
+   * standalone-inflatable ZLIB stream (the common single-segment case). Undefined
+   * if the `segment` part was absent or could not be inflated on its own.
+   */
+  records?: unknown[];
+}
+
 export interface Trace {
   env: string;
   spans: Span[];
@@ -37,9 +51,27 @@ export interface ProfilingRequest {
 
 const byType = (type: string) => (event: ReceivedEvent) => (event.body as { type?: string }).type === type;
 
+/**
+ * Splits a Buffer on a byte delimiter, returning the chunks between delimiters.
+ * Operates on raw bytes (not utf8) so binary multipart parts survive intact.
+ */
+function splitBuffer(buf: Buffer, delimiter: Buffer): Buffer[] {
+  const parts: Buffer[] = [];
+  let start = 0;
+  let idx = buf.indexOf(delimiter, start);
+  while (idx !== -1) {
+    if (idx > start) parts.push(buf.subarray(start, idx));
+    start = idx + delimiter.length;
+    idx = buf.indexOf(delimiter, start);
+  }
+  if (start < buf.length) parts.push(buf.subarray(start));
+  return parts;
+}
+
 export class Intake {
   private server: http.Server | null = null;
   private rumEvents: ReceivedEvent[] = [];
+  private replaySegments: ReplaySegment[] = [];
   private traces: Trace[] = [];
   private profilingRequests: ProfilingRequest[] = [];
   private port = 0;
@@ -57,6 +89,61 @@ export class Intake {
     }
   }
 
+  private storeReplaySegment(rawBody: Buffer, headers: Record<string, string>) {
+    const contentType = headers['content-type'] ?? '';
+    const boundaryMatch = /boundary=([^\s;]+)/.exec(contentType);
+    if (!boundaryMatch) return;
+
+    // Parse the multipart body on the raw Buffer (not a utf8 string): the `segment`
+    // part is deflate-compressed binary and would be corrupted by a utf8 round-trip.
+    const delimiter = Buffer.from(`--${boundaryMatch[1]}`);
+    let metadata: Record<string, unknown> | undefined;
+    let compressed: Buffer | undefined;
+
+    for (const part of splitBuffer(rawBody, delimiter)) {
+      // Each part is `\r\n<headers>\r\n\r\n<body>\r\n`. The preamble and the closing
+      // `--` marker have no header/body separator, so they're skipped here.
+      const headerEnd = part.indexOf('\r\n\r\n');
+      if (headerEnd === -1) continue;
+
+      const partHeaders = part.subarray(0, headerEnd).toString('utf8');
+      let body = part.subarray(headerEnd + 4);
+      // Strip the trailing CRLF that precedes the next delimiter.
+      if (body.length >= 2 && body[body.length - 2] === 0x0d && body[body.length - 1] === 0x0a) {
+        body = body.subarray(0, body.length - 2);
+      }
+
+      if (partHeaders.includes('name="event"')) {
+        try {
+          metadata = JSON.parse(body.toString('utf8')) as Record<string, unknown>;
+        } catch {
+          // malformed event part — ignore
+        }
+      } else if (partHeaders.includes('name="segment"')) {
+        compressed = body;
+      }
+    }
+
+    if (!metadata) return;
+
+    const segment: ReplaySegment = { timestamp: Date.now(), metadata, headers };
+
+    if (compressed) {
+      try {
+        // A single segment is a self-contained ZLIB stream (header + full-flush body +
+        // final block + Adler-32), so it inflates standalone. Continuation segments carry
+        // a cumulative Adler-32 that won't match a lone stream — those stay undecoded.
+        const inflated = zlib.inflateSync(compressed);
+        const parsed = JSON.parse(inflated.toString('utf8')) as { records?: unknown[] };
+        segment.records = parsed.records ?? [];
+      } catch {
+        // segment blob not standalone-inflatable — leave records undefined
+      }
+    }
+
+    this.replaySegments.push(segment);
+  }
+
   private storeTraces(parsedBody: unknown) {
     const items = Array.isArray(parsedBody) ? (parsedBody as Trace[]) : [parsedBody as Trace];
     for (const item of items) {
@@ -92,10 +179,10 @@ export class Intake {
           return;
         }
 
-        let body = '';
+        const chunks: Buffer[] = [];
 
         req.on('data', (chunk: Buffer) => {
-          body += chunk.toString();
+          chunks.push(chunk);
         });
 
         if (ddforward.startsWith('/api/v2/profile')) {
@@ -117,18 +204,28 @@ export class Intake {
         }
 
         req.on('end', () => {
-          try {
-            const parsedBody: unknown = JSON.parse(body);
-            const headers: Record<string, string> = {};
-
-            for (const [key, value] of Object.entries(req.headers)) {
-              if (typeof value === 'string') {
-                headers[key.toLowerCase()] = value;
-              } else if (Array.isArray(value)) {
-                headers[key.toLowerCase()] = value.join(', ');
-              }
+          const rawBody = Buffer.concat(chunks);
+          const headers: Record<string, string> = {};
+
+          for (const [key, value] of Object.entries(req.headers)) {
+            if (typeof value === 'string') {
+              headers[key.toLowerCase()] = value;
+            } else if (Array.isArray(value)) {
+              headers[key.toLowerCase()] = value.join(', ');
             }
+          }
+
+          const isMultipart = (headers['content-type'] ?? '').includes('multipart/form-data');
 
+          if (ddforward.startsWith('/api/v2/replay') || isMultipart) {
+            this.storeReplaySegment(rawBody, headers);
+            res.writeHead(202, { 'Content-Type': 'application/json' });
+            res.end(JSON.stringify({ status: 'accepted' }));
+            return;
+          }
+
+          try {
+            const parsedBody: unknown = JSON.parse(rawBody.toString());
             if (ddforward.startsWith('/api/v2/spans')) {
               this.storeTraces(parsedBody);
             } else {
@@ -278,8 +375,30 @@ export class Intake {
     }
   }
 
+  async waitForReplaySegment(options?: {
+    timeout?: number;
+    predicate?: (segment: ReplaySegment) => boolean;
+  }): Promise<ReplaySegment> {
+    const timeout = options?.timeout ?? 10000;
+    const predicate = options?.predicate ?? (() => true);
+    const startTime = Date.now();
+
+    while (Date.now() - startTime < timeout) {
+      const match = this.replaySegments.find(predicate);
+      if (match) return match;
+      await new Promise((resolve) => setTimeout(resolve, 100));
+    }
+
+    throw new Error(`Timed out waiting for a replay segment after ${timeout}ms.`);
+  }
+
+  getReplaySegments(): ReplaySegment[] {
+    return [...this.replaySegments];
+  }
+
   clear(): void {
     this.rumEvents = [];
+    this.replaySegments = [];
     this.traces = [];
     this.profilingRequests = [];
     this.quotaDecision = 'quota_ok';
diff --git a/e2e/scenarios/replay.scenario.ts b/e2e/scenarios/replay.scenario.ts
new file mode 100644
index 00000000..f8af98eb
--- /dev/null
+++ b/e2e/scenarios/replay.scenario.ts
@@ -0,0 +1,141 @@
+import type { ElectronApplication } from '@playwright/test';
+import { test, expect } from '../lib/helpers';
+import type { Intake } from '../lib/intake';
+import type { MainPage } from '../lib/mainPage';
+
+/**
+ * Session replay E2E scenarios.
+ *
+ * These tests verify that:
+ * 1. The Electron SDK preload exposes "records" capability so the browser RUM SDK
+ *    starts emitting BrowserRecord events through the bridge.
+ * 2. ReplayCollection buffers those records and emits a compressed segment.
+ * 3. ReplayBatchConsumer uploads the segment as multipart/form-data and the
+ *    mock intake receives it with correct metadata.
+ * 4. The main-process `defaultPrivacyLevel` reaches the renderer recorder through the
+ *    bridge: masked text is absent from the recorded segment, unmasked text is present.
+ */
+
+// A distinctive token rendered into the bridge window's DOM. It appears verbatim in the
+// recorded rrweb nodes only when text is NOT masked, making it a reliable probe for whether
+// the privacy level was honoured end-to-end.
+const SENSITIVE_TEXT = 'SensitiveReplayValue42';
+
+/**
+ * Opens a bridge window, renders {@link SENSITIVE_TEXT} into it (triggering an rrweb mutation
+ * record), flushes the transport, and returns the concatenated JSON of every decoded segment's
+ * records so a test can assert on the recorded text.
+ */
+async function recordSensitiveTextAndFlush(
+  electronApp: ElectronApplication,
+  mainPage: MainPage,
+  intake: Intake
+): Promise<string> {
+  const bridgeWindow = await mainPage.openBridgeFileWindow(electronApp);
+  expect(await bridgeWindow.getBridgeCapabilities()).toContain('records');
+
+  await bridgeWindow.page.evaluate((text) => {
+    // Runs in the renderer; the e2e tsconfig has no DOM lib, so reach `document` via globalThis
+    // (same cast pattern as BridgeWindowPage.getBridgeCapabilities).
+    const { document } = globalThis as unknown as {
+      document: {
+        createElement(tag: string): { textContent: string };
+        body: { appendChild(node: unknown): void };
+      };
+    };
+    const el = document.createElement('p');
+    el.textContent = text;
+    document.body.appendChild(el);
+  }, SENSITIVE_TEXT);
+
+  // Give the recorder time to emit the mutation and buffer it into the segment.
+  await bridgeWindow.page.waitForTimeout(2000);
+  await mainPage.flushTransport();
+
+  // Wait for a segment whose blob we could actually decode into records.
+  await intake.waitForReplaySegment({ timeout: 20_000, predicate: (s) => (s.records?.length ?? 0) > 0 });
+
+  return JSON.stringify(intake.getReplaySegments().map((s) => s.records ?? []));
+}
+
+test.describe('session replay', () => {
+  test('replay segment arrives at the intake after opening a bridge window', async ({
+    electronApp,
+    mainPage,
+    intake,
+  }) => {
+    // Open a bridge window — the browser RUM SDK initialises and, because the
+    // Electron SDK preload advertises "records" capability, starts recording rrweb events.
+    const bridgeWindow = await mainPage.openBridgeFileWindow(electronApp);
+    expect(await bridgeWindow.getBridgeCapabilities()).toContain('records');
+
+    // Give the renderer time to produce at least one full-snapshot record.
+    await bridgeWindow.page.waitForTimeout(2000);
+
+    // Flush forces ReplayCollection to flush the current segment, compress it,
+    // and push it through the batch pipeline to the intake.
+    await mainPage.flushTransport();
+
+    const segment = await intake.waitForReplaySegment({ timeout: 20_000 });
+
+    // The segment metadata should be populated with main-process context
+    expect(segment.metadata['session']).toBeDefined();
+    expect(segment.metadata['application']).toBeDefined();
+    expect(segment.metadata['view']).toBeDefined();
+    expect(segment.metadata['records_count']).toBeGreaterThan(0);
+    expect(segment.metadata['source']).toBe('browser');
+  });
+
+  test('view event includes has_replay: true after a replay segment is sent', async ({
+    electronApp,
+    mainPage,
+    intake,
+  }) => {
+    // Open bridge window and wait for recording to produce data
+    const bridgeWindow = await mainPage.openBridgeFileWindow(electronApp);
+    expect(await bridgeWindow.getBridgeCapabilities()).toContain('records');
+    await bridgeWindow.page.waitForTimeout(2000);
+
+    // First flush — sends the replay segment to the intake
+    await mainPage.flushTransport();
+    await intake.waitForReplaySegment({ timeout: 20_000 });
+
+    // Trigger renderer activity so browser-rum emits a fresh view update.
+    // generateError works in file:// context (unlike fetch-based generateResource).
+    // Assembly will enrich the resulting view update with has_replay: true now
+    // that replay stats exist for this view.
+    await bridgeWindow.generateError('replay-trigger');
+    await mainPage.flushTransport();
+
+    const viewEvents = await intake.waitForEventCount('view', 1, {
+      timeout: 20_000,
+      predicate: (e) => {
+        const body = e.body as Record<string, unknown>;
+        const session = body['session'] as Record<string, unknown> | undefined;
+        return session?.['has_replay'] === true;
+      },
+    });
+
+    expect(viewEvents.length).toBeGreaterThanOrEqual(1);
+  });
+});
+
+test.describe('session replay privacy masking', () => {
+  test.describe('with defaultPrivacyLevel: mask', () => {
+    test.use({ sdkConfigOverrides: { defaultPrivacyLevel: 'mask' } });
+
+    test('masks text nodes in the recorded segment', async ({ electronApp, mainPage, intake }) => {
+      const recordsJson = await recordSensitiveTextAndFlush(electronApp, mainPage, intake);
+      expect(recordsJson).not.toContain(SENSITIVE_TEXT);
+    });
+  });
+
+  test.describe('with defaultPrivacyLevel: allow', () => {
+    test.use({ sdkConfigOverrides: { defaultPrivacyLevel: 'allow' } });
+
+    test('keeps text nodes verbatim in the recorded segment', async ({ electronApp, mainPage, intake }) => {
+      const recordsJson = await recordSensitiveTextAndFlush(electronApp, mainPage, intake);
+      expect(recordsJson).toContain(SENSITIVE_TEXT);
+    });
+  });
+});
diff --git a/playground/src/main.ts b/playground/src/main.ts
index 57e2ed4c..6f65987e 100644
--- a/playground/src/main.ts
+++ b/playground/src/main.ts
@@ -239,6 +239,7 @@ void app.whenReady().then(async () => {
     ...CONF[ACTIVE_ENV],
     service: 'electron-playground',
     env: 'dev',
+    sessionReplaySampleRate: 100,
     profilingSampleRate: 100,
     ...(process.env.DD_SDK_PROXY ? { proxy: process.env.DD_SDK_PROXY } : {}),
   });
diff --git a/playground/src/renderer.ts b/playground/src/renderer.ts
index 356fa057..4224f849 100644
--- a/playground/src/renderer.ts
+++ b/playground/src/renderer.ts
@@ -22,6 +22,7 @@ datadogRum.init({
   service: 'electron-playground',
   env: 'dev',
   sessionSampleRate: 100,
+  sessionReplaySampleRate: 100,
   trackResources: true,
   trackLongTasks: true,
   trackUserInteractions: true,
diff --git a/src/assembly/MainAssembly.ts b/src/assembly/MainAssembly.ts
index e9a866a1..100fd0a4 100644
--- a/src/assembly/MainAssembly.ts
+++ b/src/assembly/MainAssembly.ts
@@ -9,6 +9,7 @@ import {
   EventTrack,
   type RawEvent,
   type RawProfileEvent,
+  type RawReplayEvent,
   type ServerEvent,
 } from '../event';
 import type { FormatHooks } from './hooks';
@@ -16,7 +17,7 @@ import { RumEvent } from '../domain/rum';
 import { TelemetryEvent } from '../domain/telemetry';
 
 // Raw events assembled through the standard main-process hook pipeline.
-type StandardRawEvent = Exclude<RawEvent, RawProfileEvent>;
+type StandardRawEvent = Exclude<RawEvent, RawProfileEvent | RawReplayEvent>;
 
 /**
  * Transforms main-process RawEvents into ServerEvents by enriching them with
@@ -29,7 +30,7 @@ export class MainAssembly {
   ) {
     this.eventManager.registerHandler<StandardRawEvent>({
       canHandle: (event): event is StandardRawEvent =>
-        event.kind === EventKind.RAW && event.format !== EventFormat.PROFILE,
+        event.kind === EventKind.RAW && event.format !== EventFormat.PROFILE && event.format !== EventFormat.REPLAY,
       handle: (event, notify) => {
         const result = this.assembleMainProcessEvent(event);
         if (result !== DISCARDED) {
diff --git a/src/assembly/RendererPipeline.spec.ts b/src/assembly/RendererPipeline.spec.ts
index 0b5e0aa6..e61740d9 100644
--- a/src/assembly/RendererPipeline.spec.ts
+++ b/src/assembly/RendererPipeline.spec.ts
@@ -15,6 +15,7 @@ import {
   type BrowserProfilerTrace,
   type EndUserActivityEvent,
   type RawProfileEvent,
+  type RawReplayEvent,
   type ServerRumEvent,
 } from '../event';
 import { BRIDGE_CHANNEL, CONFIG_CHANNEL } from '../common';
@@ -115,6 +116,7 @@ describe('RendererPipeline', () => {
       defaultPrivacyLevel: 'allow',
       allowedWebViewHosts: ['example.com'],
       profilingSampleRate: 0,
+      sessionReplaySampleRate: 0,
     });
     mockSetBridgeConfig.mockClear();
     new RendererPipeline(eventManager, hooks, config);
@@ -127,14 +129,21 @@ describe('RendererPipeline', () => {
 
   describe('capabilities', () => {
     it('advertises the profiles capability when profilingSampleRate > 0', () => {
-      const config = createTestConfiguration({ profilingSampleRate: 100 });
+      const config = createTestConfiguration({ profilingSampleRate: 100, sessionReplaySampleRate: 0 });
       mockSetBridgeConfig.mockClear();
       new RendererPipeline(new EventManager(), createFormatHooks(), config);
       expect((mockSetBridgeConfig.mock.calls[0]?.[0] as BridgeOptions).capabilities).toEqual(['profiles']);
     });
 
-    it('advertises no capabilities when profilingSampleRate is 0', () => {
-      const config = createTestConfiguration({ profilingSampleRate: 0 });
+    it('advertises replay and profiling capabilities when both are enabled', () => {
+      const config = createTestConfiguration({ profilingSampleRate: 100, sessionReplaySampleRate: 100 });
+      mockSetBridgeConfig.mockClear();
+      new RendererPipeline(new EventManager(), createFormatHooks(), config);
+      expect((mockSetBridgeConfig.mock.calls[0]?.[0] as BridgeOptions).capabilities).toEqual(['profiles', 'records']);
+    });
+
+    it('advertises no capabilities when profiling and replay are disabled', () => {
+      const config = createTestConfiguration({ profilingSampleRate: 0, sessionReplaySampleRate: 0 });
       mockSetBridgeConfig.mockClear();
       new RendererPipeline(new EventManager(), createFormatHooks(), config);
       expect((mockSetBridgeConfig.mock.calls[0]?.[0] as BridgeOptions).capabilities).toEqual([]);
@@ -377,6 +386,74 @@ describe('RendererPipeline', () => {
     });
   });
 
+  describe('record bridge events', () => {
+    it('dispatches RawReplayEvent when bridge sends a record message', () => {
+      const record = { type: 2, timestamp: 123 };
+      const received: RawReplayEvent[] = [];
+      eventManager.registerHandler<RawReplayEvent>({
+        canHandle: (e): e is RawReplayEvent => e.kind === EventKind.RAW && e.format === EventFormat.REPLAY,
+        handle: (e) => received.push(e),
+      });
+
+      simulateIpcMessage(JSON.stringify({ eventType: 'record', event: record, view: { id: 'view-1' } }));
+
+      expect(received).toHaveLength(1);
+      expect(received[0].format).toBe(EventFormat.REPLAY);
+      expect(received[0].data).toEqual(record);
+      expect(received[0].view).toEqual({ id: 'view-1' });
+      expect(received[0].source).toBe(EventSource.RENDERER);
+    });
+
+    it('reports telemetry error and drops records missing view', () => {
+      const spy = vi.spyOn(eventManager, 'notify');
+
+      simulateIpcMessage(JSON.stringify({ eventType: 'record', event: { type: 2, timestamp: 123 } }));
+
+      expect(spy).not.toHaveBeenCalled();
+      expect(mockAddError).toHaveBeenCalledOnce();
+      expect((mockAddError.mock.calls[0][0] as Error).message).toContain('missing view');
+    });
+
+    it('reports telemetry error and drops malformed record payloads', () => {
+      const spy = vi.spyOn(eventManager, 'notify');
+
+      simulateIpcMessage(JSON.stringify({ eventType: 'record', event: 'not-an-object', view: { id: 'view-1' } }));
+
+      expect(spy).not.toHaveBeenCalled();
+      expect(mockAddError).toHaveBeenCalledOnce();
+      expect((mockAddError.mock.calls[0][0] as Error).message).toContain('malformed replay record');
+    });
+
+    it.each([
+      ['empty view id', { id: '' }],
+      ['missing view id', {}],
+      ['non-string view id', { id: 123 }],
+    ])('reports telemetry error and drops records with an %s', (_label, view) => {
+      const spy = vi.spyOn(eventManager, 'notify');
+
+      simulateIpcMessage(JSON.stringify({ eventType: 'record', event: { type: 2, timestamp: 123 }, view }));
+
+      expect(spy).not.toHaveBeenCalled();
+      expect(mockAddError).toHaveBeenCalledOnce();
+      expect((mockAddError.mock.calls[0][0] as Error).message).toContain('missing view');
+    });
+
+    it.each([
+      ['missing timestamp', { type: 2 }],
+      ['non-numeric timestamp', { type: 2, timestamp: 'now' }],
+      ['null timestamp', { type: 2, timestamp: null }],
+      ['missing type', { timestamp: 123 }],
+    ])('reports telemetry error and drops records with %s', (_label, event) => {
+      const spy = vi.spyOn(eventManager, 'notify');
+
+      simulateIpcMessage(JSON.stringify({ eventType: 'record', event, view: { id: 'view-1' } }));
+
+      expect(spy).not.toHaveBeenCalled();
+      expect(mockAddError).toHaveBeenCalledOnce();
+      expect((mockAddError.mock.calls[0][0] as Error).message).toContain('invalid timestamp or type');
+    });
+  });
+
   describe('unimplemented event types', () => {
     it('does not emit for log events (TODO)', () => {
       const spy = vi.spyOn(eventManager, 'notify');
diff --git a/src/assembly/RendererPipeline.ts b/src/assembly/RendererPipeline.ts
index 6145d7fb..08114b0f 100644
--- a/src/assembly/RendererPipeline.ts
+++ b/src/assembly/RendererPipeline.ts
@@ -3,7 +3,7 @@ import { type TimeStamp } from '@datadog/js-core/time';
 import { combine, isIndexableObject, type RecursivePartial } from '@datadog/js-core/util';
 import { DISCARDED } from '@datadog/js-core/assembly';
 import { EventKind, EventSource, EventTrack, LifecycleKind, EventFormat } from '../event';
-import type { EventManager, ServerRumEvent, BrowserProfileEvent, BrowserProfilerTrace } from '../event';
+import type { EventManager, ServerRumEvent, BrowserProfileEvent, BrowserProfilerTrace, RawReplayEvent } from '../event';
 import { isEmptyObject } from '@datadog/browser-core';
 import { monitor, addError as addTelemetryError } from '../domain/telemetry';
 import { BRIDGE_CHANNEL, setBridgeConfig, type BridgeOptions } from '../common';
@@ -11,11 +11,12 @@ import type { FormatHooks } from './hooks';
 import type { RumEvent } from '../domain/rum';
 import { Configuration } from '../config';
 
-type BridgeEventType = 'rum' | 'log' | 'internal_telemetry' | 'profile';
+type BridgeEventType = 'rum' | 'log' | 'internal_telemetry' | 'profile' | 'record';
 
 interface BridgeEvent {
   eventType: BridgeEventType;
   event: unknown;
+  view?: { id: string };
 }
 
 /**
@@ -42,7 +43,10 @@ export class RendererPipeline {
       // Capabilities are resolved once here and advertised globally, not per session. Bridge mode has no
       // channel to notify the renderer on session renew/expire or capability changes, so the browser SDK
       // cannot adjust its per-session behavior (e.g. stop profiling a sampled-out session). Out of scope for now.
-      capabilities: config.profilingSampleRate > 0 ? ['profiles'] : [],
+      capabilities: [
+        ...(config.profilingSampleRate > 0 ? ['profiles'] : []),
+        ...(config.sessionReplaySampleRate > 0 ? ['records'] : []),
+      ],
     };
 
     ipcMain.on(
@@ -94,6 +98,42 @@ export class RendererPipeline {
         });
         break;
       }
+      case 'record': {
+        // view.id is untrusted IPC input and keys the segment metadata/stats: require a non-empty
+        // *string*. A truthy non-string (e.g. 123) would otherwise become the segment's view id and
+        // no longer match the string RUM view id, breaking stitching / losing replay stats.
+        const viewId = (bridgeEvent.view as { id?: unknown } | undefined)?.id;
+        if (typeof viewId !== 'string' || viewId.length === 0) {
+          addTelemetryError(new Error('Replay record missing view'));
+          break;
+        }
+        // Validate the renderer-supplied shape early: a malformed record would otherwise fail
+        // later at segment serialization/upload with no useful context.
+        if (!isIndexableObject(bridgeEvent.event)) {
+          addTelemetryError(new Error('Received malformed replay record'));
+          break;
+        }
+        // A bridge/SDK version mismatch can send an object that lacks a numeric timestamp/type.
+        // Segment.addRecord derives start/end from timestamp via Math.min/Math.max, so a missing
+        // or non-finite value turns segment metadata into NaN (serialized as null) and makes the
+        // uploaded segment unusable. Reject at the boundary instead, matching the profile validation.
+        if (
+          typeof bridgeEvent.event.timestamp !== 'number' ||
+          !Number.isFinite(bridgeEvent.event.timestamp) ||
+          typeof bridgeEvent.event.type !== 'number'
+        ) {
+          addTelemetryError(new Error('Received replay record with invalid timestamp or type'));
+          break;
+        }
+        this.eventManager.notify({
+          kind: EventKind.RAW,
+          source: EventSource.RENDERER,
+          format: EventFormat.REPLAY,
+          data: bridgeEvent.event,
+          view: { id: viewId },
+        } as RawReplayEvent);
+        break;
+      }
       default:
         addTelemetryError(new Error(`Unhandled bridge event type: ${String(bridgeEvent.eventType)}`));
     }
@@ -113,6 +153,7 @@ export class RendererPipeline {
       eventType: data.type,
       startTime: data.date as TimeStamp,
       source: EventSource.RENDERER,
+      rendererViewId: (data as { view?: { id?: string } }).view?.id,
     });
 
     if (hookResult === DISCARDED) {
diff --git a/src/assembly/commonContext.spec.ts b/src/assembly/commonContext.spec.ts
index e2b2ad73..a0417276 100644
--- a/src/assembly/commonContext.spec.ts
+++ b/src/assembly/commonContext.spec.ts
@@ -17,6 +17,7 @@ function makeConfig(overrides: Partial<Configuration> = {}): Configuration {
     defaultPrivacyLevel: 'mask',
     allowedWebViewHosts: [],
     sessionSampleRate: 100,
+    sessionReplaySampleRate: 0,
     profilingSampleRate: 0,
     telemetrySampleRate: 20,
     ...overrides,
@@ -141,13 +142,14 @@ describe('registerCommonContext', () => {
 
   describe('sampling rates in _dd.configuration', () => {
     it.each([EventSource.MAIN, EventSource.RENDERER])(
-      'injects the Electron SDK session and profiling sample rates for %s events',
+      'injects the Electron SDK session, replay, and profiling sample rates for %s events',
       (source) => {
-        const config = makeConfig({ sessionSampleRate: 42, profilingSampleRate: 100 });
+        const config = makeConfig({ sessionSampleRate: 42, sessionReplaySampleRate: 25, profilingSampleRate: 100 });
         const result = source === EventSource.MAIN ? triggerMainRum(config) : triggerRendererRum(config);
 
         expect((result._dd as { configuration: unknown }).configuration).toEqual({
           session_sample_rate: 42,
+          session_replay_sample_rate: 25,
           profiling_sample_rate: 100,
         });
       }
diff --git a/src/assembly/commonContext.ts b/src/assembly/commonContext.ts
index b6c627bc..d8d0f6fc 100644
--- a/src/assembly/commonContext.ts
+++ b/src/assembly/commonContext.ts
@@ -11,6 +11,7 @@ export function registerCommonContext(configuration: Configuration, hooks: Forma
   // so it is authoritative for the rates reported on every RUM event.
   const ddConfiguration = {
     session_sample_rate: configuration.sessionSampleRate,
+    session_replay_sample_rate: configuration.sessionReplaySampleRate,
     profiling_sample_rate: configuration.profilingSampleRate,
   };
 
diff --git a/src/assembly/hooks.ts b/src/assembly/hooks.ts
index c2717774..2653b474 100644
--- a/src/assembly/hooks.ts
+++ b/src/assembly/hooks.ts
@@ -12,6 +12,8 @@ export interface RumAssembleParams {
   eventType: RumEventType;
   startTime: TimeStamp;
   source: EventSource;
+  /** For renderer events: the view ID reported by the renderer's browser RUM SDK. */
+  rendererViewId?: string;
 }
 
 export interface TelemetryAssembleParams {
diff --git a/src/common/bridgeConfig.spec.ts b/src/common/bridgeConfig.spec.ts
index f883aeb5..a76e758a 100644
--- a/src/common/bridgeConfig.spec.ts
+++ b/src/common/bridgeConfig.spec.ts
@@ -10,6 +10,7 @@ describe('bridgeConfig', () => {
     expect(getBridgeConfig()).toEqual({
       defaultPrivacyLevel: 'mask',
       allowedWebViewHosts: [],
+      // 'records' is intentionally omitted from the pre-init fallback (replay must wait for the real config).
       capabilities: ['profiles'],
     });
   });
diff --git a/src/common/bridgeConfig.ts b/src/common/bridgeConfig.ts
index 1aabde88..1a64a80f 100644
--- a/src/common/bridgeConfig.ts
+++ b/src/common/bridgeConfig.ts
@@ -19,10 +19,19 @@ function getHolder(): BridgeConfigHolder {
   const store = globalThis as unknown as Record<symbol, BridgeConfigHolder | undefined>;
   let holder = store[BRIDGE_CONFIG];
   if (!holder) {
-    // Advertise the SDK's supported capabilities by default to signal support. init() replaces this with
-    // the config-derived value; that narrowing is only an optimization to save renderer work, since the
-    // Electron SDK config (not the advertised capability) governs what is actually sent to Datadog.
-    holder = { value: { defaultPrivacyLevel: 'mask', allowedWebViewHosts: [], capabilities: ['profiles'] } };
+    // Advertise supported capabilities by default to signal support for windows loaded before init().
+    // init() replaces this with the config-derived value; that narrowing is only an optimization to
+    // save renderer work, since the Electron SDK config (not the advertised capability) governs what
+    // is actually sent to Datadog.
+    //
+    // 'records' is deliberately excluded from this fallback: unlike profiling, replay recording
+    // captures DOM data and depends on the real defaultPrivacyLevel and sampling decision, neither of
+    // which is known before init(). Advertising it here would start the renderer recording (and
+    // streaming records over IPC) for a session that may not even have replay enabled. It is added
+    // once setBridgeConfig() publishes the real options.
+    holder = {
+      value: { defaultPrivacyLevel: 'mask', allowedWebViewHosts: [], capabilities: ['profiles'] },
+    };
     store[BRIDGE_CONFIG] = holder;
   }
   return holder;
diff --git a/src/config.spec.ts b/src/config.spec.ts
index 29921c06..6d430068 100644
--- a/src/config.spec.ts
+++ b/src/config.spec.ts
@@ -362,6 +362,27 @@ describe('buildConfiguration', () => {
     });
   });
 
+  describe('sessionReplaySampleRate validation', () => {
+    it('defaults to 0 when not provided', () => {
+      expect(buildConfiguration({ ...DEFAULT_CONFIG })?.sessionReplaySampleRate).toBe(0);
+    });
+
+    it.each([0, 50, 100])('accepts valid value: %d', (value) => {
+      expect(buildConfiguration({ ...DEFAULT_CONFIG, sessionReplaySampleRate: value })?.sessionReplaySampleRate).toBe(
+        value
+      );
+    });
+
+    it.each([-1, 101, 'fifty', {}, NaN])('rejects invalid value: %s', (value) => {
+      const config = { ...DEFAULT_CONFIG, sessionReplaySampleRate: value } as unknown as InitConfiguration;
+
+      expect(buildConfiguration(config)).toBeUndefined();
+      expect(display.error).toHaveBeenCalledWith(
+        "Configuration error: 'sessionReplaySampleRate' must be a number between 0 and 100"
+      );
+    });
+  });
+
   describe('profilingSampleRate validation', () => {
     it('defaults to 0 when not provided', () => {
       const result = buildConfiguration({ ...DEFAULT_CONFIG });
diff --git a/src/config.ts b/src/config.ts
index 83f3c275..2efa3d60 100644
--- a/src/config.ts
+++ b/src/config.ts
@@ -37,6 +37,7 @@ export interface InitConfiguration {
   env?: string;
   version?: string;
   sessionSampleRate?: number;
+  sessionReplaySampleRate?: number;
   profilingSampleRate?: number;
   telemetrySampleRate?: number;
   batchSize?: BatchSize;
@@ -54,6 +55,7 @@ export interface Configuration {
   version?: string;
   proxy?: string;
   sessionSampleRate: number;
+  sessionReplaySampleRate: number;
   profilingSampleRate: number;
   telemetrySampleRate: number;
   batchSize?: BatchSize;
@@ -101,6 +103,17 @@ function validateSessionSampleRate(value: unknown): number | undefined {
   return value as number;
 }
 
+function validateSessionReplaySampleRate(value: unknown): number | undefined {
+  if (value === undefined || value === null) {
+    return 0;
+  }
+  if (!Number.isFinite(value) || (value as number) < 0 || (value as number) > 100) {
+    display.error("Configuration error: 'sessionReplaySampleRate' must be a number between 0 and 100");
+    return undefined;
+  }
+  return value as number;
+}
+
 function validateProfilingSampleRate(value: unknown): number | undefined {
   if (value === undefined || value === null) {
     return 0;
@@ -163,10 +176,16 @@ export function buildConfiguration(initConfig: InitConfiguration): Configuration
 
   const proxy = validateOptionalString(initConfig.proxy);
   const sessionSampleRate = validateSessionSampleRate(initConfig.sessionSampleRate);
+  const sessionReplaySampleRate = validateSessionReplaySampleRate(initConfig.sessionReplaySampleRate);
   const profilingSampleRate = validateProfilingSampleRate(initConfig.profilingSampleRate);
   const telemetrySampleRate = validateTelemetrySampleRate(initConfig.telemetrySampleRate);
 
-  if (sessionSampleRate === undefined || profilingSampleRate === undefined || telemetrySampleRate === undefined) {
+  if (
+    sessionSampleRate === undefined ||
+    sessionReplaySampleRate === undefined ||
+    profilingSampleRate === undefined ||
+    telemetrySampleRate === undefined
+  ) {
     return undefined;
   }
 
@@ -179,6 +198,7 @@ export function buildConfiguration(initConfig: InitConfiguration): Configuration
     version: validateOptionalString(initConfig.version),
     proxy,
     sessionSampleRate,
+    sessionReplaySampleRate,
     profilingSampleRate,
     telemetrySampleRate,
     defaultPrivacyLevel: validateDefaultPrivacyLevel(initConfig.defaultPrivacyLevel),
diff --git a/src/domain/replay/ReplayCollection.spec.ts b/src/domain/replay/ReplayCollection.spec.ts
new file mode 100644
index 00000000..1d14dd23
--- /dev/null
+++ b/src/domain/replay/ReplayCollection.spec.ts
@@ -0,0 +1,312 @@
+import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest';
+import { EventFormat, EventKind, EventTrack, EventManager, EventSource, LifecycleKind } from '../../event';
+import type { ServerReplayEvent } from '../../event';
+import { createFormatHooks } from '../../assembly';
+import { addError } from '../telemetry';
+import { ReplayCollection } from './ReplayCollection';
+import type { Configuration } from '../../config';
+import type { SessionManager } from '../session';
+import { CreationReason } from './Segment';
+import type { ReplaySegmentPayload } from './Segment';
+
+vi.mock('../telemetry', () => ({
+  // eslint-disable-next-line @typescript-eslint/no-unsafe-function-type
+  monitor: (fn: Function) => fn,
+  setTimeout: (callback: () => void, delay?: number) => global.setTimeout(callback, delay),
+  clearTimeout: (timeoutId: ReturnType<typeof global.setTimeout>) => global.clearTimeout(timeoutId),
+  addError: vi.fn(),
+}));
+
+vi.mock('../../tools/StreamingDeflate', () => ({
+  StreamingDeflate: class {
+    compressSegment = vi.fn().mockResolvedValue(Buffer.from('compressed'));
+  },
+}));
+
+function makeConfig(overrides?: Partial<Configuration>): Configuration {
+  return {
+    site: 'datadoghq.com',
+    service: 'test',
+    clientToken: 'pub-test',
+    applicationId: 'app-1',
+    sessionSampleRate: 100,
+    sessionReplaySampleRate: 100,
+    profilingSampleRate: 0,
+    telemetrySampleRate: 0,
+    defaultPrivacyLevel: 'mask',
+    allowedWebViewHosts: [],
+    ...overrides,
+  };
+}
+
+function makeSessionManager(id = 'sess-1', status: 'active' | 'expired' = 'active'): SessionManager {
+  return {
+    getSession: () => ({ id, status }),
+    // Every record timestamp resolves to this session while active, so the boundary guard passes;
+    // an expired session is not tracked, so it resolves to undefined.
+    getTrackedSessionId: () => (status === 'active' ? id : undefined),
+  } as unknown as SessionManager;
+}
+
+function makeHooks() {
+  return createFormatHooks();
+}
+
+function sendRecord(
+  eventManager: EventManager,
+  record: { type: number; timestamp: number; [key: string]: unknown },
+  viewId = 'view-1'
+) {
+  eventManager.notify({
+    kind: EventKind.RAW,
+    source: EventSource.RENDERER,
+    format: EventFormat.REPLAY,
+    data: record,
+    view: { id: viewId },
+  });
+}
+
+function captureReplayEvents(eventManager: EventManager): ReplaySegmentPayload[] {
+  const captured: ReplaySegmentPayload[] = [];
+  eventManager.registerHandler<ServerReplayEvent>({
+    canHandle: (event): event is ServerReplayEvent =>
+      event.kind === EventKind.SERVER && event.track === EventTrack.REPLAY,
+    handle: (event) => captured.push(event.data),
+  });
+  return captured;
+}
+
+describe('ReplayCollection', () => {
+  let eventManager: EventManager;
+
+  beforeEach(() => {
+    vi.clearAllMocks();
+    vi.useFakeTimers();
+    eventManager = new EventManager();
+  });
+
+  afterEach(() => {
+    vi.useRealTimers();
+  });
+
+  describe('idle behaviour', () => {
+    it('emits nothing when no records are received', () => {
+      const captured = captureReplayEvents(eventManager);
+      new ReplayCollection(eventManager, makeConfig(), makeSessionManager(), makeHooks());
+
+      vi.advanceTimersByTime(10_000);
+      expect(captured).toHaveLength(0);
+    });
+
+    it('does not collect records when session is expired', () => {
+      const captured = captureReplayEvents(eventManager);
+      new ReplayCollection(eventManager, makeConfig(), makeSessionManager('sess-1', 'expired'), makeHooks());
+
+      sendRecord(eventManager, { type: 3, timestamp: 100 });
+      vi.advanceTimersByTime(10_000);
+      expect(captured).toHaveLength(0);
+    });
+  });
+
+  describe('session replay sampling', () => {
+    it('uses the parent-corrected replay rate for the session ID', () => {
+      const captured = captureReplayEvents(eventManager);
+      const highHashSessionId = '5321b54a-d6ec-4b24-996d-dd70c617e09a';
+      const config = makeConfig({ sessionSampleRate: 50, sessionReplaySampleRate: 100 });
+      new ReplayCollection(eventManager, config, makeSessionManager(highHashSessionId), makeHooks());
+
+      sendRecord(eventManager, { type: 3, timestamp: 100 });
+      vi.advanceTimersByTime(5_000);
+
+      expect(captured).toHaveLength(0);
+    });
+  });
+
+  describe('duration-based flush (5 s timer)', () => {
+    it('flushes segment after 5 s and emits a ServerReplayEvent', async () => {
+      const captured = captureReplayEvents(eventManager);
+      new ReplayCollection(eventManager, makeConfig(), makeSessionManager(), makeHooks());
+
+      sendRecord(eventManager, { type: 2, timestamp: 1000 });
+      vi.advanceTimersByTime(5_000);
+      await Promise.resolve();
+
+      expect(captured).toHaveLength(1);
+      expect(captured[0].metadata.records_count).toBe(1);
+      expect(captured[0].metadata.application.id).toBe('app-1');
+      expect(captured[0].metadata.session.id).toBe('sess-1');
+      expect(captured[0].metadata.view.id).toBe('view-1');
+    });
+
+    it('accumulates multiple records into a single segment', async () => {
+      const captured = captureReplayEvents(eventManager);
+      new ReplayCollection(eventManager, makeConfig(), makeSessionManager(), makeHooks());
+
+      sendRecord(eventManager, { type: 4, timestamp: 100 });
+      sendRecord(eventManager, { type: 2, timestamp: 200 });
+      sendRecord(eventManager, { type: 3, timestamp: 300 });
+      vi.advanceTimersByTime(5_000);
+      await Promise.resolve();
+
+      expect(captured).toHaveLength(1);
+      expect(captured[0].metadata.records_count).toBe(3);
+      expect(captured[0].metadata.start).toBe(100);
+      expect(captured[0].metadata.end).toBe(300);
+      expect(captured[0].metadata.has_full_snapshot).toBe(true);
+    });
+  });
+
+  describe('view-change flush', () => {
+    it('flushes when the view ID changes', async () => {
+      const captured = captureReplayEvents(eventManager);
+      new ReplayCollection(eventManager, makeConfig(), makeSessionManager(), makeHooks());
+
+      sendRecord(eventManager, { type: 3, timestamp: 100 }, 'view-1');
+      sendRecord(eventManager, { type: 3, timestamp: 200 }, 'view-2');
+      await Promise.resolve();
+
+      expect(captured).toHaveLength(1);
+      expect(captured[0].metadata.view.id).toBe('view-1');
+    });
+
+    it('sets creation_reason to view_change on the next segment', async () => {
+      const captured = captureReplayEvents(eventManager);
+      new ReplayCollection(eventManager, makeConfig(), makeSessionManager(), makeHooks());
+
+      sendRecord(eventManager, { type: 3, timestamp: 100 }, 'view-1');
+      sendRecord(eventManager, { type: 3, timestamp: 200 }, 'view-2');
+      vi.advanceTimersByTime(5_000);
+      await Promise.resolve();
+
+      expect(captured).toHaveLength(2);
+      expect(captured[1].metadata.creation_reason).toBe(CreationReason.VIEW_CHANGE);
+    });
+  });
+
+  describe('session lifecycle flush', () => {
+    it('flushes on SESSION_EXPIRED', async () => {
+      const captured = captureReplayEvents(eventManager);
+      new ReplayCollection(eventManager, makeConfig(), makeSessionManager(), makeHooks());
+
+      sendRecord(eventManager, { type: 3, timestamp: 100 });
+      eventManager.notify({ kind: EventKind.LIFECYCLE, lifecycle: LifecycleKind.SESSION_EXPIRED });
+      await Promise.resolve();
+
+      expect(captured).toHaveLength(1);
+    });
+  });
+
+  describe('segment indexing', () => {
+    it('increments index_in_view for successive segments in the same view', async () => {
+      const captured = captureReplayEvents(eventManager);
+      new ReplayCollection(eventManager, makeConfig(), makeSessionManager(), makeHooks());
+
+      sendRecord(eventManager, { type: 3, timestamp: 100 }, 'view-1');
+      vi.advanceTimersByTime(5_000);
+      await Promise.resolve();
+
+      sendRecord(eventManager, { type: 3, timestamp: 200 }, 'view-1');
+      vi.advanceTimersByTime(5_000);
+      await Promise.resolve();
+
+      expect(captured[0].metadata.index_in_view).toBe(0);
+      expect(captured[1].metadata.index_in_view).toBe(1);
+    });
+
+    it('resets index_in_view for a new view', async () => {
+      const captured = captureReplayEvents(eventManager);
+      new ReplayCollection(eventManager, makeConfig(), makeSessionManager(), makeHooks());
+
+      sendRecord(eventManager, { type: 3, timestamp: 100 }, 'view-1');
+      vi.advanceTimersByTime(5_000);
+      await Promise.resolve();
+
+      sendRecord(eventManager, { type: 3, timestamp: 200 }, 'view-2');
+      vi.advanceTimersByTime(5_000);
+      await Promise.resolve();
+
+      expect(captured[0].metadata.index_in_view).toBe(0);
+      expect(captured[1].metadata.index_in_view).toBe(0);
+    });
+  });
+
+  describe('getViewReplayStats()', () => {
+    it('returns stats accumulated for a view after flushing', async () => {
+      const collection = new ReplayCollection(eventManager, makeConfig(), makeSessionManager(), makeHooks());
+      captureReplayEvents(eventManager);
+
+      sendRecord(eventManager, { type: 3, timestamp: 100 }, 'view-abc');
+      vi.advanceTimersByTime(5_000);
+      await Promise.resolve();
+
+      const stats = collection.getViewReplayStats('view-abc');
+      expect(stats).toBeDefined();
+      expect(stats!.segments_count).toBe(1);
+      expect(stats!.segments_total_raw_size).toBeGreaterThan(0);
+    });
+
+    it('returns undefined for a view with no segments', () => {
+      const collection = new ReplayCollection(eventManager, makeConfig(), makeSessionManager(), makeHooks());
+      expect(collection.getViewReplayStats('unknown-view')).toBeUndefined();
+    });
+  });
+
+  describe('stop()', () => {
+    it('flushes pending segment before resolving', async () => {
+      const captured = captureReplayEvents(eventManager);
+      const collection = new ReplayCollection(eventManager, makeConfig(), makeSessionManager(), makeHooks());
+
+      sendRecord(eventManager, { type: 3, timestamp: 100 });
+      await collection.stop();
+
+      expect(captured).toHaveLength(1);
+    });
+
+    it('resolves immediately if no records are pending', async () => {
+      const collection = new ReplayCollection(eventManager, makeConfig(), makeSessionManager(), makeHooks());
+      await expect(collection.stop()).resolves.toBeUndefined();
+    });
+  });
+
+  // A record delayed across a session boundary (inactivity/timeout → renewal) must not be compressed
+  // into the current session's deflate stream — that would misattribute it and corrupt stitching.
+  describe('session boundary guard', () => {
+    it('drops a record captured under a different tracked session and reports telemetry', async () => {
+      const captured = captureReplayEvents(eventManager);
+      // Record ts=100 resolves to sess-A; "now" (the default) resolves to the current session sess-B.
+      const sessionManager = {
+        getSession: () => ({ id: 'sess-B', status: 'active' as const }),
+        getTrackedSessionId: (at = 1000) => (at < 1000 ? 'sess-A' : 'sess-B'),
+      } as unknown as SessionManager;
+      const collection = new ReplayCollection(eventManager, makeConfig(), sessionManager, makeHooks());
+
+      sendRecord(eventManager, { type: 2, timestamp: 100 });
+      await collection.stop();
+
+      expect(captured).toHaveLength(0);
+      expect(addError).toHaveBeenCalledTimes(1);
+    });
+
+    it('does not report telemetry for records of an untracked/unsampled session (no mismatch)', async () => {
+      const captured = captureReplayEvents(eventManager);
+      // Not tracked (e.g. session not sampled): both lookups are undefined, so there is no mismatch —
+      // the record is dropped silently by the replay-sampling path, without boundary telemetry.
+      const sessionManager = {
+        getSession: () => ({ id: 'sess-x', status: 'active' as const }),
+        getTrackedSessionId: () => undefined,
+      } as unknown as SessionManager;
+      const collection = new ReplayCollection(
+        eventManager,
+        makeConfig({ sessionReplaySampleRate: 0 }),
+        sessionManager,
+        makeHooks()
+      );
+
+      sendRecord(eventManager, { type: 2, timestamp: 100 });
+      await collection.stop();
+
+      expect(captured).toHaveLength(0);
+      expect(addError).not.toHaveBeenCalled();
+    });
+  });
+});
diff --git a/src/domain/replay/ReplayCollection.ts b/src/domain/replay/ReplayCollection.ts
new file mode 100644
index 00000000..0e6d11e8
--- /dev/null
+++ b/src/domain/replay/ReplayCollection.ts
@@ -0,0 +1,263 @@
+import { ONE_SECOND, type TimeStamp } from '@datadog/js-core/time';
+import { EventFormat, EventKind, EventTrack, LifecycleKind } from '../../event';
+import type { EventManager, RawReplayEvent, LifecycleEvent } from '../../event';
+import type { Configuration } from '../../config';
+import type { FormatHooks } from '../../assembly';
+import { correctedChildSampleRate, isSessionSampled } from '../../tools/Sampler';
+import { StreamingDeflate } from '../../tools/StreamingDeflate';
+import type { SessionManager } from '../session';
+import { addError, clearTimeout, monitor, setTimeout } from '../telemetry';
+import { registerReplayContext } from './replayContext';
+import { byteSizeOf, CreationReason, Segment, type BrowserRecord, type SegmentContext } from './Segment';
+
+// Matches the browser SDK flush cadence.
+const SEGMENT_DURATION_LIMIT = 5 * ONE_SECOND;
+
+// 10 MB matches the iOS SDK (DatadogSessionReplay maxObjectSize). The browser
+// SDK caps at 60 KB because of browser fetch/IPC body-size constraints that do
+// not apply in the Electron main process.
+const SEGMENT_BYTES_LIMIT = 10 * 1024 * 1024;
+
+/**
+ * Orchestrates session replay segment collection in the main process.
+ *
+ * Listens for individual BrowserRecord events from renderer processes (via
+ * the bridge), buffers them into {@link Segment} objects with proper context,
+ * and emits compressed {@link ServerReplayEvent}s for the transport layer to
+ * persist and upload.
+ *
+ * Segments are flushed when:
+ * - Duration limit (5s) is reached
+ * - Estimated byte size limit (10MB) is exceeded
+ * - The renderer view changes (different view.id)
+ * - The session expires or renews
+ */
+export interface ViewReplayStats {
+  segments_count: number;
+  segments_total_raw_size: number;
+}
+
+export class ReplayCollection {
+  private segment: Segment | null = null;
+  private currentViewId: string | undefined;
+  private nextCreationReason: CreationReason = CreationReason.INIT;
+  private segmentIndexPerView = new Map<string, number>();
+  private viewReplayStats = new Map<string, ViewReplayStats>();
+  private flushTimeoutId: ReturnType<typeof setTimeout> | null = null;
+  // One persistent deflate stream per session — required so the backend can
+  // stitch all segments into a single valid ZLIB stream for the replay player.
+  private deflate = new StreamingDeflate();
+  // Tracks the last async compress+notify so stop() can await it.
+  private pendingFlush: Promise<void> = Promise.resolve();
+
+  constructor(
+    private readonly eventManager: EventManager,
+    private readonly config: Configuration,
+    private readonly sessionManager: SessionManager,
+    hooks: FormatHooks
+  ) {
+    // Enrich renderer view events with this session's replay stats. Registered here (rather than by the
+    // caller) so all replay-specific assembly logic lives with the collection, mirroring ProfilingCollection.
+    registerReplayContext(
+      hooks,
+      (viewId) => this.getViewReplayStats(viewId),
+      (at) => this.isReplayActiveAt(at)
+    );
+
+    this.eventManager.registerHandler<RawReplayEvent>({
+      canHandle: (event): event is RawReplayEvent =>
+        event.kind === EventKind.RAW && 'format' in event && event.format === EventFormat.REPLAY,
+      handle: monitor((event: RawReplayEvent) => {
+        this.onRecord(event.data, event.view.id);
+      }),
+    });
+
+    this.eventManager.registerHandler<LifecycleEvent>({
+      canHandle: (event): event is LifecycleEvent => event.kind === EventKind.LIFECYCLE,
+      handle: monitor((event: LifecycleEvent) => {
+        if (event.lifecycle === LifecycleKind.SESSION_EXPIRED) {
+          this.flush(CreationReason.INIT);
+        } else if (event.lifecycle === LifecycleKind.SESSION_RENEW) {
+          // Fresh deflate context so the new session's segments form an
+          // independent ZLIB stream the backend can stitch separately.
+          this.deflate = new StreamingDeflate();
+          // All view data from the old session is obsolete — clear to prevent
+          // unbounded growth across long-lived sessions.
+          this.segmentIndexPerView.clear();
+          this.viewReplayStats.clear();
+        }
+      }),
+    });
+  }
+
+  private isReplaySampled(sessionId: string): boolean {
+    return isSessionSampled(
+      sessionId,
+      correctedChildSampleRate(this.config.sessionSampleRate, this.config.sessionReplaySampleRate)
+    );
+  }
+
+  /**
+   * Whether replay is being recorded for the session that covered `at` — the sampling decision for
+   * the event's *own* session, resolved by time (mirroring the session and profiling hooks), not the
+   * current session. Keeps a late-delivered view event's has_replay consistent with its session
+   * across an expiry/renewal boundary. Returns false when no tracked session covered `at`.
+   */
+  private isReplayActiveAt(at: TimeStamp): boolean {
+    const sessionId = this.sessionManager.getTrackedSessionId(at);
+    return sessionId !== undefined && this.isReplaySampled(sessionId);
+  }
+
+  private onRecord(record: BrowserRecord, viewId: string | undefined): void {
+    // Session-boundary handling: replay compresses a session's segments into one persistent deflate
+    // stream (reset on renewal), so a record must be compressed under the same session it belongs to.
+    // When IPC is delayed across an inactivity expiry/renewal, the session active at processing time
+    // may differ from the one that covered the record's capture time. We can't safely retro-attribute
+    // it here — its bytes would still land in the current session's deflate stream, corrupting the
+    // stitched replay — so we drop it (and report telemetry) rather than misattribute it into the
+    // wrong session or view. This is correct and sufficient: the only cost is losing the occasional
+    // straggler record at a boundary, and the telemetry lets us monitor how often that actually
+    // happens. If it ever proves frequent enough to matter, a fuller fix — attributing the record to
+    // its own session by resolving getTrackedSessionId(record.timestamp) in getSegmentContext (as
+    // ProfilingCollection does), with per-session deflate handling — can follow.
+    //
+    // Only a genuine *mismatch* is reported: when both resolve to undefined (session not sampled /
+    // not tracked) the record falls through to the normal silent drop in getSegmentContext, so we
+    // don't emit telemetry for every record of an unsampled session.
+    const owningSessionId = this.sessionManager.getTrackedSessionId(record.timestamp as TimeStamp);
+    if (owningSessionId !== this.sessionManager.getTrackedSessionId()) {
+      addError(new Error('Dropping replay record captured outside the current session'));
+      return;
+    }
+
+    // Detect view change
+    if (viewId && this.currentViewId && viewId !== this.currentViewId) {
+      this.flush(CreationReason.VIEW_CHANGE);
+    }
+
+    if (viewId) {
+      this.currentViewId = viewId;
+    }
+
+    let segment = this.ensureSegment();
+    if (!segment) {
+      return;
+    }
+
+    // Split *before* appending so the segment written to disk never exceeds the cap by a whole
+    // record (a full snapshot can be large). A single record bigger than the cap is unavoidable —
+    // it still gets its own segment. Flushing here starts a fresh segment for this record.
+    const recordByteSize = byteSizeOf(record);
+    if (!segment.isEmpty && segment.estimatedSize + recordByteSize > SEGMENT_BYTES_LIMIT) {
+      this.flush(CreationReason.SEGMENT_BYTES_LIMIT);
+      segment = this.ensureSegment();
+      if (!segment) {
+        return;
+      }
+    }
+
+    segment.addRecord(record, recordByteSize);
+  }
+
+  /** Returns the current segment, creating one if needed. Null when there is no valid context. */
+  private ensureSegment(): Segment | null {
+    if (this.segment) {
+      return this.segment;
+    }
+
+    const context = this.getSegmentContext();
+    if (!context) {
+      return null;
+    }
+
+    const indexInView = this.getNextSegmentIndex(context.view.id);
+    this.segment = new Segment(context, this.nextCreationReason, indexInView);
+    this.nextCreationReason = CreationReason.INIT;
+    this.scheduleFlush();
+    return this.segment;
+  }
+
+  private getSegmentContext(): SegmentContext | undefined {
+    const session = this.sessionManager.getSession();
+    if (session.status !== 'active' || !this.currentViewId || !this.isReplaySampled(session.id)) {
+      return undefined;
+    }
+
+    return {
+      application: { id: this.config.applicationId },
+      session: { id: session.id },
+      view: { id: this.currentViewId },
+    };
+  }
+
+  private getNextSegmentIndex(viewId: string): number {
+    const current = this.segmentIndexPerView.get(viewId) ?? 0;
+    this.segmentIndexPerView.set(viewId, current + 1);
+    return current;
+  }
+
+  getViewReplayStats(viewId: string): ViewReplayStats | undefined {
+    return this.viewReplayStats.get(viewId);
+  }
+
+  private flush(reason: CreationReason): void {
+    this.clearFlushTimeout();
+
+    if (this.segment && !this.segment.isEmpty) {
+      const flushResult = this.segment.flush();
+      const viewId = flushResult.metadata.view.id;
+      const current = this.viewReplayStats.get(viewId) ?? { segments_count: 0, segments_total_raw_size: 0 };
+      this.viewReplayStats.set(viewId, {
+        segments_count: current.segments_count + 1,
+        segments_total_raw_size: current.segments_total_raw_size + flushResult.rawBytesCount,
+      });
+
+      const data = Buffer.from(flushResult.serializedSegment, 'utf8');
+      const segmentFlush = this.deflate.compressSegment(data).then(
+        monitor((compressed: Buffer) => {
+          this.eventManager.notify({
+            kind: EventKind.SERVER,
+            track: EventTrack.REPLAY,
+            data: {
+              metadata: flushResult.metadata,
+              rawBytesCount: flushResult.rawBytesCount,
+              compressed,
+            },
+          });
+        })
+      );
+      // Chain rather than replace so stop() awaits all in-flight compressions.
+      // Without this, a SESSION_EXPIRED flush followed by a new-session flush
+      // before the first compression completes would overwrite pendingFlush,
+      // causing stop() to miss the expired session's last segment on quit.
+      this.pendingFlush = Promise.all([this.pendingFlush, segmentFlush]).then(() => undefined);
+    }
+
+    this.segment = null;
+    this.nextCreationReason = reason;
+  }
+
+  /**
+   * Flush any pending segment and wait for compression to complete.
+   * Call on graceful app shutdown (e.g. `app.on('before-quit')`) so the
+   * final segment is queued for the transport layer before process exit.
+   */
+  stop(): Promise<void> {
+    this.flush(CreationReason.SEGMENT_DURATION_LIMIT);
+    return this.pendingFlush;
+  }
+
+  private scheduleFlush(): void {
+    this.clearFlushTimeout();
+    this.flushTimeoutId = setTimeout(() => {
+      this.flush(CreationReason.SEGMENT_DURATION_LIMIT);
+    }, SEGMENT_DURATION_LIMIT);
+  }
+
+  private clearFlushTimeout(): void {
+    if (this.flushTimeoutId !== null) {
+      clearTimeout(this.flushTimeoutId);
+      this.flushTimeoutId = null;
+    }
+  }
+}
diff --git a/src/domain/replay/Segment.spec.ts b/src/domain/replay/Segment.spec.ts
new file mode 100644
index 00000000..e7977316
--- /dev/null
+++ b/src/domain/replay/Segment.spec.ts
@@ -0,0 +1,129 @@
+import { describe, it, expect } from 'vitest';
+import { CreationReason, Segment, type BrowserRecord } from './Segment';
+
+const CONTEXT = { application: { id: 'app-1' }, session: { id: 'sess-1' }, view: { id: 'view-1' } };
+
+function makeRecord(overrides: Partial<BrowserRecord> = {}): BrowserRecord {
+  return { type: 3, timestamp: 1000, data: { source: 0 }, ...overrides };
+}
+
+describe('Segment', () => {
+  describe('initial state', () => {
+    it('starts empty', () => {
+      const segment = new Segment(CONTEXT, CreationReason.INIT, 0);
+      expect(segment.isEmpty).toBe(true);
+      expect(segment.recordsCount).toBe(0);
+    });
+  });
+
+  describe('addRecord()', () => {
+    it('tracks record count', () => {
+      const segment = new Segment(CONTEXT, CreationReason.INIT, 0);
+      segment.addRecord(makeRecord());
+      segment.addRecord(makeRecord());
+      expect(segment.recordsCount).toBe(2);
+      expect(segment.isEmpty).toBe(false);
+    });
+
+    it('tracks start and end timestamps across out-of-order records', () => {
+      const segment = new Segment(CONTEXT, CreationReason.INIT, 0);
+      segment.addRecord(makeRecord({ timestamp: 500 }));
+      segment.addRecord(makeRecord({ timestamp: 300 }));
+      segment.addRecord(makeRecord({ timestamp: 700 }));
+      const { metadata } = segment.flush();
+      expect(metadata.start).toBe(300);
+      expect(metadata.end).toBe(700);
+    });
+
+    it('does not set has_full_snapshot for non-snapshot records', () => {
+      const segment = new Segment(CONTEXT, CreationReason.INIT, 0);
+      segment.addRecord(makeRecord({ type: 3 }));
+      expect(segment.flush().metadata.has_full_snapshot).toBe(false);
+    });
+
+    it('sets has_full_snapshot when a type-2 record is added', () => {
+      const segment = new Segment(CONTEXT, CreationReason.INIT, 0);
+      segment.addRecord(makeRecord({ type: 2 }));
+      expect(segment.flush().metadata.has_full_snapshot).toBe(true);
+    });
+
+    it('sets has_full_snapshot for a change-format record (type 12) on the first segment of a view', () => {
+      const segment = new Segment(CONTEXT, CreationReason.INIT, 0);
+      segment.addRecord(makeRecord({ type: 12 }));
+      expect(segment.flush().metadata.has_full_snapshot).toBe(true);
+    });
+
+    it('does not treat a change-format record (type 12) as a full snapshot on later segments', () => {
+      const segment = new Segment(CONTEXT, CreationReason.VIEW_CHANGE, 2);
+      segment.addRecord(makeRecord({ type: 12 }));
+      expect(segment.flush().metadata.has_full_snapshot).toBe(false);
+    });
+
+    it('accumulates estimated size proportional to record count', () => {
+      const segment = new Segment(CONTEXT, CreationReason.INIT, 0);
+      expect(segment.estimatedSize).toBe(0);
+      segment.addRecord(makeRecord());
+      const sizeAfterOne = segment.estimatedSize;
+      segment.addRecord(makeRecord());
+      expect(segment.estimatedSize).toBeGreaterThan(sizeAfterOne);
+    });
+  });
+
+  describe('flush()', () => {
+    it('embeds context and segment config in metadata', () => {
+      const segment = new Segment(CONTEXT, CreationReason.VIEW_CHANGE, 3);
+      segment.addRecord(makeRecord());
+      const { metadata } = segment.flush();
+      expect(metadata.application.id).toBe('app-1');
+      expect(metadata.session.id).toBe('sess-1');
+      expect(metadata.view.id).toBe('view-1');
+      expect(metadata.creation_reason).toBe(CreationReason.VIEW_CHANGE);
+      expect(metadata.index_in_view).toBe(3);
+      expect(metadata.source).toBe('browser');
+    });
+
+    it('produces valid JSON with records array merged with metadata', () => {
+      const segment = new Segment(CONTEXT, CreationReason.INIT, 0);
+      const record = makeRecord({ timestamp: 1000 });
+      segment.addRecord(record);
+
+      const { serializedSegment } = segment.flush();
+      const parsed = JSON.parse(serializedSegment) as Record<string, unknown>;
+
+      expect(Array.isArray(parsed['records'])).toBe(true);
+      expect((parsed['records'] as unknown[]).length).toBe(1);
+      expect((parsed['records'] as unknown[])[0]).toEqual(record);
+      // Metadata fields are merged at the top level (not nested)
+      expect(parsed['application']).toEqual({ id: 'app-1' });
+      expect(parsed['records_count']).toBe(1);
+    });
+
+    it('serialized segment ends with a newline', () => {
+      const segment = new Segment(CONTEXT, CreationReason.INIT, 0);
+      segment.addRecord(makeRecord());
+      const { serializedSegment } = segment.flush();
+      expect(serializedSegment.endsWith('\n')).toBe(true);
+    });
+
+    it('rawBytesCount matches the byte length of the serialized segment', () => {
+      const segment = new Segment(CONTEXT, CreationReason.INIT, 0);
+      segment.addRecord(makeRecord());
+      const { serializedSegment, rawBytesCount } = segment.flush();
+      expect(rawBytesCount).toBe(Buffer.byteLength(serializedSegment, 'utf8'));
+    });
+
+    it('returns a snapshot of metadata (not a live reference)', () => {
+      const segment = new Segment(CONTEXT, CreationReason.INIT, 0);
+      segment.addRecord(makeRecord({ timestamp: 1000 }));
+      const first = segment.flush();
+
+      // After flush, adding more records to a new segment should not affect the snapshot
+      const segment2 = new Segment(CONTEXT, CreationReason.INIT, 0);
+      segment2.addRecord(makeRecord({ timestamp: 9999 }));
+      const second = segment2.flush();
+
+      expect(first.metadata.start).toBe(1000);
+      expect(second.metadata.start).toBe(9999);
+    });
+  });
+});
diff --git a/src/domain/replay/Segment.ts b/src/domain/replay/Segment.ts
new file mode 100644
index 00000000..63dc7be5
--- /dev/null
+++ b/src/domain/replay/Segment.ts
@@ -0,0 +1,146 @@
+/**
+ * Buffers BrowserRecord objects into a segment, tracking metadata required
+ * by the session-replay intake (timestamps, record count, full snapshot flag).
+ *
+ * On flush, produces the complete segment JSON (records array + metadata)
+ * and the metadata object separately (for the multipart 'event' part).
+ */
+
+export interface SegmentContext {
+  application: { id: string };
+  session: { id: string };
+  view: { id: string };
+}
+
+export const CreationReason = {
+  INIT: 'init',
+  SEGMENT_DURATION_LIMIT: 'segment_duration_limit',
+  SEGMENT_BYTES_LIMIT: 'segment_bytes_limit',
+  VIEW_CHANGE: 'view_change',
+  BEFORE_UNLOAD: 'before_unload',
+  VISIBILITY_HIDDEN: 'visibility_hidden',
+  PAGE_FROZEN: 'page_frozen',
+} as const;
+
+export type CreationReason = (typeof CreationReason)[keyof typeof CreationReason];
+
+/** Record type constants from the browser SDK's rrweb-based recording (RecordType). */
+const FULL_SNAPSHOT_TYPE = 2;
+// Browser SDK 7.x change-format: the initial snapshot of a view arrives as a Change record
+// (type 12) rather than a legacy full-snapshot record. It only counts as the full snapshot on
+// the first segment of the view (index_in_view === 0).
+const CHANGE_TYPE = 12;
+
+export interface SegmentMetadata {
+  application: { id: string };
+  session: { id: string };
+  view: { id: string };
+  start: number;
+  end: number;
+  records_count: number;
+  has_full_snapshot: boolean;
+  index_in_view: number;
+  source: 'browser';
+  creation_reason: CreationReason;
+}
+
+export interface SegmentFlushResult {
+  /** Complete segment JSON (records + metadata), ready for compression. */
+  serializedSegment: string;
+  /** Metadata for the multipart 'event' part. */
+  metadata: SegmentMetadata;
+  /** Byte size of the serialized segment before compression. */
+  rawBytesCount: number;
+}
+
+/**
+ * The payload written to disk and read by {@link ReplayBatchConsumer}.
+ * Produced by {@link ReplayCollection} after compression.
+ */
+export interface ReplaySegmentPayload {
+  metadata: SegmentMetadata;
+  rawBytesCount: number;
+  compressed: Buffer;
+}
+
+export interface BrowserRecord {
+  type: number;
+  timestamp: number;
+  [key: string]: unknown;
+}
+
+/** UTF-8 byte size a record contributes to a segment (matches Segment's own accounting). */
+export function byteSizeOf(record: BrowserRecord): number {
+  return Buffer.byteLength(JSON.stringify(record), 'utf8');
+}
+
+export class Segment {
+  private records: BrowserRecord[] = [];
+  private metadata: SegmentMetadata;
+  private _estimatedSize = 0;
+
+  constructor(context: SegmentContext, creationReason: CreationReason, indexInView: number) {
+    this.metadata = {
+      ...context,
+      start: Infinity,
+      end: -Infinity,
+      records_count: 0,
+      has_full_snapshot: false,
+      index_in_view: indexInView,
+      source: 'browser',
+      creation_reason: creationReason,
+    };
+  }
+
+  get recordsCount(): number {
+    return this.records.length;
+  }
+
+  get isEmpty(): boolean {
+    return this.records.length === 0;
+  }
+
+  get estimatedSize(): number {
+    return this._estimatedSize;
+  }
+
+  addRecord(record: BrowserRecord, recordByteSize: number = byteSizeOf(record)): void {
+    // Measure UTF-8 byte length, not UTF-16 code-unit count: SEGMENT_BYTES_LIMIT is a byte
+    // cap and flush() reports rawBytesCount via Buffer.byteLength. Using String#length would
+    // undercount non-ASCII DOM text (CJK/emoji), letting a segment exceed the intended cap.
+    // The caller may pass a precomputed size (from byteSizeOf) to avoid stringifying twice.
+    this._estimatedSize += recordByteSize;
+    this.records.push(record);
+    this.metadata.start = Math.min(this.metadata.start, record.timestamp);
+    this.metadata.end = Math.max(this.metadata.end, record.timestamp);
+    this.metadata.records_count += 1;
+    // Mirror the browser SDK's segment.js: a full snapshot is a legacy type-2 record, or — in the
+    // 7.x change format — a Change record (type 12) on the view's first segment (index_in_view 0).
+    if (record.type === FULL_SNAPSHOT_TYPE || (record.type === CHANGE_TYPE && this.metadata.index_in_view === 0)) {
+      this.metadata.has_full_snapshot = true;
+    }
+  }
+
+  flush(): SegmentFlushResult {
+    // Match the browser SDK's exact wire format (from segment.js in @datadog/browser-rum):
+    //   encoder.write('{"records":[' + JSON.stringify(r1))   ← opening + first record
+    //   encoder.write(',' + JSON.stringify(rN))              ← subsequent records
+    //   encoder.write(`],${JSON.stringify(metadata).slice(1)}\n`)  ← close + metadata + newline
+    //
+    // The resulting format is: {"records":[r1,...,rN],...metadata}\n
+    //
+    // The trailing \n is critical: the backend stitches segments by concatenating
+    // DEFLATE bodies. The decompressed stitched stream is N lines of NDJSON that
+    // the player splits on \n to extract individual segment payloads. Without \n
+    // the player sees one large blob and can only parse segment 0.
+    const metadataJson = JSON.stringify(this.metadata);
+    // metadataJson = '{"start":...}' — drop the opening '{' to splice into the records object
+    const serializedSegment = `{"records":${JSON.stringify(this.records)},${metadataJson.slice(1)}\n`;
+
+    return {
+      serializedSegment,
+      metadata: { ...this.metadata },
+      rawBytesCount: Buffer.byteLength(serializedSegment, 'utf8'),
+    };
+  }
+}
diff --git a/src/domain/replay/index.ts b/src/domain/replay/index.ts
new file mode 100644
index 00000000..110cdd6e
--- /dev/null
+++ b/src/domain/replay/index.ts
@@ -0,0 +1,5 @@
+export { ReplayCollection } from './ReplayCollection';
+export type { ViewReplayStats } from './ReplayCollection';
+export { CreationReason } from './Segment';
+export type { ReplaySegmentPayload, SegmentMetadata, BrowserRecord } from './Segment';
+export { registerReplayContext } from './replayContext';
diff --git a/src/domain/replay/replayContext.spec.ts b/src/domain/replay/replayContext.spec.ts
new file mode 100644
index 00000000..c7afae3b
--- /dev/null
+++ b/src/domain/replay/replayContext.spec.ts
@@ -0,0 +1,214 @@
+import { describe, it, expect, vi } from 'vitest';
+import type { TimeStamp } from '@datadog/js-core/time';
+import { createFormatHooks } from '../../assembly';
+import { EventSource } from '../../event';
+import type { ViewReplayStats } from './ReplayCollection';
+import { registerReplayContext } from './replayContext';
+
+const STATS: ViewReplayStats = { segments_count: 3, segments_total_raw_size: 1024 };
+const ACTIVE = () => true;
+const INACTIVE = () => false;
+
+describe('registerReplayContext', () => {
+  describe('when to skip', () => {
+    it('returns SKIPPED for main-process events', () => {
+      const hooks = createFormatHooks();
+      const getStats = vi.fn().mockReturnValue(STATS);
+      registerReplayContext(hooks, getStats, ACTIVE);
+
+      const result = hooks.triggerRum({
+        eventType: 'view',
+        startTime: 0 as TimeStamp,
+        source: EventSource.MAIN,
+        rendererViewId: 'view-1',
+      });
+
+      expect(getStats).not.toHaveBeenCalled();
+      // result is undefined (SKIPPED by our hook, no other hooks registered)
+      expect(result).toBeUndefined();
+    });
+
+    it('returns SKIPPED for renderer non-view events', () => {
+      const hooks = createFormatHooks();
+      const getStats = vi.fn().mockReturnValue(STATS);
+      registerReplayContext(hooks, getStats, ACTIVE);
+
+      const result = hooks.triggerRum({
+        eventType: 'error',
+        startTime: 0 as TimeStamp,
+        source: EventSource.RENDERER,
+        rendererViewId: 'view-1',
+      });
+
+      expect(getStats).not.toHaveBeenCalled();
+      expect(result).toBeUndefined();
+    });
+
+    it('returns SKIPPED when rendererViewId is missing', () => {
+      const hooks = createFormatHooks();
+      const getStats = vi.fn().mockReturnValue(STATS);
+      registerReplayContext(hooks, getStats, ACTIVE);
+
+      const result = hooks.triggerRum({
+        eventType: 'view',
+        startTime: 0 as TimeStamp,
+        source: EventSource.RENDERER,
+      });
+
+      expect(getStats).not.toHaveBeenCalled();
+      expect(result).toBeUndefined();
+    });
+  });
+
+  describe('when replay is not active (session sampled out)', () => {
+    it('clears has_replay and zeroes replay_stats, regardless of any buffered stats', () => {
+      const hooks = createFormatHooks();
+      // Even if stats somehow exist, an inactive session must report no replay.
+      registerReplayContext(hooks, () => STATS, INACTIVE);
+
+      const result = hooks.triggerRum({
+        eventType: 'view',
+        startTime: 0 as TimeStamp,
+        source: EventSource.RENDERER,
+        rendererViewId: 'view-1',
+      });
+
+      // The session was not sampled for replay, so the RUM view must not claim a replay exists —
+      // nor carry stale renderer-provided stats — even if the renderer stamped them.
+      expect((result as Record<string, unknown>)?.['session']).toMatchObject({ has_replay: false });
+      expect((result as Record<string, unknown>)?.['_dd']).toMatchObject({
+        replay_stats: { records_count: 0, segments_count: 0, segments_total_raw_size: 0 },
+      });
+    });
+  });
+
+  describe('when replay is active for a renderer view event', () => {
+    it('calls getViewReplayStats with the renderer view ID', () => {
+      const hooks = createFormatHooks();
+      const getStats = vi.fn().mockReturnValue(STATS);
+      registerReplayContext(hooks, getStats, ACTIVE);
+
+      hooks.triggerRum({
+        eventType: 'view',
+        startTime: 0 as TimeStamp,
+        source: EventSource.RENDERER,
+        rendererViewId: 'view-abc',
+      });
+
+      expect(getStats).toHaveBeenCalledWith('view-abc');
+    });
+
+    it('returns session.has_replay: true', () => {
+      const hooks = createFormatHooks();
+      registerReplayContext(hooks, () => STATS, ACTIVE);
+
+      const result = hooks.triggerRum({
+        eventType: 'view',
+        startTime: 0 as TimeStamp,
+        source: EventSource.RENDERER,
+        rendererViewId: 'view-1',
+      });
+
+      expect((result as Record<string, unknown>)?.['session']).toMatchObject({ has_replay: true });
+    });
+
+    it('returns _dd.replay_stats with segments_count and segments_total_raw_size', () => {
+      const hooks = createFormatHooks();
+      registerReplayContext(hooks, () => STATS, ACTIVE);
+
+      const result = hooks.triggerRum({
+        eventType: 'view',
+        startTime: 0 as TimeStamp,
+        source: EventSource.RENDERER,
+        rendererViewId: 'view-1',
+      });
+
+      expect((result as Record<string, unknown>)?.['_dd']).toMatchObject({
+        replay_stats: {
+          segments_count: STATS.segments_count,
+          segments_total_raw_size: STATS.segments_total_raw_size,
+        },
+      });
+    });
+
+    it('reports has_replay: true with zeroed segment counts before the first segment is flushed', () => {
+      const hooks = createFormatHooks();
+      // Active session, but no segment flushed yet for this view (short view / route change).
+      registerReplayContext(hooks, () => undefined, ACTIVE);
+
+      const result = hooks.triggerRum({
+        eventType: 'view',
+        startTime: 0 as TimeStamp,
+        source: EventSource.RENDERER,
+        rendererViewId: 'view-buffered',
+      });
+
+      // A segment is buffered and will be uploaded, so the view must still claim a replay exists.
+      expect((result as Record<string, unknown>)?.['session']).toMatchObject({ has_replay: true });
+      expect((result as Record<string, unknown>)?.['_dd']).toMatchObject({
+        replay_stats: { segments_count: 0, segments_total_raw_size: 0 },
+      });
+    });
+  });
+
+  describe('attribution by event start time', () => {
+    it('resolves has_replay for the event start time, not the current session', () => {
+      const hooks = createFormatHooks();
+      // Replay is active only for events at/after t=1000 (e.g. an older event's session was sampled
+      // out or had expired). A late-delivered view event must be judged by its own startTime.
+      const isReplayActiveAt = (at: TimeStamp) => (at as number) >= 1000;
+      registerReplayContext(hooks, () => STATS, isReplayActiveAt);
+
+      const oldView = hooks.triggerRum({
+        eventType: 'view',
+        startTime: 500 as TimeStamp,
+        source: EventSource.RENDERER,
+        rendererViewId: 'view-old',
+      });
+      const newView = hooks.triggerRum({
+        eventType: 'view',
+        startTime: 1500 as TimeStamp,
+        source: EventSource.RENDERER,
+        rendererViewId: 'view-new',
+      });
+
+      expect((oldView as Record<string, unknown>)?.['session']).toMatchObject({ has_replay: false });
+      expect((newView as Record<string, unknown>)?.['session']).toMatchObject({ has_replay: true });
+    });
+  });
+
+  describe('integration with other hooks', () => {
+    it('combines replay attributes with other registered hook results', () => {
+      const hooks = createFormatHooks();
+      hooks.registerRum(() => ({ session: { id: 'main-session-id' } }));
+      registerReplayContext(hooks, () => STATS, ACTIVE);
+
+      const result = hooks.triggerRum({
+        eventType: 'view',
+        startTime: 0 as TimeStamp,
+        source: EventSource.RENDERER,
+        rendererViewId: 'view-1',
+      });
+
+      const session = (result as Record<string, unknown>)?.['session'] as Record<string, unknown>;
+      expect(session?.['id']).toBe('main-session-id');
+      expect(session?.['has_replay']).toBe(true);
+    });
+
+    it('does not interfere with non-replay hooks for main-process events', () => {
+      const hooks = createFormatHooks();
+      hooks.registerRum(() => ({ session: { id: 'main-session-id' } }));
+      registerReplayContext(hooks, () => STATS, ACTIVE);
+
+      const result = hooks.triggerRum({
+        eventType: 'view',
+        startTime: 0 as TimeStamp,
+        source: EventSource.MAIN,
+      });
+
+      const session = (result as Record<string, unknown>)?.['session'] as Record<string, unknown>;
+      expect(session?.['id']).toBe('main-session-id');
+      expect(session?.['has_replay']).toBeUndefined();
+    });
+  });
+});
diff --git a/src/domain/replay/replayContext.ts b/src/domain/replay/replayContext.ts
new file mode 100644
index 00000000..800a9e0a
--- /dev/null
+++ b/src/domain/replay/replayContext.ts
@@ -0,0 +1,62 @@
+import { SKIPPED } from '@datadog/js-core/assembly';
+import type { TimeStamp } from '@datadog/js-core/time';
+import { EventSource } from '../../event';
+import type { FormatHooks } from '../../assembly';
+import type { ViewReplayStats } from './ReplayCollection';
+
+/**
+ * Registers a RUM hook that injects session replay state into renderer view
+ * events. Assembly triggers this hook like any other context hook, keeping
+ * the replay-specific logic out of Assembly itself.
+ */
+export function registerReplayContext(
+  hooks: FormatHooks,
+  getViewReplayStats: (viewId: string) => ViewReplayStats | undefined,
+  isReplayActiveAt: (at: TimeStamp) => boolean
+): void {
+  hooks.registerRum(({ source, eventType, startTime, rendererViewId }) => {
+    if (source !== EventSource.RENDERER || eventType !== 'view' || !rendererViewId) {
+      return SKIPPED;
+    }
+
+    // has_replay reflects the main process's replay SAMPLING decision — mirroring the Browser SDK,
+    // where it means "replay is being recorded for this session" — not whether a segment has flushed
+    // yet. It is resolved for the event's *own* session (by startTime), like the session and
+    // profiling hooks, so a view event delivered after an expiry/renewal isn't stamped with the
+    // current session's decision. The main process is authoritative (it samples and uploads), so we
+    // always override the renderer's values. This avoids three failure modes:
+    //  - false positive: the renderer stamped has_replay but Electron sampled that session out;
+    //  - false negative: a segment is buffered but not yet flushed (short views / route changes)
+    //    when the view event is sent, which a flush-state-based flag would wrongly report as no replay;
+    //  - wrong-session: a late view event getting the current session's decision instead of its own.
+    if (!isReplayActiveAt(startTime)) {
+      // No replay for this session. Zero every replay_stats field (rather than omit) because
+      // combine() merges key-by-key and skips undefined, so omitting would let stale renderer counts
+      // survive onto a view that claims no replay.
+      return {
+        session: { has_replay: false },
+        _dd: {
+          replay_stats: {
+            records_count: 0,
+            segments_count: 0,
+            segments_total_raw_size: 0,
+          },
+        },
+      };
+    }
+
+    // Replay is active. Report the main process's authoritative segment counts (0 until the first
+    // flush for this view). records_count is left untouched: the SDK does not track it per view, and
+    // the renderer's own count is the best available.
+    const stats = getViewReplayStats(rendererViewId);
+    return {
+      session: { has_replay: true },
+      _dd: {
+        replay_stats: {
+          segments_count: stats?.segments_count ?? 0,
+          segments_total_raw_size: stats?.segments_total_raw_size ?? 0,
+        },
+      },
+    };
+  });
+}
diff --git a/src/domain/telemetry/timer.ts b/src/domain/telemetry/timer.ts
index 42a21fc4..a999774b 100644
--- a/src/domain/telemetry/timer.ts
+++ b/src/domain/telemetry/timer.ts
@@ -10,6 +10,12 @@ export function setTimeout(callback: () => void, delay?: number): ReturnType<typ
   return global.setTimeout(monitor(callback), delay);
 }
 
+// Paired with setTimeout so timer usage goes through this module rather than reaching for the
+// globals directly. There is no callback to wrap here, so it simply delegates to global.clearTimeout.
+export function clearTimeout(timeoutId: ReturnType<typeof global.setTimeout> | undefined): void {
+  global.clearTimeout(timeoutId);
+}
+
 export function setInterval(callback: () => void, delay?: number): ReturnType<typeof global.setInterval> {
   return global.setInterval(monitor(callback), delay);
 }
diff --git a/src/event/event.constants.ts b/src/event/event.constants.ts
index 8d835e03..99c98e20 100644
--- a/src/event/event.constants.ts
+++ b/src/event/event.constants.ts
@@ -9,6 +9,7 @@ export const EventTrack = {
   RUM: 'rum',
   SPANS: 'spans',
   PROFILE: 'profile',
+  REPLAY: 'replay',
 } as const;
 export type EventTrack = (typeof EventTrack)[keyof typeof EventTrack];
 
@@ -16,6 +17,7 @@ export const EventFormat = {
   RUM: 'rum',
   TELEMETRY: 'telemetry',
   PROFILE: 'profile',
+  REPLAY: 'replay',
 } as const;
 
 export const EventKind = {
diff --git a/src/event/event.types.ts b/src/event/event.types.ts
index 4c3856f1..efb2afe0 100644
--- a/src/event/event.types.ts
+++ b/src/event/event.types.ts
@@ -4,10 +4,11 @@ import { RawRumData, RumEvent } from '../domain/rum';
 import type { TimeStamp } from '@datadog/js-core/time';
 import { RawTraceData } from '../domain/tracing/rawTracingData.types';
 import type { BrowserProfileEvent, BrowserProfilerTrace } from '../domain/profiling';
+import type { ReplaySegmentPayload, BrowserRecord } from '../domain/replay';
 
 export type { BrowserProfileEvent, BrowserProfilerTrace };
 
-export type RawEvent = RawRumEvent | RawTelemetryEvent | RawProfileEvent;
+export type RawEvent = RawRumEvent | RawTelemetryEvent | RawProfileEvent | RawReplayEvent;
 
 export interface RawRumEvent {
   kind: typeof EventKind.RAW;
@@ -23,19 +24,29 @@ export interface RawTelemetryEvent {
   startTime?: TimeStamp;
 }
 
+export interface RawReplayEvent {
+  kind: typeof EventKind.RAW;
+  source: typeof EventSource.RENDERER;
+  format: typeof EventFormat.REPLAY;
+  data: BrowserRecord;
+  view: { id: string };
+  startTime?: TimeStamp;
+}
+
 export type ServerEvent =
   | ServerRumEvent
   | ServerTelemetryEvent
   | ServerLogsEvent
   | ServerSpansEvent
-  | ServerProfileEvent;
+  | ServerProfileEvent
+  | ServerReplayEvent;
 
 /**
  * Server events transported as newline-delimited JSON, i.e. every {@link ServerEvent} whose
  * `data` is the full payload to serialize. Excludes {@link ServerProfileEvent}, which carries
  * an additional `trace` field and is transported as a multipart profile.
  */
-export type StandardServerEvent = Exclude<ServerEvent, ServerProfileEvent>;
+export type StandardServerEvent = Exclude<ServerEvent, ServerProfileEvent | ServerReplayEvent>;
 
 export interface ServerRumEvent {
   kind: typeof EventKind.SERVER;
@@ -80,6 +91,12 @@ export interface ServerProfileEvent {
   trace: BrowserProfilerTrace;
 }
 
+export interface ServerReplayEvent {
+  kind: typeof EventKind.SERVER;
+  track: typeof EventTrack.REPLAY;
+  data: ReplaySegmentPayload;
+}
+
 export interface EndUserActivityEvent {
   kind: typeof EventKind.LIFECYCLE;
   lifecycle: typeof LifecycleKind.END_USER_ACTIVITY;
diff --git a/src/index.ts b/src/index.ts
index 501cbd7e..99215031 100644
--- a/src/index.ts
+++ b/src/index.ts
@@ -1,3 +1,4 @@
+import { app } from 'electron';
 import { MainAssembly, RendererPipeline, createFormatHooks, registerCommonContext } from './assembly';
 import { setDurationVitalApi } from './api';
 import type { AccountInfo, UserInfo } from './domain/customer-context';
@@ -6,8 +7,9 @@ import type { InitConfiguration } from './config';
 import { buildConfiguration } from './config';
 import type { ErrorOptions, FailureReason, FeatureOperationOptions } from './domain/rum';
 import { RumCollection } from './domain/rum';
+import { ReplayCollection } from './domain/replay';
 import { SessionManager } from './domain/session';
-import { callMonitored, startTelemetry } from './domain/telemetry';
+import { callMonitored, monitor, startTelemetry } from './domain/telemetry';
 import { SpanProcessor } from './domain/tracing/SpanProcessor';
 import { Tracing } from './domain/tracing/Tracing';
 import { ProfilingCollection } from './domain/profiling';
@@ -21,6 +23,14 @@ let rumApi: ReturnType<RumCollection['getApi']> | undefined;
 let tracing: Tracing | undefined;
 let userContext: UserContext | undefined;
 let accountContext: AccountContext | undefined;
+let replayCollection: ReplayCollection | undefined;
+// Shared across before-quit invocations so a second quit signal can't spawn a parallel flush
+// (which would return early while the first upload is still in flight and quit prematurely).
+let isQuitting = false;
+// The currently registered before-quit listener. Tracked so repeated init() calls replace it
+// instead of stacking handlers — stacked handlers share isQuitting, and a second one would
+// preventDefault() then return early without ever quitting, deadlocking shutdown.
+let onBeforeQuit: ((event: Electron.Event) => void) | undefined;
 
 /**
  * Internal SDK context
@@ -30,13 +40,6 @@ export interface InternalContext {
   session_id: string;
 }
 
-/**
- * Internal SDK context
- */
-export interface InternalContext {
-  session_id: string;
-}
-
 /**
  * Initialize the Electron SDK
  */
@@ -62,6 +65,7 @@ export async function init(configuration: InitConfiguration): Promise<boolean> {
   new RendererPipeline(eventManager, hooks, config);
 
   new ProfilingCollection(eventManager, sessionManager, config, hooks);
+  replayCollection = new ReplayCollection(eventManager, config, sessionManager, hooks);
 
   if (tracing.enabled) {
     new SpanProcessor(eventManager, hooks, config);
@@ -72,9 +76,50 @@ export async function init(configuration: InitConfiguration): Promise<boolean> {
   rumApi = rum.getApi();
   setDurationVitalApi(rumApi);
 
+  setupBeforeQuitHandler();
+
   return true;
 }
 
+/** Flushes pending SDK data before allowing Electron to quit. */
+function setupBeforeQuitHandler(): void {
+  // Replace any handler from a previous init() so only one is ever registered.
+  if (onBeforeQuit) {
+    app.removeListener('before-quit', onBeforeQuit);
+  }
+
+  // Local const so the closure below always removes *this* handler (the module-level `let` isn't
+  // narrowed inside the closure, and could be reassigned by a later init()).
+  const handler = monitor((event: Electron.Event) => {
+    event.preventDefault();
+
+    // A second quit (user hits Cmd+Q again, or the OS sends another quit signal) while the first
+    // flush is still running must not start a second flush: that flush would find the upload cycle
+    // already in progress, return immediately, and quit before the first upload settles.
+    if (isQuitting) {
+      return;
+    }
+    isQuitting = true;
+
+    // Guard against racing the fallback timer with the flush: whichever settles first quits,
+    // and the loser must not trigger a duplicate quit.
+    let done = false;
+    const doQuit = () => {
+      if (!done) {
+        done = true;
+        app.removeListener('before-quit', handler);
+        app.quit();
+      }
+    };
+
+    setTimeout(doQuit, 5000);
+    void _flushTransport().finally(doQuit);
+  });
+  onBeforeQuit = handler;
+
+  app.on('before-quit', handler);
+}
+
 /**
  * Stop the current session
  */
@@ -263,6 +308,18 @@ export function failFeatureOperation(
  * Internal API to flush all pending batches to the intake
  */
 export async function _flushTransport(): Promise<void> {
+  // Order matters because the dd-trace exporter flush can be slow or never call back, and the
+  // before-quit handler races a 5s fallback timer.
+  //
+  // 1. Produce the final replay segment (compress + hand it to the batch producer).
+  // 2. Flush the transport NOW so that segment is actually written and uploaded before the risky
+  //    tracing flush — stop() alone only queues it in the producer; the write/upload happens in
+  //    transport.flush(). If tracing hung and this ran after it, the fallback could quit with the
+  //    last segment still queued, losing exactly what this path exists to preserve.
+  // 3. Flush tracing: dd-trace turns its batched spans into RUM resource/SPANS events synchronously.
+  // 4. Flush the transport again to upload those tracing-produced events.
+  await replayCollection?.stop();
+  await transport?.flush();
   await tracing?.flush();
   await transport?.flush();
 }
diff --git a/src/instrument/registerBridgeConfig.spec.ts b/src/instrument/registerBridgeConfig.spec.ts
index e5c51853..82130f94 100644
--- a/src/instrument/registerBridgeConfig.spec.ts
+++ b/src/instrument/registerBridgeConfig.spec.ts
@@ -31,6 +31,7 @@ describe('registerBridgeConfigResponder', () => {
     expect(event.returnValue).toEqual({
       defaultPrivacyLevel: 'mask',
       allowedWebViewHosts: [],
+      // 'records' is intentionally omitted from the pre-init fallback (replay must wait for the real config).
       capabilities: ['profiles'],
     });
   });
diff --git a/src/mocks.specUtil.ts b/src/mocks.specUtil.ts
index abf1192d..5b01ee1c 100644
--- a/src/mocks.specUtil.ts
+++ b/src/mocks.specUtil.ts
@@ -37,6 +37,7 @@ export function createTestConfiguration(overrides: Partial<Configuration> = {}):
     clientToken: 'test-token',
     applicationId: 'test-app-id',
     sessionSampleRate: 100,
+    sessionReplaySampleRate: 100,
     profilingSampleRate: 100,
     telemetrySampleRate: 100,
     defaultPrivacyLevel: 'mask',
diff --git a/src/tools/StreamingDeflate.spec.ts b/src/tools/StreamingDeflate.spec.ts
new file mode 100644
index 00000000..3daae4f0
--- /dev/null
+++ b/src/tools/StreamingDeflate.spec.ts
@@ -0,0 +1,107 @@
+import { describe, it, expect } from 'vitest';
+import zlib from 'node:zlib';
+import { StreamingDeflate } from './StreamingDeflate';
+
+const ZLIB_HEADER = [0x78, 0x9c];
+const DEFLATE_FINAL_BLOCK = [0x03, 0x00];
+
+describe('StreamingDeflate', () => {
+  describe('output structure', () => {
+    it('returns a non-empty Buffer', async () => {
+      const deflate = new StreamingDeflate();
+      const result = await deflate.compressSegment(Buffer.from('hello world'));
+      expect(result).toBeInstanceOf(Buffer);
+      expect(result.length).toBeGreaterThan(0);
+    });
+
+    it('first segment starts with the ZLIB header [0x78, 0x9C]', async () => {
+      const deflate = new StreamingDeflate();
+      const result = await deflate.compressSegment(Buffer.from('first segment data'));
+      expect(result[0]).toBe(ZLIB_HEADER[0]);
+      expect(result[1]).toBe(ZLIB_HEADER[1]);
+    });
+
+    it('subsequent segments also start with the ZLIB header (manually prepended)', async () => {
+      const deflate = new StreamingDeflate();
+      await deflate.compressSegment(Buffer.from('first segment'));
+      const second = await deflate.compressSegment(Buffer.from('second segment'));
+      expect(second[0]).toBe(ZLIB_HEADER[0]);
+      expect(second[1]).toBe(ZLIB_HEADER[1]);
+    });
+
+    it('each segment ends with DEFLATE final block [0x03, 0x00] followed by 4-byte Adler-32', async () => {
+      const deflate = new StreamingDeflate();
+      const result = await deflate.compressSegment(Buffer.from('test payload'));
+      // Last 6 bytes: [0x03, 0x00] (final block) + Adler-32 (4 bytes)
+      expect(result[result.length - 6]).toBe(DEFLATE_FINAL_BLOCK[0]);
+      expect(result[result.length - 5]).toBe(DEFLATE_FINAL_BLOCK[1]);
+      // 4-byte Adler-32 follows
+      expect(result.length).toBeGreaterThanOrEqual(8);
+    });
+
+    it('single segment is a valid ZLIB stream that inflates back to the original data', async () => {
+      const deflate = new StreamingDeflate();
+      const original = Buffer.from('the quick brown fox jumps over the lazy dog');
+      const compressed = await deflate.compressSegment(original);
+      const inflated = zlib.inflateSync(compressed);
+      expect(inflated.equals(original)).toBe(true);
+    });
+  });
+
+  describe('Adler-32 checksum', () => {
+    it('is cumulative across segments — covers all data sent so far', async () => {
+      // Two separate instances with data in different orders should produce different checksums
+      const d1 = new StreamingDeflate();
+      const d2 = new StreamingDeflate();
+
+      await d1.compressSegment(Buffer.from('aaa'));
+      const r1 = await d1.compressSegment(Buffer.from('bbb'));
+
+      await d2.compressSegment(Buffer.from('bbb'));
+      const r2 = await d2.compressSegment(Buffer.from('aaa'));
+
+      // Adler-32 = last 4 bytes of each segment
+      const adler1 = r1.slice(-4);
+      const adler2 = r2.slice(-4);
+      // Different data order → different cumulative checksum
+      expect(adler1.equals(adler2)).toBe(false);
+    });
+
+    it('two instances compressing the same data in the same order produce equal Adler-32', async () => {
+      const d1 = new StreamingDeflate();
+      const d2 = new StreamingDeflate();
+      const data = Buffer.from('identical data');
+
+      const r1 = await d1.compressSegment(data);
+      const r2 = await d2.compressSegment(data);
+
+      expect(r1.slice(-4).equals(r2.slice(-4))).toBe(true);
+    });
+  });
+
+  describe('queue serialization', () => {
+    it('resolves concurrent calls in call order', async () => {
+      const deflate = new StreamingDeflate();
+      const order: number[] = [];
+
+      const p1 = deflate.compressSegment(Buffer.from('a')).then(() => order.push(1));
+      const p2 = deflate.compressSegment(Buffer.from('b')).then(() => order.push(2));
+      const p3 = deflate.compressSegment(Buffer.from('c')).then(() => order.push(3));
+
+      await Promise.all([p1, p2, p3]);
+      expect(order).toEqual([1, 2, 3]);
+    });
+
+    it('each call produces a valid result even under concurrency', async () => {
+      const deflate = new StreamingDeflate();
+      const data = Buffer.from('concurrent data');
+
+      const [r1, r2] = await Promise.all([deflate.compressSegment(data), deflate.compressSegment(data)]);
+
+      expect(r1).toBeInstanceOf(Buffer);
+      expect(r2).toBeInstanceOf(Buffer);
+      expect(r1.length).toBeGreaterThan(0);
+      expect(r2.length).toBeGreaterThan(0);
+    });
+  });
+});
diff --git a/src/tools/StreamingDeflate.ts b/src/tools/StreamingDeflate.ts
new file mode 100644
index 00000000..ff2fe061
--- /dev/null
+++ b/src/tools/StreamingDeflate.ts
@@ -0,0 +1,137 @@
+import zlib from 'node:zlib';
+import { monitor } from '../domain/telemetry';
+
+// ZLIB header for default compression (CMF=0x78, FLG=0x9C).
+// Prepended to continuation segments (index > 0) which don't have a header
+// emitted by zlib.createDeflate() since the stream is already started.
+const ZLIB_HEADER = Buffer.from([0x78, 0x9c]);
+
+// DEFLATE final empty block: BFINAL=1, BTYPE=01 (fixed Huffman), EOB symbol (256).
+// This matches the [3, 0] bytes produced by Pako's `vt()` function.
+const DEFLATE_FINAL_BLOCK = Buffer.from([0x03, 0x00]);
+
+const ADLER_MOD = 65521;
+
+/**
+ * Maintains a single continuous ZLIB-compressed stream across multiple segment
+ * flushes within a session, producing output byte-for-byte compatible with what
+ * the browser SDK's Pako-based encoder generates.
+ *
+ * Each segment's compressed output matches the Pako worker's `write` + `vt()`
+ * output format:
+ *
+ * - **Header**: ZLIB header bytes (0x78 0x9C). Only emitted by zlib.createDeflate()
+ * for segment 0; manually prepended for segments 1..N.
+ * - **Body**: Z_FULL_FLUSH compressed data — the LZ77 dictionary is reset after
+ * each flush so every segment is independently decompressable. The backend
+ * validates each segment with a standalone inflate before stitching.
+ * - **Trailer**: [0x03, 0x00] (DEFLATE final empty block, BFINAL=1) followed by
+ * a 4-byte big-endian Adler-32 checksum over ALL raw data compressed so far
+ * (cumulative, not per-segment). This is exactly what Pako's `vt()` returns.
+ *
+ * The Datadog backend's stitching algorithm strips the 2-byte ZLIB header and
+ * 6-byte trailer from every segment, concatenates the raw DEFLATE bodies, then
+ * wraps the result with segment 0's header and the last segment's trailer —
+ * producing a single valid ZLIB stream for the replay player.
+ */
+export class StreamingDeflate {
+  private stream = zlib.createDeflate();
+  private isFirstSegment = true;
+  // Running Adler-32 state (initial: A=1, B=0 per RFC 1950)
+  private adlerA = 1;
+  private adlerB = 0;
+  // Serialize concurrent calls so data events don't interleave.
+  private queue: Promise<void> = Promise.resolve();
+
+  compressSegment(data: Buffer): Promise<Buffer> {
+    const result = this.queue.then(() => this.doCompress(data));
+    this.queue = result.catch(() => undefined) as Promise<void>;
+    return result;
+  }
+
+  private doCompress(data: Buffer): Promise<Buffer> {
+    // Update running Adler-32 over the raw (uncompressed) data before compressing.
+    this.updateAdler32(data);
+
+    return new Promise((resolve, reject) => {
+      const chunks: Buffer[] = [];
+
+      const onData = monitor((chunk: Buffer) => chunks.push(chunk));
+      // eslint-disable-next-line prefer-const
+      let onError: (err: Error) => void;
+
+      // Cleans up all attached event listeners to prevent memory leaks.
+      const cleanup = () => {
+        this.stream.off('data', onData);
+        if (onError) this.stream.off('error', onError);
+      };
+
+      onError = monitor((err: Error) => {
+        cleanup();
+        reject(err);
+      });
+
+      this.stream.on('data', onData);
+      this.stream.once('error', onError);
+
+      this.stream.write(data);
+      this.stream.flush(
+        zlib.constants.Z_FULL_FLUSH,
+        monitor(() => {
+          cleanup();
+
+          const parts: Buffer[] = [];
+
+          if (!this.isFirstSegment) {
+            // Segment 0 already has the ZLIB header from zlib.createDeflate().
+            // Segments 1..N are continuations with no header — prepend one so
+            // the backend's header-stripping doesn't corrupt actual deflate data.
+            parts.push(ZLIB_HEADER);
+          }
+          this.isFirstSegment = false;
+
+          parts.push(Buffer.concat(chunks));
+
+          // Append the Pako-compatible trailer: DEFLATE final empty block +
+          // 4-byte big-endian Adler-32 of all raw data compressed so far.
+          parts.push(DEFLATE_FINAL_BLOCK);
+          parts.push(this.adler32ToBuffer());
+
+          resolve(Buffer.concat(parts));
+        })
+      );
+    });
+  }
+
+  private updateAdler32(data: Buffer): void {
+    let a = this.adlerA;
+    let b = this.adlerB;
+    const len = data.length;
+    let i = 0;
+
+    // Process in block sizes up to 5000 bytes to safely accumulate sums
+    // without modulo operations on every byte, preventing integer overflow.
+    while (i < len) {
+      const blockEnd = Math.min(i + 5000, len);
+      for (; i < blockEnd; i++) {
+        a += data[i];
+        b += a;
+      }
+      a %= ADLER_MOD;
+      b %= ADLER_MOD;
+    }
+
+    this.adlerA = a;
+    this.adlerB = b;
+  }
+
+  private adler32ToBuffer(): Buffer {
+    // Combine 'b' and 'a' into a single 32-bit signed integer.
+    const value = (this.adlerB << 16) | this.adlerA | 0;
+
+    // Fast-allocate 4 unmanaged bytes and write the value in Big-Endian order.
+    const buf = Buffer.allocUnsafe(4);
+    buf.writeUInt32BE(value >>> 0, 0);
+    return buf;
+  }
+}
diff --git a/src/transport/Transport.spec.ts b/src/transport/Transport.spec.ts
index 37ba55a7..ed07ce5a 100644
--- a/src/transport/Transport.spec.ts
+++ b/src/transport/Transport.spec.ts
@@ -66,7 +66,15 @@ describe('Transport', () => {
 
       const tracks = mockBatchCreate.mock.calls.map(([, options]) => (options as { trackType: EventTrack }).trackType);
       expect(tracks).not.toContain(EventTrack.PROFILE);
-      expect(tracks).toEqual([EventTrack.RUM, EventTrack.SPANS]);
+      expect(tracks).toEqual([EventTrack.RUM, EventTrack.SPANS, EventTrack.REPLAY]);
+    });
+
+    it('should skip the REPLAY track when replay is disabled', async () => {
+      const configWithoutReplay = createTestConfiguration({ sessionReplaySampleRate: 0 });
+      await Transport.create(configWithoutReplay, eventManager);
+
+      const tracks = mockBatchCreate.mock.calls.map(([, options]) => (options as { trackType: EventTrack }).trackType);
+      expect(tracks).not.toContain(EventTrack.REPLAY);
     });
   });
 
@@ -136,7 +144,8 @@ describe('Transport', () => {
       const transport = await Transport.create(config, eventManager);
       await transport.flush();
 
-      expect(mockBatchFlush).toHaveBeenCalledTimes(3);
+      // RUM + SPANS + PROFILE + REPLAY
+      expect(mockBatchFlush).toHaveBeenCalledTimes(4);
     });
   });
 
diff --git a/src/transport/Transport.ts b/src/transport/Transport.ts
index 19275ed5..a948b91f 100644
--- a/src/transport/Transport.ts
+++ b/src/transport/Transport.ts
@@ -39,6 +39,10 @@ export class Transport {
     if (this.config.profilingSampleRate > 0) {
       tracks.push(EventTrack.PROFILE);
     }
+
+    if (this.config.sessionReplaySampleRate > 0) {
+      tracks.push(EventTrack.REPLAY);
+    }
     return tracks;
   }
 
diff --git a/src/transport/batch/BatchManager.spec.ts b/src/transport/batch/BatchManager.spec.ts
index 5f8d11cf..3db4ae01 100644
--- a/src/transport/batch/BatchManager.spec.ts
+++ b/src/transport/batch/BatchManager.spec.ts
@@ -155,6 +155,29 @@ describe('BatchManager', () => {
       expect(mockProducerFlush).toHaveBeenCalledTimes(1);
       expect(mockConsumerUpload).toHaveBeenCalledTimes(1);
     });
+
+    it('runs a fresh cycle after an in-flight scheduled cycle instead of dropping the flush', async () => {
+      let resolveScheduled!: () => void;
+      mockProducerFlush.mockReturnValueOnce(new Promise<void>((resolve) => (resolveScheduled = resolve)));
+
+      const manager = await BatchManager.create(config, batchConfig);
+
+      // Fire the periodic cycle; its producer.flush() stays pending, simulating an in-flight upload.
+      await vi.advanceTimersByTimeAsync(batchConfig.uploadFrequency);
+      expect(mockProducerFlush).toHaveBeenCalledTimes(1);
+      expect(mockConsumerUpload).not.toHaveBeenCalled();
+
+      // A flush() arriving now (e.g. on quit, after a final segment was rotated) must not be dropped:
+      // it queues a full cycle behind the active one so the newly rotated files are uploaded.
+      const flushPromise = manager.flush();
+      resolveScheduled();
+      await flushPromise;
+
+      expect(mockProducerFlush).toHaveBeenCalledTimes(2);
+      expect(mockConsumerUpload).toHaveBeenCalledTimes(2);
+
+      manager.stop();
+    });
   });
 
   describe('stop', () => {
diff --git a/src/transport/batch/BatchManager.ts b/src/transport/batch/BatchManager.ts
index 92044978..d3a151b5 100644
--- a/src/transport/batch/BatchManager.ts
+++ b/src/transport/batch/BatchManager.ts
@@ -9,6 +9,8 @@ import { BatchConsumer } from './BatchConsumer';
 import type { BatchConsumerConfig } from './BatchConsumer';
 import { BatchProducer } from './BatchProducer';
 import { ProfileBatchConsumer, ProfileBatchProducer } from './profiling';
+import { ReplayBatchConsumer } from './replay/ReplayBatchConsumer';
+import { ReplayBatchProducer } from './replay/ReplayBatchProducer';
 import { StandardBatchConsumer } from './standard/StandardBatchConsumer';
 import { StandardBatchProducer } from './standard/StandardBatchProducer';
 import type { StandardBatchProducerConfig } from './standard/StandardBatchProducer';
@@ -24,7 +26,10 @@ export class BatchManager {
   private consumer: BatchConsumer;
   private uploadFrequency: number;
   private timeoutId: ReturnType<typeof setTimeout> | null = null;
-  private isUploading = false;
+  // The upload cycle currently running (rotate + upload), or null when idle.
+  private activeCycle: Promise<void> | null = null;
+  // A cycle queued to run after the active one. Concurrent flush() callers coalesce onto it.
+  private queuedCycle: Promise<void> | null = null;
 
   private constructor(producer: BatchProducer, consumer: BatchConsumer, uploadFrequency: number) {
     this.producer = producer;
@@ -48,9 +53,16 @@ export class BatchManager {
     this.producer.post(event);
   }
 
-  /** Drains the write queue, rotates the current batch, and uploads all pending files. */
+  /**
+   * Drains the write queue, rotates the current batch, and uploads all pending files.
+   *
+   * Guarantees a full cycle runs to completion *after* this call. A scheduled cycle already in
+   * flight may have scanned the directory before the caller rotated new files (e.g. the final
+   * replay segment flushed on quit), so we always run a fresh cycle behind it rather than
+   * short-circuiting — otherwise those files would sit on disk until the next launch.
+   */
   async flush() {
-    await this.triggerUploadCycle();
+    await this.enqueueUploadCycle();
   }
 
   /** Stops the periodic upload cycle. */
@@ -69,27 +81,56 @@ export class BatchManager {
   /** Schedules the next upload cycle after the configured frequency delay. */
   private scheduleNext() {
     this.timeoutId = setTimeout(() => {
-      void this.triggerUploadCycle()
+      void this.runPeriodicCycle()
         .catch((error) => addError(error))
         .then(() => this.scheduleNext());
     }, this.uploadFrequency);
   }
 
-  /** Flushes the producer to rotate pending files, then uploads all ready batches. */
-  private async triggerUploadCycle() {
-    if (this.isUploading) {
-      return;
+  /**
+   * Periodic tick: run a cycle only when nothing is active or queued, so ticks never stack up
+   * behind a slow upload (a fresh tick will fire next interval anyway).
+   */
+  private runPeriodicCycle(): Promise<void> {
+    if (this.activeCycle || this.queuedCycle) {
+      return this.activeCycle ?? Promise.resolve();
+    }
+    return this.enqueueUploadCycle();
+  }
+
+  /**
+   * Queues an upload cycle to run after any in-flight one. Multiple callers before the queued
+   * cycle starts share the same promise, so at most one cycle is ever pending and cycles never
+   * overlap (rotate + upload touch the same directory).
+   */
+  private enqueueUploadCycle(): Promise<void> {
+    if (this.queuedCycle) {
+      return this.queuedCycle;
     }
 
-    this.isUploading = true;
+    const previous = this.activeCycle ?? Promise.resolve();
+    const cycle = previous
+      // Swallow the prior cycle's failure — its own scheduler already reported it, and this
+      // cycle must still run so newly rotated files get uploaded.
+      .catch(() => undefined)
+      .then(() => {
+        this.queuedCycle = null;
+        this.activeCycle = this.runUploadCycle();
+        return this.activeCycle;
+      });
+    this.queuedCycle = cycle;
+    return cycle;
+  }
 
+  /** Flushes the producer to rotate pending files, then uploads all ready batches. */
+  private async runUploadCycle() {
     try {
       // Flush producer first to rotate any pending .tmp files to .log
       await this.producer.flush();
       // Then upload all .log files
       await this.consumer.upload();
     } finally {
-      this.isUploading = false;
+      this.activeCycle = null;
     }
   }
 
@@ -112,6 +153,12 @@ export class BatchManager {
 
     const consumerConfig: BatchConsumerConfig = { trackPath, intakeUrl, clientToken };
 
+    if (trackType === EventTrack.REPLAY) {
+      const producer = await ReplayBatchProducer.create({ trackPath });
+      const consumer = new ReplayBatchConsumer(consumerConfig);
+      return { producer, consumer };
+    }
+
     if (trackType === EventTrack.PROFILE) {
       const producer = await ProfileBatchProducer.create({ trackPath });
       const consumer = new ProfileBatchConsumer(consumerConfig);
diff --git a/src/transport/batch/replay/ReplayBatchConsumer.spec.ts b/src/transport/batch/replay/ReplayBatchConsumer.spec.ts
new file mode 100644
index 00000000..95976c81
--- /dev/null
+++ b/src/transport/batch/replay/ReplayBatchConsumer.spec.ts
@@ -0,0 +1,170 @@
+import { describe, it, expect, vi, beforeEach } from 'vitest';
+import { ReplayBatchConsumer } from './ReplayBatchConsumer';
+import { getUserAgent } from '../../userAgent';
+import { mockFs } from '../../../mocks.specUtil';
+
+vi.mock('node:fs/promises');
+vi.mock('../../userAgent');
+vi.mock('@datadog/browser-core', () => ({
+  generateUUID: vi.fn(() => 'test-request-id'),
+}));
+vi.stubGlobal('__SDK_VERSION__', '0.0.0-test');
+
+const fsMocks = mockFs();
+const TEST_USER_AGENT = 'TestApp/1.0.0 Electron/0';
+
+// Mirrors computeIntakeUrlForTrack output: the standard track query is already present, so the
+// consumer must merge its params in rather than appending a second `?`.
+const config = {
+  trackPath: '/mock/replay',
+  intakeUrl: 'https://browser-intake-datadoghq.com/api/v2/replay?ddsource=electron',
+  clientToken: 'test-client-token',
+};
+
+function makeFileLine(metadata: Record<string, unknown>, compressed: Buffer): string {
+  return `${JSON.stringify(metadata)}\n${compressed.toString('base64')}\n`;
+}
+
+describe('ReplayBatchConsumer — request construction', () => {
+  let consumer: ReplayBatchConsumer;
+
+  beforeEach(() => {
+    fsMocks.reset();
+    vi.mocked(getUserAgent).mockReset().mockReturnValue(TEST_USER_AGENT);
+    consumer = new ReplayBatchConsumer(config);
+    global.fetch = vi.fn().mockResolvedValue({ ok: true });
+    fsMocks.access.mockResolvedValue(undefined);
+    fsMocks.unlink.mockResolvedValue(undefined);
+  });
+
+  describe('request URL', () => {
+    it('includes the correct query parameters', async () => {
+      fsMocks.readdir.mockResolvedValue(['segment.log']);
+      fsMocks.readFile.mockResolvedValue(
+        makeFileLine(
+          { session: { id: 'sess' }, start: 0, raw_segment_size: 1, compressed_segment_size: 1 },
+          Buffer.from([0x01])
+        )
+      );
+
+      await consumer.upload();
+
+      const [request] = vi.mocked(fetch).mock.calls[0] as [Request];
+      const url = new URL(request.url);
+      // ddsource=electron from the track query is overwritten with the browser value, not duplicated.
+      expect(url.searchParams.getAll('ddsource')).toEqual(['browser']);
+      expect(url.searchParams.get('dd-api-key')).toBe(config.clientToken);
+      expect(url.searchParams.get('dd-evp-origin')).toBe('browser');
+      expect(url.searchParams.get('dd-request-id')).toBe('test-request-id');
+      expect(url.searchParams.get('ddtags')).toContain('sdk_version:0.0.0-test');
+    });
+
+    it('does not produce a double question mark when the intake URL already has a query', async () => {
+      fsMocks.readdir.mockResolvedValue(['segment.log']);
+      fsMocks.readFile.mockResolvedValue(
+        makeFileLine(
+          { session: { id: 'sess' }, start: 0, raw_segment_size: 1, compressed_segment_size: 1 },
+          Buffer.from([0x01])
+        )
+      );
+
+      await consumer.upload();
+
+      const [request] = vi.mocked(fetch).mock.calls[0] as [Request];
+      expect(request.url.match(/\?/g)).toHaveLength(1);
+    });
+
+    it('merges params inside ddforward when the intake URL is a proxy', async () => {
+      const proxyConsumer = new ReplayBatchConsumer({
+        ...config,
+        intakeUrl: 'https://proxy.example.com/?ddforward=%2Fapi%2Fv2%2Freplay%3Fddsource%3Delectron',
+      });
+      fsMocks.readdir.mockResolvedValue(['segment.log']);
+      fsMocks.readFile.mockResolvedValue(
+        makeFileLine(
+          { session: { id: 'sess' }, start: 0, raw_segment_size: 1, compressed_segment_size: 1 },
+          Buffer.from([0x01])
+        )
+      );
+
+      await proxyConsumer.upload();
+
+      const [request] = vi.mocked(fetch).mock.calls[0] as [Request];
+      const url = new URL(request.url);
+      // Auth/metadata params must live on the forwarded path, not the proxy URL itself.
+      expect(url.searchParams.get('dd-api-key')).toBeNull();
+      const forwarded = new URL(url.searchParams.get('ddforward')!, 'https://placeholder.invalid');
+      expect(forwarded.pathname).toBe('/api/v2/replay');
+      expect(forwarded.searchParams.getAll('ddsource')).toEqual(['browser']);
+      expect(forwarded.searchParams.get('dd-api-key')).toBe(config.clientToken);
+    });
+  });
+
+  describe('malformed batch files', () => {
+    it('drops (deletes without sending) a file whose metadata line is not valid JSON', async () => {
+      fsMocks.readdir.mockResolvedValue(['corrupt.log']);
+      // Truncated metadata line followed by a base64 line — the two-line shape is intact.
+      fsMocks.readFile.mockResolvedValue(`{"session":{"id":"ses\n${Buffer.from([0x01]).toString('base64')}\n`);
+
+      await consumer.upload();
+
+      expect(fetch).not.toHaveBeenCalled();
+      expect(fsMocks.unlink).toHaveBeenCalledWith('/mock/replay/corrupt.log');
+    });
+
+    it('drops a file missing session.id or start', async () => {
+      fsMocks.readdir.mockResolvedValue(['corrupt.log']);
+      fsMocks.readFile.mockResolvedValue(makeFileLine({ start: 0 }, Buffer.from([0x01])));
+
+      await consumer.upload();
+
+      expect(fetch).not.toHaveBeenCalled();
+      expect(fsMocks.unlink).toHaveBeenCalledWith('/mock/replay/corrupt.log');
+    });
+
+    it('drops a file whose segment body is truncated (shorter than compressed_segment_size)', async () => {
+      fsMocks.readdir.mockResolvedValue(['corrupt.log']);
+      // Valid metadata claiming 50 compressed bytes, but only 2 bytes of body survived the crash.
+      fsMocks.readFile.mockResolvedValue(
+        makeFileLine(
+          { session: { id: 'sess' }, start: 0, raw_segment_size: 100, compressed_segment_size: 50 },
+          Buffer.from([0x78, 0x9c])
+        )
+      );
+
+      await consumer.upload();
+
+      expect(fetch).not.toHaveBeenCalled();
+      expect(fsMocks.unlink).toHaveBeenCalledWith('/mock/replay/corrupt.log');
+    });
+  });
+
+  describe('request headers and body', () => {
+    it('sends the User-Agent header', async () => {
+      fsMocks.readdir.mockResolvedValue(['segment.log']);
+      fsMocks.readFile.mockResolvedValue(
+        makeFileLine(
+          { session: { id: 'sess' }, start: 0, raw_segment_size: 1, compressed_segment_size: 1 },
+          Buffer.from([0x01])
+        )
+      );
+
+      await consumer.upload();
+
+      const [request] = vi.mocked(fetch).mock.calls[0] as [Request];
+      expect(request.headers.get('User-Agent')).toBe(TEST_USER_AGENT);
+    });
+
+    it('sends a multipart/form-data body', async () => {
+      const metadata = { session: { id: 'sess-1' }, start: 1000, raw_segment_size: 100, compressed_segment_size: 2 };
+      fsMocks.readdir.mockResolvedValue(['segment.log']);
+      fsMocks.readFile.mockResolvedValue(makeFileLine(metadata, Buffer.from([0x78, 0x9c])));
+
+      await consumer.upload();
+
+      const [request] = vi.mocked(fetch).mock.calls[0] as [Request];
+      // FormData serialises to a ReadableStream in Request; verify via Content-Type
+      expect(request.headers.get('content-type')).toMatch(/^multipart\/form-data/);
+    });
+  });
+});
diff --git a/src/transport/batch/replay/ReplayBatchConsumer.ts b/src/transport/batch/replay/ReplayBatchConsumer.ts
new file mode 100644
index 00000000..4f3f6268
--- /dev/null
+++ b/src/transport/batch/replay/ReplayBatchConsumer.ts
@@ -0,0 +1,83 @@
+import { generateUUID } from '@datadog/browser-core';
+import { display } from '../../../tools/display';
+import { BatchConsumer } from '../BatchConsumer';
+import { appendIntakeParams } from '../../utils';
+
+declare const __SDK_VERSION__: string;
+
+/**
+ * Concrete {@link BatchConsumer} for session replay segments.
+ *
+ * Reads the two-line format written by {@link ReplayBatchProducer} and builds a
+ * multipart/form-data POST to the Datadog session replay intake:
+ *   - `segment`: deflate-compressed binary blob
+ *   - `event`: JSON metadata including raw/compressed size fields
+ *
+ * File I/O, sending, and deletion on success are handled by {@link BatchConsumer}.
+ */
+export class ReplayBatchConsumer extends BatchConsumer {
+  protected buildRequest(lines: string[]): Request | null {
+    if (lines.length < 2) {
+      return null;
+    }
+
+    // A crash mid-write can leave a recovered `.log` with a truncated metadata line. An unguarded
+    // JSON.parse would throw here — before the base class reaches its fetch error handling or deletes
+    // the file — aborting the whole upload cycle and blocking every later replay batch on retry.
+    // Validate and drop instead, matching ProfileBatchConsumer.
+    let metadataWithSizes: Record<string, unknown>;
+    try {
+      metadataWithSizes = JSON.parse(lines[0]) as Record<string, unknown>;
+    } catch {
+      display.warn('Dropping malformed replay batch: metadata line is not valid JSON');
+      return null;
+    }
+
+    const session = metadataWithSizes.session as { id?: unknown } | undefined;
+    const sessionId = session?.id;
+    const start = metadataWithSizes.start;
+    if (typeof sessionId !== 'string' || sessionId.length === 0 || typeof start !== 'number') {
+      display.warn('Dropping malformed replay batch: missing session.id or start');
+      return null;
+    }
+
+    const compressed = Buffer.from(lines[1], 'base64');
+
+    // A crash can also leave a complete metadata line but a truncated base64 body. Uploading the
+    // partial bytes yields an invalid zlib blob the intake rejects, so the corrupt file would be
+    // retried every cycle. Validate the decoded length against the size the producer recorded and
+    // drop the file when it doesn't match, matching the malformed-metadata handling above.
+    const expectedSize = metadataWithSizes.compressed_segment_size;
+    if (typeof expectedSize !== 'number' || compressed.length !== expectedSize) {
+      display.warn('Dropping malformed replay batch: segment body is truncated or incomplete');
+      return null;
+    }
+
+    const formData = new FormData();
+    formData.append('segment', new Blob([compressed], { type: 'application/octet-stream' }), `${sessionId}-${start}`);
+    formData.append('event', new Blob([JSON.stringify(metadataWithSizes)], { type: 'application/json' }));
+
+    // The replay intake uses browser SDK conventions: auth and metadata as URL
+    // query params, not headers. ddsource and dd-evp-origin are 'browser' because
+    // the records originate from @datadog/browser-rum in the renderer — the backend
+    // uses these values to determine how to parse and stitch the compressed segments.
+    //
+    // intakeUrl already carries the standard track query (`?ddsource=electron`, or a proxy
+    // `?ddforward=...`), so merge these params in — overwriting ddsource — rather than appending
+    // a second `?`, which would otherwise corrupt ddsource and proxy forwarding.
+    const url = appendIntakeParams(this.intakeUrl, {
+      ddsource: 'browser',
+      ddtags: `sdk_version:${__SDK_VERSION__}`,
+      'dd-api-key': this.clientToken,
+      'dd-evp-origin': 'browser',
+      'dd-evp-origin-version': __SDK_VERSION__,
+      'dd-request-id': generateUUID(),
+    });
+
+    return new Request(url, {
+      method: 'POST',
+      headers: { 'User-Agent': this.userAgent! },
+      body: formData,
+    });
+  }
+}
diff --git a/src/transport/batch/replay/ReplayBatchProducer.spec.ts b/src/transport/batch/replay/ReplayBatchProducer.spec.ts
new file mode 100644
index 00000000..9905b9de
--- /dev/null
+++ b/src/transport/batch/replay/ReplayBatchProducer.spec.ts
@@ -0,0 +1,133 @@
+import { describe, it, expect, vi, beforeEach } from 'vitest';
+import path from 'node:path';
+import type { BatchProducerConfig } from '../BatchProducer';
+import { mockFs } from '../../../mocks.specUtil';
+import { ReplayBatchProducer } from './ReplayBatchProducer';
+import { CreationReason } from '../../../domain/replay';
+import type { ReplaySegmentPayload, SegmentMetadata } from '../../../domain/replay';
+import { EventKind, EventTrack, type ServerReplayEvent } from '../../../event';
+
+vi.mock('node:fs/promises');
+// The batch filename generator uses dateNow from @datadog/js-core/time (not browser-core). Mock that
+// module for a deterministic timestamp, preserving its other exports; replacing @datadog/browser-core
+// here would break transitive imports that pull real browser-core exports (e.g. performDraw) in.
+vi.mock('@datadog/js-core/time', async (importOriginal) => ({
+  ...(await importOriginal<typeof import('@datadog/js-core/time')>()),
+  dateNow: vi.fn(() => 1234567890),
+}));
+
+const fsMocks = mockFs();
+
+const config: BatchProducerConfig = {
+  trackPath: '/mock/replay',
+};
+
+function makePayload(overrides: Partial<ReplaySegmentPayload> = {}): ReplaySegmentPayload {
+  const metadata: SegmentMetadata = {
+    application: { id: 'app-1' },
+    session: { id: 'sess-1' },
+    view: { id: 'view-1' },
+    start: 1000,
+    end: 2000,
+    records_count: 3,
+    has_full_snapshot: true,
+    index_in_view: 0,
+    source: 'browser',
+    creation_reason: CreationReason.INIT,
+  };
+  return {
+    metadata,
+    rawBytesCount: 256,
+    compressed: Buffer.from([0x78, 0x9c, 0x01, 0x02, 0x03]),
+    ...overrides,
+  };
+}
+
+function makeEvent(data = makePayload()): ServerReplayEvent {
+  return {
+    kind: EventKind.SERVER,
+    track: EventTrack.REPLAY,
+    data,
+  };
+}
+
+describe('ReplayBatchProducer', () => {
+  beforeEach(() => {
+    fsMocks.reset();
+    fsMocks.access.mockResolvedValue(undefined);
+    fsMocks.mkdir.mockResolvedValue(undefined);
+    fsMocks.readdir.mockResolvedValue([]);
+    fsMocks.writeFile.mockResolvedValue(undefined);
+    fsMocks.rename.mockResolvedValue(undefined);
+  });
+
+  describe('create()', () => {
+    it('creates the track directory when it does not exist', async () => {
+      fsMocks.access.mockRejectedValueOnce(new Error('ENOENT'));
+      await ReplayBatchProducer.create(config);
+      expect(fsMocks.mkdir).toHaveBeenCalledWith(config.trackPath, { recursive: true });
+    });
+
+    it('does not create directory when it already exists', async () => {
+      await ReplayBatchProducer.create(config);
+      expect(fsMocks.mkdir).not.toHaveBeenCalled();
+    });
+
+    it('rotates orphaned .tmp files from previous sessions', async () => {
+      fsMocks.readdir.mockResolvedValueOnce(['batch-111.tmp', 'batch-222.tmp']);
+      await ReplayBatchProducer.create(config);
+      expect(fsMocks.rename).toHaveBeenCalledTimes(2);
+    });
+  });
+
+  describe('writeData() — file format', () => {
+    it('writes metadata JSON as line 1 and base64-encoded compressed data as line 2', async () => {
+      const producer = await ReplayBatchProducer.create(config);
+      const payload = makePayload();
+
+      producer.post(makeEvent(payload));
+      await producer.flush();
+
+      expect(fsMocks.writeFile).toHaveBeenCalledOnce();
+      const [, content] = fsMocks.writeFile.mock.calls[0] as [string, string, string];
+      const lines = content.split('\n').filter(Boolean);
+
+      expect(lines).toHaveLength(2);
+
+      // Line 1: JSON metadata with size fields injected
+      const meta = JSON.parse(lines[0]) as Record<string, unknown>;
+      expect((meta['session'] as { id: string }).id).toBe('sess-1');
+      expect((meta['view'] as { id: string }).id).toBe('view-1');
+      expect(meta['raw_segment_size']).toBe(payload.rawBytesCount);
+      expect(meta['compressed_segment_size']).toBe(payload.compressed.byteLength);
+      expect(meta['records_count']).toBe(3);
+
+      // Line 2: base64-encoded compressed data
+      expect(Buffer.from(lines[1], 'base64').equals(payload.compressed)).toBe(true);
+    });
+
+    it('writes to a .tmp file then atomically renames it to .log', async () => {
+      const producer = await ReplayBatchProducer.create(config);
+      producer.post(makeEvent());
+      await producer.flush();
+
+      const [tmpPath] = fsMocks.writeFile.mock.calls[0] as [string, string, string];
+      const [fromPath, toPath] = fsMocks.rename.mock.calls[fsMocks.rename.mock.calls.length - 1] as [string, string];
+
+      expect(tmpPath).toMatch(/\.tmp$/);
+      expect(fromPath).toBe(tmpPath);
+      expect(toPath).toMatch(/\.log$/);
+      expect(path.dirname(toPath)).toBe(config.trackPath);
+    });
+
+    it('each post creates a separate file', async () => {
+      const producer = await ReplayBatchProducer.create(config);
+      producer.post(makeEvent());
+      producer.post(makeEvent());
+      await producer.flush();
+
+      expect(fsMocks.writeFile).toHaveBeenCalledTimes(2);
+      expect(fsMocks.rename).toHaveBeenCalledTimes(2);
+    });
+  });
+});
diff --git a/src/transport/batch/replay/ReplayBatchProducer.ts b/src/transport/batch/replay/ReplayBatchProducer.ts
new file mode 100644
index 00000000..f75deba6
--- /dev/null
+++ b/src/transport/batch/replay/ReplayBatchProducer.ts
@@ -0,0 +1,47 @@
+import fs from 'node:fs/promises';
+import path from 'node:path';
+import { BatchProducer } from '../BatchProducer';
+import type { BatchProducerConfig } from '../BatchProducer';
+import type { ServerReplayEvent } from '../../../event';
+
+/**
+ * Concrete {@link BatchProducer} for session replay segments.
+ *
+ * Unlike the standard producer, each call to {@link writeData} writes one
+ * complete, atomic file per segment (no append/rotation by size). Each file
+ * contains two lines:
+ *   - Line 1: JSON metadata + size fields (consumed by the multipart 'event' part)
+ *   - Line 2: base64-encoded compressed segment binary (the multipart 'blob' part)
+ *
+ * The `.tmp` → `.log` rename ensures atomicity: the consumer never reads a
+ * partially-written segment.
+ */
+export class ReplayBatchProducer extends BatchProducer {
+  protected override fileNamePrefix = 'replay';
+
+  /** Creates and fully initializes a ReplayBatchProducer. */
+  static async create(config: BatchProducerConfig): Promise<ReplayBatchProducer> {
+    const producer = new ReplayBatchProducer(config);
+    await producer.initialize();
+    return producer;
+  }
+
+  protected async writeData(event: ServerReplayEvent): Promise<void> {
+    const { metadata, rawBytesCount, compressed } = event.data;
+
+    await this.ensureTrackDirectoryExists();
+
+    const fileName = this.generateBatchFileName();
+    const tmpPath = path.join(this.trackPath, fileName);
+
+    const metadataWithSizes = {
+      ...metadata,
+      raw_segment_size: rawBytesCount,
+      compressed_segment_size: compressed.byteLength,
+    };
+
+    const content = `${JSON.stringify(metadataWithSizes)}\n${compressed.toString('base64')}\n`;
+    await fs.writeFile(tmpPath, content, 'utf8');
+    await this.renameBatchFile(fileName);
+  }
+}
diff --git a/src/transport/utils.spec.ts b/src/transport/utils.spec.ts
index 8eed3a2f..18f214d0 100644
--- a/src/transport/utils.spec.ts
+++ b/src/transport/utils.spec.ts
@@ -1,5 +1,5 @@
 import { describe, it, expect } from 'vitest';
-import { computeIntakeHostname, computeIntakeUrlForTrack } from './utils';
+import { appendIntakeParams, computeIntakeHostname, computeIntakeUrlForTrack } from './utils';
 
 describe('computeIntakeHostname', () => {
   it('returns the intake hostname for a simple site', () => {
@@ -57,3 +57,32 @@ describe('computeIntakeUrlForTrack', () => {
     );
   });
 });
+
+describe('appendIntakeParams', () => {
+  it('merges params into a direct intake URL, overwriting the default ddsource', () => {
+    const result = appendIntakeParams('https://browser-intake-datadoghq.com/api/v2/replay?ddsource=electron', {
+      ddsource: 'browser',
+      'dd-api-key': 'token',
+    });
+
+    const url = new URL(result);
+    expect(url.searchParams.getAll('ddsource')).toEqual(['browser']);
+    expect(url.searchParams.get('dd-api-key')).toBe('token');
+    // No second '?' introduced.
+    expect(result.match(/\?/g)).toHaveLength(1);
+  });
+
+  it('folds params into the forwarded path for a proxy URL, not the proxy query itself', () => {
+    const proxyUrl = computeIntakeUrlForTrack('datadoghq.com', 'replay', { proxy: 'http://proxy:8080' });
+
+    const result = appendIntakeParams(proxyUrl, { ddsource: 'browser', 'dd-api-key': 'token' });
+
+    const url = new URL(result);
+    // Auth params must not leak onto the proxy URL.
+    expect(url.searchParams.get('dd-api-key')).toBeNull();
+    const forwarded = new URL(url.searchParams.get('ddforward')!, 'https://placeholder.invalid');
+    expect(forwarded.pathname).toBe('/api/v2/replay');
+    expect(forwarded.searchParams.getAll('ddsource')).toEqual(['browser']);
+    expect(forwarded.searchParams.get('dd-api-key')).toBe('token');
+  });
+});
diff --git a/src/transport/utils.ts b/src/transport/utils.ts
index d073badc..222d060c 100644
--- a/src/transport/utils.ts
+++ b/src/transport/utils.ts
@@ -36,3 +36,36 @@ export function computeIntakeUrlForTrack(
   const subdomainPrefix = subdomain ? `${subdomain}.` : '';
   return `https://${subdomainPrefix}browser-intake-${computeIntakeSite(site)}${path}`;
 }
+
+/**
+ * Merges query parameters into an intake URL produced by {@link computeIntakeUrlForTrack},
+ * overwriting any parameter of the same name (e.g. replacing the default `ddsource=electron`).
+ *
+ * Handles both URL shapes:
+ *  - direct: params are set on the intake URL's own query string
+ *  - proxy (`<proxy>?ddforward=<encoded path>`): params belong on the *forwarded* intake path,
+ *    not the proxy URL, so they are merged inside `ddforward` and re-encoded.
+ *
+ * Needed because the replay intake follows Browser SDK conventions (`ddsource=browser`, auth and
+ * metadata as query params). Naively appending `?${params}` would yield a second `?`, corrupting
+ * both the direct URL and the proxy forwarding.
+ */
+export function appendIntakeParams(intakeUrl: string, params: Record<string, string>): string {
+  const url = new URL(intakeUrl);
+  const ddforward = url.searchParams.get('ddforward');
+
+  if (ddforward !== null) {
+    // Proxy form: fold the params into the forwarded intake path.
+    const forwarded = new URL(ddforward, 'https://placeholder.invalid');
+    for (const [key, value] of Object.entries(params)) {
+      forwarded.searchParams.set(key, value);
+    }
+    url.searchParams.set('ddforward', `${forwarded.pathname}${forwarded.search}`);
+    return url.toString();
+  }
+
+  for (const [key, value] of Object.entries(params)) {
+    url.searchParams.set(key, value);
+  }
+  return url.toString();
+}