diff --git a/sdk/highlight-run/src/__tests__/client-worker-stop.test.ts b/sdk/highlight-run/src/__tests__/client-worker-stop.test.ts
new file mode 100644
index 0000000000..dc22e9054f
--- /dev/null
+++ b/sdk/highlight-run/src/__tests__/client-worker-stop.test.ts
@@ -0,0 +1,50 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import { Highlight } from '../client'
+import { MessageType, StopReason } from '../client/workers/types'
+
+const shutdown = vi.fn()
+vi.mock('../client/otel', async (importOriginal) => ({
+	...(await importOriginal<typeof import('../client/otel')>()),
+	shutdown: () => shutdown(),
+}))
+
+describe('Highlight worker stop handling', () => {
+	let highlight: Highlight
+
+	beforeEach(() => {
+		vi.useFakeTimers()
+		shutdown.mockClear()
+		highlight = new Highlight({
+			organizationID: '1',
+			sessionSecureID: 'seed',
+			backendUrl: 'https://pub.observability.app.launchdarkly.com',
+		})
+	})
+
+	afterEach(() => {
+		vi.useRealTimers()
+	})
+
+	it('keeps telemetry running when the worker stops replay', () => {
+		highlight._worker.onmessage({
+			data: {
+				response: {
+					type: MessageType.Stop,
+					reason: StopReason.UnrecoverableError,
+				},
+			},
+		} as MessageEvent<any>)
+
+		expect(shutdown).not.toHaveBeenCalled()
+		expect(highlight.state).toBe('NotRecording')
+		// Keeps the page visibility listener, which outlives a stop, from restarting us.
+		expect(highlight.manualStopped).toBe(true)
+	})
+
+	it('shuts telemetry down when the SDK itself stops', () => {
+		highlight.stopRecording(true)
+
+		expect(shutdown).toHaveBeenCalledTimes(1)
+		expect(highlight.state).toBe('NotRecording')
+	})
+})
diff --git a/sdk/highlight-run/src/__tests__/record-worker-stop.test.ts b/sdk/highlight-run/src/__tests__/record-worker-stop.test.ts
new file mode 100644
index 0000000000..a76f903ffa
--- /dev/null
+++ b/sdk/highlight-run/src/__tests__/record-worker-stop.test.ts
@@ -0,0 +1,108 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import { RecordSDK } from '../sdk/record'
+import { MessageType, StopReason } from '../client/workers/types'
+
+const recordStop = vi.fn()
+const recordSpy = vi.fn(() => recordStop)
+vi.mock('@highlight-run/rrweb', () => ({
+	addCustomEvent: vi.fn(),
+	record: () => recordSpy(),
+}))
+
+vi.mock('../client/graph/generated/operations', () => ({
+	getSdk: () => ({
+		initializeSession: vi.fn().mockResolvedValue({
+			initializeSession: {
+				secure_id: 'test-session',
+				project_id: '1',
+			},
+		}),
+	}),
+}))
+
+vi.mock('../client/workers/highlight-client-worker?worker&inline', () => ({
+	default: class MockWorker {
+		onmessage: any
+		postMessage() {}
+	},
+}))
+
+async function startedSDK(
+	options: Partial<ConstructorParameters<typeof RecordSDK>[0]> = {},
+): Promise<RecordSDK> {
+	const sdk = new RecordSDK({
+		organizationID: '1',
+		sessionSecureID: 'seed',
+		...options,
+	})
+	await sdk.start()
+	return sdk
+}
+
+/** Delivers a worker response to the SDK the way the real worker would. */
+function postToSDK(sdk: RecordSDK, response: unknown) {
+	sdk._worker.onmessage({ data: { response } } as MessageEvent<any>)
+}
+
+describe('RecordSDK worker stop handling', () => {
+	beforeEach(() => {
+		vi.useFakeTimers()
+		recordStop.mockClear()
+		recordSpy.mockClear()
+	})
+
+	afterEach(() => {
+		vi.useRealTimers()
+	})
+
+	it.each([
+		StopReason.PushPayloadTimeout,
+		StopReason.UnrecoverableError,
+		StopReason.RetriesExhausted,
+	])(
+		'releases the recorder when the worker stops us (%s)',
+		async (reason) => {
+			const sdk = await startedSDK()
+			sdk.events.push({ type: 3, data: {}, timestamp: 1 } as any)
+
+			postToSDK(sdk, { type: MessageType.Stop, reason })
+
+			expect(recordStop).toHaveBeenCalledTimes(1)
+			expect(sdk.events).toEqual([])
+			expect(sdk.getRecordingState()).toBe('NotRecording')
+		},
+	)
+
+	// The page visibility listener stays attached once recording has ever started, so it is the
+	// one thing that can restart us after the worker has given up.
+	it('stays stopped when the tab becomes visible again', async () => {
+		const sdk = await startedSDK({ disableBackgroundRecording: true })
+
+		postToSDK(sdk, {
+			type: MessageType.Stop,
+			reason: StopReason.UnrecoverableError,
+		})
+		recordSpy.mockClear()
+		sdk._lastVisibilityChangeTime = 0 // clear the debounce frozen fake timers impose
+		await sdk._visibilityHandler(false)
+
+		expect(recordSpy).not.toHaveBeenCalled()
+		expect(sdk.getRecordingState()).toBe('NotRecording')
+	})
+
+	it('keeps recording while the worker only reports uploads', async () => {
+		const sdk = await startedSDK()
+		sdk.events.push({ type: 3, data: {}, timestamp: 1 } as any)
+
+		postToSDK(sdk, {
+			type: MessageType.AsyncEvents,
+			id: 1,
+			eventsSize: 100,
+			compressedSize: 50,
+		})
+
+		expect(recordStop).not.toHaveBeenCalled()
+		expect(sdk.events).toHaveLength(1)
+		expect(sdk.getRecordingState()).toBe('Recording')
+	})
+})
diff --git a/sdk/highlight-run/src/client/index.tsx b/sdk/highlight-run/src/client/index.tsx
index 8a7f803222..039ac4de5c 100644
--- a/sdk/highlight-run/src/client/index.tsx
+++ b/sdk/highlight-run/src/client/index.tsx
@@ -201,6 +201,10 @@ export class Highlight {
 	events!: eventWithTime[]
 	sessionData!: SessionData
 	ready!: boolean
+	/**
+	 * Recording must not resume on its own: either the app stopped us, or the worker gave up on
+	 * uploading. Starting again explicitly clears it.
+	 */
 	manualStopped!: boolean
 	state!: 'NotRecording' | 'Recording'
 	logger!: Logger
@@ -298,10 +302,16 @@ export class Highlight {
 				)
 			} else if (e.data.response?.type === MessageType.Stop) {
 				HighlightWarning(
-					'Stopping recording due to worker failure',
+					`Stopping recording due to worker failure: ${e.data.response.reason}`,
 					e.data.response,
 				)
-				this.stopRecording(false)
+				// Replay is what failed, so tracing and metrics keep running. `_save` no longer
+				// runs once we are not recording, so detach the recorder and drop its buffer
+				// instead of filling memory for the rest of this page load. The visibility
+				// listener outlives a stop by design, so mark the stop as one it must not undo.
+				this._stopCapture(true)
+				this.manualStopped = true
+				this.events = []
 			}
 		}
 
