diff --git a/integration-tests/debugger/snapshot-time-budget.spec.js b/integration-tests/debugger/snapshot-time-budget.spec.js
index af9dbd16c35..02140be6afb 100644
--- a/integration-tests/debugger/snapshot-time-budget.spec.js
+++ b/integration-tests/debugger/snapshot-time-budget.spec.js
@@ -2,10 +2,14 @@
 
 const assert = require('node:assert')
 const { setup } = require('./utils')
+const {
+  DEFAULT_MAX_COLLECTION_SIZE,
+  LARGE_OBJECT_SKIP_THRESHOLD
+} = require('../../packages/dd-trace/src/debugger/devtools_client/snapshot/constants')
 
 describe('Dynamic Instrumentation', function () {
   describe('input messages', function () {
-    describe('with snapshot under tight time budget', function () {
+    describe('with snapshot under time budget', function () {
       context('1ms time budget', function () {
         // Force a very small time budget in ms to trigger partial snapshots
         const budget = 1
@@ -17,57 +21,177 @@ describe('Dynamic Instrumentation', function () {
         it(
           'should include partial snapshot marked with notCapturedReason: timeout',
           // A tolerance of 15ms is used to avoid flakiness
-          test({ t, maxPausedTime: budget + 15, breakpointIndex: 0, maxReferenceDepth: 5 })
+          test({ t, maxPausedTime: budget + 15, breakpointIndex: 0, maxReferenceDepth: 5 }, (locals) => {
+            assert.strictEqual(
+              containsTimeBudget(locals),
+              true,
+              'expected at least one field/element to be marked with notCapturedReason: "timeout"'
+            )
+          })
         )
       })
 
       context('default time budget', function () {
-        const budget = 15 // default time budget in ms
         const t = setup({ dependencies: ['fastify'] })
 
-        // TODO: Make this pass
-        // eslint-disable-next-line mocha/no-pending-tests
-        it.skip(
-          'should keep budget when state includes an object with 1 million properties',
-          // A tolerance of 5ms is used to avoid flakiness
-          test({ t, maxPausedTime: budget + 5, breakpointIndex: 1, maxReferenceDepth: 1 })
+        it(
+          'should timeout first, then disable subsequent snapshots and emit error diagnostics',
+          async function () {
+            const breakpoint = t.breakpoints[1]
+            const expectedEvaluationErrors = [{
+              expr: '',
+              message: 'An object with 1000000 properties was detected while collecting a snapshot. This exceeds ' +
+                `the maximum number of allowed properties of ${LARGE_OBJECT_SKIP_THRESHOLD}. Future snapshots for ` +
+                'existing probes in this location will be skipped until the Node.js process is restarted'
+            }]
+
+            // Listen for the first snapshot payload (should contain notCapturedReason: "timeout")
+            const firstPayloadReceived = new Promise(/** @type {() => void} */ (resolve) => {
+              t.agent.once('debugger-input', ({ payload: [{ debugger: { snapshot } }] }) => {
+                const { locals } = snapshot.captures.lines[breakpoint.line]
+                assert.strictEqual(
+                  containsTimeBudget(locals),
+                  true,
+                  'expected at least one field/element to be marked with notCapturedReason: "timeout"'
+                )
+                assert.deepStrictEqual(snapshot.evaluationErrors, expectedEvaluationErrors)
+                resolve()
+              })
+            })
+
+            // Install probe with snapshot capture enabled
+            t.agent.addRemoteConfig(breakpoint.generateRemoteConfig({
+              captureSnapshot: true,
+              capture: { maxReferenceDepth: 1 }
+            }))
+
+            // Trigger once; this run is expected to be slow and mark fields with "timeout"
+            const result1 = /** @type {import('axios').AxiosResponse<{ paused: number }>} */
+              (await breakpoint.triggerBreakpoint())
+            assert.ok(
+              result1.data.paused >= 1_000,
+              `expected thread to be paused for at least 1 second, but was paused for ~${result1.data.paused}ms`
+            )
+
+            await firstPayloadReceived
+
+            // Prepare to assert that no snapshot is produced on a subsequent trigger
+            const secondPayloadReceived = new Promise(/** @type {() => void} */ (resolve) => {
+              t.agent.once('debugger-input', ({ payload: [{ debugger: { snapshot } }] }) => {
+                assert.ok(!Object.hasOwn(snapshot, 'captures'))
+                assert.deepStrictEqual(snapshot.evaluationErrors, expectedEvaluationErrors)
+                resolve()
+              })
+            })
+
+            // Trigger the same breakpoint again directly
+            const result2 = await t.axios.get(breakpoint.url)
+            assert.ok(
+              result2.data.paused <= 50,
+              `expected thread to be paused <=50ms, but was paused for ~${result2.data.paused}ms`
+            )
+
+            await secondPayloadReceived
+          }
         )
+      })
+
+      context('large object thresholds', function () {
+        // The tests in this group, should take a lot longer than 100ms to capture a snapshot if no thresholds are
+        // applied. But they should take a lot less than the 100ms budget if thresholds are applied. Using 100ms means
+        // that the tests should not be flaky, but still fail if the thresholds are not applied.
+        const budget = 100
+        const t = setup({
+          dependencies: ['fastify'],
+          env: { DD_DYNAMIC_INSTRUMENTATION_CAPTURE_TIMEOUT_MS: String(budget) }
+        })
 
-        // TODO: Make this pass
-        // eslint-disable-next-line mocha/no-pending-tests
-        it.skip(
-          'should keep budget when state includes an array of 1 million primitives',
-          // A tolerance of 5ms is used to avoid flakiness
-          test({ t, maxPausedTime: budget + 5, breakpointIndex: 2, maxReferenceDepth: 1 })
+        it(
+          'should keep budget when state includes collections with 1 million elements',
+          test({ t, maxPausedTime: budget, breakpointIndex: 2, maxReferenceDepth: 1 }, (locals) => {
+            const notCapturedReason = `Large collection with too many elements (skip threshold: ${
+              LARGE_OBJECT_SKIP_THRESHOLD
+            })`
+            assert.strictEqual(locals.arrOfPrimitives.notCapturedReason, notCapturedReason)
+            assert.strictEqual(locals.arrOfPrimitives.size, 1_000_000)
+            assert.strictEqual(locals.arrOfPrimitives.elements.length, 0)
+            assert.strictEqual(locals.arrOfObjects.notCapturedReason, notCapturedReason)
+            assert.strictEqual(locals.arrOfObjects.size, 1_000_000)
+            assert.strictEqual(locals.arrOfObjects.elements.length, 0)
+            assert.strictEqual(locals.map.notCapturedReason, notCapturedReason)
+            assert.strictEqual(locals.map.size, 1_000_000)
+            assert.strictEqual(locals.map.entries.length, 0)
+            assert.strictEqual(locals.set.notCapturedReason, notCapturedReason)
+            assert.strictEqual(locals.set.size, 1_000_000)
+            assert.strictEqual(locals.set.elements.length, 0)
+          })
         )
 
-        // TODO: Make this pass
-        // eslint-disable-next-line mocha/no-pending-tests
-        it.skip(
-          'should keep budget when state includes an array of 1 million objects',
-          // A tolerance of 5ms is used to avoid flakiness
-          test({ t, maxPausedTime: budget + 5, breakpointIndex: 3, maxReferenceDepth: 1 })
+        it(
+          'should keep budget when state includes collections with less than the size threshold',
+          test({ t, maxPausedTime: budget, breakpointIndex: 3, maxReferenceDepth: 1 }, (locals) => {
+            assert.strictEqual(locals.arrOfPrimitives.notCapturedReason, 'collectionSize')
+            assert.strictEqual(locals.arrOfPrimitives.size, LARGE_OBJECT_SKIP_THRESHOLD - 1)
+            assert.strictEqual(locals.arrOfPrimitives.elements.length, DEFAULT_MAX_COLLECTION_SIZE)
+            assert.strictEqual(locals.arrOfObjects.notCapturedReason, 'collectionSize')
+            assert.strictEqual(locals.arrOfObjects.size, LARGE_OBJECT_SKIP_THRESHOLD - 1)
+            assert.strictEqual(locals.arrOfObjects.elements.length, DEFAULT_MAX_COLLECTION_SIZE)
+            assert.strictEqual(locals.map.notCapturedReason, 'collectionSize')
+            assert.strictEqual(locals.map.size, LARGE_OBJECT_SKIP_THRESHOLD - 1)
+            assert.strictEqual(locals.map.entries.length, DEFAULT_MAX_COLLECTION_SIZE)
+            assert.strictEqual(locals.set.notCapturedReason, 'collectionSize')
+            assert.strictEqual(locals.set.size, LARGE_OBJECT_SKIP_THRESHOLD - 1)
+            assert.strictEqual(locals.set.elements.length, DEFAULT_MAX_COLLECTION_SIZE)
+          })
         )
       })
+
+      context('fuzzing', function () {
+        for (let budget = 0; budget < 20; budget++) {
+          context(`graceful handling with time budget of ${budget}ms`, function () {
+            // Anything longer than this, and the debugger worker thread most likely crashed.
+            // Run test with `DD_TRACE_DEBUG=true` to see more.
+            this.timeout(2000)
+
+            const t = setup({
+              dependencies: ['fastify'],
+              env: { DD_DYNAMIC_INSTRUMENTATION_CAPTURE_TIMEOUT_MS: String(budget) }
+            })
+
+            // If this test uncovers any issues, it will show itself as being flaky, as the exact timing of how long it
+            // takes to collect the snapshot is not deterministic.
+            it('should send a probe result to the agent', async function () {
+              t.agent.addRemoteConfig(t.generateRemoteConfig({
+                captureSnapshot: true,
+                capture: { maxReferenceDepth: 5 }
+              }))
+              t.triggerBreakpoint()
+
+              const { captures } = await t.snapshotReceived()
+              // If the snapshot contains a `captures` field, it means it was captured and processes without any issues
+              assert.ok(
+                captures !== null && typeof captures === 'object',
+                'expected snapshot to contain a `captures` object'
+              )
+              // To make this test more future-proof, we also assert that the snapshot contains at least one local
+              // property, though currently this is not necessary.
+              assert.ok(
+                Object.keys(captures.lines[t.breakpoint.line].locals).length > 0,
+                'expected snapshot to contain at least one local property'
+              )
+            })
+          })
+        }
+      })
     })
   })
 })
 
-function test ({ t, maxPausedTime, breakpointIndex, maxReferenceDepth }) {
+function test ({ t, maxPausedTime = 0, breakpointIndex, maxReferenceDepth }, assertFn) {
   const breakpoint = t.breakpoints[breakpointIndex]
 
   return async function () {
-    const payloadReceived = new Promise((/** @type {(value?: unknown) => void} */ resolve) => {
-      t.agent.on('debugger-input', ({ payload: [{ debugger: { snapshot: { captures } } }] }) => {
-        const { locals } = captures.lines[breakpoint.line]
-        assert.strictEqual(
-          containsTimeBudget(locals),
-          true,
-          'expected at least one field/element to be marked with notCapturedReason: "timeout"'
-        )
-        resolve()
-      })
-    })
+    const snapshotPromise = t.snapshotReceived()
 
     t.agent.addRemoteConfig(breakpoint.generateRemoteConfig({
       captureSnapshot: true,
@@ -81,7 +205,8 @@ function test ({ t, maxPausedTime, breakpointIndex, maxReferenceDepth }) {
       `expected thread to be paused <=${maxPausedTime}ms, but was paused for ~${data.paused}ms`
     )
 
-    await payloadReceived
+    const snapshot = await snapshotPromise
+    assertFn?.(snapshot.captures.lines[breakpoint.line].locals)
   }
 }
 
diff --git a/integration-tests/debugger/target-app/snapshot-time-budget.js b/integration-tests/debugger/target-app/snapshot-time-budget.js
index 7f396d546b1..529ec0c2190 100644
--- a/integration-tests/debugger/target-app/snapshot-time-budget.js
+++ b/integration-tests/debugger/target-app/snapshot-time-budget.js
@@ -5,6 +5,10 @@ require('dd-trace/init')
 
 // @ts-expect-error This code is running in a sandbox where fastify is available
 const Fastify = require('fastify')
+const {
+  LARGE_OBJECT_SKIP_THRESHOLD
+  // @ts-expect-error This code is running in a sandbox where dd-trace is available
+} = require('dd-trace/packages/dd-trace/src/debugger/devtools_client/snapshot/constants')
 
 const fastify = Fastify({ logger: { level: 'error' } })
 
@@ -18,24 +22,31 @@ fastify.get('/deeply-nested-large-object', function handler () {
   return { paused: Number(diff) / 1_000_000 }
 })
 
-fastify.get('/large-object-of-primitives', function handler () {
+fastify.get('/object-with-many-properties', function handler () {
   const obj = generateObject(0, 1_000_000) // eslint-disable-line no-unused-vars
   const start = process.hrtime.bigint()
-  const diff = process.hrtime.bigint() - start // BREAKPOINT: /large-object-of-primitives
+  const diff = process.hrtime.bigint() - start // BREAKPOINT: /object-with-many-properties
   return { paused: Number(diff) / 1_000_000 }
 })
 
-fastify.get('/large-array-of-primitives', function handler () {
-  const arr = Array.from({ length: 1_000_000 }, (_, i) => i) // eslint-disable-line no-unused-vars
+fastify.get('/large-collections', function handler () {
+  const arrOfPrimitives = Array.from({ length: 1_000_000 }, (_, i) => i) // eslint-disable-line no-unused-vars
+  const arrOfObjects = Array.from({ length: 1_000_000 }, (_, i) => ({ i }))
+  const map = new Map(arrOfObjects.map((obj, i) => [i, obj])) // eslint-disable-line no-unused-vars
+  const set = new Set(arrOfObjects) // eslint-disable-line no-unused-vars
   const start = process.hrtime.bigint()
-  const diff = process.hrtime.bigint() - start // BREAKPOINT: /large-array-of-primitives
+  const diff = process.hrtime.bigint() - start // BREAKPOINT: /large-collections
   return { paused: Number(diff) / 1_000_000 }
 })
 
-fastify.get('/large-array-of-objects', function handler () {
-  const arr = Array.from({ length: 1_000_000 }, (_, i) => ({ i })) // eslint-disable-line no-unused-vars
+fastify.get('/large-collections-below-size-threshold', function handler () {
+  // eslint-disable-next-line no-unused-vars
+  const arrOfPrimitives = Array.from({ length: LARGE_OBJECT_SKIP_THRESHOLD - 1 }, (_, i) => i)
+  const arrOfObjects = Array.from({ length: LARGE_OBJECT_SKIP_THRESHOLD - 1 }, (_, i) => ({ i }))
+  const map = new Map(arrOfObjects.map((obj, i) => [i, obj])) // eslint-disable-line no-unused-vars
+  const set = new Set(arrOfObjects) // eslint-disable-line no-unused-vars
   const start = process.hrtime.bigint()
-  const diff = process.hrtime.bigint() - start // BREAKPOINT: /large-array-of-objects
+  const diff = process.hrtime.bigint() - start // BREAKPOINT: /large-collections-below-size-threshold
   return { paused: Number(diff) / 1_000_000 }
 })
 
diff --git a/integration-tests/debugger/utils.js b/integration-tests/debugger/utils.js
index a086f364e73..35d4d6e5394 100644
--- a/integration-tests/debugger/utils.js
+++ b/integration-tests/debugger/utils.js
@@ -41,7 +41,7 @@ const pollInterval = 0.1
  *
  * @typedef {BreakpointInfo & {
  *   rcConfig: object|null,
- *   triggerBreakpoint: (url: string) => Promise<import('axios').AxiosResponse<unknown>>,
+ *   triggerBreakpoint: () => Promise<import('axios').AxiosResponse<unknown>>,
  *   generateRemoteConfig: (overrides?: object) => object,
  *   generateProbeConfig: BoundGenerateProbeConfigFn
  * }} EnrichedBreakpoint
@@ -66,6 +66,7 @@ const pollInterval = 0.1
  *   once installed.
  * @property {(overrides?: object) => object} generateRemoteConfig - Generates RC for the primary breakpoint.
  * @property {BoundGenerateProbeConfigFn} generateProbeConfig - Generates probe config for the primary breakpoint.
+ * @property {() => Promise<object>} snapshotReceived - Waits for a snapshot to be received from the test app.
  */
 
 module.exports = {
@@ -129,7 +130,15 @@ function setup ({ env, testApp, testAppSource, dependencies, silent, stdioHandle
     rcConfig: null,
     triggerBreakpoint: triggerBreakpoint.bind(null, breakpoints[0].url),
     generateRemoteConfig: generateRemoteConfig.bind(null, breakpoints[0]),
-    generateProbeConfig: generateProbeConfig.bind(null, breakpoints[0])
+    generateProbeConfig: generateProbeConfig.bind(null, breakpoints[0]),
+
+    snapshotReceived () {
+      return new Promise((/** @type {(value: object) => void} */ resolve) => {
+        t.agent.on('debugger-input', ({ payload: [{ debugger: { snapshot } }] }) => {
+          resolve(snapshot)
+        })
+      })
+    }
   }
 
   /**
diff --git a/packages/dd-trace/src/ci-visibility/dynamic-instrumentation/worker/index.js b/packages/dd-trace/src/ci-visibility/dynamic-instrumentation/worker/index.js
index 4cf2ab66190..e4811e9f3dc 100644
--- a/packages/dd-trace/src/ci-visibility/dynamic-instrumentation/worker/index.js
+++ b/packages/dd-trace/src/ci-visibility/dynamic-instrumentation/worker/index.js
@@ -36,7 +36,7 @@ session.on('Debugger.paused', async ({ params: { hitBreakpoints: [hitBreakpoint]
 
   const stack = getStackFromCallFrames(callFrames)
 
-  const getLocalState = await getLocalStateForCallFrame(callFrames[0])
+  const { processLocalState } = await getLocalStateForCallFrame(callFrames[0])
 
   await session.post('Debugger.resume')
 
@@ -48,17 +48,13 @@ session.on('Debugger.paused', async ({ params: { hitBreakpoints: [hitBreakpoint]
       version: '0',
       location: probe.location
     },
+    captures: {
+      lines: { [probe.location.lines[0]]: { locals: processLocalState() } }
+    },
     stack,
     language: 'javascript'
   }
 
-  const state = getLocalState()
-  if (state) {
-    snapshot.captures = {
-      lines: { [probe.location.lines[0]]: { locals: state } }
-    }
-  }
-
   breakpointHitChannel.postMessage({ snapshot })
 })
 
diff --git a/packages/dd-trace/src/debugger/devtools_client/index.js b/packages/dd-trace/src/debugger/devtools_client/index.js
index 1458fd1c343..55f6f72b75f 100644
--- a/packages/dd-trace/src/debugger/devtools_client/index.js
+++ b/packages/dd-trace/src/debugger/devtools_client/index.js
@@ -166,16 +166,17 @@ session.on('Debugger.paused', async ({ params }) => {
   }
 
   // TODO: Create unique states for each affected probe based on that probes unique `capture` settings (DEBUG-2863)
-  const processLocalState = numberOfProbesWithSnapshots !== 0 && await getLocalStateForCallFrame(
-    params.callFrames[0],
-    {
+  let processLocalState, captureErrors
+  if (numberOfProbesWithSnapshots !== 0) {
+    const opts = {
       maxReferenceDepth,
       maxCollectionSize,
       maxFieldCount,
       maxLength,
       deadlineNs: start + config.dynamicInstrumentation.captureTimeoutNs
     }
-  )
+    ;({ processLocalState, captureErrors } = await getLocalStateForCallFrame(params.callFrames[0], opts))
+  }
 
   await session.post('Debugger.resume')
   const diff = process.hrtime.bigint() - start // TODO: Recorded as telemetry (DEBUG-2858)
@@ -215,16 +216,23 @@ session.on('Debugger.paused', async ({ params }) => {
     }
 
     if (probe.captureSnapshot) {
-      const state = processLocalState()
-      if (state instanceof Error) {
-        snapshot.captureError = state.message
-      } else if (state) {
-        snapshot.captures = {
-          lines: { [probe.location.lines[0]]: { locals: state } }
-        }
+      if (captureErrors?.length > 0) {
+        // There was an error collecting the snapshot for this probe, let's not try again
+        probe.captureSnapshot = false
+        probe.permanentEvaluationErrors = captureErrors.map(error => ({
+          expr: '',
+          message: error.message
+        }))
+      }
+      snapshot.captures = {
+        lines: { [probe.location.lines[0]]: { locals: processLocalState() } }
       }
     }
 
+    if (probe.permanentEvaluationErrors !== undefined) {
+      snapshot.evaluationErrors = [...probe.permanentEvaluationErrors]
+    }
+
     let message = ''
     if (probe.templateRequiresEvaluation) {
       const results = evalResults[messageIndex++]
diff --git a/packages/dd-trace/src/debugger/devtools_client/snapshot/collector.js b/packages/dd-trace/src/debugger/devtools_client/snapshot/collector.js
index 7c993329227..ff86cea70d5 100644
--- a/packages/dd-trace/src/debugger/devtools_client/snapshot/collector.js
+++ b/packages/dd-trace/src/debugger/devtools_client/snapshot/collector.js
@@ -1,10 +1,12 @@
 'use strict'
 
-const { collectionSizeSym, fieldCountSym, timeBudgetSym } = require('./symbols')
+const { collectionSizeSym, largeCollectionSkipThresholdSym, fieldCountSym, timeBudgetSym } = require('./symbols')
+const { LARGE_OBJECT_SKIP_THRESHOLD } = require('./constants')
 const session = require('../session')
 
 const LEAF_SUBTYPES = new Set(['date', 'regexp'])
 const ITERABLE_SUBTYPES = new Set(['map', 'set', 'weakmap', 'weakset'])
+const SIZE_IN_DESCRIPTION_SUBTYPES = new Set(['array', 'typedarray', 'arraybuffer', 'dataview', 'map', 'set'])
 
 module.exports = {
   collectObjectProperties
@@ -16,9 +18,10 @@ module.exports = {
  * @property {number} maxCollectionSize - The maximum size of a collection to include in the snapshot
  * @property {number} maxFieldCount - The maximum number of properties on an object to include in the snapshot
  * @property {bigint} deadlineNs - The deadline in nanoseconds compared to `process.hrtime.bigint()`
- * @property {Object} [ctx={}] - A context object to track the state/progress of the snapshot collection.
- * @property {boolean} [ctx.deadlineReached=false] - Whether the deadline has been reached. Should not be set by the
- *   caller, but is used to signal the deadline overrun to the caller.
+ * @property {Object} ctx - A context object to track the state/progress of the snapshot collection.
+ * @property {boolean} ctx.deadlineReached - Will be set to `true` if the deadline has been reached.
+ * @property {Error[]} ctx.captureErrors - An array on which errors can be pushed if an issue is detected while
+ *   collecting the snapshot.
  */
 
 /**
@@ -51,6 +54,13 @@ async function collectObjectProperties (objectId, opts, depth = 0, collection =
   } else if (result.length > opts.maxFieldCount) {
     // Trim the number of properties on the object if there's too many.
     const size = result.length
+    if (size > LARGE_OBJECT_SKIP_THRESHOLD) {
+      opts.ctx.captureErrors.push(new Error(
+        `An object with ${size} properties was detected while collecting a snapshot. ` +
+        `This exceeds the maximum number of allowed properties of ${LARGE_OBJECT_SKIP_THRESHOLD}. ` +
+        'Future snapshots for existing probes in this location will be skipped until the Node.js process is restarted'
+      ))
+    }
     result.length = opts.maxFieldCount
     result[fieldCountSym] = size
   } else if (privateProperties) {
@@ -70,10 +80,17 @@ async function traverseGetPropertiesResult (props, opts, depth) {
 
   for (const prop of props) {
     if (prop.value === undefined) continue
-    const { value: { type, objectId, subtype } } = prop
+    const { value: { type, objectId, subtype, description } } = prop
     if (type === 'object') {
       if (objectId === undefined) continue // if `subtype` is "null"
       if (LEAF_SUBTYPES.has(subtype)) continue // don't waste time with these subtypes
+      const size = parseLengthFromDescription(description, subtype)
+      if (size !== null && size >= LARGE_OBJECT_SKIP_THRESHOLD) {
+        const empty = []
+        empty[largeCollectionSkipThresholdSym] = size
+        prop.value.properties = empty
+        continue
+      }
       work.push([
         prop.value,
         () => collectPropertiesBySubtype(subtype, objectId, opts, depth).then((properties) => {
@@ -235,6 +252,26 @@ function removeNonEnumerableProperties (props) {
   }
 }
 
+function parseLengthFromDescription (description, subtype) {
+  if (typeof description !== 'string') return null
+  if (!SIZE_IN_DESCRIPTION_SUBTYPES.has(subtype)) return null
+
+  const open = description.lastIndexOf('(')
+  if (open === -1) return null
+
+  const close = description.indexOf(')', open + 1)
+  if (close === -1) return null
+
+  const s = description.slice(open + 1, close)
+  if (s === '') return null
+
+  const n = Number(s)
+  if (!Number.isSafeInteger(n) || n < 0) return null
+  if (String(n) !== s) return null
+
+  return n
+}
+
 function overBudget (opts) {
   if (opts.ctx.deadlineReached) return true
   opts.ctx.deadlineReached = process.hrtime.bigint() >= opts.deadlineNs
diff --git a/packages/dd-trace/src/debugger/devtools_client/snapshot/constants.js b/packages/dd-trace/src/debugger/devtools_client/snapshot/constants.js
new file mode 100644
index 00000000000..1856f9242a6
--- /dev/null
+++ b/packages/dd-trace/src/debugger/devtools_client/snapshot/constants.js
@@ -0,0 +1,25 @@
+'use strict'
+
+const { getEnvironmentVariable } = require('../../../config-helper')
+
+const largeObjectSkipThreshold = Number(
+  getEnvironmentVariable('_DD_DYNAMIC_INSTRUMENTATION_EXPERIMENTAL_LARGE_OBJECT_SKIP_THRESHOLD')
+)
+
+module.exports = {
+  /**
+   * When collecting a snapshot, this constant controls what happens when objects with a large number of properties or
+   * collections (arrays, maps, sets, etc.) with a large number of elements are detected:
+   *
+   * - If a collection is detected with more than this number of elements, none of its elements will be included in the
+   *   snapshot.
+   * - If an object is detected with more than this number of properties, it will be included in the snapshot, but
+   *   snapshotting will be turned off for that probe in the future, until the probe is either updated or the Node.js
+   *   process is restarted.
+   */
+  LARGE_OBJECT_SKIP_THRESHOLD: Number.isNaN(largeObjectSkipThreshold) ? 500 : largeObjectSkipThreshold,
+  DEFAULT_MAX_COLLECTION_SIZE: 100,
+  DEFAULT_MAX_FIELD_COUNT: 20,
+  DEFAULT_MAX_LENGTH: 255,
+  DEFAULT_MAX_REFERENCE_DEPTH: 3,
+}
diff --git a/packages/dd-trace/src/debugger/devtools_client/snapshot/index.js b/packages/dd-trace/src/debugger/devtools_client/snapshot/index.js
index dec181f6ef0..a05a3f6165d 100644
--- a/packages/dd-trace/src/debugger/devtools_client/snapshot/index.js
+++ b/packages/dd-trace/src/debugger/devtools_client/snapshot/index.js
@@ -1,24 +1,20 @@
 'use strict'
 
+const {
+  DEFAULT_MAX_REFERENCE_DEPTH,
+  DEFAULT_MAX_COLLECTION_SIZE,
+  DEFAULT_MAX_FIELD_COUNT,
+  DEFAULT_MAX_LENGTH
+} = require('./constants')
 const { collectObjectProperties } = require('./collector')
 const { processRawState } = require('./processor')
-const log = require('../log')
 
 const BIGINT_MAX = (1n << 256n) - 1n
 
-const DEFAULT_MAX_REFERENCE_DEPTH = 3
-const DEFAULT_MAX_COLLECTION_SIZE = 100
-const DEFAULT_MAX_FIELD_COUNT = 20
-const DEFAULT_MAX_LENGTH = 255
-
 module.exports = {
   getLocalStateForCallFrame
 }
 
-function returnError () {
-  return new Error('Error getting local state')
-}
-
 /**
  * @typedef {Object} GetLocalStateForCallFrameOptions
  * @property {number} [maxReferenceDepth] - The maximum depth of the object to traverse. Defaults to
@@ -50,38 +46,38 @@ async function getLocalStateForCallFrame (
     deadlineNs = BIGINT_MAX
   } = {}
 ) {
+  /** @type {{ deadlineReached: boolean, captureErrors: Error[] }} */
+  const ctx = { deadlineReached: false, captureErrors: [] }
+  const opts = { maxReferenceDepth, maxCollectionSize, maxFieldCount, deadlineNs, ctx }
   const rawState = []
   let processedState = null
 
-  try {
-    const opts = {
-      maxReferenceDepth,
-      maxCollectionSize,
-      maxFieldCount,
-      deadlineNs,
-      ctx: { deadlineReached: false }
-    }
-    for (const scope of callFrame.scopeChain) {
-      if (scope.type === 'global') continue // The global scope is too noisy
-      const { objectId } = scope.object
-      if (objectId === undefined) continue // I haven't seen this happen, but according to the types it's possible
+  for (const scope of callFrame.scopeChain) {
+    if (scope.type === 'global') continue // The global scope is too noisy
+    const { objectId } = scope.object
+    if (objectId === undefined) continue // I haven't seen this happen, but according to the types it's possible
+    try {
       // The objectId for a scope points to a pseudo-object whose properties are the actual variables in the scope.
       // This is why we can just call `collectObjectProperties` directly and expect it to return the in-scope variables
       // as an array.
       // eslint-disable-next-line no-await-in-loop
       rawState.push(...await collectObjectProperties(objectId, opts))
-      if (opts.ctx.deadlineReached === true) break // TODO: Bad UX; Variables in remaining scopes are silently dropped
+    } catch (err) {
+      ctx.captureErrors.push(new Error(
+        `Error getting local state for closure scope (type: ${scope.type}). ` +
+        'Future snapshots for existing probes in this location will be skipped until the Node.js process is restarted',
+        { cause: err } // TODO: The cause is not used by the backend
+      ))
     }
-  } catch (err) {
-    // TODO: We might be able to get part of the scope chain.
-    // Consider if we could set errors just for the part of the scope chain that throws during collection.
-    log.error('[debugger:devtools_client] Error getting local state for call frame', err)
-    return returnError
+    if (ctx.deadlineReached === true) break // TODO: Bad UX; Variables in remaining scopes are silently dropped
   }
 
   // Delay calling `processRawState` so the caller gets a chance to resume the main thread before processing `rawState`
-  return () => {
-    processedState = processedState ?? processRawState(rawState, maxLength)
-    return processedState
+  return {
+    processLocalState () {
+      processedState = processedState ?? processRawState(rawState, maxLength)
+      return processedState
+    },
+    captureErrors: ctx.captureErrors
   }
 }
diff --git a/packages/dd-trace/src/debugger/devtools_client/snapshot/processor.js b/packages/dd-trace/src/debugger/devtools_client/snapshot/processor.js
index fd31d5f187a..d824d829c93 100644
--- a/packages/dd-trace/src/debugger/devtools_client/snapshot/processor.js
+++ b/packages/dd-trace/src/debugger/devtools_client/snapshot/processor.js
@@ -1,6 +1,7 @@
 'use strict'
 
-const { collectionSizeSym, fieldCountSym, timeBudgetSym } = require('./symbols')
+const { LARGE_OBJECT_SKIP_THRESHOLD } = require('./constants')
+const { collectionSizeSym, largeCollectionSkipThresholdSym, fieldCountSym, timeBudgetSym } = require('./symbols')
 const { normalizeName, REDACTED_IDENTIFIERS } = require('./redaction')
 
 module.exports = {
@@ -182,6 +183,7 @@ function toArray (type, elements, maxLength, timeBudgetReached) {
   }
 
   setNotCaptureReasonOnCollection(result, elements)
+  setNotCaptureReasonOnTooLargeCollection(result, elements)
 
   return result
 }
@@ -202,6 +204,14 @@ function toMap (type, pairs, maxLength, timeBudgetReached) {
       // This can be skipped and we can go directly to its children, of which
       // there will always be exactly two, the first containing the key, and the
       // second containing the value of this entry of the Map.
+
+      if (value[timeBudgetSym] === true) {
+        return [{ notCapturedReason: 'timeout' }, { notCapturedReason: 'timeout' }]
+      }
+      if (value.properties === undefined) {
+        return [{ notCapturedReason: 'unknown' }, { notCapturedReason: 'unknown' }]
+      }
+
       const shouldRedact = shouldRedactMapValue(value.properties[0])
       const key = getPropertyValue(value.properties[0], maxLength)
       const val = shouldRedact
@@ -212,6 +222,7 @@ function toMap (type, pairs, maxLength, timeBudgetReached) {
   }
 
   setNotCaptureReasonOnCollection(result, pairs)
+  setNotCaptureReasonOnTooLargeCollection(result, pairs)
 
   return result
 }
@@ -234,11 +245,16 @@ function toSet (type, values, maxLength, timeBudgetReached) {
       // `internal#entry`. This can be skipped and we can go directly to its
       // children, of which there will always be exactly one, which contain the
       // actual value in this entry of the Set.
+
+      if (value[timeBudgetSym] === true) return { notCapturedReason: 'timeout' }
+      if (value.properties === undefined) return { notCapturedReason: 'unknown' }
+
       return getPropertyValue(value.properties[0], maxLength)
     })
   }
 
   setNotCaptureReasonOnCollection(result, values)
+  setNotCaptureReasonOnTooLargeCollection(result, values)
 
   return result
 }
@@ -293,6 +309,15 @@ function setNotCaptureReasonOnCollection (result, collection) {
   }
 }
 
+function setNotCaptureReasonOnTooLargeCollection (result, collection) {
+  if (collection[largeCollectionSkipThresholdSym] !== undefined) {
+    result.notCapturedReason = `Large collection with too many elements (skip threshold: ${
+      LARGE_OBJECT_SKIP_THRESHOLD
+    })`
+    result.size = collection[largeCollectionSkipThresholdSym]
+  }
+}
+
 function notCapturedDepth (type) {
   return { type, notCapturedReason: 'depth' }
 }
diff --git a/packages/dd-trace/src/debugger/devtools_client/snapshot/symbols.js b/packages/dd-trace/src/debugger/devtools_client/snapshot/symbols.js
index a912b54e86f..69de8e568bb 100644
--- a/packages/dd-trace/src/debugger/devtools_client/snapshot/symbols.js
+++ b/packages/dd-trace/src/debugger/devtools_client/snapshot/symbols.js
@@ -2,6 +2,7 @@
 
 module.exports = {
   collectionSizeSym: Symbol('datadog.collectionSize'),
+  largeCollectionSkipThresholdSym: Symbol('datadog.largeCollectionSkipThresholdSym'),
   fieldCountSym: Symbol('datadog.fieldCount'),
   timeBudgetSym: Symbol('datadog.timeout')
 }
diff --git a/packages/dd-trace/test/debugger/devtools_client/snapshot/complex-types.spec.js b/packages/dd-trace/test/debugger/devtools_client/snapshot/complex-types.spec.js
index 19c3e53f77b..bafd7e85724 100644
--- a/packages/dd-trace/test/debugger/devtools_client/snapshot/complex-types.spec.js
+++ b/packages/dd-trace/test/debugger/devtools_client/snapshot/complex-types.spec.js
@@ -38,7 +38,7 @@ describe('debugger -> devtools client -> snapshot.getLocalStateForCallFrame', fu
         resolve((await getLocalStateForCallFrame(
           params.callFrames[0],
           { maxFieldCount: Number.MAX_SAFE_INTEGER })
-        )())
+        ).processLocalState())
       })
 
       await setAndTriggerBreakpoint(target, 10)
diff --git a/packages/dd-trace/test/debugger/devtools_client/snapshot/error-handling.spec.js b/packages/dd-trace/test/debugger/devtools_client/snapshot/error-handling.spec.js
index 7099b0e7c3d..1245bdb14d2 100644
--- a/packages/dd-trace/test/debugger/devtools_client/snapshot/error-handling.spec.js
+++ b/packages/dd-trace/test/debugger/devtools_client/snapshot/error-handling.spec.js
@@ -8,12 +8,29 @@ const { getLocalStateForCallFrame } = require('./utils')
 
 describe('debugger -> devtools client -> snapshot.getLocalStateForCallFrame', function () {
   describe('error handling', function () {
-    it('should generate a notCapturedReason if an error is thrown during initial collection', async function () {
-      const invalidCallFrameThatTriggersAnException = {}
-      const processLocalState = await getLocalStateForCallFrame(invalidCallFrameThatTriggersAnException)
-      const result = processLocalState()
-      assert.ok(result instanceof Error)
-      assert.strictEqual(result.message, 'Error getting local state')
+    it('should generate a notCapturedReason if getRuntimeObject rejects', async function () {
+      const mockCallFrame = {
+        // `42` isn't a valid object id, so we should get an error
+        scopeChain: [{ type: 'local', object: { objectId: '42' } }]
+      }
+      const { captureErrors, processLocalState } = await getLocalStateForCallFrame(mockCallFrame)
+
+      assert.strictEqual(captureErrors.length, 1)
+
+      for (const error of captureErrors) {
+        assert.ok(error instanceof Error)
+        assert.strictEqual(
+          error.message,
+          'Error getting local state for closure scope (type: local). ' +
+          'Future snapshots for existing probes in this location will be skipped until the Node.js process is restarted'
+        )
+
+        const { cause } = error
+        assert.ok(cause instanceof Error)
+        assert.strictEqual(cause.message, 'Inspector error -32000: Invalid remote object id')
+      }
+
+      assert.deepStrictEqual(processLocalState(), {})
     })
   })
 })
diff --git a/packages/dd-trace/test/debugger/devtools_client/snapshot/max-collection-size.spec.js b/packages/dd-trace/test/debugger/devtools_client/snapshot/max-collection-size.spec.js
index 0fe078f89a1..a3735a92d63 100644
--- a/packages/dd-trace/test/debugger/devtools_client/snapshot/max-collection-size.spec.js
+++ b/packages/dd-trace/test/debugger/devtools_client/snapshot/max-collection-size.spec.js
@@ -9,8 +9,11 @@ const { assertObjectContains } = require('../../../../../../integration-tests/he
 require('../../../setup/mocha')
 
 const { getTargetCodePath, enable, teardown, assertOnBreakpoint, setAndTriggerBreakpoint } = require('./utils')
+const {
+  LARGE_OBJECT_SKIP_THRESHOLD,
+  DEFAULT_MAX_COLLECTION_SIZE
+} = require('../../../../src/debugger/devtools_client/snapshot/constants')
 
-const DEFAULT_MAX_COLLECTION_SIZE = 100
 const target = getTargetCodePath(__filename)
 
 describe('debugger -> devtools client -> snapshot.getLocalStateForCallFrame', function () {
@@ -48,11 +51,11 @@ describe('debugger -> devtools client -> snapshot.getLocalStateForCallFrame', fu
           assertOnBreakpoint(done, config, (_state) => {
             state = _state
           })
-          setAndTriggerBreakpoint(target, 24)
+          setAndTriggerBreakpoint(target, 29)
         })
 
         it('should have expected number of elements in state', function () {
-          expect(state).to.have.keys(['arr', 'map', 'set', 'wmap', 'wset', 'typedArray'])
+          expect(state).to.have.keys(['LARGE_SIZE', 'arr', 'map', 'set', 'wmap', 'wset', 'typedArray'])
         })
 
         it('Array', function () {
@@ -60,7 +63,7 @@ describe('debugger -> devtools client -> snapshot.getLocalStateForCallFrame', fu
             type: 'Array',
             elements: expectedElements,
             notCapturedReason: 'collectionSize',
-            size: 1000
+            size: LARGE_OBJECT_SKIP_THRESHOLD - 1
           })
         })
 
@@ -69,7 +72,7 @@ describe('debugger -> devtools client -> snapshot.getLocalStateForCallFrame', fu
             type: 'Map',
             entries: expectedEntries,
             notCapturedReason: 'collectionSize',
-            size: 1000
+            size: LARGE_OBJECT_SKIP_THRESHOLD - 1
           })
         })
 
@@ -78,7 +81,7 @@ describe('debugger -> devtools client -> snapshot.getLocalStateForCallFrame', fu
             type: 'Set',
             elements: expectedElements,
             notCapturedReason: 'collectionSize',
-            size: 1000
+            size: LARGE_OBJECT_SKIP_THRESHOLD - 1
           })
         })
 
@@ -86,7 +89,7 @@ describe('debugger -> devtools client -> snapshot.getLocalStateForCallFrame', fu
           assertObjectContains(state.wmap, {
             type: 'WeakMap',
             notCapturedReason: 'collectionSize',
-            size: 1000
+            size: LARGE_OBJECT_SKIP_THRESHOLD - 1
           })
 
           assert.strictEqual(state.wmap.entries.length, maxCollectionSize)
@@ -108,7 +111,7 @@ describe('debugger -> devtools client -> snapshot.getLocalStateForCallFrame', fu
           assertObjectContains(state.wset, {
             type: 'WeakSet',
             notCapturedReason: 'collectionSize',
-            size: 1000
+            size: LARGE_OBJECT_SKIP_THRESHOLD - 1
           })
 
           assert.strictEqual(state.wset.elements.length, maxCollectionSize)
@@ -128,7 +131,7 @@ describe('debugger -> devtools client -> snapshot.getLocalStateForCallFrame', fu
             type: 'Uint16Array',
             elements: expectedElements,
             notCapturedReason: 'collectionSize',
-            size: 1000
+            size: LARGE_OBJECT_SKIP_THRESHOLD - 1
           })
         })
       })
diff --git a/packages/dd-trace/test/debugger/devtools_client/snapshot/max-field-count.spec.js b/packages/dd-trace/test/debugger/devtools_client/snapshot/max-field-count.spec.js
index e331fe15959..2b907bb6d96 100644
--- a/packages/dd-trace/test/debugger/devtools_client/snapshot/max-field-count.spec.js
+++ b/packages/dd-trace/test/debugger/devtools_client/snapshot/max-field-count.spec.js
@@ -6,8 +6,8 @@ const { describe, it, beforeEach, afterEach } = require('mocha')
 require('../../../setup/mocha')
 
 const { getTargetCodePath, enable, teardown, assertOnBreakpoint, setAndTriggerBreakpoint } = require('./utils')
+const { DEFAULT_MAX_FIELD_COUNT } = require('../../../../src/debugger/devtools_client/snapshot/constants')
 
-const DEFAULT_MAX_FIELD_COUNT = 20
 const target = getTargetCodePath(__filename)
 
 describe('debugger -> devtools client -> snapshot.getLocalStateForCallFrame', function () {
diff --git a/packages/dd-trace/test/debugger/devtools_client/snapshot/target-code/max-collection-size.js b/packages/dd-trace/test/debugger/devtools_client/snapshot/target-code/max-collection-size.js
index 69f0af85afe..f394d71c64a 100644
--- a/packages/dd-trace/test/debugger/devtools_client/snapshot/target-code/max-collection-size.js
+++ b/packages/dd-trace/test/debugger/devtools_client/snapshot/target-code/max-collection-size.js
@@ -1,15 +1,20 @@
 'use strict'
 
+const { LARGE_OBJECT_SKIP_THRESHOLD } = require('../../../../../src/debugger/devtools_client/snapshot/constants')
+
+// `LARGE_SIZE` is larger than the default maxCollectionSize, but lower than the large object skip threshold, after
+// which nothing is captured.
+const LARGE_SIZE = LARGE_OBJECT_SKIP_THRESHOLD - 1
+
 function run () {
   const arr = []
   const map = new Map()
   const set = new Set()
   const wmap = new WeakMap()
   const wset = new WeakSet()
-  const typedArray = new Uint16Array(new ArrayBuffer(2000))
+  const typedArray = new Uint16Array(new ArrayBuffer(LARGE_SIZE * 2))
 
-  // 1000 is larger the default maxCollectionSize of 100
-  for (let i = 1; i <= 1000; i++) {
+  for (let i = 1; i <= LARGE_SIZE; i++) {
     // A reference that can be used in WeakMap/WeakSet to avoid GC
     const obj = { i }
 
diff --git a/packages/dd-trace/test/debugger/devtools_client/snapshot/utils.js b/packages/dd-trace/test/debugger/devtools_client/snapshot/utils.js
index 3c7b6580ae8..998348acbd7 100644
--- a/packages/dd-trace/test/debugger/devtools_client/snapshot/utils.js
+++ b/packages/dd-trace/test/debugger/devtools_client/snapshot/utils.js
@@ -103,8 +103,8 @@ function assertOnBreakpoint (done, snapshotConfig, callback) {
   session.once('Debugger.paused', ({ params }) => {
     assert.strictEqual(params.hitBreakpoints.length, 1)
 
-    getLocalStateForCallFrame(params.callFrames[0], snapshotConfig).then((process) => {
-      callback(process())
+    getLocalStateForCallFrame(params.callFrames[0], snapshotConfig).then(({ processLocalState }) => {
+      callback(processLocalState())
       done()
     }).catch(done)
   })