diff --git a/benchmark/sirun/debugger/benchmark-worker.js b/benchmark/sirun/debugger/benchmark-worker.js
index be7fa20f548..f35de82b1bc 100644
--- a/benchmark/sirun/debugger/benchmark-worker.js
+++ b/benchmark/sirun/debugger/benchmark-worker.js
@@ -92,15 +92,17 @@ function markProbeHandled () {
  * @param {Record<string, unknown> | undefined} dd
  * @param {DebuggerSnapshot} snapshot
  * @param {string | undefined} processTags
+ * @param {number} eventType
+ * @param {number} incompleteReasons
  * @returns {void}
  */
-function sendAndCount (message, logger, dd, snapshot, processTags) {
+function sendAndCount (message, logger, dd, snapshot, processTags, eventType, incompleteReasons) {
   const captureKind = getCaptureKind(snapshot)
   if (preflightPending) {
     preflightPending = false
     Atomics.store(probeCounts, CAPTURE_KIND_INDEX, captureKind)
   } else {
-    send(message, logger, dd, snapshot, processTags)
+    send(message, logger, dd, snapshot, processTags, eventType, incompleteReasons)
     if (captureKind === EXPECTED_CAPTURE_KIND) Atomics.add(probeCounts, MATCHED_CAPTURE_KIND_INDEX, 1)
   }
 
diff --git a/integration-tests/debugger/guardrail-metrics.spec.js b/integration-tests/debugger/guardrail-metrics.spec.js
new file mode 100644
index 00000000000..25f6e7f62c9
--- /dev/null
+++ b/integration-tests/debugger/guardrail-metrics.spec.js
@@ -0,0 +1,103 @@
+'use strict'
+
+const assert = require('node:assert/strict')
+const { inspect } = require('node:util')
+
+const { setup } = require('./utils')
+
+// The guardrail counters are converted into telemetry metrics every 10 seconds, which are then sent on the next
+// telemetry heartbeat
+const GUARDRAIL_METRICS_FLUSH_INTERVAL_MS = 10_000
+const TELEMETRY_HEARTBEAT_INTERVAL_SECONDS = 1
+
+describe('Dynamic Instrumentation', function () {
+  const t = setup({
+    testApp: 'target-app/basic.js',
+    dependencies: ['fastify'],
+    env: {
+      DD_TELEMETRY_HEARTBEAT_INTERVAL: String(TELEMETRY_HEARTBEAT_INTERVAL_SECONDS),
+      // The app-started event is sent before the test can listen for it, but the extended heartbeat repeats its
+      // application payload
+      DD_TELEMETRY_EXTENDED_HEARTBEAT_INTERVAL: String(TELEMETRY_HEARTBEAT_INTERVAL_SECONDS),
+    },
+  })
+
+  describe('guardrail telemetry', function () {
+    this.timeout(GUARDRAIL_METRICS_FLUSH_INTERVAL_MS * 3)
+
+    it('should report Dynamic Instrumentation as an enabled product', async function () {
+      await t.agent.assertTelemetryReceived({
+        fn: ({ payload }) => {
+          assert.deepStrictEqual(payload.payload.products.dynamic_instrumentation, { enabled: true })
+        },
+        requestType: 'app-extended-heartbeat',
+      })
+    })
+
+    it('should report skipped events and incomplete captures', async function () {
+      // A log probe that is hit far more often than its per-probe rate limit allows
+      const rateLimitedProbe = t.breakpoints[0].generateRemoteConfig({ sampling: { snapshotsPerSecond: 1 } })
+      // A snapshot probe that cannot capture the request object within its reference depth limit
+      const shallowSnapshotProbe = t.breakpoints[1].generateRemoteConfig({
+        captureSnapshot: true,
+        capture: { maxReferenceDepth: 0 },
+      })
+
+      const installed = new Set()
+      const allInstalled = new Promise((/** @type {(value?: void) => void} */ resolve) => {
+        t.agent.on('debugger-diagnostics', ({ payload }) => {
+          for (const { debugger: { diagnostics: { probeId, status } } } of payload) {
+            if (status === 'INSTALLED') installed.add(probeId)
+          }
+          if (installed.size === 2) resolve()
+        })
+      })
+
+      const checkMetrics = t.agent.assertTelemetryReceived({
+        fn: ({ payload }) => {
+          const { series } = payload.payload
+
+          const skipped = findMetric(series, 'events.skipped', ['event_type:log', 'reason:rateLimitProbe'])
+          assert.strictEqual(skipped.type, 'count')
+          assert.strictEqual(skipped.common, true)
+          assert.strictEqual(skipped.points.length, 1)
+          assert.ok(skipped.points[0][1] >= 1, `Expected ${skipped.points[0][1]} >= 1`)
+
+          const incomplete = findMetric(series, 'capture.incomplete', ['event_type:snapshot', 'reason:depth'])
+          assert.strictEqual(incomplete.type, 'count')
+          assert.strictEqual(incomplete.common, true)
+          assert.strictEqual(incomplete.points.length, 1)
+          assert.strictEqual(incomplete.points[0][1], 1)
+        },
+        requestType: 'generate-metrics',
+        timeout: GUARDRAIL_METRICS_FLUSH_INTERVAL_MS * 2,
+        resolveAtFirstSuccess: true,
+        namespace: 'live_debugger',
+      })
+
+      t.agent.addRemoteConfig(rateLimitedProbe)
+      t.agent.addRemoteConfig(shallowSnapshotProbe)
+      await allInstalled
+
+      // Trigger the rate limited probe well within a single second, so all but the first hit are skipped
+      await Promise.all(Array.from({ length: 5 }, () => t.request(t.breakpoints[0].url)))
+      await t.request(t.breakpoints[1].url)
+
+      await checkMetrics
+    })
+  })
+})
+
+/**
+ * @param {Array<{ metric: string, tags: string[] }>} series - The telemetry metric series
+ * @param {string} metric - The metric name to find
+ * @param {string[]} tags - The exact tags the metric must carry
+ * @returns {object} The matching series entry
+ */
+function findMetric (series, metric, tags) {
+  const match = series.find((entry) => {
+    return entry.metric === metric && entry.tags.length === tags.length && tags.every((tag) => entry.tags.includes(tag))
+  })
+  assert.ok(match, `Expected metric ${metric} with tags ${inspect(tags)} in ${inspect(series)}`)
+  return match
+}
diff --git a/packages/dd-trace/src/debugger/devtools_client/guardrail-metrics.js b/packages/dd-trace/src/debugger/devtools_client/guardrail-metrics.js
new file mode 100644
index 00000000000..36c7242e6be
--- /dev/null
+++ b/packages/dd-trace/src/debugger/devtools_client/guardrail-metrics.js
@@ -0,0 +1,15 @@
+'use strict'
+
+const { workerData } = require('node:worker_threads')
+
+const { GuardrailMetrics } = require('../guardrail-metrics')
+
+// For testing purposes, we allow `workerData` to be undefined and fallback to counters that are never drained
+const buffer = workerData?.guardrailMetricsBuffer ?? GuardrailMetrics.createBuffer()
+
+/**
+ * The worker's view of the guardrail counters shared with the main thread, which drains them into telemetry.
+ *
+ * @type {GuardrailMetrics}
+ */
+module.exports = new GuardrailMetrics(buffer)
diff --git a/packages/dd-trace/src/debugger/devtools_client/index.js b/packages/dd-trace/src/debugger/devtools_client/index.js
index 7f07e134b6c..076720c93ce 100644
--- a/packages/dd-trace/src/debugger/devtools_client/index.js
+++ b/packages/dd-trace/src/debugger/devtools_client/index.js
@@ -5,6 +5,7 @@ const { workerData: { probeSamplerBuffer } } = require('worker_threads')
 const { version } = require('../../../../../package.json')
 const processTags = require('../../process-tags')
 const { INSPECT_SEGMENT_GLOBAL_PROPERTY } = require('../constants')
+const { EVENT_TYPE, INCOMPLETE_REASON } = require('../guardrail-metrics')
 const {
   MAX_SAMPLED_PROBES_PER_PAUSE,
   SAMPLED_PROBE_COUNT_INDEX,
@@ -140,17 +141,14 @@ session.on('Debugger.paused', async ({ params }) => {
   }
 
   // TODO: Create unique states for each affected probe based on that probes unique `capture` settings (DEBUG-2863)
-  let processLocalState
-  /** @type {Error[] | undefined} */
-  let fatalSnapshotErrors
+  /** @type {Awaited<ReturnType<typeof getLocalStateForCallFrame>> | undefined} */
+  let localState
   if (numberOfProbesWithSnapshots !== 0) {
-    const result = await getLocalStateForCallFrame(
+    localState = await getLocalStateForCallFrame(
       params.callFrames[0],
       { maxReferenceDepth, maxCollectionSize, maxFieldCount, maxLength },
       start + config.dynamicInstrumentation.captureTimeoutNs
     )
-    processLocalState = result.processLocalState
-    fatalSnapshotErrors = result.fatalErrors
   }
 
   // Evaluate capture expressions for probes that have them
@@ -206,19 +204,30 @@ session.on('Debugger.paused', async ({ params }) => {
       language: 'javascript',
     }
 
+    // Which guardrail bucket the event belongs to, and which capture limits were enforced while producing it. The
+    // snapshot module records the reasons, including runtime errors: a fatal error does not necessarily mean one, as
+    // the collector also raises a fatal error to disable capture when it hits its large object safety threshold.
+    /** @type {number} */
+    let eventType = EVENT_TYPE.LOG
+    let incompleteReasons = 0
+
     if (probe.captureSnapshot) {
-      if (fatalSnapshotErrors && fatalSnapshotErrors.length > 0) {
+      eventType = EVENT_TYPE.SNAPSHOT
+      const { processLocalState, fatalErrors, incomplete } = /** @type {NonNullable<typeof localState>} */ (localState)
+      if (fatalErrors.length > 0) {
         // There was an error collecting the snapshot for this probe, let's not try again
         probe.captureSnapshot = false
-        probe.permanentEvaluationErrors = fatalSnapshotErrors.map(error => ({
+        probe.permanentEvaluationErrors = fatalErrors.map(error => ({
           expr: '',
           message: error.message,
         }))
       }
       snapshot.captures = {
-        lines: { [probe.location.lines[0]]: { locals: /** @type {Function} */ (processLocalState)() } },
+        lines: { [probe.location.lines[0]]: { locals: processLocalState() } },
       }
+      incompleteReasons |= incomplete.reasons
     } else if (probe.compiledCaptureExpressions !== undefined) {
+      eventType = EVENT_TYPE.SNAPSHOT
       const expressionResult = /** @type {Map} */ (captureExpressionResults).get(probe.id)
       if (expressionResult) {
         // Handle fatal capture errors - disable capture expressions for this probe permanently
@@ -233,6 +242,7 @@ session.on('Debugger.paused', async ({ params }) => {
         snapshot.captures = {
           lines: { [probe.location.lines[0]]: { captureExpressions: expressionResult.processCaptureExpressions() } },
         }
+        incompleteReasons |= expressionResult.incomplete.reasons
 
         // Handle transient evaluation errors - include in snapshot for this capture
         if (expressionResult.evaluationErrors?.length > 0) {
@@ -249,6 +259,7 @@ session.on('Debugger.paused', async ({ params }) => {
           expr: '',
           message: 'Internal error: capture expression results not found',
         }]
+        incompleteReasons |= INCOMPLETE_REASON.RUNTIME_ERROR
       }
     }
 
@@ -283,7 +294,8 @@ session.on('Debugger.paused', async ({ params }) => {
     ackEmitting(probe)
 
     send(message, logger, dd, snapshot,
-      config.propagateProcessTags.enabled ? processTags.serialized : undefined)
+      config.propagateProcessTags.enabled ? processTags.serialized : undefined,
+      eventType, incompleteReasons)
   }
 })
 
diff --git a/packages/dd-trace/src/debugger/devtools_client/send.js b/packages/dd-trace/src/debugger/devtools_client/send.js
index 2f594c96ae6..9d66f55f4c4 100644
--- a/packages/dd-trace/src/debugger/devtools_client/send.js
+++ b/packages/dd-trace/src/debugger/devtools_client/send.js
@@ -6,9 +6,11 @@ const { stringify } = require('querystring')
 const { version } = require('../../../../../package.json')
 const request = require('../../exporters/common/request')
 const { DEBUGGER_DIAGNOSTICS_V1, DEBUGGER_INPUT_V2 } = require('../constants')
+const { INCOMPLETE_REASON } = require('../guardrail-metrics')
 const log = require('./log')
 const JSONBuffer = require('./json-buffer')
 const config = require('./config')
+const guardrailMetrics = require('./guardrail-metrics')
 const getRequestOptions = require('./request-options')
 const { pruneSnapshot } = require('./snapshot-pruner')
 const buildTags = require('./tags')
@@ -34,14 +36,28 @@ const jsonBuffer = new JSONBuffer({
   onFlush,
 })
 
-function send (message, logger, dd, snapshot, processTags) {
+/**
+ * Queue a probe result for upload.
+ *
+ * @param {string} message - The evaluated log message
+ * @param {object} logger - The logger metadata
+ * @param {object | undefined} dd - The trace and span ids of the active trace, if any
+ * @param {object} snapshot - The snapshot payload
+ * @param {string | undefined} processTags - The serialized process tags, if enabled
+ * @param {number} eventType - The guardrail event type, one of `EVENT_TYPE`
+ * @param {number} incompleteReasons - Bitmask of `INCOMPLETE_REASON` flags enforced while capturing the snapshot
+ */
+function send (message, logger, dd, snapshot, processTags, eventType, incompleteReasons) {
+  if (message?.length > MAX_MESSAGE_LENGTH) {
+    message = message.slice(0, MAX_MESSAGE_LENGTH) + '…'
+    incompleteReasons |= INCOMPLETE_REASON.STRING_LENGTH
+  }
+
   const payload = {
     ddsource,
     hostname,
     service,
-    message: message?.length > MAX_MESSAGE_LENGTH
-      ? message.slice(0, MAX_MESSAGE_LENGTH) + '…'
-      : message,
+    message,
     logger,
     dd,
     process_tags: processTags,
@@ -52,6 +68,7 @@ function send (message, logger, dd, snapshot, processTags) {
   let size = Buffer.byteLength(json)
 
   if (size > MAX_LOG_PAYLOAD_SIZE_BYTES) {
+    incompleteReasons |= INCOMPLETE_REASON.PAYLOAD_TOO_LARGE
     let pruned
     try {
       pruned = pruneSnapshot(json, size, MAX_LOG_PAYLOAD_SIZE_BYTES)
@@ -71,6 +88,8 @@ function send (message, logger, dd, snapshot, processTags) {
   }
 
   jsonBuffer.write(json, size)
+
+  if (incompleteReasons !== 0) guardrailMetrics.captureIncomplete(incompleteReasons, eventType)
 }
 
 /**
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 938331d7920..7ed9ab99849 100644
--- a/packages/dd-trace/src/debugger/devtools_client/snapshot/index.js
+++ b/packages/dd-trace/src/debugger/devtools_client/snapshot/index.js
@@ -1,8 +1,10 @@
 'use strict'
 
+const { INCOMPLETE_REASON } = require('../../guardrail-metrics')
 const session = require('../session')
 const { collectObjectProperties } = require('./collector')
 const { processRawState, processRemoteObject } = require('./processor')
+const { fieldCountSym } = require('./symbols')
 
 const BIGINT_MAX = (1n << 256n) - 1n
 
@@ -26,8 +28,13 @@ module.exports = {
  * @param {CaptureLimits} limits - The capture limits
  * @param {bigint} [deadlineNs] - The deadline in nanoseconds compared to `process.hrtime.bigint()`. Defaults to
  *   {@link BIGINT_MAX}. If the deadline is reached, the snapshot will be truncated.
- * @returns {Promise<{ processLocalState: () => ReturnType<typeof processRawState>, fatalErrors: Error[] }>} The local
- *   state for the call frame
+ * @returns {Promise<{
+ *   processLocalState: () => ReturnType<typeof processRawState>,
+ *   fatalErrors: Error[],
+ *   incomplete: import('./processor').IncompleteCapture
+ * }>} The local state for the call frame. `incomplete` is only fully populated once `processLocalState` has run.
+ *   Not every fatal error is recorded as a runtime error: the collector also raises one when it hits its large object
+ *   safety threshold, which is reported as the field count limit it is.
  */
 async function getLocalStateForCallFrame (callFrame, limits, deadlineNs = BIGINT_MAX) {
   const { maxReferenceDepth, maxCollectionSize, maxFieldCount, maxLength } = limits
@@ -35,39 +42,63 @@ async function getLocalStateForCallFrame (callFrame, limits, deadlineNs = BIGINT
   const ctx = { deadlineReached: false, fatalErrors: [] }
   const opts = { maxReferenceDepth, maxCollectionSize, maxFieldCount, deadlineNs, ctx }
   const rawState = []
+  /** @type {import('./processor').IncompleteCapture} */
+  const incomplete = { reasons: 0 }
   /** @type {ReturnType<typeof processRawState> | null} */
   let processedState = null
 
-  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
+  const { scopeChain } = callFrame
+  for (const scope of scopeChain) {
+    if (!isCollectable(scope)) continue
+    const objectId = /** @type {string} */ (scope.object.objectId)
     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))
+      const variables = await collectObjectProperties(objectId, opts)
+      // A scope with more variables than the field count limit is trimmed like any other object, but since the scope
+      // itself is not part of the snapshot, the marker would be lost once its variables are spread into the state
+      if (variables[fieldCountSym] !== undefined) incomplete.reasons |= INCOMPLETE_REASON.FIELD_COUNT
+      rawState.push(...variables)
     } catch (err) {
+      incomplete.reasons |= INCOMPLETE_REASON.RUNTIME_ERROR
       ctx.fatalErrors.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 probes are re-applied',
         { cause: err } // TODO: The cause is not used by the backend
       ))
     }
-    if (ctx.deadlineReached === true) break // TODO: Bad UX; Variables in remaining scopes are silently dropped
+    if (ctx.deadlineReached === true) {
+      // Nodes skipped within this scope carry the timeout marker and are recorded when processed (unless they end up
+      // redacted), but the remaining scopes are dropped without any marker.
+      // TODO: Bad UX; Variables in remaining scopes are silently dropped
+      // @ts-expect-error - findLast is available in Node.js 18+ but TypeScript doesn't know about it without ES2023 lib
+      if (scope !== scopeChain.findLast(isCollectable)) incomplete.reasons |= INCOMPLETE_REASON.TIMEOUT
+      break
+    }
   }
 
   // Delay calling `processRawState` so caller can resume the main thread before processing `rawState`
   return {
     processLocalState () {
-      processedState ??= processRawState(rawState, maxLength)
+      processedState ??= processRawState(rawState, maxLength, incomplete)
       return processedState
     },
     fatalErrors: ctx.fatalErrors,
+    incomplete,
   }
 }
 
+/**
+ * @param {import('inspector').Debugger.Scope} scope
+ * @returns {boolean} Whether the variables of the scope are collected into the snapshot
+ */
+function isCollectable (scope) {
+  // The global scope is too noisy, and a scope without an object id is possible according to the types
+  return scope.type !== 'global' && scope.object.objectId !== undefined
+}
+
 /**
  * @typedef {object} CompiledCaptureExpression
  * @property {string} name - The name of the expression (used as key in snapshot)
@@ -82,6 +113,9 @@ async function getLocalStateForCallFrame (callFrame, limits, deadlineNs = BIGINT
  * @property {{ expr: string, message: string }[]} evaluationErrors - Transient errors from expression evaluation
  *   (safe to retry)
  * @property {Error[]} fatalErrors - Fatal errors that should disable capture expressions for this probe permanently
+ * @property {import('./processor').IncompleteCapture} incomplete - The capture limits enforced on the expressions,
+ *   including a runtime error for every expression that threw or could not be evaluated. Only fully populated once
+ *   `processCaptureExpressions` has run.
  */
 
 /**
@@ -108,6 +142,8 @@ async function evaluateCaptureExpressions (callFrame, expressions, deadlineNs =
   const evaluationErrors = []
   /** @type {Error[]} */
   const fatalErrors = []
+  /** @type {import('./processor').IncompleteCapture} */
+  const incomplete = { reasons: 0 }
   /** @type {Record<string, ReturnType<typeof processRemoteObject>> | null} */
   let processedResult = null
 
@@ -126,6 +162,7 @@ async function evaluateCaptureExpressions (callFrame, expressions, deadlineNs =
 
       // Handle evaluation exceptions (maybe transient - bad expression, undefined var, etc.)
       if (exceptionDetails) {
+        incomplete.reasons |= INCOMPLETE_REASON.RUNTIME_ERROR
         evaluationErrors.push({ expr: name, message: extractErrorMessage(exceptionDetails) })
         continue
       }
@@ -156,9 +193,11 @@ async function evaluateCaptureExpressions (callFrame, expressions, deadlineNs =
         }
 
         if (ctx.deadlineReached) {
-          // Add the current expression (properties may be incomplete due to timeout)
+          // Add the current expression (properties may be incomplete due to timeout). Those carry the timeout marker
+          // and are recorded when processed, unless they end up redacted.
           rawResults.push({ name, remoteObject: result, maxLength })
-          // Add stub entries for remaining uncaptured expressions
+          // Add stub entries for remaining uncaptured expressions. The stubs are used as-is, so record them here.
+          if (i + 1 < expressions.length) incomplete.reasons |= INCOMPLETE_REASON.TIMEOUT
           for (let j = i + 1; j < expressions.length; j++) {
             rawResults.push({
               name: expressions[j].name,
@@ -172,6 +211,7 @@ async function evaluateCaptureExpressions (callFrame, expressions, deadlineNs =
 
       rawResults.push({ name, remoteObject: result, maxLength })
     } catch (err) {
+      incomplete.reasons |= INCOMPLETE_REASON.RUNTIME_ERROR
       fatalErrors.push(new Error(
         `Error capturing expression "${name}". ` +
         'Capture expressions for this probe will be skipped until the probe is re-applied',
@@ -189,7 +229,7 @@ async function evaluateCaptureExpressions (callFrame, expressions, deadlineNs =
       for (const { name, remoteObject, maxLength } of rawResults) {
         // If the remote object has notCapturedReason (e.g., timeout), use it as-is without processing
         processedResult[name] = remoteObject.notCapturedReason === undefined
-          ? processRemoteObject(remoteObject, maxLength)
+          ? processRemoteObject(remoteObject, maxLength, incomplete)
           : remoteObject
       }
 
@@ -197,6 +237,7 @@ async function evaluateCaptureExpressions (callFrame, expressions, deadlineNs =
     },
     evaluationErrors,
     fatalErrors,
+    incomplete,
   }
 }
 
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 54f69f28079..7d96025fa8d 100644
--- a/packages/dd-trace/src/debugger/devtools_client/snapshot/processor.js
+++ b/packages/dd-trace/src/debugger/devtools_client/snapshot/processor.js
@@ -1,5 +1,6 @@
 'use strict'
 
+const { INCOMPLETE_REASON } = require('../../guardrail-metrics')
 const { LARGE_OBJECT_SKIP_THRESHOLD } = require('./constants')
 const { collectionSizeSym, largeCollectionSkipThresholdSym, fieldCountSym, timeBudgetSym } = require('./symbols')
 const { normalizeName, REDACTED_IDENTIFIERS } = require('./redaction')
@@ -15,15 +16,24 @@ module.exports = {
  * @typedef {import('inspector').Runtime.RemoteObject & { properties?: object[] }} RemoteObjectWithProperties
  */
 
+/**
+ * Accumulates which capture limits were enforced while processing a single event, so they can be reported once per
+ * event instead of once per affected node.
+ *
+ * @typedef {object} IncompleteCapture
+ * @property {number} reasons - Bitmask of {@link INCOMPLETE_REASON} flags
+ */
+
 /**
  * Process a RemoteObject into the snapshot format.
  *
  * @param {RemoteObjectWithProperties} remoteObject
  * @param {number} maxLength - Maximum string length
+ * @param {IncompleteCapture} incomplete - Receives the capture limits enforced on the object
  * @returns {object} The processed value in snapshot format
  */
-function processRemoteObject (remoteObject, maxLength) {
-  return getPropertyValueRaw({ value: remoteObject }, maxLength)
+function processRemoteObject (remoteObject, maxLength, incomplete) {
+  return getPropertyValueRaw({ value: remoteObject }, maxLength, incomplete)
 }
 
 // Matches classes in source code, no matter how it's written:
@@ -33,7 +43,15 @@ function processRemoteObject (remoteObject, maxLength) {
 // - Anonymous, with odd whitespace: class\n{}
 const CLASS_REGEX = /^class\s([^{]*)/
 
-function processProperties (props, maxLength) {
+/**
+ * Process the collected properties of an object into the snapshot format.
+ *
+ * @param {object[]} props - The collected property descriptors
+ * @param {number} maxLength - Maximum string length
+ * @param {IncompleteCapture} incomplete - Receives the capture limits enforced on the properties
+ * @returns {Record<string, object>} The processed properties in snapshot format
+ */
+function processProperties (props, maxLength, incomplete) {
   const result = {}
 
   for (const prop of props) {
@@ -42,20 +60,26 @@ function processProperties (props, maxLength) {
     if (name.includes('.')) {
       name = name.replaceAll('.', '_')
     }
-    result[name] = getPropertyValue(prop, maxLength)
+    result[name] = getPropertyValue(prop, maxLength, incomplete)
   }
 
   return result
 }
 
+// Capture limits enforced while processing a value that ends up redacted never make it into the snapshot, so they are
+// not recorded as incomplete capture reasons either
+const discardedIncomplete = { reasons: 0 }
+
 // TODO: Improve performance of redaction algorithm.
 // This algorithm is probably slower than if we embedded the redaction logic inside the functions below.
 // That way we didn't have to traverse objects that will just be redacted anyway.
-function getPropertyValue (prop, maxLength) {
-  return redact(prop, getPropertyValueRaw(prop, maxLength))
+function getPropertyValue (prop, maxLength, incomplete) {
+  return shouldRedactProperty(prop)
+    ? notCapturedRedacted(getPropertyValueRaw(prop, maxLength, discardedIncomplete).type)
+    : getPropertyValueRaw(prop, maxLength, incomplete)
 }
 
-function getPropertyValueRaw (prop, maxLength) {
+function getPropertyValueRaw (prop, maxLength, incomplete) {
   // Special case for getters and setters which does not have a value property
   if (prop.get) {
     const hasGet = prop.get.type !== 'undefined'
@@ -69,15 +93,15 @@ function getPropertyValueRaw (prop, maxLength) {
 
   switch (prop.value?.type) {
     case 'object':
-      return getObjectValue(prop.value, maxLength)
+      return getObjectValue(prop.value, maxLength, incomplete)
     case 'string':
-      return toString(prop.value.value, maxLength)
+      return toString(prop.value.value, maxLength, incomplete)
     case 'number':
       return { type: 'number', value: prop.value.description } // use `description` to get it as string
     case 'boolean':
       return { type: 'boolean', value: prop.value.value === true ? 'true' : 'false' }
     case 'function':
-      return toFunctionOrClass(prop.value, maxLength)
+      return toFunctionOrClass(prop.value, maxLength, incomplete)
     case undefined: // TODO: Add test for when a prop has no value. I think it's if it's defined after the breakpoint?
     case 'undefined':
       return { type: 'undefined' }
@@ -88,18 +112,19 @@ function getPropertyValueRaw (prop, maxLength) {
     default:
       // As of this writing, the Chrome DevTools Protocol doesn't allow any other types than the ones listed above, but
       // in the future new ones might be added.
+      incomplete.reasons |= INCOMPLETE_REASON.OTHER
       return { type: prop.value.type, notCapturedReason: 'Unsupported property type' }
   }
 }
 
-function getObjectValue (obj, maxLength) {
+function getObjectValue (obj, maxLength, incomplete) {
   const timeBudgetReached = obj[timeBudgetSym] === true
 
   switch (obj.subtype) {
     case undefined:
-      return toObject(obj.className, obj.properties, maxLength, timeBudgetReached)
+      return toObject(obj.className, obj.properties, maxLength, timeBudgetReached, incomplete)
     case 'array':
-      return toArray(obj.className, obj.properties, maxLength, timeBudgetReached)
+      return toArray(obj.className, obj.properties, maxLength, timeBudgetReached, incomplete)
     case 'null':
       return { type: 'null', isNull: true }
     // case 'node': // TODO: What does this subtype represent?
@@ -113,28 +138,28 @@ function getObjectValue (obj, maxLength) {
       return { type: obj.className, value }
     }
     case 'map':
-      return toMap(obj.className, obj.properties, maxLength, timeBudgetReached)
+      return toMap(obj.className, obj.properties, maxLength, timeBudgetReached, incomplete)
     case 'set':
-      return toSet(obj.className, obj.properties, maxLength, timeBudgetReached)
+      return toSet(obj.className, obj.properties, maxLength, timeBudgetReached, incomplete)
     case 'error':
       // TODO: Convert stack trace to array to avoid string truncation or disable truncation in this case?
-      return toObject(obj.className, obj.properties, maxLength, timeBudgetReached)
+      return toObject(obj.className, obj.properties, maxLength, timeBudgetReached, incomplete)
     case 'proxy':
       // Use `description` instead of `className` as the `type` to get type of target object (`Proxy(Error)` vs `proxy`)
-      return toObject(obj.description, obj.properties, maxLength, timeBudgetReached)
+      return toObject(obj.description, obj.properties, maxLength, timeBudgetReached, incomplete)
     case 'promise':
-      return toObject(obj.className, obj.properties, maxLength, timeBudgetReached)
+      return toObject(obj.className, obj.properties, maxLength, timeBudgetReached, incomplete)
     case 'typedarray':
-      return toArray(obj.className, obj.properties, maxLength, timeBudgetReached)
+      return toArray(obj.className, obj.properties, maxLength, timeBudgetReached, incomplete)
     case 'generator':
       // Use `subtype` instead of `className` to make it obvious it's a generator
-      return toObject(obj.subtype, obj.properties, maxLength, timeBudgetReached)
+      return toObject(obj.subtype, obj.properties, maxLength, timeBudgetReached, incomplete)
     case 'arraybuffer':
-      return toArrayBuffer(obj.className, obj.properties, maxLength, timeBudgetReached)
+      return toArrayBuffer(obj.className, obj.properties, maxLength, timeBudgetReached, incomplete)
     case 'weakmap':
-      return toMap(obj.className, obj.properties, maxLength, timeBudgetReached)
+      return toMap(obj.className, obj.properties, maxLength, timeBudgetReached, incomplete)
     case 'weakset':
-      return toSet(obj.className, obj.properties, maxLength, timeBudgetReached)
+      return toSet(obj.className, obj.properties, maxLength, timeBudgetReached, incomplete)
     // case 'iterator': // TODO: I've not been able to trigger this subtype
     // case 'dataview': // TODO: Looks like the internal ArrayBuffer is only accessible via the `buffer` getter
     // case 'webassemblymemory': // TODO: Looks like the internal ArrayBuffer is only accessible via the `buffer` getter
@@ -142,31 +167,34 @@ function getObjectValue (obj, maxLength) {
     default:
       // As of this writing, the Chrome DevTools Protocol doesn't allow any other subtypes than the ones listed above,
       // but in the future new ones might be added.
+      incomplete.reasons |= INCOMPLETE_REASON.OTHER
       return { type: obj.subtype, notCapturedReason: 'Unsupported object type' }
   }
 }
 
-function toFunctionOrClass (value, maxLength) {
+function toFunctionOrClass (value, maxLength, incomplete) {
   const classMatch = value.description.match(CLASS_REGEX)
 
   if (classMatch === null) {
     // This is a function
     const timeBudgetReached = value[timeBudgetSym] === true
     // TODO: Would it make sense to detect if it's an arrow function or not?
-    return toObject(value.className, value.properties, maxLength, timeBudgetReached)
+    return toObject(value.className, value.properties, maxLength, timeBudgetReached, incomplete)
   }
   // This is a class
   const className = classMatch[1].trim()
   return { type: className ? `class ${className}` : 'class' }
 }
 
-function toString (str, maxLength) {
+function toString (str, maxLength, incomplete) {
   const size = str.length
 
   if (size <= maxLength) {
     return { type: 'string', value: str }
   }
 
+  incomplete.reasons |= INCOMPLETE_REASON.STRING_LENGTH
+
   return {
     type: 'string',
     value: str.slice(0, maxLength),
@@ -175,16 +203,17 @@ function toString (str, maxLength) {
   }
 }
 
-function toObject (type, props, maxLength, timeBudgetReached) {
-  if (timeBudgetReached === true) return notCapturedTimeBudget(type)
-  if (props === undefined) return notCapturedDepth(type)
+function toObject (type, props, maxLength, timeBudgetReached, incomplete) {
+  if (timeBudgetReached === true) return notCapturedTimeBudget(type, incomplete)
+  if (props === undefined) return notCapturedDepth(type, incomplete)
 
   const result = {
     type,
-    fields: processProperties(props, maxLength),
+    fields: processProperties(props, maxLength, incomplete),
   }
 
   if (props[fieldCountSym] !== undefined) {
+    incomplete.reasons |= INCOMPLETE_REASON.FIELD_COUNT
     result.notCapturedReason = 'fieldCount'
     result.size = props[fieldCountSym]
   }
@@ -192,27 +221,29 @@ function toObject (type, props, maxLength, timeBudgetReached) {
   return result
 }
 
-function toArray (type, elements, maxLength, timeBudgetReached) {
-  if (timeBudgetReached === true) return notCapturedTimeBudget(type)
-  if (elements === undefined) return notCapturedDepth(type)
+function toArray (type, elements, maxLength, timeBudgetReached, incomplete) {
+  if (timeBudgetReached === true) return notCapturedTimeBudget(type, incomplete)
+  if (elements === undefined) return notCapturedDepth(type, incomplete)
 
   const result = {
     type,
     elements: elements.map((element) => {
-      return getPropertyValue(element, maxLength)
+      return getPropertyValue(element, maxLength, incomplete)
     }),
   }
 
-  setNotCaptureReasonOnCollection(result, elements)
-  setNotCaptureReasonOnTooLargeCollection(result, elements)
+  setNotCaptureReasonOnCollection(result, elements, incomplete)
+  setNotCaptureReasonOnTooLargeCollection(result, elements, incomplete)
 
   return result
 }
 
-function toMap (type, pairs, maxLength, timeBudgetReached) {
-  if (timeBudgetReached === true) return notCapturedTimeBudget(type)
-  if (pairs === undefined) return notCapturedDepth(type)
-  if (pairs.length > 0 && pairs.every(({ value }) => value[timeBudgetSym] === true)) return notCapturedTimeBudget(type)
+function toMap (type, pairs, maxLength, timeBudgetReached, incomplete) {
+  if (timeBudgetReached === true) return notCapturedTimeBudget(type, incomplete)
+  if (pairs === undefined) return notCapturedDepth(type, incomplete)
+  if (pairs.length > 0 && pairs.every(({ value }) => value[timeBudgetSym] === true)) {
+    return notCapturedTimeBudget(type, incomplete)
+  }
 
   const result = {
     type,
@@ -227,32 +258,34 @@ function toMap (type, pairs, maxLength, timeBudgetReached) {
       // second containing the value of this entry of the Map.
 
       if (value[timeBudgetSym] === true) {
+        incomplete.reasons |= INCOMPLETE_REASON.TIMEOUT
         return [{ notCapturedReason: 'timeout' }, { notCapturedReason: 'timeout' }]
       }
       if (value.properties === undefined) {
+        incomplete.reasons |= INCOMPLETE_REASON.OTHER
         return [{ notCapturedReason: 'unknown' }, { notCapturedReason: 'unknown' }]
       }
 
       const shouldRedact = shouldRedactMapValue(value.properties[0])
-      const key = getPropertyValue(value.properties[0], maxLength)
+      const key = getPropertyValue(value.properties[0], maxLength, incomplete)
       const val = shouldRedact
         ? notCapturedRedacted(value.properties[1].value.type)
-        : getPropertyValue(value.properties[1], maxLength)
+        : getPropertyValue(value.properties[1], maxLength, incomplete)
       return [key, val]
     }),
   }
 
-  setNotCaptureReasonOnCollection(result, pairs)
-  setNotCaptureReasonOnTooLargeCollection(result, pairs)
+  setNotCaptureReasonOnCollection(result, pairs, incomplete)
+  setNotCaptureReasonOnTooLargeCollection(result, pairs, incomplete)
 
   return result
 }
 
-function toSet (type, values, maxLength, timeBudgetReached) {
-  if (timeBudgetReached === true) return notCapturedTimeBudget(type)
-  if (values === undefined) return notCapturedDepth(type)
+function toSet (type, values, maxLength, timeBudgetReached, incomplete) {
+  if (timeBudgetReached === true) return notCapturedTimeBudget(type, incomplete)
+  if (values === undefined) return notCapturedDepth(type, incomplete)
   if (values.length > 0 && values.every(({ value }) => value[timeBudgetSym] === true)) {
-    return notCapturedTimeBudget(type)
+    return notCapturedTimeBudget(type, incomplete)
   }
 
   const result = {
@@ -267,33 +300,42 @@ function toSet (type, values, maxLength, timeBudgetReached) {
       // 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' }
+      if (value[timeBudgetSym] === true) {
+        incomplete.reasons |= INCOMPLETE_REASON.TIMEOUT
+        return { notCapturedReason: 'timeout' }
+      }
+      if (value.properties === undefined) {
+        incomplete.reasons |= INCOMPLETE_REASON.OTHER
+        return { notCapturedReason: 'unknown' }
+      }
 
-      return getPropertyValue(value.properties[0], maxLength)
+      return getPropertyValue(value.properties[0], maxLength, incomplete)
     }),
   }
 
-  setNotCaptureReasonOnCollection(result, values)
-  setNotCaptureReasonOnTooLargeCollection(result, values)
+  setNotCaptureReasonOnCollection(result, values, incomplete)
+  setNotCaptureReasonOnTooLargeCollection(result, values, incomplete)
 
   return result
 }
 
-function toArrayBuffer (type, bytes, maxLength, timeBudgetReached) {
-  if (timeBudgetReached === true) return notCapturedTimeBudget(type)
-  if (bytes === undefined) return notCapturedDepth(type)
+function toArrayBuffer (type, bytes, maxLength, timeBudgetReached, incomplete) {
+  if (timeBudgetReached === true) return notCapturedTimeBudget(type, incomplete)
+  if (bytes === undefined) return notCapturedDepth(type, incomplete)
 
   const size = bytes.length
 
-  return size > maxLength
-    ? {
-        type,
-        value: arrayBufferToString(bytes, maxLength),
-        truncated: true,
-        size: bytes.length,
-      }
-    : { type, value: arrayBufferToString(bytes, size) }
+  if (size > maxLength) {
+    incomplete.reasons |= INCOMPLETE_REASON.STRING_LENGTH
+    return {
+      type,
+      value: arrayBufferToString(bytes, maxLength),
+      truncated: true,
+      size: bytes.length,
+    }
+  }
+
+  return { type, value: arrayBufferToString(bytes, size) }
 }
 
 function arrayBufferToString (bytes, size) {
@@ -304,9 +346,8 @@ function arrayBufferToString (bytes, size) {
   return buf.toString()
 }
 
-function redact (prop, obj) {
-  const name = getNormalizedNameFromProp(prop)
-  return REDACTED_IDENTIFIERS.has(name) ? notCapturedRedacted(obj.type) : obj
+function shouldRedactProperty (prop) {
+  return REDACTED_IDENTIFIERS.has(getNormalizedNameFromProp(prop))
 }
 
 function shouldRedactMapValue (key) {
@@ -323,15 +364,19 @@ function getNormalizedNameFromProp (prop) {
   return normalizeName(prop.name, prop.symbol !== undefined)
 }
 
-function setNotCaptureReasonOnCollection (result, collection) {
+function setNotCaptureReasonOnCollection (result, collection, incomplete) {
   if (collection[collectionSizeSym] !== undefined) {
+    incomplete.reasons |= INCOMPLETE_REASON.COLLECTION_SIZE
     result.notCapturedReason = 'collectionSize'
     result.size = collection[collectionSizeSym]
   }
 }
 
-function setNotCaptureReasonOnTooLargeCollection (result, collection) {
+function setNotCaptureReasonOnTooLargeCollection (result, collection, incomplete) {
   if (collection[largeCollectionSkipThresholdSym] !== undefined) {
+    // The collection was skipped entirely because it exceeded the skip threshold, which is a different limit than the
+    // `maxCollectionSize` limit that `collectionSize` represents
+    incomplete.reasons |= INCOMPLETE_REASON.OTHER
     result.notCapturedReason = `Large collection with too many elements (skip threshold: ${
       LARGE_OBJECT_SKIP_THRESHOLD
     })`
@@ -339,7 +384,8 @@ function setNotCaptureReasonOnTooLargeCollection (result, collection) {
   }
 }
 
-function notCapturedDepth (type) {
+function notCapturedDepth (type, incomplete) {
+  incomplete.reasons |= INCOMPLETE_REASON.DEPTH
   return { type, notCapturedReason: 'depth' }
 }
 
@@ -347,6 +393,7 @@ function notCapturedRedacted (type) {
   return { type, notCapturedReason: 'redactedIdent' }
 }
 
-function notCapturedTimeBudget (type) {
+function notCapturedTimeBudget (type, incomplete) {
+  incomplete.reasons |= INCOMPLETE_REASON.TIMEOUT
   return { type, notCapturedReason: 'timeout' }
 }
diff --git a/packages/dd-trace/src/debugger/guardrail-metrics.js b/packages/dd-trace/src/debugger/guardrail-metrics.js
new file mode 100644
index 00000000000..f5b0d0a6dfc
--- /dev/null
+++ b/packages/dd-trace/src/debugger/guardrail-metrics.js
@@ -0,0 +1,165 @@
+'use strict'
+
+/**
+ * Guardrail telemetry counters shared between the main thread and the debugger worker thread.
+ *
+ * Guardrail decisions are made in two places: the probe sampler that runs inside the V8 breakpoint condition on the
+ * main thread (rate limits) and the debugger worker (capture limits, queue overflow). Neither should pay for a message
+ * or a telemetry API call per skipped, dropped or incomplete event, so both increment counters in a shared buffer using
+ * atomics, and the main thread periodically drains the counters into instrumentation telemetry metrics.
+ *
+ * The metric names, tags and reason codes are defined by the "Debugger Observability for GA" specification and are
+ * shared with the other tracers. All metrics are counts in the `live_debugger` telemetry namespace.
+ */
+
+const TELEMETRY_NAMESPACE = 'live_debugger'
+
+const EVENT_TYPE = Object.freeze({
+  SNAPSHOT: 0,
+  LOG: 1,
+  DIAGNOSTIC: 2,
+})
+const EVENT_TYPE_NAMES = ['snapshot', 'log', 'diagnostic']
+
+/** Attempts stopped before event creation (`events.skipped`) */
+const SKIPPED_REASON = Object.freeze({
+  RATE_LIMIT_GLOBAL: 0,
+  RATE_LIMIT_PROBE: 1,
+  EVALUATION_TIMEOUT: 2,
+})
+const SKIPPED_REASON_NAMES = ['rateLimitGlobal', 'rateLimitProbe', 'evaluationTimeout']
+
+/** Complete events dropped before transport (`events.dropped`) */
+const DROPPED_REASON = Object.freeze({
+  QUEUE_FULL: 0,
+  PAYLOAD_TOO_LARGE: 1,
+})
+const DROPPED_REASON_NAMES = ['queueFull', 'payloadTooLarge']
+
+/**
+ * Capture enforced limits (`capture.incomplete`). These are bit flags so that the reasons hit while producing a single
+ * event can be accumulated in a bitmask and reported once per event per reason.
+ */
+const INCOMPLETE_REASON = Object.freeze({
+  RUNTIME_ERROR: 1 << 0,
+  TIMEOUT: 1 << 1,
+  DEPTH: 1 << 2,
+  FIELD_COUNT: 1 << 3,
+  COLLECTION_SIZE: 1 << 4,
+  STRING_LENGTH: 1 << 5,
+  PAYLOAD_TOO_LARGE: 1 << 6,
+  OTHER: 1 << 7,
+})
+const INCOMPLETE_REASON_NAMES = [
+  'runtimeError', 'timeout', 'depth', 'fieldCount', 'collectionSize', 'stringLength', 'payloadTooLarge', 'other',
+]
+
+const EVENT_TYPE_COUNT = EVENT_TYPE_NAMES.length
+const SKIPPED_OFFSET = 0
+const DROPPED_OFFSET = SKIPPED_OFFSET + SKIPPED_REASON_NAMES.length * EVENT_TYPE_COUNT
+const INCOMPLETE_OFFSET = DROPPED_OFFSET + DROPPED_REASON_NAMES.length * EVENT_TYPE_COUNT
+const SLOT_COUNT = INCOMPLETE_OFFSET + INCOMPLETE_REASON_NAMES.length * EVENT_TYPE_COUNT
+
+/**
+ * @typedef {object} MetricSlot
+ * @property {string} metric - The telemetry metric name
+ * @property {string[]} tags - The telemetry tags, sorted
+ */
+
+/** @type {MetricSlot[]} */
+const SLOTS = []
+addSlots('events.skipped', SKIPPED_REASON_NAMES)
+addSlots('events.dropped', DROPPED_REASON_NAMES)
+addSlots('capture.incomplete', INCOMPLETE_REASON_NAMES)
+
+class GuardrailMetrics {
+  /** @type {Int32Array} */
+  #counters
+
+  /**
+   * @param {SharedArrayBuffer} buffer - A buffer created with {@link GuardrailMetrics.createBuffer}
+   */
+  constructor (buffer) {
+    this.#counters = new Int32Array(buffer)
+  }
+
+  /**
+   * Create the shared buffer backing the counters.
+   *
+   * @returns {SharedArrayBuffer}
+   */
+  static createBuffer () {
+    return new SharedArrayBuffer(SLOT_COUNT * Int32Array.BYTES_PER_ELEMENT)
+  }
+
+  /**
+   * Record an attempt that was stopped before an event was created.
+   *
+   * @param {number} reason - One of {@link SKIPPED_REASON}
+   * @param {number} eventType - One of {@link EVENT_TYPE}
+   */
+  eventSkipped (reason, eventType) {
+    Atomics.add(this.#counters, SKIPPED_OFFSET + reason * EVENT_TYPE_COUNT + eventType, 1)
+  }
+
+  /**
+   * Record a complete event that was dropped before transport.
+   *
+   * @param {number} reason - One of {@link DROPPED_REASON}
+   * @param {number} eventType - One of {@link EVENT_TYPE}
+   * @param {number} [count] - The number of dropped events
+   */
+  eventDropped (reason, eventType, count = 1) {
+    Atomics.add(this.#counters, DROPPED_OFFSET + reason * EVENT_TYPE_COUNT + eventType, count)
+  }
+
+  /**
+   * Record the capture limits enforced while producing a single event, once per reason.
+   *
+   * @param {number} reasons - A bitmask of {@link INCOMPLETE_REASON} flags
+   * @param {number} eventType - One of {@link EVENT_TYPE}
+   */
+  captureIncomplete (reasons, eventType) {
+    for (let bit = 0; reasons !== 0; bit++, reasons >>>= 1) {
+      if ((reasons & 1) === 1) {
+        Atomics.add(this.#counters, INCOMPLETE_OFFSET + bit * EVENT_TYPE_COUNT + eventType, 1)
+      }
+    }
+  }
+
+  /**
+   * Reset all counters and report the ones that were non-zero.
+   *
+   * @param {(metric: string, tags: string[], count: number) => void} report - Called once per non-zero counter
+   */
+  drain (report) {
+    for (let i = 0; i < SLOT_COUNT; i++) {
+      const count = Atomics.exchange(this.#counters, i, 0)
+      if (count !== 0) {
+        const { metric, tags } = SLOTS[i]
+        report(metric, tags, count)
+      }
+    }
+  }
+}
+
+/**
+ * @param {string} metric
+ * @param {string[]} reasonNames
+ */
+function addSlots (metric, reasonNames) {
+  for (const reason of reasonNames) {
+    for (const eventType of EVENT_TYPE_NAMES) {
+      SLOTS.push({ metric, tags: [`event_type:${eventType}`, `reason:${reason}`] })
+    }
+  }
+}
+
+module.exports = {
+  DROPPED_REASON,
+  EVENT_TYPE,
+  GuardrailMetrics,
+  INCOMPLETE_REASON,
+  SKIPPED_REASON,
+  TELEMETRY_NAMESPACE,
+}
diff --git a/packages/dd-trace/src/debugger/index.js b/packages/dd-trace/src/debugger/index.js
index 3430ee43fa1..8aa27b5e935 100644
--- a/packages/dd-trace/src/debugger/index.js
+++ b/packages/dd-trace/src/debugger/index.js
@@ -4,8 +4,10 @@ const { readFile } = require('fs')
 const { types } = require('util')
 const { join } = require('path')
 const { Worker, MessageChannel, threadId: parentThreadId } = require('worker_threads')
+const dc = require('dc-polyfill')
 const log = require('../log')
 const { fetchAgentInfo } = require('../agent/info')
+const telemetryMetrics = require('../telemetry/metrics')
 const getDebuggerConfig = require('./config')
 const {
   DEBUGGER_DIAGNOSTICS_V1,
@@ -13,6 +15,7 @@ const {
   DEBUGGER_INPUT_V2,
   INSPECT_SEGMENT_GLOBAL_PROPERTY,
 } = require('./constants')
+const { GuardrailMetrics, TELEMETRY_NAMESPACE } = require('./guardrail-metrics')
 const { installProbeSampler, uninstallProbeSampler } = require('./probe_sampler')
 
 /**
@@ -23,12 +26,24 @@ const { installProbeSampler, uninstallProbeSampler } = require('./probe_sampler'
  * @typedef {import('../remote_config')} RemoteConfig
  */
 
+// Guardrail counters are aggregated in shared memory and only converted into telemetry metrics at this interval, so
+// the interval bounds the delay before a guardrail hit becomes visible, not the cost of recording it.
+const GUARDRAIL_METRICS_FLUSH_INTERVAL_MS = 10_000
+
+// Published by telemetry right before it sends its final metrics on process exit. The flush interval timer is unref'ed
+// and the worker does not keep the process alive, so without this hook everything counted since the last tick would
+// be lost when the application exits on its own.
+const TELEMETRY_APP_CLOSING_CHANNEL = 'datadog:telemetry:app-closing'
+
 let worker = null
 let configChannel = null
 let ackId = 0
 let rcAckCallbacks = null
 let rc = null
 let inputPath = null
+/** @type {GuardrailMetrics | null} */
+let guardrailMetrics = null
+let guardrailMetricsTimer = null
 
 // eslint-disable-next-line eslint-rules/eslint-process-env
 const { NODE_OPTIONS, ...env } = process.env
@@ -77,7 +92,13 @@ function start (config, rcInstance) {
   debuggerGlobals.utilTypes = types
   debuggerGlobals[INSPECT_SEGMENT_GLOBAL_PROPERTY] = require('./inspect-segment')
 
-  const probeSamplerBuffer = installProbeSampler()
+  const guardrailMetricsBuffer = GuardrailMetrics.createBuffer()
+  guardrailMetrics = new GuardrailMetrics(guardrailMetricsBuffer)
+  guardrailMetricsTimer = setInterval(flushGuardrailMetrics, GUARDRAIL_METRICS_FLUSH_INTERVAL_MS)
+  guardrailMetricsTimer.unref?.()
+  dc.subscribe(TELEMETRY_APP_CLOSING_CHANNEL, flushGuardrailMetrics)
+
+  const probeSamplerBuffer = installProbeSampler(guardrailMetrics)
 
   readProbeFile(config.dynamicInstrumentation.probeFile, (probes) => {
     const action = 'apply'
@@ -125,6 +146,7 @@ function start (config, rcInstance) {
           logPort: logChannel.port1,
           configPort: configChannel.port1,
           probeSamplerBuffer,
+          guardrailMetricsBuffer,
         },
         transferList: [probeChannel.port1, logChannel.port1, configChannel.port1],
       }
@@ -211,6 +233,20 @@ function cleanup (error) {
   configChannel = null
   inputPath = null
 
+  if (guardrailMetricsTimer !== null) {
+    clearInterval(guardrailMetricsTimer)
+    guardrailMetricsTimer = null
+  }
+  if (guardrailMetrics !== null) {
+    dc.unsubscribe(TELEMETRY_APP_CLOSING_CHANNEL, flushGuardrailMetrics)
+    // Report what the worker counted up until it was stopped. Known limitation: `Worker#terminate()` interrupts the
+    // worker asynchronously, so anything it counts between this drain and its actual termination is lost. That only
+    // concerns events still sitting in the worker's upload buffer, which die with the worker anyway, so it isn't worth
+    // deferring the drain until the worker has exited.
+    flushGuardrailMetrics()
+    guardrailMetrics = null
+  }
+
   // Call any pending ack callbacks
   // Pass error for unexpected exits, or undefined for graceful shutdown
   if (rcAckCallbacks) {
@@ -223,6 +259,17 @@ function cleanup (error) {
   }
 }
 
+/**
+ * Convert the guardrail counters accumulated by the probe sampler and the worker into telemetry metrics.
+ */
+function flushGuardrailMetrics () {
+  if (guardrailMetrics === null) return
+  const namespace = telemetryMetrics.manager.namespace(TELEMETRY_NAMESPACE)
+  guardrailMetrics.drain((metric, tags, count) => {
+    namespace.count(metric, tags).inc(count)
+  })
+}
+
 /**
  * Detect which debugger endpoint is available on the agent
  *
diff --git a/packages/dd-trace/src/debugger/probe_sampler.js b/packages/dd-trace/src/debugger/probe_sampler.js
index ab6563118a4..b9b153da33d 100644
--- a/packages/dd-trace/src/debugger/probe_sampler.js
+++ b/packages/dd-trace/src/debugger/probe_sampler.js
@@ -1,6 +1,7 @@
 'use strict'
 
 const { MAX_SNAPSHOTS_PER_SECOND_GLOBALLY } = require('./devtools_client/defaults')
+const { EVENT_TYPE, SKIPPED_REASON } = require('./guardrail-metrics')
 const {
   DD_TRACE_SYMBOL,
   MAX_SAMPLED_PROBES_PER_PAUSE,
@@ -22,9 +23,10 @@ module.exports = {
 /**
  * Install the runtime sampler in the debuggee context.
  *
+ * @param {import('./guardrail-metrics').GuardrailMetrics} guardrailMetrics - Counters for skipped probe hits.
  * @returns {SharedArrayBuffer} The shared sampler buffer to pass to the debugger worker.
  */
-function installProbeSampler () {
+function installProbeSampler (guardrailMetrics) {
   const buffer = createProbeSamplerBuffer()
 
   const lastCaptureNsByProbeId = new Map()
@@ -49,13 +51,20 @@ function installProbeSampler () {
     makeSampleDecision (probeIndex, probeId, nsBetweenSampling, isSnapshotProducingProbe) {
       const now = process.hrtime.bigint()
       const lastCaptureNs = lastCaptureNsByProbeId.get(probeId)
-      if (lastCaptureNs !== undefined && now - lastCaptureNs < nsBetweenSampling) return false
+      if (lastCaptureNs !== undefined && now - lastCaptureNs < nsBetweenSampling) {
+        guardrailMetrics.eventSkipped(
+          SKIPPED_REASON.RATE_LIMIT_PROBE,
+          isSnapshotProducingProbe === true ? EVENT_TYPE.SNAPSHOT : EVENT_TYPE.LOG
+        )
+        return false
+      }
 
       let shouldResetGlobalSnapshotRateWindow = false
       if (isSnapshotProducingProbe === true) {
         if (now - globalSnapshotSamplingRateWindowStart > oneSecondNs) {
           shouldResetGlobalSnapshotRateWindow = true
         } else if (snapshotsSampledWithinTheLastSecond >= MAX_SNAPSHOTS_PER_SECOND_GLOBALLY) {
+          guardrailMetrics.eventSkipped(SKIPPED_REASON.RATE_LIMIT_GLOBAL, EVENT_TYPE.SNAPSHOT)
           return false
         }
       }
diff --git a/packages/dd-trace/src/telemetry/telemetry.js b/packages/dd-trace/src/telemetry/telemetry.js
index cce74abb312..4622fc26975 100644
--- a/packages/dd-trace/src/telemetry/telemetry.js
+++ b/packages/dd-trace/src/telemetry/telemetry.js
@@ -156,6 +156,9 @@ function getProducts (config) {
       version: tracerVersion,
       enabled: profilingEnabledToBoolean(config.profiling.DD_PROFILING_ENABLED),
     },
+    dynamic_instrumentation: {
+      enabled: config.dynamicInstrumentation.enabled,
+    },
   }
 }
 
diff --git a/packages/dd-trace/test/appsec/iast/telemetry/logs.spec.js b/packages/dd-trace/test/appsec/iast/telemetry/logs.spec.js
index 1794a2c7a2e..ac19c93422f 100644
--- a/packages/dd-trace/test/appsec/iast/telemetry/logs.spec.js
+++ b/packages/dd-trace/test/appsec/iast/telemetry/logs.spec.js
@@ -43,6 +43,7 @@ describe('Telemetry logs', () => {
       version: '1.2.3-beta4',
       appsec: { enabled: false, DD_API_SECURITY_ENDPOINT_COLLECTION_ENABLED: false },
       profiling: { DD_PROFILING_ENABLED: false },
+      dynamicInstrumentation: { enabled: false },
       env: 'preprod',
       tags: {
         'runtime-id': '1a2b3c',
diff --git a/packages/dd-trace/test/debugger/devtools_client/index.spec.js b/packages/dd-trace/test/debugger/devtools_client/index.spec.js
index d28582555df..61d4f594b30 100644
--- a/packages/dd-trace/test/debugger/devtools_client/index.spec.js
+++ b/packages/dd-trace/test/debugger/devtools_client/index.spec.js
@@ -9,6 +9,8 @@ const sinon = require('sinon')
 
 require('../../setup/mocha')
 
+const { LARGE_OBJECT_SKIP_THRESHOLD } = require('../../../src/debugger/devtools_client/snapshot/constants')
+const { EVENT_TYPE, GuardrailMetrics, INCOMPLETE_REASON } = require('../../../src/debugger/guardrail-metrics')
 const { installProbeSampler } = require('../../../src/debugger/probe_sampler')
 const { MAX_SAMPLED_PROBES_PER_PAUSE } = require('../../../src/debugger/probe_sampler_constants')
 
@@ -94,7 +96,7 @@ describe('onPause', function () {
 
     send = sinon.spy()
     send['@noCallThru'] = true
-    sampledProbeIndexes = new Int32Array(installProbeSampler())
+    sampledProbeIndexes = new Int32Array(installProbeSampler(new GuardrailMetrics(GuardrailMetrics.createBuffer())))
 
     state = proxyquire('../../../src/debugger/devtools_client/state', { './session': session })
     const loadStatus = proxyquire.noCallThru()
@@ -129,6 +131,18 @@ describe('onPause', function () {
     onPaused = onPausedCall[1]
   })
 
+  /**
+   * Attach a probe to the hit breakpoint and mark it as sampled for the next pause.
+   *
+   * @param {ReturnType<typeof genProcessedProbe>} probe - The probe to sample.
+   */
+  function sampleProbe (probe) {
+    state.breakpointToProbes.set(breakpointId, new Map([[probe.id, probe]]))
+    state.samplingIndexToProbe.set(1, probe)
+    Atomics.store(sampledProbeIndexes, 0, 1)
+    Atomics.store(sampledProbeIndexes, 2, 1)
+  }
+
   it('should not fail if there is no probe for at the breakpoint', async function () {
     await onPaused(event)
     sinon.assert.calledOnceWithExactly(session.post, 'Debugger.resume')
@@ -180,6 +194,190 @@ describe('onPause', function () {
     assert.strictEqual(send.firstCall.args[2], undefined)
   })
 
+  it('should send log probe results as log events', async function () {
+    const probe = genProcessedProbe('probe-1')
+    sampleProbe(probe)
+
+    await onPaused(event)
+
+    sinon.assert.calledOnce(send)
+    const [, , , , , eventType, incompleteReasons] = send.firstCall.args
+    assert.strictEqual(eventType, EVENT_TYPE.LOG)
+    assert.strictEqual(incompleteReasons, 0)
+  })
+
+  it('should send snapshot probe results as snapshot events with the enforced capture limits', async function () {
+    const probe = genProcessedProbe('probe-1')
+    probe.captureSnapshot = true
+    probe.capture = { maxReferenceDepth: 0, maxCollectionSize: 100, maxFieldCount: 20, maxLength: 255 }
+    sampleProbe(probe)
+
+    session.post = sinon.stub().callsFake((method, params) => {
+      if (method === 'Debugger.evaluateOnCallFrame') return Promise.resolve({ result: { value: [{}] } })
+      if (method === 'Runtime.getProperties' && params.objectId === 'scope-object-id') {
+        return Promise.resolve({
+          result: [{ name: 'obj', value: { type: 'object', className: 'Object', objectId: 'nested-object-id' } }],
+        })
+      }
+      return Promise.resolve({})
+    })
+    const eventWithScope = {
+      params: {
+        ...event.params,
+        callFrames: [{
+          ...event.params.callFrames[0],
+          scopeChain: [{ type: 'local', object: { objectId: 'scope-object-id' } }],
+        }],
+      },
+    }
+
+    await onPaused(eventWithScope)
+
+    sinon.assert.calledOnce(send)
+    const [, , , snapshot, , eventType, incompleteReasons] = send.firstCall.args
+    assert.deepStrictEqual(snapshot.captures, {
+      lines: { 1: { locals: { obj: { type: 'Object', notCapturedReason: 'depth' } } } },
+    })
+    assert.strictEqual(eventType, EVENT_TYPE.SNAPSHOT)
+    assert.strictEqual(incompleteReasons, INCOMPLETE_REASON.DEPTH)
+  })
+
+  it('should record a runtime error when the snapshot cannot be collected', async function () {
+    const probe = genProcessedProbe('probe-1')
+    probe.captureSnapshot = true
+    probe.capture = { maxReferenceDepth: 3, maxCollectionSize: 100, maxFieldCount: 20, maxLength: 255 }
+    sampleProbe(probe)
+
+    session.post = sinon.stub().callsFake((method) => {
+      if (method === 'Debugger.evaluateOnCallFrame') return Promise.resolve({ result: { value: [{}] } })
+      if (method === 'Runtime.getProperties') return Promise.reject(new Error('boom'))
+      return Promise.resolve({})
+    })
+    const eventWithScope = {
+      params: {
+        ...event.params,
+        callFrames: [{
+          ...event.params.callFrames[0],
+          scopeChain: [{ type: 'local', object: { objectId: 'scope-object-id' } }],
+        }],
+      },
+    }
+
+    await onPaused(eventWithScope)
+
+    sinon.assert.calledOnce(send)
+    const [, , , snapshot, , eventType, incompleteReasons] = send.firstCall.args
+    assert.strictEqual(snapshot.evaluationErrors.length, 1)
+    assert.strictEqual(eventType, EVENT_TYPE.SNAPSHOT)
+    assert.strictEqual(incompleteReasons, INCOMPLETE_REASON.RUNTIME_ERROR)
+    assert.strictEqual(probe.captureSnapshot, false, 'should disable future snapshots for the probe')
+  })
+
+  it('should not record a runtime error when the large object safety threshold disables the snapshot',
+    async function () {
+      const probe = genProcessedProbe('probe-1')
+      probe.captureSnapshot = true
+      probe.capture = { maxReferenceDepth: 3, maxCollectionSize: 100, maxFieldCount: 20, maxLength: 255 }
+      sampleProbe(probe)
+
+      const hugeObjectProperties = Array.from({ length: LARGE_OBJECT_SKIP_THRESHOLD + 1 }, (_, i) => ({
+        name: `property${i}`, value: { type: 'number', value: i }, enumerable: true,
+      }))
+      session.post = sinon.stub().callsFake((method, params) => {
+        if (method === 'Debugger.evaluateOnCallFrame') return Promise.resolve({ result: { value: [{}] } })
+        if (method === 'Runtime.getProperties') {
+          return Promise.resolve(params.objectId === 'scope-object-id'
+            ? {
+                result: [{
+                  name: 'huge',
+                  value: { type: 'object', className: 'Object', description: 'Object', objectId: 'huge-object-id' },
+                  enumerable: true,
+                }],
+              }
+            : { result: hugeObjectProperties })
+        }
+        return Promise.resolve({})
+      })
+      const eventWithScope = {
+        params: {
+          ...event.params,
+          callFrames: [{
+            ...event.params.callFrames[0],
+            scopeChain: [{ type: 'local', object: { objectId: 'scope-object-id' } }],
+          }],
+        },
+      }
+
+      await onPaused(eventWithScope)
+
+      sinon.assert.calledOnce(send)
+      const [, , , snapshot, , eventType, incompleteReasons] = send.firstCall.args
+      assert.strictEqual(snapshot.captures.lines[1].locals.huge.notCapturedReason, 'fieldCount')
+      assert.strictEqual(snapshot.evaluationErrors.length, 1, 'should tell the user why future captures are skipped')
+      assert.strictEqual(eventType, EVENT_TYPE.SNAPSHOT)
+      assert.strictEqual(incompleteReasons, INCOMPLETE_REASON.FIELD_COUNT, 'should not count it as a runtime error')
+      assert.strictEqual(probe.captureSnapshot, false, 'should disable future snapshots for the probe')
+    })
+
+  it('should send capture expression results as snapshot events', async function () {
+    const probe = genProcessedProbe('probe-1')
+    probe.compiledCaptureExpressions = [{
+      name: 'foo',
+      expression: 'foo',
+      limits: { maxReferenceDepth: 3, maxCollectionSize: 100, maxFieldCount: 20, maxLength: 255 },
+    }]
+    sampleProbe(probe)
+
+    session.post = sinon.stub().callsFake((method, params) => {
+      if (method === 'Debugger.evaluateOnCallFrame') {
+        return params.expression === 'foo'
+          ? Promise.resolve({ result: { type: 'string', value: 'x'.repeat(300) } })
+          : Promise.resolve({ result: { value: [{}] } })
+      }
+      return Promise.resolve({})
+    })
+
+    await onPaused(event)
+
+    sinon.assert.calledOnce(send)
+    const [, , , snapshot, , eventType, incompleteReasons] = send.firstCall.args
+    assert.deepStrictEqual(snapshot.captures.lines[1].captureExpressions.foo, {
+      type: 'string', value: 'x'.repeat(255), truncated: true, size: 300,
+    })
+    assert.strictEqual(eventType, EVENT_TYPE.SNAPSHOT)
+    assert.strictEqual(incompleteReasons, INCOMPLETE_REASON.STRING_LENGTH)
+  })
+
+  it('should record a runtime error when a capture expression fails to evaluate', async function () {
+    const probe = genProcessedProbe('probe-1')
+    probe.compiledCaptureExpressions = [{
+      name: 'foo',
+      expression: 'foo',
+      limits: { maxReferenceDepth: 3, maxCollectionSize: 100, maxFieldCount: 20, maxLength: 255 },
+    }]
+    sampleProbe(probe)
+
+    session.post = sinon.stub().callsFake((method, params) => {
+      if (method === 'Debugger.evaluateOnCallFrame') {
+        return params.expression === 'foo'
+          ? Promise.resolve({
+            result: { type: 'object', subtype: 'error' },
+            exceptionDetails: { exception: { description: 'ReferenceError: foo is not defined' } },
+          })
+          : Promise.resolve({ result: { value: [{}] } })
+      }
+      return Promise.resolve({})
+    })
+
+    await onPaused(event)
+
+    sinon.assert.calledOnce(send)
+    const [, , , snapshot, , eventType, incompleteReasons] = send.firstCall.args
+    assert.deepStrictEqual(snapshot.evaluationErrors, [{ expr: 'foo', message: 'ReferenceError: foo is not defined' }])
+    assert.strictEqual(eventType, EVENT_TYPE.SNAPSHOT)
+    assert.strictEqual(incompleteReasons, INCOMPLETE_REASON.RUNTIME_ERROR)
+  })
+
   it('should log sampler overflow', async function () {
     state.breakpointToProbes.set(breakpointId, new Map())
     Atomics.store(sampledProbeIndexes, 1, 1)
diff --git a/packages/dd-trace/test/debugger/devtools_client/send.spec.js b/packages/dd-trace/test/debugger/devtools_client/send.spec.js
index 6296263bdeb..e7bbfa7d7df 100644
--- a/packages/dd-trace/test/debugger/devtools_client/send.spec.js
+++ b/packages/dd-trace/test/debugger/devtools_client/send.spec.js
@@ -7,6 +7,7 @@ const { afterEach, beforeEach, describe, it } = require('mocha')
 const proxyquire = require('proxyquire')
 const sinon = require('sinon')
 
+const { EVENT_TYPE, INCOMPLETE_REASON } = require('../../../src/debugger/guardrail-metrics')
 const JSONBuffer = require('../../../src/debugger/devtools_client/json-buffer')
 const { version: debuggerVersion } = require('../../../../../package.json')
 const { getRequestOptions } = require('./utils')
@@ -21,10 +22,11 @@ const repositoryUrl = 'my-repository-url'
 const url = 'my-url'
 const ddsource = 'dd_debugger'
 const hostname = getHostname()
-const message = { message: true }
+const message = 'my-message'
 const logger = { logger: true }
 const dd = { dd: true }
 const snapshot = { snapshot: true }
+const MAX_MESSAGE_LENGTH = 8 * 1024 // Mirrors the limit in send.js
 
 describe('input message http requests', function () {
   /** @type {sinon.SinonFakeTimers} */
@@ -37,6 +39,8 @@ describe('input message http requests', function () {
   let jsonBufferWrite
   /** @type {sinon.SinonStub} */
   let pruneSnapshotStub
+  /** @type {{ captureIncomplete: sinon.SinonStub, '@noCallThru': boolean }} */
+  let guardrailMetrics
 
   beforeEach(function () {
     clock = sinon.useFakeTimers({
@@ -49,6 +53,8 @@ describe('input message http requests', function () {
     pruneSnapshotStub = sinon.stub()
     pruneSnapshotStub['@noCallThru'] = true
 
+    guardrailMetrics = { captureIncomplete: sinon.stub(), '@noCallThru': true }
+
     class JSONBufferSpy extends JSONBuffer {
       constructor (...args) {
         super(...args)
@@ -61,6 +67,7 @@ describe('input message http requests', function () {
       './json-buffer': JSONBufferSpy,
       '../../exporters/common/request': request,
       './snapshot-pruner': { pruneSnapshot: pruneSnapshotStub },
+      './guardrail-metrics': guardrailMetrics,
     })
   })
 
@@ -69,23 +76,23 @@ describe('input message http requests', function () {
   })
 
   it('should buffer instead of calling request directly', function () {
-    send(message, logger, dd, snapshot)
+    send(message, logger, dd, snapshot, undefined, EVENT_TYPE.LOG, 0)
     sinon.assert.notCalled(request)
     sinon.assert.calledOnceWithMatch(jsonBufferWrite, JSON.stringify(getPayload()))
   })
 
   it('should call request with the expected payload once the buffer is flushed', function (done) {
-    send({ message: 1 }, logger, dd, snapshot)
-    send({ message: 2 }, logger, dd, snapshot)
-    send({ message: 3 }, logger, dd, snapshot)
+    send('message-1', logger, dd, snapshot, undefined, EVENT_TYPE.LOG, 0)
+    send('message-2', logger, dd, snapshot, undefined, EVENT_TYPE.LOG, 0)
+    send('message-3', logger, dd, snapshot, undefined, EVENT_TYPE.LOG, 0)
     sinon.assert.notCalled(request)
 
     clock.tick(1000)
 
     sinon.assert.calledOnceWithMatch(request, JSON.stringify([
-      getPayload({ message: 1 }),
-      getPayload({ message: 2 }),
-      getPayload({ message: 3 }),
+      getPayload('message-1'),
+      getPayload('message-2'),
+      getPayload('message-3'),
     ]))
 
     const opts = getRequestOptions(request)
@@ -116,9 +123,10 @@ describe('input message http requests', function () {
       './log': logStub,
       '../../exporters/common/request': request,
       './snapshot-pruner': { pruneSnapshot: pruneSnapshotStub },
+      './guardrail-metrics': guardrailMetrics,
     })
 
-    sendWithInvalidTag(message, logger, dd, snapshot)
+    sendWithInvalidTag(message, logger, dd, snapshot, undefined, EVENT_TYPE.LOG, 0)
     clock.tick(1000)
 
     sinon.assert.calledOnce(request)
@@ -144,9 +152,10 @@ describe('input message http requests', function () {
       './json-buffer': JSONBuffer,
       '../../exporters/common/request': request,
       './snapshot-pruner': { pruneSnapshot: pruneSnapshotStub },
+      './guardrail-metrics': guardrailMetrics,
     })
 
-    sendWithNumericTag(message, logger, dd, snapshot)
+    sendWithNumericTag(message, logger, dd, snapshot, undefined, EVENT_TYPE.LOG, 0)
     clock.tick(1000)
 
     sinon.assert.calledOnce(request)
@@ -166,6 +175,7 @@ describe('input message http requests', function () {
       './json-buffer': JSONBuffer,
       '../../exporters/common/request': request,
       './snapshot-pruner': { pruneSnapshot: pruneSnapshotStub },
+      './guardrail-metrics': guardrailMetrics,
     })
 
     sendV2(message, logger, dd, snapshot)
@@ -194,9 +204,10 @@ describe('input message http requests', function () {
       './json-buffer': JSONBuffer,
       '../../exporters/common/request': request,
       './snapshot-pruner': { pruneSnapshot: pruneSnapshotStub },
+      './guardrail-metrics': guardrailMetrics,
     })
 
-    sendAgentless(message, logger, dd, snapshot)
+    sendAgentless(message, logger, dd, snapshot, undefined, EVENT_TYPE.LOG, 0)
     clock.tick(1000)
 
     sinon.assert.calledOnce(request)
@@ -230,6 +241,7 @@ describe('input message http requests', function () {
       './json-buffer': JSONBuffer,
       '../../exporters/common/request': requestWith404,
       './snapshot-pruner': { pruneSnapshot: pruneSnapshotStub },
+      './guardrail-metrics': guardrailMetrics,
     })
 
     sendV2(message, logger, dd, snapshot)
@@ -279,6 +291,7 @@ describe('input message http requests', function () {
       './json-buffer': JSONBuffer,
       '../../exporters/common/request': requestWith404,
       './snapshot-pruner': { pruneSnapshot: pruneSnapshotStub },
+      './guardrail-metrics': guardrailMetrics,
     })
 
     // First send - should trigger v2 → diagnostics fallback
@@ -307,7 +320,7 @@ describe('input message http requests', function () {
 
   it('should include process_tags at root level when provided', function () {
     const processTags = 'entrypoint.name:banana,entrypoint.type:script'
-    send(message, logger, dd, snapshot, processTags)
+    send(message, logger, dd, snapshot, processTags, EVENT_TYPE.LOG, 0)
 
     const writtenJson = jsonBufferWrite.getCall(0).args[0]
     const written = JSON.parse(writtenJson)
@@ -317,7 +330,7 @@ describe('input message http requests', function () {
   })
 
   it('should not include process_tags when not provided', function () {
-    send(message, logger, dd, snapshot)
+    send(message, logger, dd, snapshot, undefined, EVENT_TYPE.LOG, 0)
 
     const writtenJson = jsonBufferWrite.getCall(0).args[0]
     const written = JSON.parse(writtenJson)
@@ -326,19 +339,8 @@ describe('input message http requests', function () {
   })
 
   describe('snapshot pruning', function () {
-    const largeSnapshot = {
-      id: '123',
-      stack: [{ function: 'test' }],
-      captures: {
-        lines: {
-          10: {
-            locals: {
-              largeData: { type: 'string', value: 'x'.repeat(2 * 1024 * 1024) },
-            },
-          },
-        },
-      },
-    }
+    /** @type {object} */
+    let largeSnapshot
     const prunedPayload = {
       ...getPayload(message),
       debugger: {
@@ -358,8 +360,25 @@ describe('input message http requests', function () {
       },
     }
 
+    beforeEach(function () {
+      // Recreated for each test since the pruning fallback mutates the snapshot in place
+      largeSnapshot = {
+        id: '123',
+        stack: [{ function: 'test' }],
+        captures: {
+          lines: {
+            10: {
+              locals: {
+                largeData: { type: 'string', value: 'x'.repeat(2 * 1024 * 1024) },
+              },
+            },
+          },
+        },
+      }
+    })
+
     it('should not attempt to prune if payload is under size limit', function () {
-      send(message, logger, dd, snapshot)
+      send(message, logger, dd, snapshot, undefined, EVENT_TYPE.LOG, 0)
       sinon.assert.notCalled(pruneSnapshotStub)
     })
 
@@ -367,7 +386,7 @@ describe('input message http requests', function () {
       const prunedJson = JSON.stringify(getPayload(message, largeSnapshot))
       pruneSnapshotStub.returns(prunedJson)
 
-      send(message, logger, dd, largeSnapshot)
+      send(message, logger, dd, largeSnapshot, undefined, EVENT_TYPE.LOG, 0)
 
       sinon.assert.calledOnce(pruneSnapshotStub)
       const call = pruneSnapshotStub.getCall(0)
@@ -380,7 +399,7 @@ describe('input message http requests', function () {
       const prunedJson = JSON.stringify(prunedPayload)
       pruneSnapshotStub.returns(prunedJson)
 
-      send(message, logger, dd, largeSnapshot)
+      send(message, logger, dd, largeSnapshot, undefined, EVENT_TYPE.LOG, 0)
 
       sinon.assert.calledOnce(pruneSnapshotStub)
       sinon.assert.calledOnceWithMatch(jsonBufferWrite, prunedJson)
@@ -389,7 +408,7 @@ describe('input message http requests', function () {
     it('should fall back to deleting captures if pruning fails', function () {
       pruneSnapshotStub.returns(undefined)
 
-      send(message, logger, dd, largeSnapshot)
+      send(message, logger, dd, largeSnapshot, undefined, EVENT_TYPE.LOG, 0)
 
       sinon.assert.calledOnce(pruneSnapshotStub)
 
@@ -399,11 +418,71 @@ describe('input message http requests', function () {
 
       assert.deepStrictEqual(written.debugger.snapshot.captures.lines[10], { pruned: true })
     })
+
+    it('should record the snapshot as incomplete due to its size when pruned', function () {
+      pruneSnapshotStub.returns(JSON.stringify(prunedPayload))
+
+      send(message, logger, dd, largeSnapshot, undefined, EVENT_TYPE.SNAPSHOT, 0)
+
+      sinon.assert.calledOnceWithExactly(
+        guardrailMetrics.captureIncomplete, INCOMPLETE_REASON.PAYLOAD_TOO_LARGE, EVENT_TYPE.SNAPSHOT
+      )
+    })
+
+    it('should record the snapshot as incomplete due to its size when pruning fails', function () {
+      pruneSnapshotStub.returns(undefined)
+
+      send(message, logger, dd, largeSnapshot, undefined, EVENT_TYPE.SNAPSHOT, INCOMPLETE_REASON.DEPTH)
+
+      sinon.assert.calledOnceWithExactly(
+        guardrailMetrics.captureIncomplete,
+        INCOMPLETE_REASON.DEPTH | INCOMPLETE_REASON.PAYLOAD_TOO_LARGE,
+        EVENT_TYPE.SNAPSHOT
+      )
+    })
+  })
+
+  describe('guardrail metrics', function () {
+    it('should not record complete captures', function () {
+      send(message, logger, dd, snapshot, undefined, EVENT_TYPE.SNAPSHOT, 0)
+
+      sinon.assert.notCalled(guardrailMetrics.captureIncomplete)
+    })
+
+    it('should record the enforced capture limits once the event is queued', function () {
+      const reasons = INCOMPLETE_REASON.DEPTH | INCOMPLETE_REASON.STRING_LENGTH
+
+      send(message, logger, dd, snapshot, undefined, EVENT_TYPE.LOG, reasons)
+
+      sinon.assert.calledOnce(jsonBufferWrite)
+      sinon.assert.calledOnceWithExactly(guardrailMetrics.captureIncomplete, reasons, EVENT_TYPE.LOG)
+      assert.ok(
+        guardrailMetrics.captureIncomplete.calledAfter(jsonBufferWrite),
+        'should record after the event is queued'
+      )
+    })
+
+    it('should record a truncated message as an enforced string length limit', function () {
+      const longMessage = 'x'.repeat(MAX_MESSAGE_LENGTH + 1)
+
+      send(longMessage, logger, dd, snapshot, undefined, EVENT_TYPE.LOG, INCOMPLETE_REASON.DEPTH)
+
+      assert.strictEqual(JSON.parse(jsonBufferWrite.firstCall.args[0]).message, 'x'.repeat(MAX_MESSAGE_LENGTH) + '…')
+      sinon.assert.calledOnceWithExactly(
+        guardrailMetrics.captureIncomplete, INCOMPLETE_REASON.DEPTH | INCOMPLETE_REASON.STRING_LENGTH, EVENT_TYPE.LOG
+      )
+    })
+
+    it('should not record a message that fits as truncated', function () {
+      send('x'.repeat(MAX_MESSAGE_LENGTH), logger, dd, snapshot, undefined, EVENT_TYPE.LOG, 0)
+
+      sinon.assert.notCalled(guardrailMetrics.captureIncomplete)
+    })
   })
 })
 
 /**
- * @param {object} [_message] - The message to get the payload for. Defaults to the {@link message} object.
+ * @param {string} [_message] - The message to get the payload for. Defaults to the {@link message} string.
  * @param {object} [_snapshot] - The snapshot to get the payload for. Defaults to the {@link snapshot} object.
  * @returns {object} - The payload.
  */
diff --git a/packages/dd-trace/test/debugger/devtools_client/snapshot/collector-deadline.spec.js b/packages/dd-trace/test/debugger/devtools_client/snapshot/collector-deadline.spec.js
index a5820a6f636..b995f0b45a8 100644
--- a/packages/dd-trace/test/debugger/devtools_client/snapshot/collector-deadline.spec.js
+++ b/packages/dd-trace/test/debugger/devtools_client/snapshot/collector-deadline.spec.js
@@ -29,7 +29,8 @@ describe('debugger -> devtools client -> snapshot collector deadline', function
     session.removeAllListeners('Debugger.scriptParsed')
     session.removeAllListeners('Debugger.paused')
     await session.post('Debugger.disable')
-    clock.restore()
+    // Also restores the `process.hrtime.bigint` stubs, so they don't leak into other specs
+    sinon.restore()
   })
 
   it('should not mark properties with timeout when deadline is not exceeded', async function () {
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 12ec08ab951..73044a7ddf1 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
@@ -8,6 +8,8 @@ const NODE_20_PLUS = require('semver').gte(process.version, '20.0.0')
 
 const { assertObjectContains } = require('../../../../../../integration-tests/helpers')
 require('../../../setup/mocha')
+
+const { INCOMPLETE_REASON } = require('../../../../src/debugger/guardrail-metrics')
 const {
   session,
   getTargetCodePath,
@@ -23,6 +25,7 @@ const target = getTargetCodePath(__filename)
 describe('debugger -> devtools client -> snapshot.getLocalStateForCallFrame', function () {
   describe('complex types', function () {
     let state
+    let incomplete
 
     beforeEach(enable(__filename))
 
@@ -35,10 +38,12 @@ describe('debugger -> devtools client -> snapshot.getLocalStateForCallFrame', fu
       session.once('Debugger.paused', async ({ params }) => {
         assert.strictEqual(params.hitBreakpoints.length, 1)
 
-        resolve((await getLocalStateForCallFrame(
+        const result = await getLocalStateForCallFrame(
           params.callFrames[0],
-          { ...DEFAULT_CAPTURE_LIMITS, maxFieldCount: Number.MAX_SAFE_INTEGER })
-        ).processLocalState())
+          { ...DEFAULT_CAPTURE_LIMITS, maxFieldCount: Number.MAX_SAFE_INTEGER }
+        )
+        incomplete = result.incomplete
+        resolve(result.processLocalState())
       })
 
       await setAndTriggerBreakpoint(target, 10)
@@ -240,6 +245,10 @@ describe('debugger -> devtools client -> snapshot.getLocalStateForCallFrame', fu
       })
     })
 
+    it('should record the string length limit as an incomplete capture reason', function () {
+      assert.strictEqual(incomplete.reasons & INCOMPLETE_REASON.STRING_LENGTH, INCOMPLETE_REASON.STRING_LENGTH)
+    })
+
     it('Function', function () {
       assert.ok('fn' in state)
       assert.deepStrictEqual(state.fn, {
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 611cd77b638..26eda47413d 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
@@ -6,6 +6,7 @@ const assert = require('node:assert')
 const { inspect } = require('node:util')
 const sinon = require('sinon')
 
+const { INCOMPLETE_REASON } = require('../../../../src/debugger/guardrail-metrics')
 const { getLocalStateForCallFrame, evaluateCaptureExpressions, DEFAULT_CAPTURE_LIMITS, session } = require('./utils')
 
 describe('debugger -> devtools client -> snapshot', function () {
@@ -29,12 +30,13 @@ describe('debugger -> devtools client -> snapshot', function () {
         sessionPostStub = sinon.stub(session, 'post')
         sessionPostStub.withArgs('Runtime.getProperties').rejects(new Error('Protocol error'))
 
-        const { fatalErrors, processLocalState } = await getLocalStateForCallFrame(
+        const { fatalErrors, incomplete, processLocalState } = await getLocalStateForCallFrame(
           mockCallFrame,
           DEFAULT_CAPTURE_LIMITS
         )
 
         assert.strictEqual(fatalErrors.length, 1)
+        assert.strictEqual(incomplete.reasons, INCOMPLETE_REASON.RUNTIME_ERROR)
 
         for (const error of fatalErrors) {
           assert.ok(error instanceof Error)
@@ -72,6 +74,7 @@ describe('debugger -> devtools client -> snapshot', function () {
 
         assert.strictEqual(result.fatalErrors.length, 1)
         assert.strictEqual(result.evaluationErrors.length, 0)
+        assert.strictEqual(result.incomplete.reasons, INCOMPLETE_REASON.RUNTIME_ERROR)
 
         const error = result.fatalErrors[0]
         assert.ok(error instanceof Error)
@@ -170,6 +173,9 @@ describe('debugger -> devtools client -> snapshot', function () {
           'Error capturing expression "protocolError". ' +
           'Capture expressions for this probe will be skipped until the probe is re-applied'
         )
+
+        // Both count as a runtime error once
+        assert.strictEqual(result.incomplete.reasons, INCOMPLETE_REASON.RUNTIME_ERROR)
       })
     })
   })
diff --git a/packages/dd-trace/test/debugger/devtools_client/snapshot/incomplete-capture.spec.js b/packages/dd-trace/test/debugger/devtools_client/snapshot/incomplete-capture.spec.js
new file mode 100644
index 00000000000..5eb703b5174
--- /dev/null
+++ b/packages/dd-trace/test/debugger/devtools_client/snapshot/incomplete-capture.spec.js
@@ -0,0 +1,239 @@
+'use strict'
+
+const assert = require('node:assert/strict')
+
+const { afterEach, beforeEach, describe, it } = require('mocha')
+
+require('../../../setup/mocha')
+
+const { INCOMPLETE_REASON } = require('../../../../src/debugger/guardrail-metrics')
+const { LARGE_OBJECT_SKIP_THRESHOLD } = require('../../../../src/debugger/devtools_client/snapshot/constants')
+const {
+  DEFAULT_CAPTURE_LIMITS,
+  enable,
+  evaluateCaptureExpressions,
+  getLocalStateForCallFrame,
+  getTargetCodePath,
+  session,
+  setAndTriggerBreakpoint,
+  teardown,
+} = require('./utils')
+
+const target = getTargetCodePath(__filename)
+const {
+  runWithHugeObject,
+  runWithManyLocals,
+  runWithRedactedValues,
+  runWithRedactedObject,
+  runWithRedactedObjectAndClosure,
+} = require(target)
+
+// A deadline that has already passed, so the collector stops at the first opportunity
+const EXPIRED_DEADLINE_NS = 0n
+
+describe('debugger -> devtools client -> snapshot incomplete capture reasons', function () {
+  beforeEach(enable(__filename))
+
+  afterEach(teardown)
+
+  describe('getLocalStateForCallFrame', function () {
+    it('should not record any reason for a complete capture', async function () {
+      const { processLocalState, incomplete } = await whilePaused((callFrame) => {
+        return getLocalStateForCallFrame(callFrame, DEFAULT_CAPTURE_LIMITS)
+      })
+
+      processLocalState()
+
+      assert.strictEqual(incomplete.reasons, 0)
+    })
+
+    it('should only populate the reasons once the local state has been processed', async function () {
+      const { processLocalState, incomplete } = await whilePaused((callFrame) => {
+        return getLocalStateForCallFrame(callFrame, { ...DEFAULT_CAPTURE_LIMITS, maxReferenceDepth: 1 })
+      })
+
+      assert.strictEqual(incomplete.reasons, 0)
+      processLocalState()
+      assert.strictEqual(incomplete.reasons, INCOMPLETE_REASON.DEPTH)
+    })
+
+    it('should record a timeout when the capture deadline is reached', async function () {
+      const { processLocalState, incomplete } = await whilePaused((callFrame) => {
+        return getLocalStateForCallFrame(callFrame, DEFAULT_CAPTURE_LIMITS, EXPIRED_DEADLINE_NS)
+      })
+
+      const state = processLocalState()
+
+      assert.deepStrictEqual(state.nested, { type: 'Object', notCapturedReason: 'timeout' })
+      assert.strictEqual(incomplete.reasons, INCOMPLETE_REASON.TIMEOUT)
+    })
+
+    it('should record the field count limit when a scope has more variables than allowed', async function () {
+      const { processLocalState, incomplete } = await whilePaused((callFrame) => {
+        return getLocalStateForCallFrame(callFrame, { ...DEFAULT_CAPTURE_LIMITS, maxFieldCount: 2 })
+      }, { line: 21, trigger: runWithManyLocals })
+
+      const state = processLocalState()
+
+      assert.deepStrictEqual(Object.keys(state), ['first', 'second'], 'should drop the variables over the limit')
+      assert.strictEqual(incomplete.reasons, INCOMPLETE_REASON.FIELD_COUNT)
+    })
+
+    it('should not record the field count limit when a scope has exactly as many variables as allowed',
+      async function () {
+        const { processLocalState, incomplete } = await whilePaused((callFrame) => {
+          return getLocalStateForCallFrame(callFrame, { ...DEFAULT_CAPTURE_LIMITS, maxFieldCount: 3 })
+        }, { line: 21, trigger: runWithManyLocals })
+
+        const state = processLocalState()
+
+        assert.deepStrictEqual(Object.keys(state), ['first', 'second', 'third'])
+        assert.strictEqual(incomplete.reasons, 0)
+      })
+
+    it('should not record a timeout that only cut short a value that ends up redacted', async function () {
+      const { processLocalState, incomplete } = await whilePaused((callFrame) => {
+        return getLocalStateForCallFrame(callFrame, DEFAULT_CAPTURE_LIMITS, EXPIRED_DEADLINE_NS)
+      }, { line: 35, trigger: runWithRedactedObject })
+
+      const state = processLocalState()
+
+      assert.deepStrictEqual(state, { password: { type: 'Object', notCapturedReason: 'redactedIdent' } })
+      assert.strictEqual(incomplete.reasons, 0)
+    })
+
+    it('should record a timeout that drops the remaining scopes', async function () {
+      const { processLocalState, incomplete } = await whilePaused((callFrame) => {
+        return getLocalStateForCallFrame(callFrame, DEFAULT_CAPTURE_LIMITS, EXPIRED_DEADLINE_NS)
+      }, { line: 43, trigger: runWithRedactedObjectAndClosure() })
+
+      const state = processLocalState()
+
+      // The closure scope holding `fromClosure` is never collected, which is only observable through the metric
+      assert.deepStrictEqual(state, { password: { type: 'Object', notCapturedReason: 'redactedIdent' } })
+      assert.strictEqual(incomplete.reasons, INCOMPLETE_REASON.TIMEOUT)
+    })
+
+    it('should not record limits enforced on values that end up redacted', async function () {
+      const { processLocalState, incomplete } = await whilePaused((callFrame) => {
+        return getLocalStateForCallFrame(callFrame, { ...DEFAULT_CAPTURE_LIMITS, maxReferenceDepth: 1 })
+      }, { line: 29, trigger: runWithRedactedValues })
+
+      const state = processLocalState()
+
+      assert.deepStrictEqual(state, {
+        password: { type: 'string', notCapturedReason: 'redactedIdent' },
+        secret: { type: 'Object', notCapturedReason: 'redactedIdent' },
+      })
+      assert.strictEqual(incomplete.reasons, 0)
+    })
+
+    it('should not record a runtime error when the large object safety threshold disables the capture',
+      async function () {
+        const { processLocalState, fatalErrors, incomplete } = await whilePaused((callFrame) => {
+          return getLocalStateForCallFrame(callFrame, DEFAULT_CAPTURE_LIMITS)
+        }, { line: 12, trigger: () => runWithHugeObject(LARGE_OBJECT_SKIP_THRESHOLD + 1) })
+
+        const state = processLocalState()
+
+        assert.strictEqual(fatalErrors.length, 1, 'should still disable future captures for the probe')
+        assert.match(fatalErrors[0].message, /exceeds the maximum number of allowed properties/)
+        assert.strictEqual(state.huge.notCapturedReason, 'fieldCount')
+        assert.strictEqual(incomplete.reasons, INCOMPLETE_REASON.FIELD_COUNT)
+      })
+  })
+
+  describe('evaluateCaptureExpressions', function () {
+    it('should not record any reason for a complete capture', async function () {
+      const { processCaptureExpressions, incomplete } = await whilePaused((callFrame) => {
+        return evaluateCaptureExpressions(callFrame, [
+          { name: 'nested', expression: 'nested', limits: DEFAULT_CAPTURE_LIMITS },
+        ])
+      })
+
+      processCaptureExpressions()
+
+      assert.strictEqual(incomplete.reasons, 0)
+    })
+
+    it('should record the enforced capture limits', async function () {
+      const { processCaptureExpressions, incomplete } = await whilePaused((callFrame) => {
+        return evaluateCaptureExpressions(callFrame, [
+          { name: 'nested', expression: 'nested', limits: { ...DEFAULT_CAPTURE_LIMITS, maxReferenceDepth: 1 } },
+        ])
+      })
+
+      const captured = processCaptureExpressions()
+
+      assert.deepStrictEqual(captured.nested, {
+        type: 'Object',
+        fields: { foo: { type: 'Object', notCapturedReason: 'depth' } },
+      })
+      assert.strictEqual(incomplete.reasons, INCOMPLETE_REASON.DEPTH)
+    })
+
+    it('should record a timeout when the capture deadline is reached', async function () {
+      const { processCaptureExpressions, incomplete } = await whilePaused((callFrame) => {
+        return evaluateCaptureExpressions(callFrame, [
+          { name: 'nested', expression: 'nested', limits: DEFAULT_CAPTURE_LIMITS },
+          { name: 'skipped', expression: 'nested', limits: DEFAULT_CAPTURE_LIMITS },
+        ], EXPIRED_DEADLINE_NS)
+      })
+
+      const captured = processCaptureExpressions()
+
+      assert.deepStrictEqual(captured.skipped, { notCapturedReason: 'timeout' })
+      assert.strictEqual(incomplete.reasons, INCOMPLETE_REASON.TIMEOUT)
+    })
+
+    it('should not record a timeout that only cut short a value that ends up redacted', async function () {
+      const { processCaptureExpressions, incomplete } = await whilePaused((callFrame) => {
+        return evaluateCaptureExpressions(callFrame, [
+          { name: 'wrapped', expression: '({ password: nested })', limits: DEFAULT_CAPTURE_LIMITS },
+        ], EXPIRED_DEADLINE_NS)
+      })
+
+      const captured = processCaptureExpressions()
+
+      assert.deepStrictEqual(captured.wrapped, {
+        type: 'Object',
+        fields: { password: { type: 'Object', notCapturedReason: 'redactedIdent' } },
+      })
+      assert.strictEqual(incomplete.reasons, 0)
+    })
+
+    it('should record a timeout that skips the remaining expressions', async function () {
+      const { processCaptureExpressions, incomplete } = await whilePaused((callFrame) => {
+        return evaluateCaptureExpressions(callFrame, [
+          { name: 'wrapped', expression: '({ password: nested })', limits: DEFAULT_CAPTURE_LIMITS },
+          { name: 'skipped', expression: 'nested', limits: DEFAULT_CAPTURE_LIMITS },
+        ], EXPIRED_DEADLINE_NS)
+      })
+
+      const captured = processCaptureExpressions()
+
+      assert.deepStrictEqual(captured.skipped, { notCapturedReason: 'timeout' })
+      assert.strictEqual(incomplete.reasons, INCOMPLETE_REASON.TIMEOUT)
+    })
+  })
+})
+
+/**
+ * Trigger a breakpoint in the target code and run `fn` while the debuggee is paused on it.
+ *
+ * @template T
+ * @param {(callFrame: import('inspector').Debugger.CallFrame) => Promise<T>} fn - Work to do while paused
+ * @param {object} [options]
+ * @param {number} [options.line] - The line to break on. Defaults to the line in `run`.
+ * @param {() => void} [options.trigger] - The function hitting the breakpoint. Defaults to `run`.
+ * @returns {Promise<T>}
+ */
+function whilePaused (fn, { line = 6, trigger } = {}) {
+  return new Promise((resolve, reject) => {
+    session.once('Debugger.paused', ({ params }) => {
+      assert.strictEqual(params.hitBreakpoints.length, 1)
+      fn(params.callFrames[0]).then(resolve, reject)
+    })
+    setAndTriggerBreakpoint(target, line, trigger).catch(reject)
+  })
+}
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 b8bb28e947a..cf564d04dbe 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
@@ -8,6 +8,7 @@ const { afterEach, beforeEach, describe, it } = require('mocha')
 const { assertObjectContains } = require('../../../../../../integration-tests/helpers')
 require('../../../setup/mocha')
 
+const { INCOMPLETE_REASON } = require('../../../../src/debugger/guardrail-metrics')
 const {
   LARGE_OBJECT_SKIP_THRESHOLD,
   DEFAULT_MAX_COLLECTION_SIZE,
@@ -33,6 +34,7 @@ describe('debugger -> devtools client -> snapshot.getLocalStateForCallFrame', fu
 
       describe(`should respect the default maxCollectionSize if ${postfix}`, function () {
         let state
+        let incomplete
 
         const expectedElements = []
         const expectedEntries = []
@@ -48,12 +50,17 @@ describe('debugger -> devtools client -> snapshot.getLocalStateForCallFrame', fu
         }
 
         beforeEach(function (done) {
-          assertOnBreakpoint(done, config, (_state) => {
+          assertOnBreakpoint(done, config, (_state, _incomplete) => {
             state = _state
+            incomplete = _incomplete
           })
           setAndTriggerBreakpoint(target, 29)
         })
 
+        it('should record the collection size limit as an incomplete capture reason', function () {
+          assert.strictEqual(incomplete.reasons, INCOMPLETE_REASON.COLLECTION_SIZE)
+        })
+
         it('should have expected number of elements in state', function () {
           assert.deepStrictEqual(
             Object.keys(state).sort(),
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 39950c58655..7e5929fe6a4 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
@@ -7,6 +7,7 @@ const { afterEach, beforeEach, describe, it } = require('mocha')
 require('../../../setup/mocha')
 
 const { DEFAULT_MAX_FIELD_COUNT } = require('../../../../src/debugger/devtools_client/snapshot/constants')
+const { INCOMPLETE_REASON } = require('../../../../src/debugger/guardrail-metrics')
 const { getTargetCodePath, enable, teardown, assertOnBreakpoint, setAndTriggerBreakpoint } = require('./utils')
 
 const target = getTargetCodePath(__filename)
@@ -26,6 +27,7 @@ describe('debugger -> devtools client -> snapshot.getLocalStateForCallFrame', fu
 function generateTestCases (config) {
   const maxFieldCount = config?.maxFieldCount ?? DEFAULT_MAX_FIELD_COUNT
   let state
+  let incomplete
 
   const expectedFields = {}
   for (let i = 1; i <= maxFieldCount; i++) {
@@ -34,12 +36,17 @@ function generateTestCases (config) {
 
   return function () {
     beforeEach(function (done) {
-      assertOnBreakpoint(done, config, (_state) => {
+      assertOnBreakpoint(done, config, (_state, _incomplete) => {
         state = _state
+        incomplete = _incomplete
       })
       setAndTriggerBreakpoint(target, 11)
     })
 
+    it('should record the field count limit as an incomplete capture reason', function () {
+      assert.strictEqual(incomplete.reasons, INCOMPLETE_REASON.FIELD_COUNT)
+    })
+
     it('should capture expected snapshot', function () {
       assert.strictEqual(Object.keys(state).length, ((Array.isArray(['obj']) ? ['obj'] : [['obj']])).length)
       assert.ok(
diff --git a/packages/dd-trace/test/debugger/devtools_client/snapshot/max-reference-depth.spec.js b/packages/dd-trace/test/debugger/devtools_client/snapshot/max-reference-depth.spec.js
index a33f9fdd425..7b7830c9fb2 100644
--- a/packages/dd-trace/test/debugger/devtools_client/snapshot/max-reference-depth.spec.js
+++ b/packages/dd-trace/test/debugger/devtools_client/snapshot/max-reference-depth.spec.js
@@ -6,6 +6,7 @@ const { inspect } = require('node:util')
 const { afterEach, beforeEach, describe, it } = require('mocha')
 require('../../../setup/mocha')
 
+const { INCOMPLETE_REASON } = require('../../../../src/debugger/guardrail-metrics')
 const { getTargetCodePath, enable, teardown, assertOnBreakpoint, setAndTriggerBreakpoint } = require('./utils')
 
 const target = getTargetCodePath(__filename)
@@ -17,7 +18,8 @@ describe('debugger -> devtools client -> snapshot.getLocalStateForCallFrame', fu
     afterEach(teardown)
 
     it('should return expected object for nested objects with maxReferenceDepth: 1', function (done) {
-      assertOnBreakpoint(done, { maxReferenceDepth: 1 }, (state) => {
+      assertOnBreakpoint(done, { maxReferenceDepth: 1 }, (state, incomplete) => {
+        assert.strictEqual(incomplete.reasons, INCOMPLETE_REASON.DEPTH)
         assert.strictEqual(Object.keys(state).length, 1)
 
         assert.ok(Object.hasOwn(state, 'myNestedObj'), `Available keys: ${inspect(Object.keys(state))}`)
@@ -43,7 +45,8 @@ describe('debugger -> devtools client -> snapshot.getLocalStateForCallFrame', fu
     })
 
     it('should return expected object for nested objects with maxReferenceDepth: 5', function (done) {
-      assertOnBreakpoint(done, { maxReferenceDepth: 5 }, (state) => {
+      assertOnBreakpoint(done, { maxReferenceDepth: 5 }, (state, incomplete) => {
+        assert.strictEqual(incomplete.reasons, INCOMPLETE_REASON.DEPTH)
         assert.strictEqual(Object.entries(state).length, 1)
 
         assert.ok(Object.hasOwn(state, 'myNestedObj'), `Available keys: ${inspect(Object.keys(state))}`)
diff --git a/packages/dd-trace/test/debugger/devtools_client/snapshot/target-code/incomplete-capture.js b/packages/dd-trace/test/debugger/devtools_client/snapshot/target-code/incomplete-capture.js
new file mode 100644
index 00000000000..a5a6e856b68
--- /dev/null
+++ b/packages/dd-trace/test/debugger/devtools_client/snapshot/target-code/incomplete-capture.js
@@ -0,0 +1,54 @@
+'use strict'
+
+function run () {
+  // eslint-disable-next-line no-unused-vars
+  const nested = { foo: { bar: { baz: 42 } } }
+  return 'my return value' // breakpoint at this line
+}
+
+function runWithHugeObject (propertyCount) {
+  // eslint-disable-next-line no-unused-vars
+  const huge = Object.fromEntries(Array.from({ length: propertyCount }, (_, i) => [`property${i}`, i]))
+  return 'my return value' // breakpoint at this line
+}
+
+function runWithManyLocals () {
+  /* eslint-disable no-unused-vars */
+  const first = 1
+  const second = 2
+  const third = 3
+  /* eslint-enable no-unused-vars */
+  return 'my return value' // breakpoint at this line
+}
+
+function runWithRedactedValues () {
+  /* eslint-disable no-unused-vars */
+  const password = 'x'.repeat(300)
+  const secret = { nested: { deeper: { deepest: 42 } } }
+  /* eslint-enable no-unused-vars */
+  return 'my return value' // breakpoint at this line
+}
+
+function runWithRedactedObject () {
+  // eslint-disable-next-line no-unused-vars
+  const password = { nested: { deeper: 42 } }
+  return 'my return value' // breakpoint at this line
+}
+
+function runWithRedactedObjectAndClosure () {
+  const fromClosure = { foo: 'bar' }
+  return function inner () {
+    // eslint-disable-next-line no-unused-vars
+    const password = { nested: { deeper: 42 } }
+    return fromClosure // breakpoint at this line
+  }
+}
+
+module.exports = {
+  run,
+  runWithHugeObject,
+  runWithManyLocals,
+  runWithRedactedValues,
+  runWithRedactedObject,
+  runWithRedactedObjectAndClosure,
+}
diff --git a/packages/dd-trace/test/debugger/devtools_client/snapshot/time-budget.spec.js b/packages/dd-trace/test/debugger/devtools_client/snapshot/time-budget.spec.js
index d88b8e5b03a..e39628014c6 100644
--- a/packages/dd-trace/test/debugger/devtools_client/snapshot/time-budget.spec.js
+++ b/packages/dd-trace/test/debugger/devtools_client/snapshot/time-budget.spec.js
@@ -2,12 +2,23 @@
 
 const assert = require('node:assert/strict')
 const proxyquire = require('proxyquire')
+const { INCOMPLETE_REASON } = require('../../../../src/debugger/guardrail-metrics')
 const { timeBudgetSym } = require('../../../../src/debugger/devtools_client/snapshot/symbols')
 
 const MAX_LENGTH = 255
 
 describe('Debugger snapshot time budget', () => {
   let processRawState
+  /** @type {import('../../../../src/debugger/devtools_client/snapshot/processor').IncompleteCapture} */
+  let incomplete
+
+  beforeEach(() => {
+    incomplete = { reasons: 0 }
+  })
+
+  afterEach(() => {
+    assert.strictEqual(incomplete.reasons, INCOMPLETE_REASON.TIMEOUT, 'should record the timeout as incomplete reason')
+  })
 
   before(() => {
     const loadRedaction = proxyquire.noCallThru()
@@ -38,7 +49,7 @@ describe('Debugger snapshot time budget', () => {
       },
     }]
 
-    const out = processRawState(raw, MAX_LENGTH)
+    const out = processRawState(raw, MAX_LENGTH, incomplete)
     assert.deepStrictEqual(out.fn, {
       type: 'Function',
       notCapturedReason: 'timeout',
@@ -55,7 +66,7 @@ describe('Debugger snapshot time budget', () => {
       },
     }]
 
-    const out = processRawState(raw, MAX_LENGTH)
+    const out = processRawState(raw, MAX_LENGTH, incomplete)
     assert.deepStrictEqual(out.obj, {
       type: 'Object',
       notCapturedReason: 'timeout',
@@ -73,7 +84,7 @@ describe('Debugger snapshot time budget', () => {
       },
     }]
 
-    const out = processRawState(raw, MAX_LENGTH)
+    const out = processRawState(raw, MAX_LENGTH, incomplete)
     assert.deepStrictEqual(out.arr, {
       type: 'Array',
       notCapturedReason: 'timeout',
@@ -101,7 +112,7 @@ describe('Debugger snapshot time budget', () => {
       },
     }]
 
-    const out = processRawState(raw, MAX_LENGTH)
+    const out = processRawState(raw, MAX_LENGTH, incomplete)
     assert.deepStrictEqual(out.map, {
       type: 'Map',
       notCapturedReason: 'timeout',
@@ -129,7 +140,7 @@ describe('Debugger snapshot time budget', () => {
       },
     }]
 
-    const out = processRawState(raw, MAX_LENGTH)
+    const out = processRawState(raw, MAX_LENGTH, incomplete)
     assert.deepStrictEqual(out.set, {
       type: 'Set',
       notCapturedReason: 'timeout',
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 88477a014d2..54620cf8604 100644
--- a/packages/dd-trace/test/debugger/devtools_client/snapshot/utils.js
+++ b/packages/dd-trace/test/debugger/devtools_client/snapshot/utils.js
@@ -101,15 +101,19 @@ async function teardown () {
   await session.post('Debugger.disable')
 }
 
-async function setAndTriggerBreakpoint (path, line) {
-  const { run, scriptId } = require(path)
+/**
+ * @param {string} path - The target code file
+ * @param {number} line - The line to break on
+ * @param {() => void} [trigger] - The function hitting the breakpoint. Defaults to the target's `run` export.
+ */
+async function setAndTriggerBreakpoint (path, line, trigger = require(path).run) {
   await session.post('Debugger.setBreakpoint', {
     location: {
-      scriptId: await scriptId,
+      scriptId: await require(path).scriptId,
       lineNumber: line - 1, // Beware! lineNumber is zero-indexed
     },
   })
-  run()
+  trigger()
 }
 
 function assertOnBreakpoint (done, snapshotConfig, callback) {
@@ -123,8 +127,9 @@ function assertOnBreakpoint (done, snapshotConfig, callback) {
   session.once('Debugger.paused', ({ params }) => {
     assert.strictEqual(params.hitBreakpoints.length, 1)
 
-    getLocalStateForCallFrame(params.callFrames[0], snapshotConfig).then(({ processLocalState }) => {
-      callback(processLocalState())
+    getLocalStateForCallFrame(params.callFrames[0], snapshotConfig).then(({ processLocalState, incomplete }) => {
+      const state = processLocalState()
+      callback(state, incomplete)
       done()
     }).catch(done)
   })
diff --git a/packages/dd-trace/test/debugger/guardrail-metrics.spec.js b/packages/dd-trace/test/debugger/guardrail-metrics.spec.js
new file mode 100644
index 00000000000..29e3f9fd9de
--- /dev/null
+++ b/packages/dd-trace/test/debugger/guardrail-metrics.spec.js
@@ -0,0 +1,121 @@
+'use strict'
+
+const assert = require('node:assert/strict')
+
+const { beforeEach, describe, it } = require('mocha')
+require('../setup/mocha')
+
+const {
+  DROPPED_REASON,
+  EVENT_TYPE,
+  GuardrailMetrics,
+  INCOMPLETE_REASON,
+  SKIPPED_REASON,
+  TELEMETRY_NAMESPACE,
+} = require('../../src/debugger/guardrail-metrics')
+
+describe('debugger/guardrail-metrics', () => {
+  /** @type {GuardrailMetrics} */
+  let metrics
+  /** @type {SharedArrayBuffer} */
+  let buffer
+
+  beforeEach(() => {
+    buffer = GuardrailMetrics.createBuffer()
+    metrics = new GuardrailMetrics(buffer)
+  })
+
+  it('should use the telemetry namespace shared with the other tracers', () => {
+    assert.strictEqual(TELEMETRY_NAMESPACE, 'live_debugger')
+  })
+
+  it('should create a shared buffer', () => {
+    assert.ok(buffer instanceof SharedArrayBuffer)
+  })
+
+  it('should report nothing when no counter was incremented', () => {
+    assert.deepStrictEqual(drain(), [])
+  })
+
+  it('should report skipped events by reason and event type', () => {
+    metrics.eventSkipped(SKIPPED_REASON.RATE_LIMIT_GLOBAL, EVENT_TYPE.SNAPSHOT)
+    metrics.eventSkipped(SKIPPED_REASON.RATE_LIMIT_PROBE, EVENT_TYPE.SNAPSHOT)
+    metrics.eventSkipped(SKIPPED_REASON.RATE_LIMIT_PROBE, EVENT_TYPE.SNAPSHOT)
+    metrics.eventSkipped(SKIPPED_REASON.RATE_LIMIT_PROBE, EVENT_TYPE.LOG)
+    metrics.eventSkipped(SKIPPED_REASON.EVALUATION_TIMEOUT, EVENT_TYPE.LOG)
+
+    assert.deepStrictEqual(drain(), [
+      ['events.skipped', ['event_type:snapshot', 'reason:rateLimitGlobal'], 1],
+      ['events.skipped', ['event_type:snapshot', 'reason:rateLimitProbe'], 2],
+      ['events.skipped', ['event_type:log', 'reason:rateLimitProbe'], 1],
+      ['events.skipped', ['event_type:log', 'reason:evaluationTimeout'], 1],
+    ])
+  })
+
+  it('should report dropped events by reason and event type', () => {
+    metrics.eventDropped(DROPPED_REASON.QUEUE_FULL, EVENT_TYPE.DIAGNOSTIC)
+    metrics.eventDropped(DROPPED_REASON.QUEUE_FULL, EVENT_TYPE.SNAPSHOT, 5)
+    metrics.eventDropped(DROPPED_REASON.PAYLOAD_TOO_LARGE, EVENT_TYPE.LOG)
+
+    assert.deepStrictEqual(drain(), [
+      ['events.dropped', ['event_type:snapshot', 'reason:queueFull'], 5],
+      ['events.dropped', ['event_type:diagnostic', 'reason:queueFull'], 1],
+      ['events.dropped', ['event_type:log', 'reason:payloadTooLarge'], 1],
+    ])
+  })
+
+  it('should report each incomplete capture reason once per event', () => {
+    metrics.captureIncomplete(
+      INCOMPLETE_REASON.DEPTH | INCOMPLETE_REASON.STRING_LENGTH | INCOMPLETE_REASON.OTHER,
+      EVENT_TYPE.SNAPSHOT
+    )
+    metrics.captureIncomplete(INCOMPLETE_REASON.DEPTH, EVENT_TYPE.SNAPSHOT)
+    metrics.captureIncomplete(
+      INCOMPLETE_REASON.RUNTIME_ERROR | INCOMPLETE_REASON.TIMEOUT | INCOMPLETE_REASON.FIELD_COUNT |
+        INCOMPLETE_REASON.COLLECTION_SIZE | INCOMPLETE_REASON.PAYLOAD_TOO_LARGE,
+      EVENT_TYPE.LOG
+    )
+    metrics.captureIncomplete(0, EVENT_TYPE.LOG)
+
+    assert.deepStrictEqual(drain(), [
+      ['capture.incomplete', ['event_type:log', 'reason:runtimeError'], 1],
+      ['capture.incomplete', ['event_type:log', 'reason:timeout'], 1],
+      ['capture.incomplete', ['event_type:snapshot', 'reason:depth'], 2],
+      ['capture.incomplete', ['event_type:log', 'reason:fieldCount'], 1],
+      ['capture.incomplete', ['event_type:log', 'reason:collectionSize'], 1],
+      ['capture.incomplete', ['event_type:snapshot', 'reason:stringLength'], 1],
+      ['capture.incomplete', ['event_type:log', 'reason:payloadTooLarge'], 1],
+      ['capture.incomplete', ['event_type:snapshot', 'reason:other'], 1],
+    ])
+  })
+
+  it('should reset the counters when drained', () => {
+    metrics.eventSkipped(SKIPPED_REASON.RATE_LIMIT_GLOBAL, EVENT_TYPE.SNAPSHOT)
+    drain()
+
+    assert.deepStrictEqual(drain(), [])
+  })
+
+  it('should share the counters between instances backed by the same buffer', () => {
+    const other = new GuardrailMetrics(buffer)
+
+    metrics.eventSkipped(SKIPPED_REASON.RATE_LIMIT_GLOBAL, EVENT_TYPE.SNAPSHOT)
+    other.eventSkipped(SKIPPED_REASON.RATE_LIMIT_GLOBAL, EVENT_TYPE.SNAPSHOT)
+
+    assert.deepStrictEqual(drain(), [
+      ['events.skipped', ['event_type:snapshot', 'reason:rateLimitGlobal'], 2],
+    ])
+    assert.deepStrictEqual(drain(other), [])
+  })
+
+  /**
+   * @param {GuardrailMetrics} [instance]
+   * @returns {Array<[string, string[], number]>}
+   */
+  function drain (instance = metrics) {
+    /** @type {Array<[string, string[], number]>} */
+    const reported = []
+    instance.drain((metric, tags, count) => reported.push([metric, tags, count]))
+    return reported
+  }
+})
diff --git a/packages/dd-trace/test/debugger/index.spec.js b/packages/dd-trace/test/debugger/index.spec.js
index 31d5a6671c4..215d40a92e8 100644
--- a/packages/dd-trace/test/debugger/index.spec.js
+++ b/packages/dd-trace/test/debugger/index.spec.js
@@ -3,12 +3,16 @@
 const assert = require('node:assert/strict')
 const { inspect } = require('node:util')
 
+const dc = require('dc-polyfill')
 const { describe, it, beforeEach, afterEach } = require('mocha')
 const sinon = require('sinon')
 const proxyquire = require('proxyquire')
 
 require('../setup/mocha')
 
+const telemetryMetrics = require('../../src/telemetry/metrics')
+const { GuardrailMetrics, TELEMETRY_NAMESPACE } = require('../../src/debugger/guardrail-metrics')
+
 describe('debugger/index', () => {
   let DynamicInstrumentation
   let Worker
@@ -339,6 +343,110 @@ describe('debugger/index', () => {
     })
   })
 
+  describe('guardrail metrics', () => {
+    const appClosingChannel = dc.channel('datadog:telemetry:app-closing')
+    /** @type {sinon.SinonFakeTimers} */
+    let clock
+
+    beforeEach(() => {
+      clock = sinon.useFakeTimers({ toFake: ['setInterval', 'clearInterval'] })
+      telemetryMetrics.manager.delete(TELEMETRY_NAMESPACE)
+    })
+
+    afterEach(() => {
+      clock.restore()
+      telemetryMetrics.manager.delete(TELEMETRY_NAMESPACE)
+    })
+
+    it('should share the guardrail counters with the worker', () => {
+      DynamicInstrumentation.start(config, rc)
+
+      const { workerData } = Worker.firstCall.args[1]
+      assert.ok(workerData.guardrailMetricsBuffer instanceof SharedArrayBuffer)
+    })
+
+    it('should periodically report the counters as telemetry metrics', () => {
+      DynamicInstrumentation.start(config, rc)
+      const workerMetrics = new GuardrailMetrics(Worker.firstCall.args[1].workerData.guardrailMetricsBuffer)
+
+      workerMetrics.eventDropped(0, 0, 3) // queueFull, snapshot
+      workerMetrics.captureIncomplete(0b100, 1) // depth, log
+
+      assert.deepStrictEqual(getTelemetryMetrics(), [], 'should not report before the flush interval')
+
+      clock.tick(10_000)
+
+      assert.deepStrictEqual(getTelemetryMetrics(), [
+        { metric: 'events.dropped', tags: ['event_type:snapshot', 'reason:queueFull'], value: 3 },
+        { metric: 'capture.incomplete', tags: ['event_type:log', 'reason:depth'], value: 1 },
+      ])
+
+      workerMetrics.eventDropped(0, 0, 1)
+      clock.tick(10_000)
+
+      assert.deepStrictEqual(getTelemetryMetrics(), [
+        { metric: 'events.dropped', tags: ['event_type:snapshot', 'reason:queueFull'], value: 4 },
+        { metric: 'capture.incomplete', tags: ['event_type:log', 'reason:depth'], value: 1 },
+      ], 'should accumulate into the same telemetry counters')
+    })
+
+    it('should report the remaining counters and stop the timer when stopped', () => {
+      DynamicInstrumentation.start(config, rc)
+      const workerMetrics = new GuardrailMetrics(Worker.firstCall.args[1].workerData.guardrailMetricsBuffer)
+
+      workerMetrics.eventSkipped(1, 0) // rateLimitProbe, snapshot
+      DynamicInstrumentation.stop()
+
+      assert.deepStrictEqual(getTelemetryMetrics(), [
+        { metric: 'events.skipped', tags: ['event_type:snapshot', 'reason:rateLimitProbe'], value: 1 },
+      ])
+      assert.strictEqual(clock.countTimers(), 0)
+    })
+
+    it('should report the counters when telemetry is about to send its final metrics', () => {
+      DynamicInstrumentation.start(config, rc)
+      const workerMetrics = new GuardrailMetrics(Worker.firstCall.args[1].workerData.guardrailMetricsBuffer)
+
+      workerMetrics.eventDropped(0, 1, 2) // queueFull, log
+      appClosingChannel.publish()
+
+      assert.deepStrictEqual(getTelemetryMetrics(), [
+        { metric: 'events.dropped', tags: ['event_type:log', 'reason:queueFull'], value: 2 },
+      ], 'should report without waiting for the flush interval')
+
+      DynamicInstrumentation.stop()
+      workerMetrics.eventDropped(0, 1, 1)
+      appClosingChannel.publish()
+
+      assert.deepStrictEqual(getTelemetryMetrics(), [
+        { metric: 'events.dropped', tags: ['event_type:log', 'reason:queueFull'], value: 2 },
+      ], 'should stop listening once stopped')
+    })
+
+    it('should not keep the process alive', () => {
+      const setIntervalSpy = sinon.spy(global, 'setInterval')
+      try {
+        DynamicInstrumentation.start(config, rc)
+      } finally {
+        setIntervalSpy.restore()
+      }
+
+      sinon.assert.calledOnce(setIntervalSpy)
+      assert.strictEqual(setIntervalSpy.firstCall.returnValue.hasRef(), false)
+    })
+
+    /**
+     * @returns {Array<{ metric: string, tags: string[], value: number }>}
+     */
+    function getTelemetryMetrics () {
+      const namespace = telemetryMetrics.manager.get(TELEMETRY_NAMESPACE)
+      if (namespace === undefined) return []
+      return [...namespace.metrics.values()]
+        .filter((metric) => metric.hasPoints())
+        .map(({ metric, tags, points }) => ({ metric, tags, value: points[0][1] }))
+    }
+  })
+
   describe('readProbeFile', () => {
     it('should do nothing when path is not provided', () => {
       // probeFile is undefined by default (not set in config)
diff --git a/packages/dd-trace/test/debugger/probe_sampler.spec.js b/packages/dd-trace/test/debugger/probe_sampler.spec.js
index 71bf9b31d65..e3f89ad4e56 100644
--- a/packages/dd-trace/test/debugger/probe_sampler.spec.js
+++ b/packages/dd-trace/test/debugger/probe_sampler.spec.js
@@ -11,6 +11,7 @@ const {
   SAMPLED_PROBE_INDEXES_START,
   SAMPLED_PROBE_OVERFLOW_INDEX,
 } = require('../../src/debugger/probe_sampler_constants')
+const { GuardrailMetrics } = require('../../src/debugger/guardrail-metrics')
 const { installProbeSampler, uninstallProbeSampler } = require('../../src/debugger/probe_sampler')
 const {
   compileBreakpointCondition,
@@ -21,6 +22,9 @@ const { MAX_SNAPSHOTS_PER_SECOND_GLOBALLY } = require('../../src/debugger/devtoo
 const ddTraceSymbol = Symbol.for('dd-trace')
 const samplerSymbol = Symbol.for('dd-trace.debugger.probeSampler')
 
+/** @type {GuardrailMetrics} */
+let guardrailMetrics
+
 describe('probe sampler', function () {
   /** @type {typeof process.hrtime.bigint} */
   let originalHrtimeBigint
@@ -29,6 +33,7 @@ describe('probe sampler', function () {
 
   beforeEach(function () {
     delete getDatadogGlobal()[samplerSymbol]
+    guardrailMetrics = new GuardrailMetrics(GuardrailMetrics.createBuffer())
     originalHrtimeBigint = process.hrtime.bigint
     now = 1_000_000_000n
     process.hrtime.bigint = () => now
@@ -41,7 +46,7 @@ describe('probe sampler', function () {
 
   describe('shared buffer', function () {
     it('should create a shared buffer with the expected layout', function () {
-      const buffer = installProbeSampler()
+      const buffer = installProbeSampler(guardrailMetrics)
       const sampledProbeIndexes = new Int32Array(buffer)
 
       assert(buffer instanceof SharedArrayBuffer)
@@ -49,7 +54,7 @@ describe('probe sampler', function () {
     })
 
     it('should initialize the shared buffer', function () {
-      const installedBuffer = installProbeSampler()
+      const installedBuffer = installProbeSampler(guardrailMetrics)
       const installedSampledProbeIndexes = new Int32Array(installedBuffer)
 
       assert.strictEqual(Atomics.load(installedSampledProbeIndexes, SAMPLED_PROBE_COUNT_INDEX), 0)
@@ -151,6 +156,9 @@ describe('probe sampler', function () {
 
       assert.strictEqual(sampler.makeSampleDecision(7, 'probe-1', 200000n, false), false)
       assert.strictEqual(Atomics.load(sampledProbeIndexes, SAMPLED_PROBE_COUNT_INDEX), 1)
+      assert.deepStrictEqual(drainGuardrailMetrics(), [
+        ['events.skipped', ['event_type:log', 'reason:rateLimitProbe'], 1],
+      ])
 
       now += 1n
 
@@ -159,6 +167,22 @@ describe('probe sampler', function () {
 
       assert.strictEqual(sampler.makeSampleDecision(7, 'probe-1', 200000n, false), false)
       assert.strictEqual(Atomics.load(sampledProbeIndexes, SAMPLED_PROBE_COUNT_INDEX), 2)
+      assert.deepStrictEqual(drainGuardrailMetrics(), [
+        ['events.skipped', ['event_type:log', 'reason:rateLimitProbe'], 1],
+      ])
+    })
+
+    it('should count snapshot-producing probes skipped by the per-probe rate limit as snapshots', function () {
+      installSampler()
+      const sampler = getSampler()
+
+      assert.strictEqual(sampler.makeSampleDecision(7, 'probe-1', 200000n, true), true)
+      assert.strictEqual(sampler.makeSampleDecision(7, 'probe-1', 200000n, true), false)
+      assert.strictEqual(sampler.makeSampleDecision(7, 'probe-1', 200000n, true), false)
+
+      assert.deepStrictEqual(drainGuardrailMetrics(), [
+        ['events.skipped', ['event_type:snapshot', 'reason:rateLimitProbe'], 2],
+      ])
     })
 
     it('should allow a removed probe to sample again immediately', function () {
@@ -190,6 +214,9 @@ describe('probe sampler', function () {
         Atomics.load(sampledProbeIndexes, SAMPLED_PROBE_COUNT_INDEX),
         MAX_SNAPSHOTS_PER_SECOND_GLOBALLY + 1
       )
+      assert.deepStrictEqual(drainGuardrailMetrics(), [
+        ['events.skipped', ['event_type:snapshot', 'reason:rateLimitGlobal'], 1],
+      ])
     })
 
     it('should not advance the sampled probe count when global snapshot rate rejects a probe', function () {
@@ -221,6 +248,7 @@ describe('probe sampler', function () {
 
       now += 1_000_000_001n
       assert.strictEqual(sampler.makeSampleDecision(99, 'snapshot-next-window', 0n, true), true)
+      assert.deepStrictEqual(drainGuardrailMetrics(), [])
     })
 
     it('should set overflow and skip probes when the shared buffer is full', function () {
@@ -234,6 +262,8 @@ describe('probe sampler', function () {
         false
       )
       assert.strictEqual(Atomics.load(sampledProbeIndexes, SAMPLED_PROBE_OVERFLOW_INDEX), 1)
+      // The overflow guard is an internal limit without a canonical skip reason, so it's not reported as a skip
+      assert.deepStrictEqual(drainGuardrailMetrics(), [])
     })
   })
 })
@@ -242,7 +272,19 @@ describe('probe sampler', function () {
  * Install the runtime sampler for tests.
  */
 function installSampler () {
-  return new Int32Array(installProbeSampler())
+  return new Int32Array(installProbeSampler(guardrailMetrics))
+}
+
+/**
+ * Drain the guardrail counters recorded by the runtime sampler.
+ *
+ * @returns {Array<[string, string[], number]>} The non-zero counters as `[metric, tags, count]` tuples.
+ */
+function drainGuardrailMetrics () {
+  /** @type {Array<[string, string[], number]>} */
+  const reported = []
+  guardrailMetrics.drain((metric, tags, count) => reported.push([metric, tags, count]))
+  return reported
 }
 
 /**
diff --git a/packages/dd-trace/test/telemetry/index.spec.js b/packages/dd-trace/test/telemetry/index.spec.js
index 493b8a35cd8..2379c3eaeac 100644
--- a/packages/dd-trace/test/telemetry/index.spec.js
+++ b/packages/dd-trace/test/telemetry/index.spec.js
@@ -138,6 +138,7 @@ describe('telemetry', () => {
       circularObject,
       appsec: { enabled: true, DD_API_SECURITY_ENDPOINT_COLLECTION_ENABLED: false },
       profiling: { DD_PROFILING_ENABLED: 'true' },
+      dynamicInstrumentation: { enabled: true },
       peerServiceMapping: {
         service_1: 'remapped_service_1',
         service_2: 'remapped_service_2',
@@ -159,6 +160,7 @@ describe('telemetry', () => {
       assert.deepStrictEqual(payload.products, {
         appsec: { enabled: true },
         profiler: { version: tracerVersion, enabled: true },
+        dynamic_instrumentation: { enabled: true },
       })
       assert.deepStrictEqual(payload.install_signature, {
         install_id: '68e75c48-57ca-4a12-adfc-575c4b05fcbe',
@@ -286,6 +288,7 @@ describe('telemetry', () => {
       },
       appsec: { enabled: false, DD_APPSEC_SCA_ENABLED: undefined },
       profiling: { DD_PROFILING_ENABLED: false },
+      dynamicInstrumentation: { enabled: false },
     }, {
       _pluginsByName: pluginsByName,
     })
@@ -342,6 +345,7 @@ describe('telemetry app-heartbeat', () => {
       version: '1.2.3-beta4',
       appsec: { enabled: true, DD_API_SECURITY_ENDPOINT_COLLECTION_ENABLED: false },
       profiling: { DD_PROFILING_ENABLED: true },
+      dynamicInstrumentation: { enabled: false },
       env: 'preprod',
       tags: {
         'runtime-id': '1a2b3c',
@@ -410,6 +414,7 @@ describe('Telemetry extended heartbeat', () => {
       version: '1.2.3-beta4',
       appsec: { enabled: true, DD_API_SECURITY_ENDPOINT_COLLECTION_ENABLED: false },
       profiling: { DD_PROFILING_ENABLED: true },
+      dynamicInstrumentation: { enabled: false },
       env: 'preprod',
       tags: {
         'runtime-id': '1a2b3c',
@@ -455,6 +460,7 @@ describe('Telemetry extended heartbeat', () => {
       version: '1.2.3-beta4',
       appsec: { enabled: true, DD_API_SECURITY_ENDPOINT_COLLECTION_ENABLED: false },
       profiling: { DD_PROFILING_ENABLED: true },
+      dynamicInstrumentation: { enabled: false },
       env: 'preprod',
       tags: {
         'runtime-id': '1a2b3c',
@@ -562,6 +568,7 @@ describe('Telemetry extended heartbeat', () => {
       version: '1.2.3-beta4',
       appsec: { enabled: true, DD_API_SECURITY_ENDPOINT_COLLECTION_ENABLED: false },
       profiling: { DD_PROFILING_ENABLED: true },
+      dynamicInstrumentation: { enabled: false },
       env: 'preprod',
       tags: {
         'runtime-id': '1a2b3c',
@@ -658,6 +665,7 @@ describe('Telemetry retry', () => {
       version: '1.2.3-beta4',
       appsec: { enabled: true, DD_API_SECURITY_ENDPOINT_COLLECTION_ENABLED: false },
       profiling: { DD_PROFILING_ENABLED: true },
+      dynamicInstrumentation: { enabled: false },
       env: 'preprod',
       tags: {
         'runtime-id': '1a2b3c',
@@ -751,6 +759,7 @@ describe('Telemetry retry', () => {
       version: '1.2.3-beta4',
       appsec: { enabled: true, DD_API_SECURITY_ENDPOINT_COLLECTION_ENABLED: false },
       profiling: { DD_PROFILING_ENABLED: true },
+      dynamicInstrumentation: { enabled: false },
       env: 'preprod',
       tags: {
         'runtime-id': '1a2b3c',
@@ -826,6 +835,7 @@ describe('Telemetry retry', () => {
       version: '1.2.3-beta4',
       appsec: { enabled: true, DD_API_SECURITY_ENDPOINT_COLLECTION_ENABLED: false },
       profiling: { DD_PROFILING_ENABLED: true },
+      dynamicInstrumentation: { enabled: false },
       env: 'preprod',
       tags: {
         'runtime-id': '1a2b3c',
@@ -892,6 +902,7 @@ describe('Telemetry retry', () => {
       version: '1.2.3-beta4',
       appsec: { enabled: true, DD_API_SECURITY_ENDPOINT_COLLECTION_ENABLED: false },
       profiling: { DD_PROFILING_ENABLED: true },
+      dynamicInstrumentation: { enabled: false },
       env: 'preprod',
       tags: {
         'runtime-id': '1a2b3c',
@@ -980,6 +991,7 @@ describe('Telemetry retry', () => {
       version: '1.2.3-beta4',
       appsec: { enabled: true, DD_API_SECURITY_ENDPOINT_COLLECTION_ENABLED: false },
       profiling: { DD_PROFILING_ENABLED: true },
+      dynamicInstrumentation: { enabled: false },
       env: 'preprod',
       tags: {
         'runtime-id': '1a2b3c',
@@ -1083,6 +1095,7 @@ describe('AVM OSS', () => {
             },
             appsec: { enabled: false, DD_API_SECURITY_ENDPOINT_COLLECTION_ENABLED: false },
             profiling: { DD_PROFILING_ENABLED: false },
+            dynamicInstrumentation: { enabled: false },
           }
         })