diff --git a/sentry-android-replay/src/main/java/io/sentry/android/replay/capture/BaseCaptureStrategy.kt b/sentry-android-replay/src/main/java/io/sentry/android/replay/capture/BaseCaptureStrategy.kt
index dab98ec4e2..bacf672291 100644
--- a/sentry-android-replay/src/main/java/io/sentry/android/replay/capture/BaseCaptureStrategy.kt
+++ b/sentry-android-replay/src/main/java/io/sentry/android/replay/capture/BaseCaptureStrategy.kt
@@ -58,11 +58,21 @@ internal abstract class BaseCaptureStrategy(
     private const val MAX_TRACE_IDS = 100
   }
 
-  private val persistingExecutor: ScheduledExecutorService by lazy {
-    val delegate =
-      Executors.newSingleThreadScheduledExecutor(ReplayPersistingExecutorServiceThreadFactory())
-    ReplayExecutorService(delegate, options)
-  }
+  // Explicit holder so we can detect whether the executor was ever initialised and shut it down
+  // without initialising it as a side-effect (which is what a plain `lazy` delegate would do).
+  private var persistingExecutorHolder: ScheduledExecutorService? = null
+  private val persistingExecutor: ScheduledExecutorService
+    get() {
+      return persistingExecutorHolder
+        ?: run {
+          val delegate =
+            Executors.newSingleThreadScheduledExecutor(
+              ReplayPersistingExecutorServiceThreadFactory()
+            )
+          ReplayExecutorService(delegate, options).also { persistingExecutorHolder = it }
+        }
+    }
+
   private val gestureConverter = ReplayGestureConverter(dateProvider)
 
   protected val isTerminating = AtomicBoolean(false)
@@ -123,6 +133,13 @@ internal abstract class BaseCaptureStrategy(
     replayStartTimestamp.set(0)
     segmentTimestamp = null
     currentReplayId = SentryId.EMPTY_ID
+    // Shut down the persisting executor only if it was actually initialised.  We must not call the
+    // blocking ReplayExecutorService.shutdown() here because stop() may run on the main thread,
+    // and blocking there risks an ANR.  shutdownNow() is non-blocking: it cancels queued tasks and
+    // interrupts any running one, which is acceptable because the tasks are best-effort persistence
+    // writes and the cache has already been closed above.
+    persistingExecutorHolder?.shutdownNow()
+    persistingExecutorHolder = null
   }
 
   protected fun createSegmentInternal(
diff --git a/sentry-android-replay/src/test/java/io/sentry/android/replay/capture/SessionCaptureStrategyTest.kt b/sentry-android-replay/src/test/java/io/sentry/android/replay/capture/SessionCaptureStrategyTest.kt
index b5a00bc624..6343ca6f42 100644
--- a/sentry-android-replay/src/test/java/io/sentry/android/replay/capture/SessionCaptureStrategyTest.kt
+++ b/sentry-android-replay/src/test/java/io/sentry/android/replay/capture/SessionCaptureStrategyTest.kt
@@ -31,6 +31,7 @@ import io.sentry.rrweb.RRWebMetaEvent
 import io.sentry.rrweb.RRWebOptionsEvent
 import io.sentry.transport.CurrentDateProvider
 import io.sentry.transport.ICurrentDateProvider
+import io.sentry.util.thread.IThreadChecker
 import java.io.File
 import java.util.Date
 import kotlin.test.Test
@@ -554,4 +555,35 @@ class SessionCaptureStrategyTest {
         any(),
       )
   }
+
+  @Test
+  fun `stop shuts down persisting executor so no SentryReplayPersister threads leak across cycles`() {
+    // Force isMainThread() to return true so that property-reset writes in stop() are dispatched
+    // to the persistingExecutor, which is the code path that used to leak threads.
+    fixture.options.threadChecker =
+      mock<IThreadChecker> {
+        on { isMainThread() }.thenReturn(true)
+        on { isMainThread(any<Thread>()) }.thenReturn(true)
+        on { isMainThread(anyLong()) }.thenReturn(false)
+        on { currentThreadName }.thenReturn("main")
+        on { currentThreadSystemId() }.thenReturn(0L)
+      }
+
+    repeat(3) {
+      val strategy = fixture.getSut()
+      strategy.start(0, SentryId())
+      strategy.onConfigurationChanged(fixture.recorderConfig)
+      strategy.stop()
+    }
+
+    // Allow time for thread termination after shutdownNow().
+    Thread.sleep(200)
+
+    val leakedThreads =
+      Thread.getAllStackTraces().keys.filter { it.name.startsWith("SentryReplayPersister-") }
+    assertTrue(
+      leakedThreads.isEmpty(),
+      "Expected no SentryReplayPersister threads after stop(), found: ${leakedThreads.map { it.name }}",
+    )
+  }
 }