@@ -1440,16 +1450,36 @@ SessionSecureID: ${this.sessionData.sessionSecureID}`,
 				'H.stop() was called which stops Highlight from recording.',
 			)
 		}
+		this._stopCapture(manual)
+		void shutdown()
+	}
+
+	/**
+	 * Stops session replay and leaves telemetry alone. `stopRecording` shuts OTel down as well,
+	 * which is right when the whole SDK is stopping but not when only replay uploads have failed.
+	 *
+	 * @param detachRecorder also release rrweb's observers.
+	 */
+	_stopCapture(detachRecorder?: boolean) {
 		this.state = 'NotRecording'
-		// stop rrweb recording mutation observers
-		if (manual && this._recordStop) {
-			this._recordStop()
-			this._recordStop = undefined
+		if (detachRecorder) {
+			this._stopRecorder()
 		}
 		// stop all other event listeners, to be restarted on initialize()
 		this.listeners.forEach((stop) => stop())
 		this.listeners = []
-		void shutdown()
+	}
+
+	/**
+	 * Stops rrweb's recording mutation observers. `stopRecording` only does this on a manual stop
+	 * because rrweb's stop -> restart path is unreliable (eg. iframe listeners), so call this only
+	 * when recording will not resume during this page load.
+	 */
+	_stopRecorder() {
+		if (this._recordStop) {
+			this._recordStop()
+			this._recordStop = undefined
+		}
 	}
 
 	getCurrentSessionTimestamp() {
diff --git a/sdk/highlight-run/src/client/utils/error-recoverability.test.ts b/sdk/highlight-run/src/client/utils/error-recoverability.test.ts
new file mode 100644
index 0000000000..955ca3749b
--- /dev/null
+++ b/sdk/highlight-run/src/client/utils/error-recoverability.test.ts
@@ -0,0 +1,113 @@
+import { describe, expect, it } from 'vitest'
+import { ClientError } from 'graphql-request'
+import { PublicGraphError } from '../graph/generated/schemas'
+import {
+	isErrorRecoverable,
+	isHttpErrorRecoverable,
+} from './error-recoverability'
+
+type TestGraphQLError = {
+	message: string
+	extensions?: { retryable?: boolean }
+}
+
+function clientError(status: number, errors?: TestGraphQLError[]): ClientError {
+	return new ClientError(
+		{ status, errors } as unknown as ClientError['response'],
+		{ query: 'mutation pushPayload { pushPayload }' },
+	)
+}
+
+describe('isHttpErrorRecoverable', () => {
+	it('treats transient 4xx statuses as recoverable', () => {
+		expect(isHttpErrorRecoverable(400)).toBe(true)
+		expect(isHttpErrorRecoverable(408)).toBe(true)
+		expect(isHttpErrorRecoverable(429)).toBe(true)
+	})
+
+	it('treats every other 4xx status as unrecoverable', () => {
+		expect(isHttpErrorRecoverable(401)).toBe(false)
+		expect(isHttpErrorRecoverable(402)).toBe(false)
+		expect(isHttpErrorRecoverable(403)).toBe(false)
+		expect(isHttpErrorRecoverable(404)).toBe(false)
+		expect(isHttpErrorRecoverable(422)).toBe(false)
+	})
+
+	it('treats server errors and non-error statuses as recoverable', () => {
+		expect(isHttpErrorRecoverable(500)).toBe(true)
+		expect(isHttpErrorRecoverable(502)).toBe(true)
+		expect(isHttpErrorRecoverable(503)).toBe(true)
+		expect(isHttpErrorRecoverable(504)).toBe(true)
+		expect(isHttpErrorRecoverable(200)).toBe(true)
+		expect(isHttpErrorRecoverable(0)).toBe(true)
+	})
+})
+
+describe('isErrorRecoverable', () => {
+	it('treats errors of unknown origin as recoverable', () => {
+		expect(isErrorRecoverable(new Error('Failed to fetch'))).toBe(true)
+		expect(isErrorRecoverable(new TypeError('NetworkError'))).toBe(true)
+		expect(isErrorRecoverable(undefined)).toBe(true)
+	})
+
+	it('classifies a rejected request by its status', () => {
+		expect(isErrorRecoverable(clientError(403))).toBe(false)
+		expect(isErrorRecoverable(clientError(404))).toBe(false)
+		expect(isErrorRecoverable(clientError(429))).toBe(true)
+		expect(isErrorRecoverable(clientError(503))).toBe(true)
+	})
+
+	it('treats a 200 carrying GraphQL errors as unrecoverable', () => {
+		expect(
+			isErrorRecoverable(clientError(200, [{ message: 'not allowed' }])),
+		).toBe(false)
+	})
+
+	it('honors an explicit retryable flag over the status', () => {
+		expect(
+			isErrorRecoverable(
+				clientError(403, [
+					{ message: 'try again', extensions: { retryable: true } },
+				]),
+			),
+		).toBe(true)
+		expect(
+			isErrorRecoverable(
+				clientError(429, [
+					{ message: 'give up', extensions: { retryable: false } },
+				]),
+			),
+		).toBe(false)
+		expect(
+			isErrorRecoverable(
+				clientError(200, [
+					{ message: 'try again', extensions: { retryable: true } },
+				]),
+			),
+		).toBe(true)
+	})
+
+	it('takes the most pessimistic retryable flag when errors disagree', () => {
+		expect(
+			isErrorRecoverable(
+				clientError(200, [
+					{ message: 'try again', extensions: { retryable: true } },
+					{ message: 'give up', extensions: { retryable: false } },
+				]),
+			),
+		).toBe(false)
+	})
+
+	it('ignores a retryable flag on a permanent public graph error', () => {
+		expect(
+			isErrorRecoverable(
+				clientError(200, [
+					{
+						message: PublicGraphError.BillingQuotaExceeded,
+						extensions: { retryable: true },
+					},
+				]),
+			),
+		).toBe(false)
+	})
+})
diff --git a/sdk/highlight-run/src/client/utils/error-recoverability.ts b/sdk/highlight-run/src/client/utils/error-recoverability.ts
new file mode 100644
index 0000000000..46a772f2c7
--- /dev/null
+++ b/sdk/highlight-run/src/client/utils/error-recoverability.ts
@@ -0,0 +1,74 @@
+import { ClientError } from 'graphql-request'
+import { PublicGraphError } from '../graph/generated/schemas'
+
+type GraphQLErrors = NonNullable<ClientError['response']['errors']>
+
+// Public graph errors that are permanent whatever the transport reports.
+const UNRECOVERABLE_ERRORS: string[] = [
+	PublicGraphError.BillingQuotaExceeded.toString(),
+]
+
+/**
+ * Tests whether an HTTP error status represents a condition that might resolve on its own if we
+ * retry. 4xx statuses are permanent apart from the three that describe a transient condition;
+ * anything else, including 5xx, is worth another attempt.
+ */
+export const isHttpErrorRecoverable = (statusCode: number): boolean => {
+	if (statusCode < 400 || statusCode >= 500) {
+		return true
+	}
+
+	switch (statusCode) {
+		case 400: // bad request
+		case 408: // request timeout
+		case 429: // too many requests
+			return true
+		default:
+			return false // all other 4xx errors are unrecoverable
+	}
+}
+
+/**
+ * Classifies a failed public graph request as recoverable (retrying may succeed) or unrecoverable
+ * (permanent for this page load). Errors of unknown origin count as recoverable: retrying costs a
+ * backed-off request, while a wrong permanent verdict silently disables recording.
+ */
+export const isErrorRecoverable = (error: unknown): boolean => {
+	if (!(error instanceof ClientError)) {
+		// Offline, DNS and aborted-fetch failures reject with the raw error and carry no
+		// permanent signal.
+		return true
+	}
+
+	const errors = error.response.errors
+	if (errors?.some((e) => UNRECOVERABLE_ERRORS.includes(e.message))) {
+		return false
+	}
+
+	// An explicit `retryable` is more specific than the status code, so it wins.
+	const retryable = retryableFlag(errors)
+	if (retryable !== undefined) {
+		return retryable
+	}
+
+	if (error.response.status >= 400) {
+		return isHttpErrorRecoverable(error.response.status)
+	}
+
+	// The public graph also reports a rejected request as `200` + `errors`. Classify those like a
+	// generic 4xx: permanent unless the server marked an error retryable.
+	return !errors?.length
+}
+
+/** The server's retry verdict for a set of GraphQL errors, or `undefined` when none states one. */
+const retryableFlag = (
+	errors: GraphQLErrors | undefined,
+): boolean | undefined => {
+	if (errors?.some((e) => e.extensions?.retryable === false)) {
+		return false
+	}
+	if (errors?.some((e) => e.extensions?.retryable === true)) {
+		return true
+	}
+	return undefined
+}
diff --git a/sdk/highlight-run/src/client/utils/graph.test.ts b/sdk/highlight-run/src/client/utils/graph.test.ts
new file mode 100644
index 0000000000..d743707348
--- /dev/null
+++ b/sdk/highlight-run/src/client/utils/graph.test.ts
@@ -0,0 +1,90 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import { ClientError } from 'graphql-request'
+import {
+	getGraphQLRequestWrapper,
+	MAX_PUBLIC_GRAPH_RETRY_ATTEMPTS,
+} from './graph'
+
+function clientError(status: number): ClientError {
+	return new ClientError({ status } as unknown as ClientError['response'], {
+		query: 'mutation pushPayload { pushPayload }',
+	})
+}
+
+/** Runs the wrapper to completion, letting every backoff timer fire immediately. */
+async function withoutBackoff<T>(request: Promise<T>): Promise<T> {
+	const settled = request.then(
+		(value) => ({ value }),
+		(error) => ({ error }),
+	)
+	await vi.runAllTimersAsync()
+	const outcome = await settled
+	if ('error' in outcome) {
+		throw outcome.error
+	}
+	return outcome.value
+}
+
+describe('getGraphQLRequestWrapper', () => {
+	beforeEach(() => {
+		vi.useFakeTimers()
+		vi.spyOn(console, 'error').mockImplementation(() => {})
+	})
+
+	afterEach(() => {
+		vi.useRealTimers()
+		vi.restoreAllMocks()
+	})
+
+	it('does not retry a request that succeeds', async () => {
+		const requestFn = vi.fn().mockResolvedValue('ok')
+
+		await expect(
+			withoutBackoff(
+				getGraphQLRequestWrapper()(requestFn, 'pushPayload'),
+			),
+		).resolves.toBe('ok')
+		expect(requestFn).toHaveBeenCalledTimes(1)
+	})
+
+	it('retries a recoverable failure until it succeeds', async () => {
+		const requestFn = vi
+			.fn()
+			.mockRejectedValueOnce(clientError(503))
+			.mockRejectedValueOnce(new Error('Failed to fetch'))
+			.mockResolvedValue('ok')
+
+		await expect(
+			withoutBackoff(
+				getGraphQLRequestWrapper()(requestFn, 'pushPayload'),
+			),
+		).resolves.toBe('ok')
+		expect(requestFn).toHaveBeenCalledTimes(3)
+	})
+
+	it('gives up on a recoverable failure after the retry budget', async () => {
+		const error = clientError(503)
+		const requestFn = vi.fn().mockRejectedValue(error)
+
+		await expect(
+			withoutBackoff(
+				getGraphQLRequestWrapper()(requestFn, 'pushPayload'),
+			),
+		).rejects.toBe(error)
+		expect(requestFn).toHaveBeenCalledTimes(
+			MAX_PUBLIC_GRAPH_RETRY_ATTEMPTS + 1,
+		)
+	})
+
+	it('throws an unrecoverable failure without retrying', async () => {
+		const error = clientError(403)
+		const requestFn = vi.fn().mockRejectedValue(error)
+
+		await expect(
+			withoutBackoff(
+				getGraphQLRequestWrapper()(requestFn, 'pushPayload'),
+			),
+		).rejects.toBe(error)
+		expect(requestFn).toHaveBeenCalledTimes(1)
+	})
+})
diff --git a/sdk/highlight-run/src/client/utils/graph.ts b/sdk/highlight-run/src/client/utils/graph.ts
index 08d18fb5f7..28c6d19b55 100644
--- a/sdk/highlight-run/src/client/utils/graph.ts
+++ b/sdk/highlight-run/src/client/utils/graph.ts
@@ -1,5 +1,4 @@
-import { ClientError } from 'graphql-request'
-import { PublicGraphError } from '../graph/generated/schemas'
+import { isErrorRecoverable } from './error-recoverability'
 
 export const MAX_PUBLIC_GRAPH_RETRY_ATTEMPTS = 3
 
@@ -7,17 +6,6 @@ export const MAX_PUBLIC_GRAPH_RETRY_ATTEMPTS = 3
 export const BASE_DELAY_MS = 1000
 export const BACKOFF_DELAY_MS = 500
 
-// Do not retry if any of these public graph errors are thrown
-const NON_RETRYABLE_ERRORS = [PublicGraphError.BillingQuotaExceeded.toString()]
-
-// A `ClientError` is retryable if none of the response errors is non-retryable
-const isErrorRetryable = (error: ClientError): boolean => {
-	const match = error.response.errors?.find((e) =>
-		NON_RETRYABLE_ERRORS.includes(e.message),
-	)
-	return match === undefined
-}
-
 export const getGraphQLRequestWrapper = () => {
 	const graphQLRequestWrapper = async <T>(
 		requestFn: () => Promise<T>,
@@ -29,7 +17,9 @@ export const getGraphQLRequestWrapper = () => {
 		try {
 			return await requestFn()
 		} catch (error: any) {
-			if (error instanceof ClientError && !isErrorRetryable(error)) {
+			// Retrying an unrecoverable failure only delays the caller and adds load the backend
+			// already rejected.
+			if (!isErrorRecoverable(error)) {
 				throw error
 			}
 
diff --git a/sdk/highlight-run/src/client/workers/highlight-client-worker-errors.test.ts b/sdk/highlight-run/src/client/workers/highlight-client-worker-errors.test.ts
new file mode 100644
index 0000000000..a985f98c98
--- /dev/null
+++ b/sdk/highlight-run/src/client/workers/highlight-client-worker-errors.test.ts
@@ -0,0 +1,227 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
+import { ClientError } from 'graphql-request'
+import { MAX_PUBLIC_GRAPH_RETRY_ATTEMPTS } from './constants'
+import {
+	CustomEventResponse,
+	HighlightClientWorkerParams,
+	HighlightClientWorkerResponse,
+	MessageType,
+	StatusResponse,
+	StopEventResponse,
+	StopReason,
+} from './types'
+
+// Every operation the worker performs rejects with this, so a single knob decides whether the
+// worker sees a recoverable or an unrecoverable failure.
+const graph = vi.hoisted(() => ({ error: undefined as unknown }))
+
+vi.mock('../graph/generated/operations', async (importOriginal) => {
+	const actual =
+		await importOriginal<typeof import('../graph/generated/operations')>()
+	const reject = () => Promise.reject(graph.error)
+	return {
+		...actual,
+		getSdk: () => ({
+			identifySession: reject,
+			addSessionProperties: reject,
+			PushSessionEvents: reject,
+			pushMetrics: reject,
+		}),
+	}
+})
+
+interface TestWorker {
+	postMessage: (params: HighlightClientWorkerParams) => void
+	onmessage:
+		| ((event: MessageEvent<HighlightClientWorkerResponse>) => void)
+		| null
+	terminate: () => void
+}
+
+function clientError(status: number): ClientError {
+	return new ClientError({ status } as unknown as ClientError['response'], {
+		query: 'mutation identifySession { identifySession }',
+	})
+}
+
+function initializeMessage(): HighlightClientWorkerParams {
+	return {
+		message: {
+			type: MessageType.Initialize,
+			backend: 'https://test.highlight.io/graphql',
+			sessionSecureID: 'test-session-123',
+			debug: false,
+			recordingStartTime: Date.now(),
+		},
+	}
+}
+
+function identifyMessage(userIdentifier: string): HighlightClientWorkerParams {
+	return {
+		message: {
+			type: MessageType.Identify,
+			userIdentifier,
+			userObject: { name: 'Test User' },
+		},
+	}
+}
+
+describe('highlight-client-worker error handling', () => {
+	let worker: TestWorker
+	let responses: HighlightClientWorkerResponse[]
+
+	const responsesOfType = <T>(type: MessageType): T[] =>
+		responses
+			.filter((r) => r.response?.type === type)
+			.map((r) => r.response as T)
+
+	const status = async (): Promise<StatusResponse> => {
+		const before = responsesOfType<StatusResponse>(
+			MessageType.GetStatus,
+		).length
+		worker.postMessage({ message: { type: MessageType.GetStatus } })
+		await vi.waitFor(() => {
+			expect(
+				responsesOfType<StatusResponse>(MessageType.GetStatus).length,
+			).toBeGreaterThan(before)
+		})
+		const all = responsesOfType<StatusResponse>(MessageType.GetStatus)
+		return all[all.length - 1]
+	}
+
+	beforeEach(() => {
+		vi.resetModules()
+		vi.spyOn(console, 'warn').mockImplementation(() => {})
+		// See highlight-client-worker.test.ts: the source module (rather than the `?worker&inline`
+		// build output) is what @vitest/web-worker can run for real.
+		worker = new Worker(
+			new URL('./highlight-client-worker.ts', import.meta.url),
+			{ type: 'module' },
+		) as unknown as TestWorker
+		responses = []
+		worker.onmessage = (event) => {
+			responses.push(event.data)
+		}
+	})
+
+	afterEach(() => {
+		worker.terminate()
+		vi.restoreAllMocks()
+	})
+
+	it('stops recording when the backend rejects a request permanently', async () => {
+		graph.error = clientError(403)
+
+		worker.postMessage(initializeMessage())
+		worker.postMessage(identifyMessage('refused-user'))
+
+		await vi.waitFor(() => {
+			const stops = responsesOfType<StopEventResponse>(MessageType.Stop)
+			expect(stops).toHaveLength(1)
+			expect(stops[0].reason).toBe(StopReason.UnrecoverableError)
+		})
+
+		expect((await status()).initialized).toBe(false)
+	})
+
+	it('does not report the stop as a timeline event', async () => {
+		graph.error = clientError(403)
+
+		worker.postMessage(initializeMessage())
+		worker.postMessage(identifyMessage('refused-user'))
+
+		await vi.waitFor(() => {
+			expect(
+				responsesOfType<StopEventResponse>(MessageType.Stop),
+			).toHaveLength(1)
+		})
+
+		// The client is NotRecording by the time it handles this, so a Track event could never be
+		// captured; it would only leave `addCustomEvent` polling every 500ms for the rest of the
+		// page load.
+		expect(
+			responsesOfType<CustomEventResponse>(MessageType.CustomEvent).map(
+				(e) => e.tag,
+			),
+		).not.toContain('Track')
+	})
+
+	it('drops later messages after a permanent rejection', async () => {
+		graph.error = clientError(403)
+
+		worker.postMessage(initializeMessage())
+		worker.postMessage(identifyMessage('refused-user'))
+
+		await vi.waitFor(() => {
+			expect(
+				responsesOfType<StopEventResponse>(MessageType.Stop),
+			).toHaveLength(1)
+		})
+
+		worker.postMessage(identifyMessage('dropped-user'))
+
+		const { pendingCount } = await status()
+		expect(pendingCount).toBe(0)
+		expect(
+			responsesOfType<CustomEventResponse>(MessageType.CustomEvent).some(
+				(e) => e.payload?.includes('dropped-user'),
+			),
+		).toBe(false)
+	})
+
+	it('keeps recording when the failure may resolve on its own', async () => {
+		graph.error = clientError(503)
+
+		worker.postMessage(initializeMessage())
+		worker.postMessage(identifyMessage('retried-user'))
+
+		await vi.waitFor(() => {
+			expect(
+				responsesOfType<CustomEventResponse>(
+					MessageType.CustomEvent,
+				).some((e) => e.payload?.includes('retried-user')),
+			).toBe(true)
+		})
+
+		expect(responsesOfType<StopEventResponse>(MessageType.Stop)).toEqual([])
+		expect((await status()).initialized).toBe(true)
+	})
+
+	it('stops recording once the retry budget is spent', async () => {
+		graph.error = clientError(503)
+
+		worker.postMessage(initializeMessage())
+		for (let i = 0; i < MAX_PUBLIC_GRAPH_RETRY_ATTEMPTS; i++) {
+			worker.postMessage(identifyMessage(`retried-user-${i}`))
+		}
+
+		await vi.waitFor(() => {
+			const stops = responsesOfType<StopEventResponse>(MessageType.Stop)
+			expect(stops).toHaveLength(1)
+			expect(stops[0].reason).toBe(StopReason.RetriesExhausted)
+		})
+
+		worker.postMessage(identifyMessage('dropped-user'))
+
+		const { pendingCount, initialized } = await status()
+		expect(pendingCount).toBe(0)
+		expect(initialized).toBe(false)
+	})
+
+	it('resumes after a new session initializes', async () => {
+		graph.error = clientError(403)
+
+		worker.postMessage(initializeMessage())
+		worker.postMessage(identifyMessage('refused-user'))
+
+		await vi.waitFor(() => {
+			expect(
+				responsesOfType<StopEventResponse>(MessageType.Stop),
+			).toHaveLength(1)
+		})
+
+		worker.postMessage(initializeMessage())
+
+		expect((await status()).initialized).toBe(true)
+	})
+})
diff --git a/sdk/highlight-run/src/client/workers/highlight-client-worker.ts b/sdk/highlight-run/src/client/workers/highlight-client-worker.ts
index c6dcfa573c..2384899fe1 100644
--- a/sdk/highlight-run/src/client/workers/highlight-client-worker.ts
+++ b/sdk/highlight-run/src/client/workers/highlight-client-worker.ts
@@ -10,6 +10,7 @@ import { payloadToBase64 } from '../utils/payload'
 import { ReplayEventsInput } from '../graph/generated/schemas'
 import { Logger } from '../logger'
 import { MetricCategory } from '../types/client'
+import { isErrorRecoverable } from '../utils/error-recoverability'
 import { getGraphQLRequestWrapper } from '../utils/graph'
 import {
 	MAX_PUBLIC_GRAPH_RETRY_ATTEMPTS,
@@ -26,6 +27,8 @@ import {
 	MessageType,
 	MetricsMessage,
 	PropertiesMessage,
+	StopEventResponse,
+	StopReason,
 } from './types'
 
 export interface HighlightClientRequestWorker {
@@ -100,6 +103,7 @@ function stringifyProperties(
 	let sessionSecureID: string
 	let numberOfFailedRequests: number = 0
 	let numberOfFailedPushPayloads: number = 0
+	let hasStoppedRecording: boolean = false
 	let debug: boolean = false
 	let recordingStartTime: number = 0
 	let logger = new Logger(false, '[worker]')
@@ -116,8 +120,8 @@ function stringifyProperties(
 
 	const shouldSendRequest = (): boolean => {
 		return (
+			!hasStoppedRecording &&
 			recordingStartTime !== 0 &&
-			numberOfFailedRequests < MAX_PUBLIC_GRAPH_RETRY_ATTEMPTS &&
 			!!sessionSecureID?.length
 		)
 	}
@@ -132,6 +136,65 @@ function stringifyProperties(
 		})
 	}
 
+	/**
+	 * Asks the client to stop recording and stops accepting work until it initializes again.
+	 * Everything buffered here is dropped: the client stops handing us events, so a queue kept
+	 * across a stop would only hold memory for data that will never be uploaded.
+	 *
+	 * Only the first caller is served. Requests already in flight when we stop keep failing and
+	 * arrive here afterwards, and the client has nothing left to tear down by then.
+	 */
+	const stopRecording = (
+		reason: StopReason,
+		details?: Pick<
+			StopEventResponse,
+			'requestStart' | 'asyncEventsResponse'
+		>,
+	) => {
+		if (hasStoppedRecording) {
+			return
+		}
+
+		hasStoppedRecording = true
+		pendingMessages.length = 0
+		metricsPayload.length = 0
+
+		worker.postMessage({
+			response: {
+				type: MessageType.Stop,
+				reason,
+				...details,
+			},
+		})
+	}
+
+	/**
+	 * Accounts for a message we could not send. An unrecoverable failure ends recording right
+	 * away, because the backend would reject every later request the same way. A recoverable one
+	 * spends part of the retry budget, and exhausting that budget also ends recording: the client
+	 * would otherwise keep producing payloads that pile up here unsent.
+	 */
+	const handleFailedMessage = (e: unknown) => {
+		if (debug) {
+			console.error(e)
+		}
+
+		if (!isErrorRecoverable(e)) {
+			console.warn(`Session data was rejected, stopping recording.`, e)
+			stopRecording(StopReason.UnrecoverableError)
+			return
+		}
+
+		numberOfFailedRequests += 1
+		if (numberOfFailedRequests >= MAX_PUBLIC_GRAPH_RETRY_ATTEMPTS) {
+			console.warn(
+				`Session data failed to upload ${numberOfFailedRequests} times, stopping recording.`,
+				e,
+			)
+			stopRecording(StopReason.RetriesExhausted)
+		}
+	}
+
 	const processAsyncEventsMessage = async (msg: AsyncEventsMessage) => {
 		const {
 			id,
@@ -229,20 +292,9 @@ function stringifyProperties(
 						`Uploading pushPayload took too long, stopping recording to avoid OOM.`,
 					)
 
-					worker.postMessage({
-						response: {
-							type: MessageType.Stop,
-							requestStart,
-							asyncEventsResponse: response,
-						},
-					})
-
-					processPropertiesMessage({
-						type: MessageType.Properties,
-						propertiesObject: {
-							stopReason: 'Push Payload Timeout',
-						},
-						propertyType: { type: 'track' },
+					stopRecording(StopReason.PushPayloadTimeout, {
+						requestStart,
+						asyncEventsResponse: response,
 					})
 				}
 			}
@@ -374,10 +426,7 @@ function stringifyProperties(
 				await processMessage(msg)
 				numberOfFailedRequests = 0
 			} catch (e) {
-				if (debug) {
-					console.error(e)
-				}
-				numberOfFailedRequests += 1
+				handleFailedMessage(e)
 			}
 		}
 	}
@@ -389,6 +438,11 @@ function stringifyProperties(
 			debug = e.data.message.debug
 			recordingStartTime = e.data.message.recordingStartTime
 			logger.debug = debug
+			// The client only initializes after `initializeSession` succeeded, so whatever made
+			// us stop no longer applies.
+			hasStoppedRecording = false
+			numberOfFailedRequests = 0
+			numberOfFailedPushPayloads = 0
 			graphqlSDK = getSdk(
 				new GraphQLClient(backend, {
 					headers: {},
@@ -407,6 +461,7 @@ function stringifyProperties(
 			metricsPayload.length = 0
 			numberOfFailedRequests = 0
 			numberOfFailedPushPayloads = 0
+			hasStoppedRecording = false
 			// Reset sessionSecureID and recordingStartTime so that messages arriving
 			// between Reset and new Initialize are queued rather than processed with
 			// the old session SecureID
@@ -427,6 +482,12 @@ function stringifyProperties(
 			return
 		}
 
+		// Drop messages once recording has stopped: queueing them would grow without bound
+		// because nothing is going to send them.
+		if (hasStoppedRecording) {
+			return
+		}
+
 		if (!shouldSendRequest()) {
 			pendingMessages.push(e.data.message)
 			return
@@ -437,10 +498,7 @@ function stringifyProperties(
 			numberOfFailedRequests = 0
 			await drainPendingMessages()
 		} catch (e) {
-			if (debug) {
-				console.error(e)
-			}
-			numberOfFailedRequests += 1
+			handleFailedMessage(e)
 		}
 	}
 }
diff --git a/sdk/highlight-run/src/client/workers/types.ts b/sdk/highlight-run/src/client/workers/types.ts
index 752b5216a8..f4faceb5e8 100644
--- a/sdk/highlight-run/src/client/workers/types.ts
+++ b/sdk/highlight-run/src/client/workers/types.ts
@@ -100,10 +100,20 @@ export type CustomEventResponse = {
 	payload: any
 }
 
+export enum StopReason {
+	/** Uploads are outrunning the recording, so events would pile up in memory. */
+	PushPayloadTimeout = 'Push Payload Timeout',
+	/** The backend rejected the session data permanently; retrying cannot help. */
+	UnrecoverableError = 'Unrecoverable Error',
+	/** Enough uploads failed in a row that the backend is not answering us at all. */
+	RetriesExhausted = 'Retries Exhausted',
+}
+
 export type StopEventResponse = {
 	type: MessageType.Stop
-	requestStart: number
-	asyncEventsResponse: AsyncEventsResponse
+	reason: StopReason
+	requestStart?: number
+	asyncEventsResponse?: AsyncEventsResponse
 }
 
 export type HighlightClientWorkerParams = {
diff --git a/sdk/highlight-run/src/sdk/record.ts b/sdk/highlight-run/src/sdk/record.ts
index 79c0420f72..be723f051e 100644
--- a/sdk/highlight-run/src/sdk/record.ts
+++ b/sdk/highlight-run/src/sdk/record.ts
@@ -116,6 +116,10 @@ export class RecordSDK implements Record {
 	events!: eventWithTime[]
 	sessionData!: SessionData
 	ready!: boolean
+	/**
+	 * Recording must not resume on its own: either the app stopped us, or the worker gave up on
+	 * uploading. Starting again explicitly clears it.
+	 */
 	manualStopped!: boolean
 	state!: 'NotRecording' | 'Recording'
 	logger!: Logger
@@ -187,10 +191,18 @@ export class RecordSDK implements Record {
 				internalLog(
 					'worker.onmessage',
 					'warn',
-					'Stopping recording due to worker failure',
+					`Stopping recording due to worker failure: ${e.data.response.reason}`,
 					e.data.response,
 				)
 				this.stop(false)
+				// The worker only asks us to stop when it cannot upload what we record, and
+				// `_save` no longer runs once we are not recording, so detach the recorder and
+				// drop its buffer instead of filling memory for the rest of this page load.
+				// The visibility listener outlives a stop by design, so mark the stop as one it
+				// must not undo.
+				this.manualStopped = true
+				this._stopRecorder()
+				this.events = []
 			}
 		}
 
@@ -950,16 +962,26 @@ SessionSecureID: ${this.sessionData.sessionSecureID}`,
 			)
 		}
 		this.state = 'NotRecording'
-		// stop rrweb recording mutation observers
-		if (manual && this._recordStop) {
-			this._recordStop()
-			this._recordStop = undefined
+		if (manual) {
+			this._stopRecorder()
 		}
 		// stop all other event listeners, to be restarted on initialize()
 		this.listeners.forEach((stop) => stop())
 		this.listeners = []
 	}
 
+	/**
+	 * Stops rrweb's recording mutation observers. `stop` only does this on a manual stop because
+	 * rrweb's stop -> restart path is unreliable (eg. iframe listeners), so call this only when
+	 * recording will not resume during this page load.
+	 */
+	_stopRecorder() {
+		if (this._recordStop) {
+			this._recordStop()
+			this._recordStop = undefined
+		}
+	}
+
 	/**
 	 * Returns the current timestamp for the current session.
 	 */