diff --git a/packages/datadog-instrumentations/src/http/client.js b/packages/datadog-instrumentations/src/http/client.js
index d12ff5fb62..c266c84479 100644
--- a/packages/datadog-instrumentations/src/http/client.js
+++ b/packages/datadog-instrumentations/src/http/client.js
@@ -14,6 +14,7 @@ const finishChannel = channel('apm:http:client:request:finish')
 const endChannel = channel('apm:http:client:request:end')
 const asyncStartChannel = channel('apm:http:client:request:asyncStart')
 const errorChannel = channel('apm:http:client:request:error')
+const responseStartChannel = channel('apm:http:client:response:start')
 const responseFinishChannel = channel('apm:http:client:response:finish')
 
 addHook({ name: 'http' }, hookFn)
@@ -78,10 +79,15 @@ function setupResponseInstrumentation (ctx, res) {
   let dataReadStarted = false
 
   const { shouldCollectBody } = ctx
-  const bodyChunks = shouldCollectBody ? [] : null
+
+  let bodyChunks = null
+
+  if (shouldCollectBody) {
+    bodyChunks = []
+  }
 
   const collectChunk = chunk => {
-    if (!shouldCollectBody || !chunk) return
+    if (!bodyChunks || !chunk) return
 
     if (typeof chunk === 'string') {
       bodyChunks.push(chunk)
@@ -229,6 +235,10 @@ function patch (http, methodName) {
                 res.once('end', finish)
                 res.once(errorMonitor, finish)
 
+                if (responseStartChannel.hasSubscribers) {
+                  responseStartChannel.publish({ ctx, res })
+                }
+
                 const instrumentation = setupResponseInstrumentation(ctx, res)
 
                 if (!instrumentation) {
diff --git a/packages/datadog-instrumentations/test/http.spec.js b/packages/datadog-instrumentations/test/http.spec.js
index 8cc194ceba..f97f6431e0 100644
--- a/packages/datadog-instrumentations/test/http.spec.js
+++ b/packages/datadog-instrumentations/test/http.spec.js
@@ -1,6 +1,7 @@
 'use strict'
 
 const assert = require('node:assert')
+const { URL } = require('node:url')
 const { inspect } = require('node:util')
 const dc = require('dc-polyfill')
 const { describe, it, beforeEach, afterEach } = require('mocha')
@@ -16,6 +17,7 @@ describe('client', () => {
   const asyncStartChannel = dc.channel('apm:http:client:request:asyncStart')
   const errorChannel = dc.channel('apm:http:client:request:error')
   const responseFinishChannel = dc.channel('apm:http:client:response:finish')
+  const responseStartChannel = dc.channel('apm:http:client:response:start')
 
   before(async () => {
     await agent.load('http')
@@ -44,6 +46,15 @@ describe('client', () => {
     errorChannel.unsubscribe(errorChannelCb)
   })
 
+  // originalUrl is the raw http.request first arg (string, URL, or options); uri is always a string.
+  function getRequestUrlString (args) {
+    if (!args) return undefined
+    const { originalUrl, uri } = args
+    if (typeof originalUrl === 'string') return originalUrl
+    if (originalUrl instanceof URL) return originalUrl.href
+    return uri
+  }
+
   /*
    * Necessary because the tracer makes extra requests to the agent
    * and the same stub could be called multiple times
@@ -51,7 +62,7 @@ describe('client', () => {
   function getContextFromStubByUrl (url, stub) {
     for (const args of stub.args) {
       const arg = args[0]
-      if (arg.args?.originalUrl === url) {
+      if (getRequestUrlString(arg.args) === url) {
         return arg
       }
     }
@@ -60,9 +71,7 @@ describe('client', () => {
 
   function stubHasResponseForUrl (url, stub) {
     return stub.args.some(([payload]) => {
-      const ctx = payload?.ctx
-      const originalUrl = ctx?.args?.originalUrl || ctx?.args?.uri
-      return originalUrl === url
+      return getRequestUrlString(payload?.ctx?.args) === url
     })
   }
 
@@ -194,7 +203,7 @@ describe('client', () => {
       })
 
       describe('response finish channel', () => {
-        let responseFinishChannelCb
+        let responseFinishChannelCb, responseStartChannelCb
 
         before(() => {
           http = require(httpSchema)
@@ -204,23 +213,49 @@ describe('client', () => {
         beforeEach(() => {
           responseFinishChannelCb = sinon.stub()
           responseFinishChannel.subscribe(responseFinishChannelCb)
+
+          responseStartChannelCb = sinon.stub()
+          responseStartChannel.subscribe(responseStartChannelCb)
         })
 
         afterEach(() => {
           responseFinishChannel.unsubscribe(responseFinishChannelCb)
+          responseStartChannel.unsubscribe(responseStartChannelCb)
+        })
+
+        it('publishes response:start before response:finish for the same request', (done) => {
+          http.get(url, (res) => {
+            res.resume()
+            res.on('end', () => {
+              try {
+                sinon.assert.called(responseStartChannelCb)
+                sinon.assert.called(responseFinishChannelCb)
+                assert.ok(responseStartChannelCb.calledBefore(responseFinishChannelCb))
+                done()
+              } catch (e) {
+                done(e)
+              }
+            })
+          })
         })
 
+        function isLocalServerRequest (args) {
+          const requestUrl = getRequestUrlString(args)
+          return typeof requestUrl === 'string' && requestUrl.startsWith('http://127.0.0.1:')
+        }
+
         function setCollectBody (ctx) {
-          if (ctx.args.originalUrl === url) {
+          const requestUrl = getRequestUrlString(ctx.args)
+          // External tests use `url` (datadoghq); local server tests use 127.0.0.1 and never match it.
+          if (requestUrl === url || isLocalServerRequest(ctx.args)) {
             ctx.shouldCollectBody = true
           }
         }
 
-        function getResponseFinishPayload (url, stub) {
+        function getResponseFinishPayload (requestUrl, stub) {
           for (const args of stub.args) {
             const payload = args[0]
-            const originalUrl = payload?.ctx?.args?.originalUrl || payload?.ctx?.args?.uri
-            if (originalUrl === url) {
+            if (getRequestUrlString(payload?.ctx?.args) === requestUrl) {
               return payload
             }
           }
@@ -252,121 +287,295 @@ describe('client', () => {
           })
         })
 
-        it('collects and concatenates all chunks when ctx.shouldCollectBody is true', (done) => {
-          startChannelCb.callsFake(setCollectBody)
+        // Local server tests use plain http:// URLs; skip under the https schema loop.
+        const describeIfHttp = httpSchema === 'http' ? describe : describe.skip
+
+        describeIfHttp('with local http server', () => {
+          function requestWithLocalServer ({ responseHeaders, responseBody, onResponse }, done) {
+            const server = http.createServer((req, res) => {
+              res.writeHead(200, responseHeaders)
+              if (responseBody != null) {
+                res.end(responseBody)
+              } else {
+                res.end()
+              }
+            })
 
-          const chunks = []
-          http.get(url, (res) => {
-            res.on('data', (chunk) => {
-              chunks.push(chunk)
+            server.listen(0, () => {
+              const localUrl = `http://127.0.0.1:${server.address().port}/`
+              startChannelCb.callsFake(setCollectBody)
+
+              http.get(localUrl, (res) => {
+                onResponse(res, server, localUrl, done)
+              }).on('error', (err) => {
+                server.close(() => done(err))
+              })
             })
-            res.on('end', () => {
-              try {
-                const payload = getResponseFinishPayload(url, responseFinishChannelCb)
-                assert(Buffer.isBuffer(payload.body), `Expected Buffer, got ${inspect(payload.body)}`)
+          }
 
-                const expectedBody = Buffer.concat(chunks)
-                assert(payload.body.equals(expectedBody), `Got: ${inspect(payload.body)}`)
+          function requestWithLocalJsonBody (onResponse, done) {
+            const body = JSON.stringify({ ok: true })
+            requestWithLocalServer({
+              responseHeaders: {
+                'content-type': 'application/json',
+                'content-length': String(Buffer.byteLength(body)),
+              },
+              responseBody: body,
+              onResponse,
+            }, done)
+          }
 
-                done()
-              } catch (e) {
-                done(e)
+          it('collects and concatenates all chunks when ctx.shouldCollectBody is true', (done) => {
+            const chunks = []
+            requestWithLocalJsonBody((res, server, localUrl, done) => {
+              res.on('data', (chunk) => {
+                chunks.push(chunk)
+              })
+              res.on('end', () => {
+                try {
+                  const payload = getResponseFinishPayload(localUrl, responseFinishChannelCb)
+                  assert(Buffer.isBuffer(payload.body), `Expected Buffer, got ${inspect(payload.body)}`)
+
+                  const expectedBody = Buffer.concat(chunks)
+                  assert.deepStrictEqual(payload.body, expectedBody)
+
+                  server.close(() => done())
+                } catch (e) {
+                  server.close(() => done(e))
+                }
+              })
+            }, done)
+          })
+
+          it('collects and concatenates string chunks when using setEncoding', (done) => {
+            const chunks = []
+            requestWithLocalJsonBody((res, server, localUrl, done) => {
+              res.setEncoding('utf8')
+              const consume = () => {
+                let chunk
+                while ((chunk = res.read()) !== null) {
+                  chunks.push(chunk)
+                }
               }
-            })
+
+              res.on('readable', consume)
+              res.on('end', () => {
+                try {
+                  const payload = getResponseFinishPayload(localUrl, responseFinishChannelCb)
+                  assert.strictEqual(typeof payload.body, 'string')
+
+                  const expectedBody = chunks.join('')
+                  assert.strictEqual(payload.body, expectedBody)
+
+                  server.close(() => done())
+                } catch (e) {
+                  server.close(() => done(e))
+                }
+              })
+            }, done)
           })
-        })
 
-        it('collects and concatenates string chunks when using setEncoding', (done) => {
-          startChannelCb.callsFake(setCollectBody)
+          it('should collect data correctly when read and data are both used', (done) => {
+            const chunks = []
+            requestWithLocalJsonBody((res, server, localUrl, done) => {
+              res.setEncoding('utf8')
+              // eslint-disable-next-line sonarjs/no-identical-functions -- per-test chunks buffer
+              const consume = () => {
+                let chunk
+                while ((chunk = res.read()) !== null) {
+                  chunks.push(chunk)
+                }
+              }
+              res.on('data', () => {})
 
-          const chunks = []
-          http.get(url, (res) => {
-            res.setEncoding('utf8')
-            const consume = () => {
-              let chunk
-              while ((chunk = res.read()) !== null) {
-                chunks.push(chunk)
+              res.on('readable', consume)
+              res.on('end', () => {
+                try {
+                  const payload = getResponseFinishPayload(localUrl, responseFinishChannelCb)
+                  assert.strictEqual(typeof payload.body, 'string')
+
+                  const expectedBody = chunks.join('')
+                  assert.strictEqual(payload.body, expectedBody)
+
+                  server.close(() => done())
+                } catch (e) {
+                  server.close(() => done(e))
+                }
+              })
+            }, done)
+          })
+
+          it('should collect data correctly when read and data are both used in different order', (done) => {
+            const chunks = []
+            requestWithLocalJsonBody((res, server, localUrl, done) => {
+              let onDataAdded = false
+              res.setEncoding('utf8')
+              const consume = () => {
+                let chunk
+                while ((chunk = res.read(100)) !== null) {
+                  if (!onDataAdded) {
+                    onDataAdded = true
+                    res.on('data', () => {})
+                  }
+                  chunks.push(chunk)
+                }
               }
+              res.on('readable', consume)
+              res.on('end', () => {
+                try {
+                  const payload = getResponseFinishPayload(localUrl, responseFinishChannelCb)
+                  assert.strictEqual(typeof payload.body, 'string')
+                  const expectedBody = chunks.join('')
+                  assert.strictEqual(payload.body, expectedBody)
+
+                  server.close(() => done())
+                } catch (e) {
+                  server.close(() => done(e))
+                }
+              })
+            }, done)
+          })
+
+          describe('response body limits', () => {
+            const instrumentationTestSupportedMime = new Set([
+              'application/json',
+              'text/json',
+              'application/x-www-form-urlencoded',
+            ])
+
+            let instrumentationBodyTestMaxBytes = 1024 * 1024
+
+            function extractMimeTypeForInstrumentationTest (headerVal) {
+              if (headerVal == null) return null
+              const raw = Array.isArray(headerVal) ? headerVal[0] : headerVal
+              const base = String(raw).split(';')[0].trim().toLowerCase()
+              return base || null
             }
 
-            res.on('readable', consume)
-            res.on('end', () => {
-              try {
-                const payload = getResponseFinishPayload(url, responseFinishChannelCb)
-                assert.strictEqual(typeof payload.body, 'string')
+            function parseContentLengthForInstrumentationTest (headerVal) {
+              if (headerVal == null) return null
+              const value = Array.isArray(headerVal) ? headerVal[0] : headerVal
+              const parsed = Number.parseInt(String(value), 10)
+              if (!Number.isFinite(parsed) || parsed < 0) return null
+              return parsed
+            }
 
-                const expectedBody = chunks.join('')
-                assert.strictEqual(payload.body, expectedBody)
+            function planCtxBodyCollectionForInstrumentationTest ({ ctx, res }) {
+              if (!ctx.shouldCollectBody) return
 
-                done()
-              } catch (e) {
-                done(e)
+              const mime = extractMimeTypeForInstrumentationTest(res.headers?.['content-type'])
+              if (!mime || !instrumentationTestSupportedMime.has(mime)) {
+                delete ctx.shouldCollectBody
+                return
               }
-            })
-          })
-        })
 
-        it('should collect data correctly when read and data are both used', (done) => {
-          startChannelCb.callsFake(setCollectBody)
+              const declared = parseContentLengthForInstrumentationTest(res.headers?.['content-length'])
+              if (declared == null || declared === 0) {
+                delete ctx.shouldCollectBody
+                return
+              }
 
-          const chunks = []
-          http.get(url, (res) => {
-            res.setEncoding('utf8')
-            // eslint-disable-next-line sonarjs/no-identical-functions -- per-test chunks buffer
-            const consume = () => {
-              let chunk
-              while ((chunk = res.read()) !== null) {
-                chunks.push(chunk)
+              if (declared > instrumentationBodyTestMaxBytes) {
+                delete ctx.shouldCollectBody
               }
             }
-            res.on('data', () => {})
 
-            res.on('readable', consume)
-            res.on('end', () => {
-              try {
-                const payload = getResponseFinishPayload(url, responseFinishChannelCb)
-                assert.strictEqual(typeof payload.body, 'string')
+            beforeEach(() => {
+              instrumentationBodyTestMaxBytes = 1024 * 1024
+              responseStartChannel.subscribe(planCtxBodyCollectionForInstrumentationTest)
+            })
 
-                const expectedBody = chunks.join('')
-                assert.strictEqual(payload.body, expectedBody)
+            afterEach(() => {
+              responseStartChannel.unsubscribe(planCtxBodyCollectionForInstrumentationTest)
+            })
 
-                done()
-              } catch (e) {
-                done(e)
-              }
+            it('ignores body when content-type is unsupported', (done) => {
+              requestWithLocalServer({
+                responseHeaders: {
+                  'content-type': 'image/png',
+                  'content-length': '4',
+                },
+                responseBody: 'test',
+                onResponse (res, server, localUrl, done) {
+                  res.on('data', () => {})
+                  res.on('end', () => {
+                    try {
+                      const payload = getResponseFinishPayload(localUrl, responseFinishChannelCb)
+                      assert.strictEqual(payload.body, null)
+                      server.close(() => done())
+                    } catch (e) {
+                      server.close(() => done(e))
+                    }
+                  })
+                },
+              }, done)
             })
-          })
-        })
 
-        it('should collect data correctly when read and data are both used in different order', (done) => {
-          startChannelCb.callsFake(setCollectBody)
+            it('ignores body when content-length is missing', (done) => {
+              requestWithLocalServer({
+                responseHeaders: {
+                  'content-type': 'application/json',
+                },
+                responseBody: '{"ok":true}',
+                onResponse (res, server, localUrl, done) {
+                  res.on('data', () => {})
+                  res.on('end', () => {
+                    try {
+                      const payload = getResponseFinishPayload(localUrl, responseFinishChannelCb)
+                      assert.strictEqual(payload.body, null)
+                      server.close(() => done())
+                    } catch (e) {
+                      server.close(() => done(e))
+                    }
+                  })
+                },
+              }, done)
+            })
 
-          const chunks = []
-          http.get(url, (res) => {
-            let onDataAdded = false
-            res.setEncoding('utf8')
-            const consume = () => {
-              let chunk
-              while ((chunk = res.read(100)) !== null) {
-                if (!onDataAdded) {
-                  onDataAdded = true
+            it('ignores body when content-length is zero', (done) => {
+              requestWithLocalServer({
+                responseHeaders: {
+                  'content-type': 'application/json',
+                  'content-length': '0',
+                },
+                responseBody: '',
+                onResponse (res, server, localUrl, done) {
                   res.on('data', () => {})
-                }
-                chunks.push(chunk)
-              }
-            }
-            res.on('readable', consume)
-            res.on('end', () => {
-              try {
-                const payload = getResponseFinishPayload(url, responseFinishChannelCb)
-                assert.strictEqual(typeof payload.body, 'string')
-                const expectedBody = chunks.join('')
-                assert.strictEqual(payload.body, expectedBody)
+                  res.on('end', () => {
+                    try {
+                      const payload = getResponseFinishPayload(localUrl, responseFinishChannelCb)
+                      assert.strictEqual(payload.body, null)
+                      server.close(() => done())
+                    } catch (e) {
+                      server.close(() => done(e))
+                    }
+                  })
+                },
+              }, done)
+            })
 
-                done()
-              } catch (e) {
-                done(e)
-              }
+            it('ignores body when content-length exceeds maxBytes', (done) => {
+              instrumentationBodyTestMaxBytes = 10
+
+              requestWithLocalServer({
+                responseHeaders: {
+                  'content-type': 'application/json',
+                  'content-length': '100',
+                },
+                responseBody: 'x'.repeat(100),
+                onResponse (res, server, localUrl, done) {
+                  res.on('data', () => {})
+                  res.on('end', () => {
+                    try {
+                      const payload = getResponseFinishPayload(localUrl, responseFinishChannelCb)
+                      assert.strictEqual(payload.body, null)
+                      server.close(() => done())
+                    } catch (e) {
+                      server.close(() => done(e))
+                    }
+                  })
+                },
+              }, done)
             })
           })
         })
diff --git a/packages/dd-trace/src/appsec/channels.js b/packages/dd-trace/src/appsec/channels.js
index 545a2dd764..6e5deab3ff 100644
--- a/packages/dd-trace/src/appsec/channels.js
+++ b/packages/dd-trace/src/appsec/channels.js
@@ -23,6 +23,7 @@ module.exports = {
   fsOperationStart: dc.channel('apm:fs:operation:start'),
   graphqlMiddlewareChannel: dc.tracingChannel('datadog:apollo:middleware'),
   httpClientRequestStart: dc.channel('apm:http:client:request:start'),
+  httpClientResponseStart: dc.channel('apm:http:client:response:start'),
   httpClientResponseFinish: dc.channel('apm:http:client:response:finish'),
   incomingHttpRequestEnd: dc.channel('dd-trace:incomingHttpRequestEnd'),
   incomingHttpRequestStart: dc.channel('dd-trace:incomingHttpRequestStart'),
diff --git a/packages/dd-trace/src/appsec/downstream_requests.js b/packages/dd-trace/src/appsec/downstream_requests.js
index 4d6be26e6f..ff73b2fe81 100644
--- a/packages/dd-trace/src/appsec/downstream_requests.js
+++ b/packages/dd-trace/src/appsec/downstream_requests.js
@@ -13,19 +13,32 @@ const {
 const KNUTH_FACTOR = 11400714819323199488n // eslint-disable-line unicorn/numeric-separators-style
 const UINT64_MAX = (1n << 64n) - 1n
 
+const SUPPORTED_RESPONSE_BODY_MIME_TYPES = new Set([
+  'application/json',
+  'text/json',
+  'application/x-www-form-urlencoded',
+])
+
+const RESPONSE_BODY_IGNORED_TAG_CONTENT_TYPE =
+  '_dd.appsec.downstream_request.response_body_ignored.content_type_invalid'
+const RESPONSE_BODY_IGNORED_TAG_CONTENT_LENGTH_MISSING =
+  '_dd.appsec.downstream_request.response_body_ignored.content_length_missing'
+const RESPONSE_BODY_IGNORED_TAG_CONTENT_LENGTH_TOO_BIG =
+  '_dd.appsec.downstream_request.response_body_ignored.content_length_too_big'
+
 let config
 let samplingRate
 let globalRequestCounter
 let bodyAnalysisCount
 let downstreamAnalysisCount
-let redirectBodyCollectionDecisions
+let responseBodyIgnoredCount
 
 function enable (_config) {
   config = _config
   globalRequestCounter = 0n
   bodyAnalysisCount = new WeakMap()
   downstreamAnalysisCount = new WeakMap()
-  redirectBodyCollectionDecisions = new WeakMap()
+  responseBodyIgnoredCount = new WeakMap()
 
   const bodyAnalysisSampleRate = config.appsec.apiSecurity?.downstreamBodyAnalysisSampleRate
   samplingRate = Math.min(Math.max(bodyAnalysisSampleRate, 0), 1)
@@ -42,49 +55,85 @@ function disable () {
   globalRequestCounter = null
   bodyAnalysisCount = null
   downstreamAnalysisCount = null
-  redirectBodyCollectionDecisions = null
+  responseBodyIgnoredCount = null
 }
 
 /**
- * Check we have a stored redirect body collection decision for a given URL.
- * @param {import('http').IncomingMessage} req outgoing request.
- * @param {string} outgoingUrl the URL being requested.
- * @returns {boolean} the stored decision
+ * @param {string|string[]|undefined} contentLength raw content-length header value.
+ * @returns {number|null} parsed content length or null when invalid.
  */
-function consumeRedirectBodyCollectionDecision (req, outgoingUrl) {
-  const decisions = redirectBodyCollectionDecisions.get(req)
-  if (!decisions) return false
+function parseContentLengthHeader (contentLength) {
+  if (contentLength == null) {
+    return null
+  }
+
+  const value = Array.isArray(contentLength) ? contentLength[0] : contentLength
+  const parsed = Number.parseInt(String(value), 10)
 
-  return decisions.delete(outgoingUrl)
+  if (!Number.isFinite(parsed) || parsed < 0) {
+    return null
+  }
+
+  return parsed
 }
 
 /**
- * Stores a redirect body collection decision for a follow-up request.
- * @param {import('http').IncomingMessage} req outgoing request.
- * @param {string} redirectUrl the URL to redirect to.
+ * Increments a response-body-ignored counter on the service-entry span.
+ * @param {import('http').IncomingMessage} req originating request.
+ * @param {string} tag full `_dd.appsec.downstream_request.response_body_ignored.*` span tag.
  */
-function storeRedirectBodyCollectionDecision (req, redirectUrl) {
-  let decisions = redirectBodyCollectionDecisions.get(req)
+function recordResponseBodyIgnored (req, tag) {
+  const span = web.root(req)
+  if (!span) return
 
-  if (!decisions) {
-    decisions = new Set()
-    redirectBodyCollectionDecisions.set(req, decisions)
+  let counts = responseBodyIgnoredCount.get(req)
+  if (!counts) {
+    counts = {}
+    responseBodyIgnoredCount.set(req, counts)
   }
 
-  decisions.add(redirectUrl)
+  const current = counts[tag] || 0
+  const next = current + 1
+  counts[tag] = next
+  span.setTag(tag, next)
 }
 
 /**
- * Determines whether the current downstream request/responses bodies should be sampled for analysis.
- * @param {import('http').IncomingMessage} req outgoing request.
- * @param {string} outgoingUrl the URL being requested (to check for redirect decisions).
- * @returns {boolean} true when the downstream response body should be captured.
+ * @param {import('http').IncomingMessage} originatingReq inbound request (for metrics).
+ * @param {import('http').IncomingMessage} res downstream response.
+ * @returns {boolean} whether downstream response body should be collected for AppSec.
  */
-function shouldSampleBody (req, outgoingUrl) {
-  // Check if there's a stored decision from a previous redirect
-  const storedDecision = consumeRedirectBodyCollectionDecision(req, outgoingUrl)
-  if (storedDecision) return true
+function evaluateResponseBodyCollection (originatingReq, res) {
+  const maxBytes = config.appsec.apiSecurity.maxDownstreamBodyBytes
+
+  const mime = extractMimeType(res.headers?.['content-type'])
+  if (!mime || !SUPPORTED_RESPONSE_BODY_MIME_TYPES.has(mime)) {
+    recordResponseBodyIgnored(originatingReq, RESPONSE_BODY_IGNORED_TAG_CONTENT_TYPE)
+    return false
+  }
+
+  const declaredContentLength = parseContentLengthHeader(res.headers?.['content-length'])
+  if (declaredContentLength == null || declaredContentLength === 0) {
+    recordResponseBodyIgnored(originatingReq, RESPONSE_BODY_IGNORED_TAG_CONTENT_LENGTH_MISSING)
+    return false
+  }
+
+  if (declaredContentLength > maxBytes) {
+    recordResponseBodyIgnored(originatingReq, RESPONSE_BODY_IGNORED_TAG_CONTENT_LENGTH_TOO_BIG)
+    return false
+  }
+
+  return true
+}
 
+/**
+ * Probabilistic gate for downstream response body capture (rate + per-request cap).
+ * Only used from {@link planResponseBodyCollection}; does not increment {@link bodyAnalysisCount}.
+ * @param {import('http').IncomingMessage} req originating server request.
+ * @param {string} [_outgoingUrl] reserved for future use.
+ * @returns {boolean}
+ */
+function shouldSampleBody (req, _outgoingUrl) {
   globalRequestCounter = (globalRequestCounter + 1n) & UINT64_MAX
 
   const currentCount = bodyAnalysisCount.get(req) || 0
@@ -96,14 +145,44 @@ function shouldSampleBody (req, outgoingUrl) {
   // Replace 1000n with the accuraccy that we want to maintain
   const threshold = (UINT64_MAX * BigInt(Math.round(samplingRate * 1000))) / 1000n
 
-  const shouldCollectBody = hashed <= threshold
+  return hashed <= threshold
+}
+
+/**
+ * @param {import('http').IncomingMessage} res downstream HTTP response.
+ * @returns {boolean}
+ */
+function isRedirectResponse (res) {
+  const location = res.headers?.location || res.headers?.Location
+  return res.statusCode >= 300 && res.statusCode < 400 && !!location
+}
+
+/**
+ * Plans downstream response body capture on the instrumentation ctx when response headers arrive.
+ * Redirect responses (3xx + Location) are ignored; each outbound hop is evaluated independently
+ * when its own non-redirect response arrives.
+ * @param {import('http').IncomingMessage} originatingReq incoming server request.
+ * @param {string} _outgoingUrl downstream URL for this hop (unused; redirect hops exit earlier).
+ * @param {import('http').IncomingMessage} res downstream response.
+ * @param {object} ctx http client instrumentation context (mutated).
+ */
+function planResponseBodyCollection (originatingReq, _outgoingUrl, res, ctx) {
+  if (!config?.appsec?.apiSecurity) {
+    return
+  }
 
-  // Track body analysis count if we're sampling the response body
-  if (shouldCollectBody) {
-    incrementBodyAnalysisCount(req)
+  if (isRedirectResponse(res)) {
+    return
   }
 
-  return shouldCollectBody
+  if (!shouldSampleBody(originatingReq, _outgoingUrl)) {
+    return
+  }
+
+  if (evaluateResponseBodyCollection(originatingReq, res)) {
+    ctx.shouldCollectBody = true
+    incrementBodyAnalysisCount(originatingReq)
+  }
 }
 
 /**
@@ -144,25 +223,6 @@ function extractRequestData (ctx) {
   return addresses
 }
 
-/**
- * Checks if a response is a redirect
- * @param {import('http').IncomingMessage} req incoming server request.
- * @param {import('http').IncomingMessage} res downstream response object.
- * @returns {boolean} is redirect.
- */
-function handleRedirectResponse (req, res) {
-  const isRedirect = res.statusCode >= 300 && res.statusCode < 400
-  const redirectLocation = res.headers?.location || ''
-
-  if (isRedirect && redirectLocation) {
-    // Store the body collection decision for the redirect target
-    storeRedirectBodyCollectionDecision(req, redirectLocation)
-    return true
-  }
-
-  return false
-}
-
 /**
  * Extracts response data for WAF analysis.
  * @param {import('http').IncomingMessage} res downstream response object.
@@ -290,13 +350,8 @@ function extractMimeType (contentType) {
 module.exports = {
   enable,
   disable,
-  shouldSampleBody,
-  handleRedirectResponse,
+  planResponseBodyCollection,
   incrementDownstreamAnalysisCount,
   extractRequestData,
   extractResponseData,
-  // exports for tests
-  parseBody,
-  getMethod,
-  storeRedirectBodyCollectionDecision,
 }
diff --git a/packages/dd-trace/src/appsec/rasp/ssrf.js b/packages/dd-trace/src/appsec/rasp/ssrf.js
index 0d774635f3..720ee3c792 100644
--- a/packages/dd-trace/src/appsec/rasp/ssrf.js
+++ b/packages/dd-trace/src/appsec/rasp/ssrf.js
@@ -3,6 +3,7 @@
 const { format } = require('url')
 const {
   httpClientRequestStart,
+  httpClientResponseStart,
   httpClientResponseFinish,
 } = require('../channels')
 const addresses = require('../addresses')
@@ -20,6 +21,7 @@ function enable (_config) {
   downstream.enable(_config)
 
   httpClientRequestStart.subscribe(analyzeSsrf)
+  httpClientResponseStart.subscribe(planResponseBodyCollection)
   httpClientResponseFinish.subscribe(handleResponseFinish)
 }
 
@@ -27,6 +29,7 @@ function disable () {
   downstream.disable()
 
   if (httpClientRequestStart.hasSubscribers) httpClientRequestStart.unsubscribe(analyzeSsrf)
+  if (httpClientResponseStart.hasSubscribers) httpClientResponseStart.unsubscribe(planResponseBodyCollection)
   if (httpClientResponseFinish.hasSubscribers) httpClientResponseFinish.unsubscribe(handleResponseFinish)
 }
 
@@ -36,9 +39,6 @@ function analyzeSsrf (ctx) {
 
   if (!req || !outgoingUrl) return
 
-  // Determine if we should collect the response body based on sampling rate and redirect URL
-  ctx.shouldCollectBody = downstream.shouldSampleBody(req, outgoingUrl)
-
   const requestAddresses = downstream.extractRequestData(ctx)
 
   const ephemeral = {
@@ -55,13 +55,26 @@ function analyzeSsrf (ctx) {
   downstream.incrementDownstreamAnalysisCount(req)
 }
 
+/**
+ * Channel handler: plans downstream response body capture once response headers are available.
+ * @param {{ ctx: object, res: import('http').IncomingMessage }} payload channel payload.
+ */
+function planResponseBodyCollection ({ ctx, res }) {
+  const originatingRequest = getActiveRequest()
+  if (!originatingRequest || !res) return
+
+  const outgoingUrl = (ctx.args.options?.uri && format(ctx.args.options.uri)) ?? ctx.args.uri
+  if (!outgoingUrl) return
+
+  downstream.planResponseBodyCollection(originatingRequest, outgoingUrl, res, ctx)
+}
+
 /**
  * Finalizes body collection for the response and triggers RASP analysis.
- * @param {{
- *   ctx: object,
- *   res: import('http').IncomingMessage,
- *   body: string|Buffer|null
- * }} payload event payload from the channel.
+ * @param {object} params event payload from the channel.
+ * @param {object} params.ctx instrumentation context.
+ * @param {import('http').IncomingMessage} params.res downstream response.
+ * @param {string|Buffer|null} params.body collected body.
  */
 function handleResponseFinish ({ ctx, res, body }) {
   // downstream response object
@@ -70,9 +83,7 @@ function handleResponseFinish ({ ctx, res, body }) {
   const originatingRequest = getActiveRequest()
   if (!originatingRequest) return
 
-  // Skip body analysis for redirect responses
-  const evaluateBody = ctx.shouldCollectBody && !downstream.handleRedirectResponse(originatingRequest, res)
-  const responseBody = evaluateBody ? body : null
+  const responseBody = ctx.shouldCollectBody ? body : null
   runResponseEvaluation(res, originatingRequest, responseBody)
 }
 
diff --git a/packages/dd-trace/src/config/generated-config-types.d.ts b/packages/dd-trace/src/config/generated-config-types.d.ts
index 8fcc25ba50..2d2aa565b5 100644
--- a/packages/dd-trace/src/config/generated-config-types.d.ts
+++ b/packages/dd-trace/src/config/generated-config-types.d.ts
@@ -11,6 +11,7 @@ export interface GeneratedConfig {
       enabled: boolean;
       endpointCollectionEnabled: boolean;
       endpointCollectionMessageLimit: number;
+      maxDownstreamBodyBytes: number;
       maxDownstreamRequestBodyAnalysis: number;
       sampleDelay: number;
     };
diff --git a/packages/dd-trace/src/config/supported-configurations.json b/packages/dd-trace/src/config/supported-configurations.json
index 15b337b203..6da026f5e6 100644
--- a/packages/dd-trace/src/config/supported-configurations.json
+++ b/packages/dd-trace/src/config/supported-configurations.json
@@ -166,6 +166,14 @@
         "default": "1"
       }
     ],
+    "DD_API_SECURITY_MAX_DOWNSTREAM_BODY_BYTES": [
+      {
+        "implementation": "A",
+        "type": "int",
+        "internalPropertyName": "appsec.apiSecurity.maxDownstreamBodyBytes",
+        "default": "10485760"
+      }
+    ],
     "DD_API_SECURITY_SAMPLE_DELAY": [
       {
         "implementation": "A",
diff --git a/packages/dd-trace/test/appsec/downstream_requests.spec.js b/packages/dd-trace/test/appsec/downstream_requests.spec.js
index c908de04c8..f2b0f4b061 100644
--- a/packages/dd-trace/test/appsec/downstream_requests.spec.js
+++ b/packages/dd-trace/test/appsec/downstream_requests.spec.js
@@ -20,6 +20,7 @@ describe('appsec downstream_requests', () => {
           enabled: true,
           downstreamBodyAnalysisSampleRate: 1,
           maxDownstreamRequestBodyAnalysis: 1,
+          maxDownstreamBodyBytes: 1024,
         },
       },
     }
@@ -34,55 +35,7 @@ describe('appsec downstream_requests', () => {
     logWarnStub.restore()
   })
 
-  describe('shouldSampleBody', () => {
-    const testUrl = 'http://example.com/api'
-
-    it('returns true when enabled with sample rate 1', () => {
-      assert.strictEqual(downstream.shouldSampleBody(req, testUrl), true)
-    })
-
-    it('returns false when per-request limit reached', () => {
-      assert.strictEqual(downstream.shouldSampleBody(req, testUrl), true)
-      assert.strictEqual(downstream.shouldSampleBody(req, 'http://example.com/api2'), false)
-    })
-
-    it('returns false when sample rate is zero', () => {
-      downstream.disable()
-      config.appsec.apiSecurity.downstreamBodyAnalysisSampleRate = 0
-      downstream.enable(config)
-
-      assert.strictEqual(downstream.shouldSampleBody(req, testUrl), false)
-    })
-
-    it('returns stored decision from redirect', () => {
-      const redirectUrl = 'http://example.com/redirect-target'
-      downstream.storeRedirectBodyCollectionDecision(req, redirectUrl)
-
-      assert.strictEqual(downstream.shouldSampleBody(req, redirectUrl), true)
-    })
-
-    it('returns stored decision even when sample rate is zero', () => {
-      downstream.disable()
-      config.appsec.apiSecurity.downstreamBodyAnalysisSampleRate = 0
-      config.appsec.apiSecurity.maxDownstreamRequestBodyAnalysis = 0
-      downstream.enable(config)
-
-      const redirectUrl = 'http://example.com/redirect-target'
-      downstream.storeRedirectBodyCollectionDecision(req, redirectUrl)
-
-      assert.strictEqual(downstream.shouldSampleBody(req, redirectUrl), true)
-    })
-
-    it('returns stored decision even after limit reached', () => {
-      assert.strictEqual(downstream.shouldSampleBody(req, 'http://example.com/first'), true)
-      // limit reached
-      assert.strictEqual(downstream.shouldSampleBody(req, 'http://example.com/second'), false)
-      // stored decision should still work
-      const redirectUrl = 'http://example.com/redirect-target'
-      downstream.storeRedirectBodyCollectionDecision(req, redirectUrl)
-      assert.strictEqual(downstream.shouldSampleBody(req, redirectUrl), true)
-    })
-
+  describe('apiSecurity downstream body analysis sample rate', () => {
     it('logs warning and clamps value when sample rate is above 1', () => {
       downstream.disable()
       config.appsec.apiSecurity.downstreamBodyAnalysisSampleRate = 1.5
@@ -114,56 +67,146 @@ describe('appsec downstream_requests', () => {
       config.appsec.apiSecurity.downstreamBodyAnalysisSampleRate = 0.5
       downstream.enable(config)
 
-      downstream.shouldSampleBody(req, testUrl)
+      const validRes = {
+        statusCode: 200,
+        headers: { 'content-type': 'application/json', 'content-length': '2' },
+      }
+      const ctx = {}
+      downstream.planResponseBodyCollection(req, 'http://example.com/api', validRes, ctx)
 
       sinon.assert.notCalled(logWarnStub)
     })
   })
 
-  describe('handleRedirectResponse', () => {
-    it('detects redirect with location header', () => {
+  describe('planResponseBodyCollection', () => {
+    const validJsonRes = {
+      statusCode: 200,
+      headers: { 'content-type': 'application/json', 'content-length': '2' },
+    }
+    it('does nothing on redirect response (no ctx flags, no sampling until a non-redirect response)', () => {
+      const inboundReq = {}
+      const ctx = {}
       const res = {
         statusCode: 302,
-        headers: { location: 'http://example.com/redirect' },
+        headers: { location: 'http://example.com/next' },
       }
 
-      const isRedirect = downstream.handleRedirectResponse(req, res)
+      downstream.planResponseBodyCollection(inboundReq, 'http://example.com/first', res, ctx)
+
+      assert.strictEqual(ctx.shouldCollectBody, undefined)
 
-      assert.strictEqual(isRedirect, true)
+      const ctxAfter = {}
+      downstream.planResponseBodyCollection(inboundReq, 'http://example.com/next', validJsonRes, ctxAfter)
+      assert.strictEqual(ctxAfter.shouldCollectBody, true)
     })
 
-    it('returns false for non redirect status codes', () => {
+    it('sets shouldCollectBody when sampling allows and response headers allow collection', () => {
+      const inboundReq = {}
+      const ctx = {}
       const res = {
         statusCode: 200,
-        headers: {},
+        headers: {
+          'content-type': 'application/json',
+          'content-length': '12',
+        },
       }
 
-      const isRedirect = downstream.handleRedirectResponse(req, res)
+      downstream.planResponseBodyCollection(inboundReq, 'http://example.com/api', res, ctx)
 
-      assert.strictEqual(isRedirect, false)
+      assert.strictEqual(ctx.shouldCollectBody, true)
     })
 
-    it('returns false for redirect without location header', () => {
-      const res = {
-        statusCode: 302,
-        headers: {},
+    it('records response body ignored metric when sampling allows but content-length is missing', () => {
+      const web = require('../../src/plugins/util/web')
+      const span = { setTag: sinon.stub() }
+      const webRootStub = sinon.stub(web, 'root').returns(span)
+      try {
+        const inboundReq = {}
+        const ctx = {}
+        const res = {
+          statusCode: 200,
+          headers: { 'content-type': 'application/json' },
+        }
+
+        downstream.planResponseBodyCollection(inboundReq, 'http://example.com/api', res, ctx)
+
+        assert.strictEqual(ctx.shouldCollectBody, undefined)
+        const tag = '_dd.appsec.downstream_request.response_body_ignored.content_length_missing'
+        sinon.assert.calledOnceWithExactly(span.setTag, tag, 1)
+      } finally {
+        webRootStub.restore()
       }
+    })
 
-      const isRedirect = downstream.handleRedirectResponse(req, res)
-
-      assert.strictEqual(isRedirect, false)
+    it('increments same ignored-body metric twice when two hops fail content-type on the same request', () => {
+      const web = require('../../src/plugins/util/web')
+      const span = { setTag: sinon.stub() }
+      const webRootStub = sinon.stub(web, 'root').returns(span)
+      try {
+        const inboundReq = {}
+        const badRes = {
+          statusCode: 200,
+          headers: { 'content-type': 'image/png', 'content-length': '4' },
+        }
+        const ctx1 = {}
+        const ctx2 = {}
+        downstream.planResponseBodyCollection(inboundReq, 'http://example.com/a', badRes, ctx1)
+        downstream.planResponseBodyCollection(inboundReq, 'http://example.com/b', badRes, ctx2)
+
+        const tag = '_dd.appsec.downstream_request.response_body_ignored.content_type_invalid'
+        sinon.assert.calledTwice(span.setTag)
+        sinon.assert.calledWith(span.setTag, tag, 1)
+        sinon.assert.calledWith(span.setTag, tag, 2)
+      } finally {
+        webRootStub.restore()
+      }
     })
 
-    it('stores body collection decision for redirect', () => {
-      const res = {
-        statusCode: 302,
-        headers: { location: 'http://example.com/target' },
+    it('records response body ignored metric when content-length exceeds configured max', () => {
+      const web = require('../../src/plugins/util/web')
+      const span = { setTag: sinon.stub() }
+      const webRootStub = sinon.stub(web, 'root').returns(span)
+      try {
+        const inboundReq = {}
+        const ctx = {}
+        const res = {
+          statusCode: 200,
+          headers: {
+            'content-type': 'application/json',
+            'content-length': '5000',
+          },
+        }
+
+        downstream.planResponseBodyCollection(inboundReq, 'http://example.com/api', res, ctx)
+
+        assert.strictEqual(ctx.shouldCollectBody, undefined)
+        const tag = '_dd.appsec.downstream_request.response_body_ignored.content_length_too_big'
+        sinon.assert.calledOnceWithExactly(span.setTag, tag, 1)
+      } finally {
+        webRootStub.restore()
       }
+    })
+
+    it('does not plan body collection when sample rate is zero', () => {
+      downstream.disable()
+      config.appsec.apiSecurity.downstreamBodyAnalysisSampleRate = 0
+      downstream.enable(config)
+
+      const inboundReq = {}
+      const ctx = {}
+      downstream.planResponseBodyCollection(inboundReq, 'http://example.com/api', validJsonRes, ctx)
+      assert.strictEqual(ctx.shouldCollectBody, undefined)
+    })
 
-      downstream.handleRedirectResponse(req, res)
+    it('stops planning body collection when per-request analysis limit is reached', () => {
+      const inboundReq = {}
+      const ctx1 = {}
+      downstream.planResponseBodyCollection(inboundReq, 'http://example.com/api', validJsonRes, ctx1)
+      assert.strictEqual(ctx1.shouldCollectBody, true)
 
-      const storedDecision = downstream.shouldSampleBody(req, 'http://example.com/target')
-      assert.strictEqual(storedDecision, true)
+      const ctx2 = {}
+      downstream.planResponseBodyCollection(inboundReq, 'http://example.com/api2', validJsonRes, ctx2)
+      assert.strictEqual(ctx2.shouldCollectBody, undefined)
     })
   })
 
@@ -244,6 +287,17 @@ describe('appsec downstream_requests', () => {
       assert.deepStrictEqual(addressesMap[addresses.HTTP_OUTGOING_RESPONSE_BODY], { ok: true })
     })
 
+    it('parses urlencoded response body when provided', () => {
+      const urlRes = {
+        statusCode: 200,
+        headers: { 'content-type': 'application/x-www-form-urlencoded' },
+      }
+      const body = Buffer.from('a=1&b=2')
+      const addressesMap = downstream.extractResponseData(urlRes, body)
+
+      assert.deepStrictEqual(addressesMap[addresses.HTTP_OUTGOING_RESPONSE_BODY], { a: '1', b: '2' })
+    })
+
     it('omits body when not provided', () => {
       const addressesMap = downstream.extractResponseData(res)
 
@@ -254,113 +308,6 @@ describe('appsec downstream_requests', () => {
     })
   })
 
-  describe('parseBody', () => {
-    describe('JSON parsing', () => {
-      it('parses JSON strings', () => {
-        assert.deepStrictEqual(downstream.parseBody('{"foo":1}', 'application/json'), { foo: 1 })
-      })
-
-      it('parses JSON buffers', () => {
-        const buffer = Buffer.from('{"foo":1}')
-        assert.deepStrictEqual(downstream.parseBody(buffer, 'application/json'), { foo: 1 })
-      })
-
-      it('handles text/json content type', () => {
-        assert.deepStrictEqual(downstream.parseBody('{"foo":1}', 'text/json'), { foo: 1 })
-      })
-
-      it('handles content-type with charset', () => {
-        assert.deepStrictEqual(downstream.parseBody('{"foo":1}', 'application/json; charset=utf-8'), { foo: 1 })
-      })
-
-      it('returns null for invalid JSON', () => {
-        assert.strictEqual(downstream.parseBody('{invalid}', 'application/json'), null)
-      })
-
-      it('returns null for non-object JSON', () => {
-        assert.strictEqual(downstream.parseBody(123, 'application/json'), null)
-      })
-    })
-
-    describe('URL-encoded parsing', () => {
-      it('parses urlencoded strings', () => {
-        const parsed = downstream.parseBody('a=1&b=2', 'application/x-www-form-urlencoded')
-        assert.deepStrictEqual(parsed, { a: '1', b: '2' })
-      })
-
-      it('parses urlencoded buffers', () => {
-        const buffer = Buffer.from('a=1&b=2')
-        const parsed = downstream.parseBody(buffer, 'application/x-www-form-urlencoded')
-        assert.deepStrictEqual(parsed, { a: '1', b: '2' })
-      })
-
-      it('handles multiple values for same key', () => {
-        const parsed = downstream.parseBody('a=1&a=2&b=3', 'application/x-www-form-urlencoded')
-        assert.deepStrictEqual(parsed, { a: ['1', '2'], b: '3' })
-      })
-
-      it('handles URL encoded values', () => {
-        const parsed = downstream.parseBody('name=John%20Doe&city=New%20York', 'application/x-www-form-urlencoded')
-        assert.deepStrictEqual(parsed, { name: 'John Doe', city: 'New York' })
-      })
-
-      it('handles empty values', () => {
-        const parsed = downstream.parseBody('a=&b=2', 'application/x-www-form-urlencoded')
-        assert.deepStrictEqual(parsed, { a: '', b: '2' })
-      })
-    })
-
-    describe('Unsupported content types', () => {
-      it('returns null for text/plain', () => {
-        assert.strictEqual(downstream.parseBody('text', 'text/plain'), null)
-      })
-
-      it('returns null for multipart/form-data', () => {
-        assert.strictEqual(downstream.parseBody('data', 'multipart/form-data'), null)
-      })
-
-      it('returns null for text/html', () => {
-        assert.strictEqual(downstream.parseBody('<html></html>', 'text/html'), null)
-      })
-
-      it('returns null for application/xml', () => {
-        assert.strictEqual(downstream.parseBody('<xml></xml>', 'application/xml'), null)
-      })
-    })
-
-    describe('Edge cases', () => {
-      it('returns null when body is null', () => {
-        assert.strictEqual(downstream.parseBody(null, 'application/json'), null)
-      })
-
-      it('returns null when body is undefined', () => {
-        assert.strictEqual(downstream.parseBody(undefined, 'application/json'), null)
-      })
-
-      it('returns null when contentType is null', () => {
-        assert.strictEqual(downstream.parseBody('{"foo":1}', null), null)
-      })
-
-      it('returns null when parsing fails', () => {
-        assert.strictEqual(downstream.parseBody('not json', 'application/json'), null)
-      })
-    })
-  })
-
-  describe('getMethod', () => {
-    it('returns method when valid string', () => {
-      assert.strictEqual(downstream.getMethod('POST'), 'POST')
-    })
-
-    it('returns GET when method is null', () => {
-      assert.strictEqual(downstream.getMethod(null), 'GET')
-    })
-
-    it('returns GET when method is not a string', () => {
-      assert.strictEqual(downstream.getMethod(123), 'GET')
-    })
-  })
-
   describe('incrementDownstreamAnalysisCount', () => {
     let web
     let span
@@ -422,30 +369,41 @@ describe('appsec downstream_requests', () => {
     })
   })
 
-  describe('sampling behavior', () => {
-    it('returns true for sample rate 1.0 (100%)', () => {
+  describe('sampling behavior via planResponseBodyCollection', () => {
+    const validJsonRes = {
+      statusCode: 200,
+      headers: { 'content-type': 'application/json', 'content-length': '2' },
+    }
+
+    it('collects body on every hop at sample rate 1.0 (100%)', () => {
       downstream.disable()
       config.appsec.apiSecurity.downstreamBodyAnalysisSampleRate = 1.0
       config.appsec.apiSecurity.maxDownstreamRequestBodyAnalysis = 100
       downstream.enable(config)
 
       for (let i = 0; i < 10; i++) {
-        assert.strictEqual(downstream.shouldSampleBody({}), true)
+        const inboundReq = {}
+        const ctx = {}
+        downstream.planResponseBodyCollection(inboundReq, `http://example.com/${i}`, validJsonRes, ctx)
+        assert.strictEqual(ctx.shouldCollectBody, true)
       }
     })
 
-    it('returns false for sample rate 0.0 (0%)', () => {
+    it('never collects at sample rate 0.0 (0%)', () => {
       downstream.disable()
       config.appsec.apiSecurity.downstreamBodyAnalysisSampleRate = 0.0
       config.appsec.apiSecurity.maxDownstreamRequestBodyAnalysis = 100
       downstream.enable(config)
 
       for (let i = 0; i < 10; i++) {
-        assert.strictEqual(downstream.shouldSampleBody({}), false)
+        const inboundReq = {}
+        const ctx = {}
+        downstream.planResponseBodyCollection(inboundReq, `http://example.com/${i}`, validJsonRes, ctx)
+        assert.strictEqual(ctx.shouldCollectBody, undefined)
       }
     })
 
-    it('produces some true and some false with rate 0.5', () => {
+    it('produces some collects and some skips with rate 0.5', () => {
       downstream.disable()
       config.appsec.apiSecurity.downstreamBodyAnalysisSampleRate = 0.5
       config.appsec.apiSecurity.maxDownstreamRequestBodyAnalysis = 1000
@@ -453,7 +411,10 @@ describe('appsec downstream_requests', () => {
 
       const results = []
       for (let i = 0; i < 100; i++) {
-        results.push(downstream.shouldSampleBody({}))
+        const inboundReq = {}
+        const ctx = {}
+        downstream.planResponseBodyCollection(inboundReq, `http://example.com/${i}`, validJsonRes, ctx)
+        results.push(ctx.shouldCollectBody === true)
       }
 
       const trueCount = results.filter(r => r).length
@@ -472,15 +433,28 @@ describe('appsec downstream_requests', () => {
       const req1 = {}
       const req2 = {}
 
-      assert.strictEqual(downstream.shouldSampleBody(req1, 'http://example.com/1'), true)
-      assert.strictEqual(downstream.shouldSampleBody(req2, 'http://example.com/2'), true)
-      assert.strictEqual(downstream.shouldSampleBody(req1, 'http://example.com/3'), true)
+      const c1 = {}
+      downstream.planResponseBodyCollection(req1, 'http://example.com/1', validJsonRes, c1)
+      assert.strictEqual(c1.shouldCollectBody, true)
+
+      const c2 = {}
+      downstream.planResponseBodyCollection(req2, 'http://example.com/2', validJsonRes, c2)
+      assert.strictEqual(c2.shouldCollectBody, true)
 
-      assert.strictEqual(downstream.shouldSampleBody(req1, 'http://example.com/4'), false)
-      assert.strictEqual(downstream.shouldSampleBody(req2, 'http://example.com/5'), true)
+      const c3 = {}
+      downstream.planResponseBodyCollection(req1, 'http://example.com/3', validJsonRes, c3)
+      assert.strictEqual(c3.shouldCollectBody, true)
+
+      const c4 = {}
+      downstream.planResponseBodyCollection(req1, 'http://example.com/4', validJsonRes, c4)
+      assert.strictEqual(c4.shouldCollectBody, undefined)
+
+      const c5 = {}
+      downstream.planResponseBodyCollection(req2, 'http://example.com/5', validJsonRes, c5)
+      assert.strictEqual(c5.shouldCollectBody, true)
     })
 
-    it('increments counter correctly', () => {
+    it('increments counter only after successful header-based collection plan', () => {
       downstream.disable()
       config.appsec.apiSecurity.downstreamBodyAnalysisSampleRate = 1.0
       config.appsec.apiSecurity.maxDownstreamRequestBodyAnalysis = 3
@@ -488,13 +462,15 @@ describe('appsec downstream_requests', () => {
 
       const testReq = {}
 
-      // Should sample 3 times
-      assert.strictEqual(downstream.shouldSampleBody(testReq), true)
-      assert.strictEqual(downstream.shouldSampleBody(testReq), true)
-      assert.strictEqual(downstream.shouldSampleBody(testReq), true)
+      for (let i = 0; i < 3; i++) {
+        const ctx = {}
+        downstream.planResponseBodyCollection(testReq, `http://example.com/${i}`, validJsonRes, ctx)
+        assert.strictEqual(ctx.shouldCollectBody, true)
+      }
 
-      // Fourth time should be false
-      assert.strictEqual(downstream.shouldSampleBody(testReq), false)
+      const ctxLast = {}
+      downstream.planResponseBodyCollection(testReq, 'http://example.com/last', validJsonRes, ctxLast)
+      assert.strictEqual(ctxLast.shouldCollectBody, undefined)
     })
   })
 })
diff --git a/packages/dd-trace/test/appsec/rasp/ssrf.spec.js b/packages/dd-trace/test/appsec/rasp/ssrf.spec.js
index c03d04562c..be75d5ca2a 100644
--- a/packages/dd-trace/test/appsec/rasp/ssrf.spec.js
+++ b/packages/dd-trace/test/appsec/rasp/ssrf.spec.js
@@ -1,5 +1,6 @@
 'use strict'
 
+const assert = require('node:assert/strict')
 const { EventEmitter } = require('events')
 const { describe, it, beforeEach, afterEach } = require('mocha')
 const sinon = require('sinon')
@@ -8,6 +9,7 @@ const proxyquire = require('proxyquire')
 const { storage } = require('../../../../datadog-core')
 const {
   httpClientRequestStart,
+  httpClientResponseStart,
   httpClientResponseFinish,
 } = require('../../../src/appsec/channels')
 const addresses = require('../../../src/appsec/addresses')
@@ -21,6 +23,8 @@ describe('RASP - ssrf.js', () => {
   let ssrf
   let telemetry
 
+  let nextIncludeBodies
+
   const makeCtx = (overrides = {}) => ({
     args: {
       uri: DEFAULT_URL,
@@ -40,11 +44,15 @@ describe('RASP - ssrf.js', () => {
   }
 
   const publishRequestStart = ({ ctx, includeBodies = false, requestAddresses = {} }) => {
-    downstream.shouldSampleBody.returns(includeBodies)
+    nextIncludeBodies = includeBodies
     downstream.extractRequestData.returns(requestAddresses)
     httpClientRequestStart.publish(ctx)
   }
 
+  const publishResponseStart = (ctx, res) => {
+    httpClientResponseStart.publish({ ctx, res })
+  }
+
   const createResponse = ({ statusCode = 200, headers = {} } = {}) => {
     const response = new EventEmitter()
     response.statusCode = statusCode
@@ -53,6 +61,8 @@ describe('RASP - ssrf.js', () => {
   }
 
   beforeEach(() => {
+    nextIncludeBodies = false
+
     waf = {
       run: sinon.stub(),
     }
@@ -60,12 +70,22 @@ describe('RASP - ssrf.js', () => {
     downstream = {
       enable: sinon.stub(),
       disable: sinon.stub(),
-      shouldSampleBody: sinon.stub().returns(true),
-      handleRedirectResponse: sinon.stub().returns(false),
       extractRequestData: sinon.stub().returns({}),
       extractResponseData: sinon.stub().returns({}),
       incrementDownstreamAnalysisCount: sinon.stub(),
-      storeRedirectBodyCollectionDecision: sinon.stub(),
+      planResponseBodyCollection: sinon.stub().callsFake((originatingReq, outgoingUrl, res, ctx) => {
+        delete ctx.shouldCollectBody
+
+        const location = res.headers?.location
+        const isRedirect = res.statusCode >= 300 && res.statusCode < 400 && location
+        if (isRedirect) {
+          return
+        }
+
+        if (nextIncludeBodies) {
+          ctx.shouldCollectBody = true
+        }
+      }),
     }
 
     telemetry = {
@@ -159,13 +179,26 @@ describe('RASP - ssrf.js', () => {
       sinon.assert.notCalled(waf.run)
     })
 
-    it('does not set shouldCollectBody flag when sampling disabled', () => {
+    it('does not set shouldCollectBody until response headers when sampling disabled', () => {
       const ctx = makeCtx()
       stubStore({}, {})
 
       publishRequestStart({ ctx, includeBodies: false })
+      sinon.assert.match(ctx.shouldCollectBody, sinon.match.undefined)
+
+      publishResponseStart(ctx, createResponse())
+      assert.strictEqual(ctx.shouldCollectBody, undefined)
+    })
+
+    it('sets shouldCollectBody when body sampling is enabled at response', () => {
+      const ctx = makeCtx()
+      stubStore({}, {})
+
+      publishRequestStart({ ctx, includeBodies: true })
+      publishResponseStart(ctx, createResponse())
 
-      sinon.assert.match(ctx.shouldCollectBody, false)
+      sinon.assert.calledOnce(downstream.planResponseBodyCollection)
+      sinon.assert.match(ctx.shouldCollectBody, true)
     })
 
     it('evaluates response and passes body through to extractResponseData', () => {
@@ -183,8 +216,8 @@ describe('RASP - ssrf.js', () => {
       waf.run.onSecondCall().returns({ events: [{ id: 'ssrf' }] })
 
       publishRequestStart({ ctx, includeBodies: true, requestAddresses })
-
       const response = createResponse({ headers: { 'content-type': 'application/json' } })
+      publishResponseStart(ctx, response)
       const body = Buffer.from('{"ok":true}')
 
       httpClientResponseFinish.publish({ ctx, res: response, body })
@@ -206,6 +239,7 @@ describe('RASP - ssrf.js', () => {
       waf.run.returns({ events: [] })
 
       publishRequestStart({ ctx, includeBodies: false })
+      publishResponseStart(ctx, createResponse())
 
       const response = createResponse()
       httpClientResponseFinish.publish({ ctx, res: response, body: null })
@@ -223,6 +257,7 @@ describe('RASP - ssrf.js', () => {
       waf.run.returns({ events: [] })
 
       publishRequestStart({ ctx, includeBodies: false })
+      publishResponseStart(ctx, createResponse())
 
       const response = createResponse()
       httpClientResponseFinish.publish({ ctx, res: response, body: null })
@@ -233,9 +268,8 @@ describe('RASP - ssrf.js', () => {
 
     it('handles redirect responses and skips body analysis', () => {
       const ctx = makeCtx()
-      const { req } = stubStore({}, {})
+      stubStore({}, {})
 
-      downstream.handleRedirectResponse.returns(true)
       downstream.extractResponseData.returns({
         [addresses.HTTP_OUTGOING_RESPONSE_STATUS]: '302',
         [addresses.HTTP_OUTGOING_RESPONSE_HEADERS]: { location: 'http://example.com/redirect' },
@@ -245,11 +279,13 @@ describe('RASP - ssrf.js', () => {
       publishRequestStart({ ctx, includeBodies: true })
 
       const response = createResponse({ statusCode: 302, headers: { location: 'http://example.com/redirect' } })
+      publishResponseStart(ctx, response)
+
+      assert.strictEqual(ctx.shouldCollectBody, undefined)
+
       const body = Buffer.from('redirect body')
       httpClientResponseFinish.publish({ ctx, res: response, body })
 
-      sinon.assert.calledOnceWithExactly(downstream.handleRedirectResponse, req, response)
-
       sinon.assert.calledWith(downstream.extractResponseData, response, null)
 
       sinon.assert.calledTwice(waf.run)
@@ -259,7 +295,6 @@ describe('RASP - ssrf.js', () => {
       const ctx = makeCtx()
       stubStore({}, {})
 
-      downstream.handleRedirectResponse.returns(false)
       downstream.extractResponseData.returns({
         [addresses.HTTP_OUTGOING_RESPONSE_STATUS]: '200',
         [addresses.HTTP_OUTGOING_RESPONSE_BODY]: { ok: true },
@@ -269,35 +304,36 @@ describe('RASP - ssrf.js', () => {
       publishRequestStart({ ctx, includeBodies: true })
 
       const response = createResponse({ statusCode: 200 })
+      publishResponseStart(ctx, response)
       const body = Buffer.from('{"ok":true}')
       httpClientResponseFinish.publish({ ctx, res: response, body })
 
       sinon.assert.calledWith(downstream.extractResponseData, response, body)
     })
 
-    it('does not store redirect decision when shouldCollectBody is false', () => {
+    it('redirect response does not enable body collection when sampling is disabled', () => {
       const ctx = makeCtx()
       stubStore({}, {})
 
       publishRequestStart({ ctx, includeBodies: false })
 
       const response = createResponse({ statusCode: 302, headers: { location: 'http://example.com/redirect' } })
-      httpClientResponseFinish.publish({ ctx, res: response, body: null })
+      publishResponseStart(ctx, response)
 
-      sinon.assert.notCalled(downstream.storeRedirectBodyCollectionDecision)
+      assert.strictEqual(ctx.shouldCollectBody, undefined)
     })
 
     it('passes null body when shouldCollectBody is false even for non-redirect', () => {
       const ctx = makeCtx()
       stubStore({}, {})
 
-      downstream.handleRedirectResponse.returns(false)
       downstream.extractResponseData.returns({
         [addresses.HTTP_OUTGOING_RESPONSE_STATUS]: '200',
       })
       waf.run.returns({ events: [] })
 
       publishRequestStart({ ctx, includeBodies: false })
+      publishResponseStart(ctx, createResponse({ statusCode: 200 }))
 
       const response = createResponse({ statusCode: 200 })
       const body = Buffer.from('{"data":"test"}')
@@ -305,5 +341,51 @@ describe('RASP - ssrf.js', () => {
 
       sinon.assert.calledWith(downstream.extractResponseData, response, null)
     })
+
+    it('skips body analysis for redirect response at finish', () => {
+      const ctx = makeCtx()
+      stubStore({}, {})
+
+      downstream.extractResponseData.returns({
+        [addresses.HTTP_OUTGOING_RESPONSE_STATUS]: '302',
+      })
+      waf.run.returns({ events: [] })
+
+      publishRequestStart({ ctx, includeBodies: true })
+
+      const response = createResponse({ statusCode: 302, headers: { location: 'http://example.com/redirect' } })
+      publishResponseStart(ctx, response)
+      httpClientResponseFinish.publish({
+        ctx,
+        res: response,
+        body: null,
+      })
+
+      sinon.assert.calledWith(downstream.extractResponseData, response, null)
+    })
+
+    it('skips WAF body when shouldCollectBody is false even if finish payload includes body', () => {
+      const ctx = makeCtx()
+      stubStore({}, {})
+
+      downstream.extractResponseData.returns({
+        [addresses.HTTP_OUTGOING_RESPONSE_STATUS]: '200',
+      })
+      waf.run.returns({ events: [] })
+
+      publishRequestStart({ ctx, includeBodies: false })
+      publishResponseStart(ctx, createResponse({ statusCode: 200 }))
+
+      const response = createResponse()
+      const body = Buffer.from('{"ok":true}')
+
+      httpClientResponseFinish.publish({
+        ctx,
+        res: response,
+        body,
+      })
+
+      sinon.assert.calledWith(downstream.extractResponseData, response, null)
+    })
   })
 })
diff --git a/packages/dd-trace/test/config/index.spec.js b/packages/dd-trace/test/config/index.spec.js
index 14e997cc9f..607a852831 100644
--- a/packages/dd-trace/test/config/index.spec.js
+++ b/packages/dd-trace/test/config/index.spec.js
@@ -682,6 +682,7 @@ describe('Config', () => {
           endpointCollectionMessageLimit: 300,
           downstreamBodyAnalysisSampleRate: 0.5,
           maxDownstreamRequestBodyAnalysis: 1,
+          maxDownstreamBodyBytes: 10485760,
         },
         blockedTemplateHtml: undefined,
         blockedTemplateJson: undefined,
@@ -822,6 +823,7 @@ describe('Config', () => {
       { name: 'DD_API_SECURITY_ENDPOINT_COLLECTION_MESSAGE_LIMIT', value: 300, origin: 'default' },
       { name: 'DD_API_SECURITY_DOWNSTREAM_BODY_ANALYSIS_SAMPLE_RATE', value: 0.5, origin: 'default' },
       { name: 'DD_API_SECURITY_MAX_DOWNSTREAM_REQUEST_BODY_ANALYSIS', value: 1, origin: 'default' },
+      { name: 'DD_API_SECURITY_MAX_DOWNSTREAM_BODY_BYTES', value: 10485760, origin: 'default' },
       { name: 'DD_APPSEC_HTTP_BLOCKED_TEMPLATE_HTML', value: null, origin: 'default' },
       { name: 'DD_APPSEC_HTTP_BLOCKED_TEMPLATE_JSON', value: null, origin: 'default' },
       { name: 'DD_APPSEC_ENABLED', value: null, origin: 'default' },
@@ -1032,6 +1034,7 @@ describe('Config', () => {
     process.env.DD_API_SECURITY_ENDPOINT_COLLECTION_MESSAGE_LIMIT = '500'
     process.env.DD_API_SECURITY_DOWNSTREAM_BODY_ANALYSIS_SAMPLE_RATE = '0.75'
     process.env.DD_API_SECURITY_MAX_DOWNSTREAM_REQUEST_BODY_ANALYSIS = '2'
+    process.env.DD_API_SECURITY_MAX_DOWNSTREAM_BODY_BYTES = '2048'
     process.env.DD_APM_TRACING_ENABLED = 'false'
     process.env.DD_APP_KEY = 'myAppKey'
     process.env.DD_APPSEC_AUTOMATED_USER_EVENTS_TRACKING = 'extended'
@@ -1156,6 +1159,7 @@ describe('Config', () => {
           endpointCollectionMessageLimit: 500,
           downstreamBodyAnalysisSampleRate: 0.75,
           maxDownstreamRequestBodyAnalysis: 2,
+          maxDownstreamBodyBytes: 2048,
         },
         blockedTemplateGraphql: BLOCKED_TEMPLATE_GRAPHQL,
         blockedTemplateHtml: BLOCKED_TEMPLATE_HTML,
@@ -1317,6 +1321,7 @@ describe('Config', () => {
       { name: 'DD_API_SECURITY_ENDPOINT_COLLECTION_MESSAGE_LIMIT', value: 500, origin: 'env_var' },
       { name: 'DD_API_SECURITY_DOWNSTREAM_BODY_ANALYSIS_SAMPLE_RATE', value: 0.75, origin: 'env_var' },
       { name: 'DD_API_SECURITY_MAX_DOWNSTREAM_REQUEST_BODY_ANALYSIS', value: 2, origin: 'env_var' },
+      { name: 'DD_API_SECURITY_MAX_DOWNSTREAM_BODY_BYTES', value: 2048, origin: 'env_var' },
       { name: 'DD_APPSEC_HTTP_BLOCKED_TEMPLATE_HTML', value: BLOCKED_TEMPLATE_HTML_PATH, origin: 'env_var' },
       { name: 'DD_APPSEC_HTTP_BLOCKED_TEMPLATE_JSON', value: BLOCKED_TEMPLATE_JSON_PATH, origin: 'env_var' },
       { name: 'DD_APPSEC_ENABLED', value: true, origin: 'env_var' },
@@ -2500,6 +2505,7 @@ describe('Config', () => {
         endpointCollectionMessageLimit: 500,
         downstreamBodyAnalysisSampleRate: 0.5,
         maxDownstreamRequestBodyAnalysis: 1,
+        maxDownstreamBodyBytes: 10485760,
       },
       blockedTemplateGraphql: BLOCKED_TEMPLATE_GRAPHQL,
       blockedTemplateHtml: BLOCKED_TEMPLATE_HTML,