diff --git a/.changesets/feat_caroline_subgraph_connector_response_size_limits.md b/.changesets/feat_caroline_subgraph_connector_response_size_limits.md
new file mode 100644
index 0000000000..d903a4b37a
--- /dev/null
+++ b/.changesets/feat_caroline_subgraph_connector_response_size_limits.md
@@ -0,0 +1,35 @@
+### Add per-subgraph and per-connector HTTP response size limits ([PR #9160](https://github.com/apollographql/router/pull/9160))
+
+The router can now cap the number of bytes it reads from subgraph and connector HTTP response bodies, protecting against out-of-memory conditions when a downstream service returns an unexpectedly large payload.
+
+The limit is enforced as the response body streams in — the router stops reading and returns a GraphQL error as soon as the limit is exceeded, without buffering the full body first.
+
+Configure a global default and optional per-subgraph or per-source overrides:
+
+```yaml
+limits:
+  subgraph:
+    all:
+      http_max_response_size: 10MB # 10 MB for all subgraphs
+    subgraphs:
+      products:
+        http_max_response_size: 20MB # 20 MB override for 'products'
+
+  connector:
+    all:
+      http_max_response_size: 5MB # 5 MB for all connector sources
+    sources:
+      products.rest:
+        http_max_response_size: 10MB # 10 MB override for 'products.rest'
+```
+
+There is no default limit; responses are unrestricted unless you configure this option.
+
+When a response is aborted due to the limit, the router:
+- Returns a GraphQL error to the client with extension code `SUBREQUEST_HTTP_ERROR`
+- Increments the `apollo.router.limits.subgraph_response_size.exceeded` or `apollo.router.limits.connector_response_size.exceeded` counter
+- Records `apollo.subgraph.response.aborted: "response_size_limit"` or `apollo.connector.response.aborted: "response_size_limit"` on the relevant span
+
+**Configuration migration**: Existing `limits` fields (previously at the top level of `limits`) are now nested under `limits.router`. A configuration migration is included that updates your config file automatically.
+
+By [@carodewig](https://github.com/carodewig) in https://github.com/apollographql/router/pull/9160
diff --git a/apollo-router/benches/deeply_nested/router.yaml b/apollo-router/benches/deeply_nested/router.yaml
index 9938b15a2d..547752b97f 100644
--- a/apollo-router/benches/deeply_nested/router.yaml
+++ b/apollo-router/benches/deeply_nested/router.yaml
@@ -1,7 +1,8 @@
 supergraph:
   listen: 127.0.0.1:44167
 limits:
-  parser_max_recursion: ${env.PARSER_MAX_RECURSION}
+  router:
+    parser_max_recursion: ${env.PARSER_MAX_RECURSION}
 include_subgraph_errors:
   all: true
 headers:
diff --git a/apollo-router/src/axum_factory/listeners.rs b/apollo-router/src/axum_factory/listeners.rs
index 3c52d63fbf..8be7057bc9 100644
--- a/apollo-router/src/axum_factory/listeners.rs
+++ b/apollo-router/src/axum_factory/listeners.rs
@@ -315,9 +315,9 @@ pub(super) fn serve_router_on_listen_addr(
     configuration: Arc<Configuration>,
     all_connections_stopped_sender: mpsc::Sender<()>,
 ) -> (impl Future<Output = Listener>, oneshot::Sender<()>) {
-    let opt_max_http1_headers = configuration.limits.http1_max_request_headers;
-    let opt_max_http1_buf_size = configuration.limits.http1_max_request_buf_size;
-    let opt_max_http2_headers_list_bytes = configuration.limits.http2_max_headers_list_bytes;
+    let opt_max_http1_headers = configuration.limits.router.http1_max_request_headers;
+    let opt_max_http1_buf_size = configuration.limits.router.http1_max_request_buf_size;
+    let opt_max_http2_headers_list_bytes = configuration.limits.router.http2_max_headers_list_bytes;
     let connection_shutdown_timeout = configuration.supergraph.connection_shutdown_timeout;
     let header_read_timeout = configuration.server.http.header_read_timeout;
     let tls_handshake_timeout = configuration.server.http.tls_handshake_timeout;
diff --git a/apollo-router/src/configuration/metrics.rs b/apollo-router/src/configuration/metrics.rs
index 28e531086c..7828705720 100644
--- a/apollo-router/src/configuration/metrics.rs
+++ b/apollo-router/src/configuration/metrics.rs
@@ -249,7 +249,7 @@ impl InstrumentData {
 
         populate_config_instrument!(
             apollo.router.config.limits,
-            "$.limits",
+            "$.limits.router",
             opt.operation.max_depth,
             "$[?(@.max_depth)]",
             opt.operation.max_aliases,
diff --git a/apollo-router/src/configuration/migrations/2045-limits-router-subgraph.yaml b/apollo-router/src/configuration/migrations/2045-limits-router-subgraph.yaml
new file mode 100644
index 0000000000..cb5adeac94
--- /dev/null
+++ b/apollo-router/src/configuration/migrations/2045-limits-router-subgraph.yaml
@@ -0,0 +1,40 @@
+description: >
+  limits config restructured: existing fields moved under `limits.router`,
+  new `limits.subgraph` section added for per-subgraph response size limits.
+actions:
+  - type: move
+    from: limits.http_max_request_bytes
+    to: limits.router.http_max_request_bytes
+  - type: move
+    from: limits.max_depth
+    to: limits.router.max_depth
+  - type: move
+    from: limits.max_height
+    to: limits.router.max_height
+  - type: move
+    from: limits.max_root_fields
+    to: limits.router.max_root_fields
+  - type: move
+    from: limits.max_aliases
+    to: limits.router.max_aliases
+  - type: move
+    from: limits.warn_only
+    to: limits.router.warn_only
+  - type: move
+    from: limits.parser_max_recursion
+    to: limits.router.parser_max_recursion
+  - type: move
+    from: limits.parser_max_tokens
+    to: limits.router.parser_max_tokens
+  - type: move
+    from: limits.http1_max_request_headers
+    to: limits.router.http1_max_request_headers
+  - type: move
+    from: limits.http1_max_request_buf_size
+    to: limits.router.http1_max_request_buf_size
+  - type: move
+    from: limits.http2_max_headers_list_bytes
+    to: limits.router.http2_max_headers_list_bytes
+  - type: move
+    from: limits.introspection_max_depth
+    to: limits.router.introspection_max_depth
diff --git a/apollo-router/src/configuration/snapshots/apollo_router__configuration__tests__schema_generation.snap b/apollo-router/src/configuration/snapshots/apollo_router__configuration__tests__schema_generation.snap
index 5c9b630c70..686118f939 100644
--- a/apollo-router/src/configuration/snapshots/apollo_router__configuration__tests__schema_generation.snap
+++ b/apollo-router/src/configuration/snapshots/apollo_router__configuration__tests__schema_generation.snap
@@ -2215,6 +2215,30 @@ expression: "&schema"
       },
       "type": "object"
     },
+    "ConnectorConfiguration2": {
+      "properties": {
+        "all": {
+          "allOf": [
+            {
+              "$ref": "#/definitions/ConnectorLimits"
+            }
+          ],
+          "default": {
+            "http_max_response_size": null
+          },
+          "description": "Options applying to all sources"
+        },
+        "sources": {
+          "additionalProperties": {
+            "$ref": "#/definitions/ConnectorLimits"
+          },
+          "default": {},
+          "description": "Map of subgraph_name.connector_source_name to configuration",
+          "type": "object"
+        }
+      },
+      "type": "object"
+    },
     "ConnectorHeadersConfiguration": {
       "additionalProperties": false,
       "properties": {
@@ -2239,6 +2263,21 @@ expression: "&schema"
       },
       "type": "object"
     },
+    "ConnectorLimits": {
+      "additionalProperties": false,
+      "description": "Per-connector-source response size limits.",
+      "properties": {
+        "http_max_response_size": {
+          "default": null,
+          "description": "Limit the size of incoming connector response bodies read from the network,\nto protect against running out of memory. Default: no limit.",
+          "type": [
+            "string",
+            "null"
+          ]
+        }
+      },
+      "type": "object"
+    },
     "ConnectorRequestConf": {
       "additionalProperties": false,
       "description": "What information is passed to a connector request stage",
@@ -6743,102 +6782,55 @@ expression: "&schema"
       "additionalProperties": false,
       "description": "Configuration for operation limits, parser limits, HTTP limits, etc.",
       "properties": {
-        "http1_max_request_buf_size": {
-          "default": null,
-          "description": "Limit the maximum buffer size for the HTTP1 connection.\n\nDefault is ~400kib.",
-          "type": [
-            "string",
-            "null"
-          ]
-        },
-        "http1_max_request_headers": {
-          "default": null,
-          "description": "Limit the maximum number of headers of incoming HTTP1 requests. Default is 100.\n\nIf router receives more headers than the buffer size, it responds to the client with\n\"431 Request Header Fields Too Large\".",
-          "format": "uint",
-          "minimum": 0,
-          "type": [
-            "integer",
-            "null"
-          ]
-        },
-        "http2_max_headers_list_bytes": {
-          "default": null,
-          "description": "For HTTP2, limit the header list to a threshold of bytes. Default is 16kb.\n\nIf router receives more headers than allowed size of the header list, it responds to the client with\n\"431 Request Header Fields Too Large\".",
-          "type": [
-            "string",
-            "null"
-          ]
-        },
-        "http_max_request_bytes": {
-          "default": 2000000,
-          "description": "Limit the size of incoming HTTP requests read from the network,\nto protect against running out of memory. Default: 2000000 (2 MB)",
-          "format": "uint",
-          "minimum": 0,
-          "type": "integer"
-        },
-        "introspection_max_depth": {
-          "default": true,
-          "description": "Limit the depth of nested list fields in introspection queries\nto protect avoid generating huge responses. Returns a GraphQL\nerror with `{ message: \"Maximum introspection depth exceeded\" }`\nwhen nested fields exceed the limit.\nDefault: true",
-          "type": "boolean"
-        },
-        "max_aliases": {
-          "default": null,
-          "description": "If set, requests with operations with more aliases than this maximum\nare rejected with a HTTP 400 Bad Request response and GraphQL error with\n`\"extensions\": {\"code\": \"MAX_ALIASES_LIMIT\"}`",
-          "format": "uint32",
-          "minimum": 0,
-          "type": [
-            "integer",
-            "null"
-          ]
-        },
-        "max_depth": {
-          "default": null,
-          "description": "If set, requests with operations deeper than this maximum\nare rejected with a HTTP 400 Bad Request response and GraphQL error with\n`\"extensions\": {\"code\": \"MAX_DEPTH_LIMIT\"}`\n\nCounts depth of an operation, looking at its selection sets,˛\nincluding fields in fragments and inline fragments. The following\nexample has a depth of 3.\n\n```graphql\nquery getProduct {\n  book { # 1\n    ...bookDetails\n  }\n}\n\nfragment bookDetails on Book {\n  details { # 2\n    ... on ProductDetailsBook {\n      country # 3\n    }\n  }\n}\n```",
-          "format": "uint32",
-          "minimum": 0,
-          "type": [
-            "integer",
-            "null"
-          ]
-        },
-        "max_height": {
-          "default": null,
-          "description": "If set, requests with operations higher than this maximum\nare rejected with a HTTP 400 Bad Request response and GraphQL error with\n`\"extensions\": {\"code\": \"MAX_DEPTH_LIMIT\"}`\n\nHeight is based on simple merging of fields using the same name or alias,\nbut only within the same selection set.\nFor example `name` here is only counted once and the query has height 3, not 4:\n\n```graphql\nquery {\n    name { first }\n    name { last }\n}\n```\n\nThis may change in a future version of Apollo Router to do\n[full field merging across fragments][merging] instead.\n\n[merging]: https://spec.graphql.org/October2021/#sec-Field-Selection-Merging]",
-          "format": "uint32",
-          "minimum": 0,
-          "type": [
-            "integer",
-            "null"
-          ]
-        },
-        "max_root_fields": {
-          "default": null,
-          "description": "If set, requests with operations with more root fields than this maximum\nare rejected with a HTTP 400 Bad Request response and GraphQL error with\n`\"extensions\": {\"code\": \"MAX_ROOT_FIELDS_LIMIT\"}`\n\nThis limit counts only the top level fields in a selection set,\nincluding fragments and inline fragments.",
-          "format": "uint32",
-          "minimum": 0,
-          "type": [
-            "integer",
-            "null"
-          ]
-        },
-        "parser_max_recursion": {
-          "default": 500,
-          "description": "Limit recursion in the GraphQL parser to protect against stack overflow.\ndefault: 500",
-          "format": "uint",
-          "minimum": 0,
-          "type": "integer"
+        "connector": {
+          "allOf": [
+            {
+              "$ref": "#/definitions/ConnectorConfiguration2"
+            }
+          ],
+          "default": {
+            "all": {
+              "http_max_response_size": null
+            },
+            "sources": {}
+          },
+          "description": "Limits that apply to outbound connector responses."
         },
-        "parser_max_tokens": {
-          "default": 15000,
-          "description": "Limit the number of tokens the GraphQL parser processes before aborting.",
-          "format": "uint",
-          "minimum": 0,
-          "type": "integer"
+        "router": {
+          "allOf": [
+            {
+              "$ref": "#/definitions/RouterLimitsConfig"
+            }
+          ],
+          "default": {
+            "http1_max_request_buf_size": null,
+            "http1_max_request_headers": null,
+            "http2_max_headers_list_bytes": null,
+            "http_max_request_bytes": 2000000,
+            "introspection_max_depth": true,
+            "max_aliases": null,
+            "max_depth": null,
+            "max_height": null,
+            "max_root_fields": null,
+            "parser_max_recursion": 500,
+            "parser_max_tokens": 15000,
+            "warn_only": false
+          },
+          "description": "Limits that apply to inbound requests to the router."
         },
-        "warn_only": {
-          "default": false,
-          "description": "If set to true (which is the default is dev mode),\nrequests that exceed a `max_*` limit are *not* rejected.\nInstead they are executed normally, and a warning is logged.",
-          "type": "boolean"
+        "subgraph": {
+          "allOf": [
+            {
+              "$ref": "#/definitions/SubgraphSubgraphLimitsConfiguration"
+            }
+          ],
+          "default": {
+            "all": {
+              "http_max_response_size": null
+            },
+            "subgraphs": {}
+          },
+          "description": "Limits that apply to outbound subgraph responses."
         }
       },
       "type": "object"
@@ -8368,6 +8360,110 @@ expression: "&schema"
       },
       "type": "object"
     },
+    "RouterLimitsConfig": {
+      "additionalProperties": false,
+      "description": "Limits that apply to inbound requests to the router.",
+      "properties": {
+        "http1_max_request_buf_size": {
+          "default": null,
+          "description": "Limit the maximum buffer size for the HTTP1 connection.\n\nDefault is ~400kib.",
+          "type": [
+            "string",
+            "null"
+          ]
+        },
+        "http1_max_request_headers": {
+          "default": null,
+          "description": "Limit the maximum number of headers of incoming HTTP1 requests. Default is 100.\n\nIf router receives more headers than the buffer size, it responds to the client with\n\"431 Request Header Fields Too Large\".",
+          "format": "uint",
+          "minimum": 0,
+          "type": [
+            "integer",
+            "null"
+          ]
+        },
+        "http2_max_headers_list_bytes": {
+          "default": null,
+          "description": "For HTTP2, limit the header list to a threshold of bytes. Default is 16kb.\n\nIf router receives more headers than allowed size of the header list, it responds to the client with\n\"431 Request Header Fields Too Large\".",
+          "type": [
+            "string",
+            "null"
+          ]
+        },
+        "http_max_request_bytes": {
+          "default": 2000000,
+          "description": "Limit the size of incoming HTTP requests read from the network,\nto protect against running out of memory. Default: 2000000 (2 MB)",
+          "format": "uint",
+          "minimum": 0,
+          "type": "integer"
+        },
+        "introspection_max_depth": {
+          "default": true,
+          "description": "Limit the depth of nested list fields in introspection queries\nto protect avoid generating huge responses. Returns a GraphQL\nerror with `{ message: \"Maximum introspection depth exceeded\" }`\nwhen nested fields exceed the limit.\nDefault: true",
+          "type": "boolean"
+        },
+        "max_aliases": {
+          "default": null,
+          "description": "If set, requests with operations with more aliases than this maximum\nare rejected with a HTTP 400 Bad Request response and GraphQL error with\n`\"extensions\": {\"code\": \"MAX_ALIASES_LIMIT\"}`",
+          "format": "uint32",
+          "minimum": 0,
+          "type": [
+            "integer",
+            "null"
+          ]
+        },
+        "max_depth": {
+          "default": null,
+          "description": "If set, requests with operations deeper than this maximum\nare rejected with a HTTP 400 Bad Request response and GraphQL error with\n`\"extensions\": {\"code\": \"MAX_DEPTH_LIMIT\"}`\n\nCounts depth of an operation, looking at its selection sets,˛\nincluding fields in fragments and inline fragments. The following\nexample has a depth of 3.\n\n```graphql\nquery getProduct {\n  book { # 1\n    ...bookDetails\n  }\n}\n\nfragment bookDetails on Book {\n  details { # 2\n    ... on ProductDetailsBook {\n      country # 3\n    }\n  }\n}\n```",
+          "format": "uint32",
+          "minimum": 0,
+          "type": [
+            "integer",
+            "null"
+          ]
+        },
+        "max_height": {
+          "default": null,
+          "description": "If set, requests with operations higher than this maximum\nare rejected with a HTTP 400 Bad Request response and GraphQL error with\n`\"extensions\": {\"code\": \"MAX_DEPTH_LIMIT\"}`\n\nHeight is based on simple merging of fields using the same name or alias,\nbut only within the same selection set.\nFor example `name` here is only counted once and the query has height 3, not 4:\n\n```graphql\nquery {\n    name { first }\n    name { last }\n}\n```\n\nThis may change in a future version of Apollo Router to do\n[full field merging across fragments][merging] instead.\n\n[merging]: https://spec.graphql.org/October2021/#sec-Field-Selection-Merging]",
+          "format": "uint32",
+          "minimum": 0,
+          "type": [
+            "integer",
+            "null"
+          ]
+        },
+        "max_root_fields": {
+          "default": null,
+          "description": "If set, requests with operations with more root fields than this maximum\nare rejected with a HTTP 400 Bad Request response and GraphQL error with\n`\"extensions\": {\"code\": \"MAX_ROOT_FIELDS_LIMIT\"}`\n\nThis limit counts only the top level fields in a selection set,\nincluding fragments and inline fragments.",
+          "format": "uint32",
+          "minimum": 0,
+          "type": [
+            "integer",
+            "null"
+          ]
+        },
+        "parser_max_recursion": {
+          "default": 500,
+          "description": "Limit recursion in the GraphQL parser to protect against stack overflow.\ndefault: 500",
+          "format": "uint",
+          "minimum": 0,
+          "type": "integer"
+        },
+        "parser_max_tokens": {
+          "default": 15000,
+          "description": "Limit the number of tokens the GraphQL parser processes before aborting.",
+          "format": "uint",
+          "minimum": 0,
+          "type": "integer"
+        },
+        "warn_only": {
+          "default": false,
+          "description": "If set to true (which is the default is dev mode),\nrequests that exceed a `max_*` limit are *not* rejected.\nInstead they are executed normally, and a warning is logged.",
+          "type": "boolean"
+        }
+      },
+      "type": "object"
+    },
     "RouterRequestConf": {
       "additionalProperties": false,
       "description": "What information is passed to a router request/response stage",
@@ -10009,6 +10105,21 @@ expression: "&schema"
       },
       "type": "object"
     },
+    "SubgraphLimits": {
+      "additionalProperties": false,
+      "description": "Per-subgraph response size limits.",
+      "properties": {
+        "http_max_response_size": {
+          "default": null,
+          "description": "Limit the size of incoming subgraph response bodies read from the network,\nto protect against running out of memory. Default: no limit.",
+          "type": [
+            "string",
+            "null"
+          ]
+        }
+      },
+      "type": "object"
+    },
     "SubgraphPassthroughMode": {
       "additionalProperties": false,
       "properties": {
@@ -11041,6 +11152,31 @@ expression: "&schema"
       },
       "type": "object"
     },
+    "SubgraphSubgraphLimitsConfiguration": {
+      "description": "Configuration options pertaining to the subgraph server component.",
+      "properties": {
+        "all": {
+          "allOf": [
+            {
+              "$ref": "#/definitions/SubgraphLimits"
+            }
+          ],
+          "default": {
+            "http_max_response_size": null
+          },
+          "description": "options applying to all subgraphs"
+        },
+        "subgraphs": {
+          "additionalProperties": {
+            "$ref": "#/definitions/SubgraphLimits"
+          },
+          "default": {},
+          "description": "per subgraph options",
+          "type": "object"
+        }
+      },
+      "type": "object"
+    },
     "SubgraphSubgraphStrategyConfigConfiguration": {
       "description": "Configuration options pertaining to the subgraph server component.",
       "properties": {
diff --git a/apollo-router/src/configuration/snapshots/apollo_router__configuration__tests__upgrade_old_configuration@http_max_request_bytes.router.yaml.snap b/apollo-router/src/configuration/snapshots/apollo_router__configuration__tests__upgrade_old_configuration@http_max_request_bytes.router.yaml.snap
index de57ef9fdc..e73681671c 100644
--- a/apollo-router/src/configuration/snapshots/apollo_router__configuration__tests__upgrade_old_configuration@http_max_request_bytes.router.yaml.snap
+++ b/apollo-router/src/configuration/snapshots/apollo_router__configuration__tests__upgrade_old_configuration@http_max_request_bytes.router.yaml.snap
@@ -4,5 +4,6 @@ expression: new_config
 ---
 ---
 limits:
-  http_max_request_bytes: 4000000
+  router:
+    http_max_request_bytes: 4000000
 
diff --git a/apollo-router/src/configuration/snapshots/apollo_router__configuration__tests__upgrade_old_configuration@operation-limits-preview.yaml.snap b/apollo-router/src/configuration/snapshots/apollo_router__configuration__tests__upgrade_old_configuration@operation-limits-preview.yaml.snap
index 4faff4711d..ff66b88f4e 100644
--- a/apollo-router/src/configuration/snapshots/apollo_router__configuration__tests__upgrade_old_configuration@operation-limits-preview.yaml.snap
+++ b/apollo-router/src/configuration/snapshots/apollo_router__configuration__tests__upgrade_old_configuration@operation-limits-preview.yaml.snap
@@ -4,5 +4,6 @@ expression: new_config
 ---
 ---
 limits:
-  parser_max_recursion: 1000
+  router:
+    parser_max_recursion: 1000
 
diff --git a/apollo-router/src/configuration/snapshots/apollo_router__configuration__tests__upgrade_old_configuration@parser_recursion.router.yaml.snap b/apollo-router/src/configuration/snapshots/apollo_router__configuration__tests__upgrade_old_configuration@parser_recursion.router.yaml.snap
index 2c9c072cd9..faddd330fa 100644
--- a/apollo-router/src/configuration/snapshots/apollo_router__configuration__tests__upgrade_old_configuration@parser_recursion.router.yaml.snap
+++ b/apollo-router/src/configuration/snapshots/apollo_router__configuration__tests__upgrade_old_configuration@parser_recursion.router.yaml.snap
@@ -4,5 +4,6 @@ expression: new_config
 ---
 ---
 limits:
-  parser_max_recursion: 100
+  router:
+    parser_max_recursion: 100
 
diff --git a/apollo-router/src/configuration/testdata/metrics/limits.router.yaml b/apollo-router/src/configuration/testdata/metrics/limits.router.yaml
index 6da506d1d5..32a973b470 100644
--- a/apollo-router/src/configuration/testdata/metrics/limits.router.yaml
+++ b/apollo-router/src/configuration/testdata/metrics/limits.router.yaml
@@ -1,9 +1,10 @@
 limits:
-  max_depth: 1
-  http_max_request_bytes: 2000000
-  warn_only: true
-  max_root_fields: 1
-  parser_max_tokens: 15000
-  parser_max_recursion: 500
-  max_height: 2
-  max_aliases: 2
+  router:
+    max_depth: 1
+    http_max_request_bytes: 2000000
+    warn_only: true
+    max_root_fields: 1
+    parser_max_tokens: 15000
+    parser_max_recursion: 500
+    max_height: 2
+    max_aliases: 2
diff --git a/apollo-router/src/introspection.rs b/apollo-router/src/introspection.rs
index f77c34e8e4..3efa7a8386 100644
--- a/apollo-router/src/introspection.rs
+++ b/apollo-router/src/introspection.rs
@@ -48,7 +48,7 @@ impl IntrospectionCache {
             storage.activate();
             Self(Mode::Enabled {
                 storage,
-                max_depth: if configuration.limits.introspection_max_depth {
+                max_depth: if configuration.limits.router.introspection_max_depth {
                     MaxDepth::Check
                 } else {
                     MaxDepth::Ignore
diff --git a/apollo-router/src/plugins/connectors/handle_responses.rs b/apollo-router/src/plugins/connectors/handle_responses.rs
index 9900b6508e..ca3b68555f 100644
--- a/apollo-router/src/plugins/connectors/handle_responses.rs
+++ b/apollo-router/src/plugins/connectors/handle_responses.rs
@@ -19,6 +19,8 @@ use apollo_federation::connectors::runtime::responses::handle_raw_response;
 use axum::body::HttpBody;
 use http::response::Parts;
 use http_body_util::BodyExt;
+use http_body_util::LengthLimitError;
+use http_body_util::Limited;
 use opentelemetry::KeyValue;
 use parking_lot::Mutex;
 use serde_json_bytes::Map;
@@ -28,6 +30,7 @@ use tracing::Span;
 use crate::Context;
 use crate::graphql;
 use crate::json_ext::Path;
+use crate::plugins::limits::ConnectorResponseSizeLimit;
 use crate::plugins::telemetry::config_new::attributes::HTTP_RESPONSE_BODY;
 use crate::plugins::telemetry::config_new::attributes::HTTP_RESPONSE_HEADERS;
 use crate::plugins::telemetry::config_new::attributes::HTTP_RESPONSE_STATUS;
@@ -66,7 +69,7 @@ impl From<RuntimeError> for graphql::Error {
 // --- handle_responses --------------------------------------------------------
 
 #[allow(clippy::too_many_arguments)]
-pub(crate) async fn process_response<T: HttpBody>(
+pub(crate) async fn process_response<T>(
     result: Result<http::Response<T>, Error>,
     response_key: ResponseKey,
     connector: Arc<Connector>,
@@ -75,7 +78,11 @@ pub(crate) async fn process_response<T: HttpBody>(
     debug_context: Option<&Arc<Mutex<ConnectorContext>>>,
     supergraph_request: Arc<http::Request<crate::graphql::Request>>,
     operation: Option<Arc<Valid<ExecutableDocument>>>,
-) -> connector::request_service::Response {
+) -> connector::request_service::Response
+where
+    T: HttpBody,
+    T::Error: Into<tower::BoxError>,
+{
     let (mapped_response, result) = match result {
         // This occurs when we short-circuit the request when over the limit
         Err(error) => {
@@ -114,10 +121,35 @@ pub(crate) async fn process_response<T: HttpBody>(
                 err
             };
 
-            let deserialized_body = body
-                .collect()
-                .await
-                .map_err(|_| ())
+            let response_size_limit = context
+                .extensions()
+                .with_lock(|e| e.get::<ConnectorResponseSizeLimit>().copied());
+
+            let body_result: Result<_, ()> = match response_size_limit {
+                Some(ConnectorResponseSizeLimit(limit)) => {
+                    Limited::new(body, limit)
+                        .collect()
+                        .await
+                        .map_err(|e| {
+                            if e.downcast_ref::<LengthLimitError>().is_some() {
+                                u64_counter!(
+                                    "apollo.router.limits.connector_response_size.exceeded",
+                                    "Number of connector responses aborted because they exceeded the configured response size limit",
+                                    1,
+                                    "connector.source" = connector.source_config_key()
+                                );
+                                tracing::Span::current()
+                                    .record("apollo.connector.response.aborted", "response_size_limit");
+                            }
+                        })
+                }
+                None => body
+                    .collect()
+                    .await
+                    .map_err(|_| ()),
+            };
+
+            let deserialized_body = body_result
                 .and_then(|body| {
                     let body = body.to_bytes();
                     let raw = deserialize_response(&body, &parts.headers).map_err(|_| {
@@ -1332,4 +1364,126 @@ mod tests {
             &Some(json!({"hello": json!(400)}))
         );
     }
+
+    fn make_connector() -> Arc<Connector> {
+        Arc::new(Connector {
+            spec: ConnectSpec::V0_1,
+            schema_subtypes_map: Default::default(),
+            id: ConnectId::new(
+                "subgraph_name".into(),
+                None,
+                name!(Query),
+                name!(hello),
+                None,
+                0,
+            ),
+            transport: HttpJsonTransport {
+                source_template: "http://localhost/api".parse().ok(),
+                connect_template: "/path".parse().unwrap(),
+                ..Default::default()
+            },
+            selection: JSONSelection::parse("$.data").unwrap(),
+            entity_resolver: None,
+            config: Default::default(),
+            max_requests: None,
+            batch_settings: None,
+            request_headers: Default::default(),
+            response_headers: Default::default(),
+            request_variable_keys: Default::default(),
+            response_variable_keys: Default::default(),
+            error_settings: Default::default(),
+            label: "test label".into(),
+        })
+    }
+
+    fn make_supergraph_request() -> Arc<http::Request<graphql::Request>> {
+        Arc::new(
+            http::Request::builder()
+                .body(graphql::Request::builder().build())
+                .unwrap(),
+        )
+    }
+
+    #[tokio::test]
+    async fn process_response_under_size_limit() {
+        use crate::plugins::limits::ConnectorResponseSizeLimit;
+
+        let ctx = Context::new();
+        ctx.extensions()
+            .with_lock(|e| e.insert(ConnectorResponseSizeLimit(1000)));
+
+        let key = ResponseKey::RootField {
+            name: "hello".to_string(),
+            inputs: Default::default(),
+            selection: Arc::new(JSONSelection::parse("$.data").unwrap()),
+        };
+        let response = http::Response::builder()
+            .body(router::body::from_bytes(r#"{"data":"world"}"#))
+            .unwrap();
+
+        let result = process_response(
+            Ok(response),
+            key,
+            make_connector(),
+            &ctx,
+            (None, Default::default()),
+            None,
+            make_supergraph_request(),
+            Default::default(),
+        )
+        .await;
+
+        let graphql_response =
+            super::aggregate_responses(vec![result.mapped_response], Context::new())
+                .unwrap()
+                .response;
+        assert!(
+            graphql_response.body().errors.is_empty(),
+            "expected no errors when response is under the limit"
+        );
+    }
+
+    #[tokio::test]
+    async fn process_response_exceeds_size_limit() {
+        use crate::plugins::limits::ConnectorResponseSizeLimit;
+
+        let ctx = Context::new();
+        // Limit of 5 bytes — well under the response body size
+        ctx.extensions()
+            .with_lock(|e| e.insert(ConnectorResponseSizeLimit(5)));
+
+        let key = ResponseKey::RootField {
+            name: "hello".to_string(),
+            inputs: Default::default(),
+            selection: Arc::new(JSONSelection::parse("$.data").unwrap()),
+        };
+        let response = http::Response::builder()
+            .body(router::body::from_bytes(r#"{"data":"world"}"#))
+            .unwrap();
+
+        let result = process_response(
+            Ok(response),
+            key,
+            make_connector(),
+            &ctx,
+            (None, Default::default()),
+            None,
+            make_supergraph_request(),
+            Default::default(),
+        )
+        .await;
+
+        let graphql_response =
+            super::aggregate_responses(vec![result.mapped_response], Context::new())
+                .unwrap()
+                .response;
+        let errors = &graphql_response.body().errors;
+        assert!(!errors.is_empty(), "expected an error for exceeded limit");
+        assert!(
+            errors[0].message.contains("exceeded limit of 5 bytes")
+                || errors[0].message.contains("unexpected format"),
+            "unexpected error message: {}",
+            errors[0].message
+        );
+    }
 }
diff --git a/apollo-router/src/plugins/limits/fixtures/content_length_limit.router.yaml b/apollo-router/src/plugins/limits/fixtures/content_length_limit.router.yaml
index e95015cbdd..3ec99ce020 100644
--- a/apollo-router/src/plugins/limits/fixtures/content_length_limit.router.yaml
+++ b/apollo-router/src/plugins/limits/fixtures/content_length_limit.router.yaml
@@ -1,2 +1,3 @@
 limits:
-  http_max_request_bytes: 10
\ No newline at end of file
+  router:
+    http_max_request_bytes: 10
\ No newline at end of file
diff --git a/apollo-router/src/plugins/limits/mod.rs b/apollo-router/src/plugins/limits/mod.rs
index f077a7055c..def1746db1 100644
--- a/apollo-router/src/plugins/limits/mod.rs
+++ b/apollo-router/src/plugins/limits/mod.rs
@@ -14,20 +14,40 @@ use tower::ServiceBuilder;
 use tower::ServiceExt;
 
 use crate::Context;
+use crate::configuration::connector::ConnectorConfiguration;
+use crate::configuration::subgraph::SubgraphConfiguration;
 use crate::graphql;
 use crate::layers::ServiceBuilderExt;
-use crate::plugin::Plugin;
 use crate::plugin::PluginInit;
+use crate::plugin::PluginPrivate;
 use crate::plugins::limits::layer::BodyLimitError;
 use crate::plugins::limits::layer::RequestBodyLimitLayer;
+use crate::services::SubgraphRequest;
+use crate::services::connector;
 use crate::services::router;
 use crate::services::router::BoxService;
+use crate::services::subgraph;
 
 /// Configuration for operation limits, parser limits, HTTP limits, etc.
-#[derive(Debug, Clone, Deserialize, Serialize, JsonSchema)]
+#[derive(Debug, Clone, Default, Deserialize, Serialize, JsonSchema)]
 #[serde(deny_unknown_fields, default)]
 #[schemars(rename = "LimitsConfig")]
 pub(crate) struct Config {
+    /// Limits that apply to inbound requests to the router.
+    pub(crate) router: RouterLimitsConfig,
+
+    /// Limits that apply to outbound subgraph responses.
+    pub(crate) subgraph: SubgraphConfiguration<SubgraphLimits>,
+
+    /// Limits that apply to outbound connector responses.
+    pub(crate) connector: ConnectorConfiguration<ConnectorLimits>,
+}
+
+/// Limits that apply to inbound requests to the router.
+#[derive(Debug, Clone, Deserialize, Serialize, JsonSchema)]
+#[serde(deny_unknown_fields, default)]
+#[schemars(rename = "RouterLimitsConfig")]
+pub(crate) struct RouterLimitsConfig {
     /// If set, requests with operations deeper than this maximum
     /// are rejected with a HTTP 400 Bad Request response and GraphQL error with
     /// `"extensions": {"code": "MAX_DEPTH_LIMIT"}`
@@ -130,7 +150,7 @@ pub(crate) struct Config {
     pub(crate) introspection_max_depth: bool,
 }
 
-impl Default for Config {
+impl Default for RouterLimitsConfig {
     fn default() -> Self {
         Self {
             // These limits are opt-in
@@ -145,7 +165,7 @@ impl Default for Config {
             http2_max_headers_list_bytes: None,
             parser_max_tokens: 15_000,
 
-            // This is `apollo-parser`’s default, which protects against stack overflow
+            // This is `apollo-parser`'s default, which protects against stack overflow
             // but is still very high for "reasonable" queries.
             // https://github.com/apollographql/apollo-rs/blob/apollo-parser%400.7.3/crates/apollo-parser/src/parser/mod.rs#L93-L104
             parser_max_recursion: 500,
@@ -155,12 +175,76 @@ impl Default for Config {
     }
 }
 
+/// Per-subgraph response size limits.
+#[derive(Debug, Default, Clone, Deserialize, Serialize, JsonSchema)]
+#[serde(deny_unknown_fields, default)]
+#[schemars(rename = "SubgraphLimits")]
+pub(crate) struct SubgraphLimits {
+    /// Limit the size of incoming subgraph response bodies read from the network,
+    /// to protect against running out of memory. Default: no limit.
+    #[schemars(with = "Option<String>", default)]
+    pub(crate) http_max_response_size: Option<ByteSize>,
+}
+
+/// Extension type placed on the request context to signal the subgraph response size limit.
+#[derive(Clone, Copy, Debug, Ord, PartialOrd, PartialEq, Eq)]
+pub(crate) struct SubgraphResponseSizeLimit(pub usize);
+
+/// Per-connector-source response size limits.
+#[derive(Debug, Clone, Default, Deserialize, Serialize, JsonSchema)]
+#[serde(deny_unknown_fields, default)]
+#[schemars(rename = "ConnectorLimits")]
+pub(crate) struct ConnectorLimits {
+    /// Limit the size of incoming connector response bodies read from the network,
+    /// to protect against running out of memory. Default: no limit.
+    #[schemars(with = "Option<String>", default)]
+    pub(crate) http_max_response_size: Option<ByteSize>,
+}
+
+/// Extension type placed on the request context to signal the connector response size limit.
+#[derive(Clone, Copy, Debug, Ord, PartialOrd, PartialEq, Eq)]
+pub(crate) struct ConnectorResponseSizeLimit(pub usize);
+
+impl Config {
+    fn subgraph_response_size_limit(
+        &self,
+        subgraph_name: &str,
+    ) -> Option<SubgraphResponseSizeLimit> {
+        // check for non-null subgraph.http_max_response_size or all.http_max_response_size
+        let subgraph_limit = self
+            .subgraph
+            .subgraphs
+            .get(subgraph_name)
+            .and_then(|s| s.http_max_response_size);
+        let limit = subgraph_limit.or_else(|| self.subgraph.all.http_max_response_size)?;
+
+        // convert to usize (needed for limits plugin)
+        Some(SubgraphResponseSizeLimit(limit.as_u64().try_into().ok()?))
+    }
+
+    fn connector_response_size_limit(
+        &self,
+        source_name: &str,
+    ) -> Option<ConnectorResponseSizeLimit> {
+        // check for non-null subgraph.http_max_response_size or all.http_max_response_size
+        let source_limit = self
+            .connector
+            .sources
+            .get(source_name)
+            .and_then(|s| s.http_max_response_size);
+        let limit = source_limit.or_else(|| self.connector.all.http_max_response_size)?;
+
+        // convert to usize (needed for limits plugin)
+        Some(ConnectorResponseSizeLimit(limit.as_u64().try_into().ok()?))
+    }
+}
+
 struct LimitsPlugin {
     config: Config,
 }
 
 #[async_trait]
-impl Plugin for LimitsPlugin {
+impl PluginPrivate for LimitsPlugin {
     type Config = Config;
 
     async fn new(init: PluginInit<Self::Config>) -> Result<Self, BoxError>
@@ -182,13 +266,43 @@ impl Plugin for LimitsPlugin {
             .map_request(Into::into)
             .map_response(Into::into)
             .layer(RequestBodyLimitLayer::new(
-                self.config.http_max_request_bytes,
+                self.config.router.http_max_request_bytes,
             ))
             .map_request(Into::into)
             .map_response(Into::into)
             .service(service)
             .boxed()
     }
+
+    fn subgraph_service(&self, name: &str, service: subgraph::BoxService) -> subgraph::BoxService {
+        match self.config.subgraph_response_size_limit(name) {
+            None => service,
+            Some(limit) => ServiceBuilder::new()
+                .map_request(move |req: SubgraphRequest| {
+                    req.context.extensions().with_lock(|e| e.insert(limit));
+                    req
+                })
+                .service(service)
+                .boxed(),
+        }
+    }
+
+    fn connector_request_service(
+        &self,
+        service: connector::request_service::BoxService,
+        source_name: String,
+    ) -> connector::request_service::BoxService {
+        match self.config.connector_response_size_limit(&source_name) {
+            None => service,
+            Some(limit) => ServiceBuilder::new()
+                .map_request(move |req: connector::request_service::Request| {
+                    req.context.extensions().with_lock(|e| e.insert(limit));
+                    req
+                })
+                .service(service)
+                .boxed(),
+        }
+    }
 }
 
 impl LimitsPlugin {
@@ -243,14 +357,36 @@ impl BodyLimitError {
     }
 }
 
-register_plugin!("apollo", "limits", LimitsPlugin);
+register_private_plugin!("apollo", "limits", LimitsPlugin);
+
+#[cfg(test)]
+impl From<SubgraphConfiguration<SubgraphLimits>> for Config {
+    fn from(subgraph: SubgraphConfiguration<SubgraphLimits>) -> Self {
+        Self {
+            subgraph,
+            ..Self::default()
+        }
+    }
+}
+
+#[cfg(test)]
+impl From<ConnectorConfiguration<ConnectorLimits>> for Config {
+    fn from(connector: ConnectorConfiguration<ConnectorLimits>) -> Self {
+        Self {
+            connector,
+            ..Self::default()
+        }
+    }
+}
 
 #[cfg(test)]
 mod test {
     use http::StatusCode;
     use tower::BoxError;
 
+    use crate::Context;
     use crate::plugins::limits::LimitsPlugin;
+    use crate::plugins::limits::SubgraphResponseSizeLimit;
     use crate::plugins::limits::layer::BodyLimitControl;
     use crate::plugins::test::PluginTestHarness;
     use crate::services::router;
@@ -445,4 +581,374 @@ mod test {
             .expect("test harness");
         plugin
     }
+
+    /// Check configuration for subgraph_response_limit
+    mod subgraph_response_limit {
+        use bytesize::ByteSize;
+
+        use crate::configuration::subgraph::SubgraphConfiguration;
+        use crate::plugins::limits::Config;
+        use crate::plugins::limits::SubgraphLimits;
+        use crate::plugins::limits::SubgraphResponseSizeLimit;
+
+        #[test]
+        fn get_response_limit_no_config() {
+            let subgraph_config = SubgraphConfiguration::<SubgraphLimits>::default();
+            let config: Config = subgraph_config.into();
+            assert_eq!(config.subgraph_response_size_limit("products"), None);
+        }
+
+        #[test]
+        fn get_response_limit_all() {
+            let mut subgraph_config = SubgraphConfiguration::<SubgraphLimits>::default();
+            subgraph_config.all.http_max_response_size = Some(ByteSize::kb(1));
+
+            let config: Config = subgraph_config.into();
+            assert_eq!(
+                config.subgraph_response_size_limit("products"),
+                Some(SubgraphResponseSizeLimit(1000))
+            );
+            assert_eq!(
+                config.subgraph_response_size_limit("reviews"),
+                Some(SubgraphResponseSizeLimit(1000))
+            );
+        }
+
+        #[test]
+        fn get_response_limit_subgraph_specific() {
+            let mut subgraph_config = SubgraphConfiguration::<SubgraphLimits>::default();
+            subgraph_config.subgraphs.insert(
+                "products".to_string(),
+                SubgraphLimits {
+                    http_max_response_size: Some(ByteSize::b(512)),
+                },
+            );
+
+            let config: Config = subgraph_config.into();
+            assert_eq!(
+                config.subgraph_response_size_limit("products"),
+                Some(SubgraphResponseSizeLimit(512))
+            );
+            assert_eq!(config.subgraph_response_size_limit("reviews"), None);
+        }
+
+        #[test]
+        fn get_response_limit_subgraph_overrides_all() {
+            let mut subgraph_config = SubgraphConfiguration::<SubgraphLimits>::default();
+            subgraph_config.all.http_max_response_size = Some(ByteSize::kib(1));
+            subgraph_config.subgraphs.insert(
+                "products".to_string(),
+                SubgraphLimits {
+                    http_max_response_size: Some(ByteSize::b(500)),
+                },
+            );
+            subgraph_config.subgraphs.insert(
+                "reviews".to_string(),
+                SubgraphLimits {
+                    http_max_response_size: None,
+                },
+            );
+
+            let config: Config = subgraph_config.into();
+            // per-subgraph override wins
+            assert_eq!(
+                config.subgraph_response_size_limit("products"),
+                Some(SubgraphResponseSizeLimit(500))
+            );
+            // fallback to all despite having an entry in the map
+            assert_eq!(
+                config.subgraph_response_size_limit("reviews"),
+                Some(SubgraphResponseSizeLimit(1024))
+            );
+        }
+    }
+
+    /// Check configuration for connector_response_limit
+    mod connector_response_limit {
+        use bytesize::ByteSize;
+
+        use crate::configuration::connector::ConnectorConfiguration;
+        use crate::plugins::limits::Config;
+        use crate::plugins::limits::ConnectorLimits;
+        use crate::plugins::limits::ConnectorResponseSizeLimit;
+
+        #[test]
+        fn get_response_limit_no_config() {
+            let connector_config = ConnectorConfiguration::<ConnectorLimits>::default();
+            let config: Config = connector_config.into();
+            assert_eq!(config.connector_response_size_limit("products.rest"), None);
+        }
+
+        #[test]
+        fn get_response_limit_all() {
+            let mut connector_config = ConnectorConfiguration::<ConnectorLimits>::default();
+            connector_config.all.http_max_response_size = Some(ByteSize::kb(1));
+
+            let config: Config = connector_config.into();
+            assert_eq!(
+                config.connector_response_size_limit("products.rest"),
+                Some(ConnectorResponseSizeLimit(1000))
+            );
+            assert_eq!(
+                config.connector_response_size_limit("reviews.api"),
+                Some(ConnectorResponseSizeLimit(1000))
+            );
+        }
+
+        #[test]
+        fn get_response_limit_subgraph_specific() {
+            let mut connector_config = ConnectorConfiguration::<ConnectorLimits>::default();
+            connector_config.sources.insert(
+                "products.rest".to_string(),
+                ConnectorLimits {
+                    http_max_response_size: Some(ByteSize::b(512)),
+                },
+            );
+
+            let config: Config = connector_config.into();
+            assert_eq!(
+                config.connector_response_size_limit("products.rest"),
+                Some(ConnectorResponseSizeLimit(512))
+            );
+            assert_eq!(config.connector_response_size_limit("reviews.api"), None);
+        }
+
+        #[test]
+        fn get_response_limit_subgraph_overrides_all() {
+            let mut connector_config = ConnectorConfiguration::<ConnectorLimits>::default();
+            connector_config.all.http_max_response_size = Some(ByteSize::kib(1));
+            connector_config.sources.insert(
+                "products.rest".to_string(),
+                ConnectorLimits {
+                    http_max_response_size: Some(ByteSize::b(500)),
+                },
+            );
+            connector_config.sources.insert(
+                "reviews.api".to_string(),
+                ConnectorLimits {
+                    http_max_response_size: None,
+                },
+            );
+
+            let config: Config = connector_config.into();
+            // per-subgraph override wins
+            assert_eq!(
+                config.connector_response_size_limit("products.rest"),
+                Some(ConnectorResponseSizeLimit(500))
+            );
+            // fallback to all despite having an entry in the map
+            assert_eq!(
+                config.connector_response_size_limit("reviews.api"),
+                Some(ConnectorResponseSizeLimit(1024))
+            );
+        }
+    }
+
+    // --- LimitsPlugin::connector_request_service ---
+
+    fn make_connector_request(
+        ctx: Context,
+    ) -> crate::services::connector::request_service::Request {
+        use std::sync::Arc;
+
+        use apollo_compiler::name;
+        use apollo_federation::connectors::ConnectId;
+        use apollo_federation::connectors::ConnectSpec;
+        use apollo_federation::connectors::Connector;
+        use apollo_federation::connectors::HttpJsonTransport;
+        use apollo_federation::connectors::JSONSelection;
+        use apollo_federation::connectors::runtime::http_json_transport::HttpRequest;
+        use apollo_federation::connectors::runtime::key::ResponseKey;
+
+        let connector = Connector {
+            spec: ConnectSpec::V0_1,
+            schema_subtypes_map: Default::default(),
+            id: ConnectId::new(
+                "subgraph_name".into(),
+                None,
+                name!(Query),
+                name!(hello),
+                None,
+                0,
+            ),
+            transport: HttpJsonTransport {
+                source_template: "http://localhost/api".parse().ok(),
+                connect_template: "/path".parse().unwrap(),
+                ..Default::default()
+            },
+            selection: JSONSelection::parse("$.data").unwrap(),
+            entity_resolver: None,
+            config: Default::default(),
+            max_requests: None,
+            batch_settings: None,
+            request_headers: Default::default(),
+            response_headers: Default::default(),
+            request_variable_keys: Default::default(),
+            response_variable_keys: Default::default(),
+            error_settings: Default::default(),
+            label: "test label".into(),
+        };
+        let key = ResponseKey::RootField {
+            name: "hello".to_string(),
+            inputs: Default::default(),
+            selection: Arc::new(JSONSelection::parse("$.data").unwrap()),
+        };
+        let http_request = HttpRequest {
+            inner: http::Request::builder().body("{}".to_string()).unwrap(),
+            debug: Default::default(),
+        };
+        crate::services::connector::request_service::Request {
+            context: ctx,
+            connector: Arc::new(connector),
+            transport_request: http_request.into(),
+            key,
+            mapping_problems: Default::default(),
+            supergraph_request: Arc::new(
+                http::Request::builder()
+                    .body(crate::graphql::Request::builder().build())
+                    .expect("valid request"),
+            ),
+            operation: Default::default(),
+        }
+    }
+
+    fn make_stub_connector_response(
+        req: &crate::services::connector::request_service::Request,
+    ) -> crate::services::connector::request_service::Response {
+        use apollo_federation::connectors::runtime::http_json_transport::HttpResponse;
+        use apollo_federation::connectors::runtime::http_json_transport::TransportResponse;
+        use apollo_federation::connectors::runtime::responses::MappedResponse;
+        use serde_json_bytes::Value;
+
+        let (parts, _) = http::Response::builder().body(()).unwrap().into_parts();
+        crate::services::connector::request_service::Response {
+            context: req.context.clone(),
+            transport_result: Ok(TransportResponse::Http(HttpResponse { inner: parts })),
+            mapped_response: MappedResponse::Data {
+                data: Value::Null,
+                key: req.key.clone(),
+                problems: vec![],
+            },
+        }
+    }
+
+    #[tokio::test]
+    async fn connector_request_service_sets_limit_on_context() {
+        use crate::plugins::limits::ConnectorResponseSizeLimit;
+
+        let plugin: PluginTestHarness<LimitsPlugin> = PluginTestHarness::builder()
+            .config("limits:\n  connector:\n    all:\n      http_max_response_size: 2kib")
+            .build()
+            .await
+            .expect("test harness");
+
+        let result = plugin
+            .call_connector_request_service(
+                make_connector_request(Context::new()),
+                |req: crate::services::connector::request_service::Request| {
+                    let limit = req
+                        .context
+                        .extensions()
+                        .with_lock(|e| e.get::<ConnectorResponseSizeLimit>().copied());
+                    assert_eq!(
+                        limit.map(|l| l.0),
+                        Some(2048),
+                        "limit should be set on context"
+                    );
+                    make_stub_connector_response(&req)
+                },
+            )
+            .await;
+
+        assert!(result.is_ok());
+    }
+
+    #[tokio::test]
+    async fn connector_request_service_no_limit_does_not_set_extension() {
+        use crate::plugins::limits::ConnectorResponseSizeLimit;
+
+        let plugin: PluginTestHarness<LimitsPlugin> = PluginTestHarness::builder()
+            .config(include_str!("fixtures/content_length_limit.router.yaml"))
+            .build()
+            .await
+            .expect("test harness");
+
+        let result = plugin
+            .call_connector_request_service(
+                make_connector_request(Context::new()),
+                |req: crate::services::connector::request_service::Request| {
+                    let limit = req
+                        .context
+                        .extensions()
+                        .with_lock(|e| e.get::<ConnectorResponseSizeLimit>().copied());
+                    assert!(limit.is_none(), "no limit should be set on context");
+                    make_stub_connector_response(&req)
+                },
+            )
+            .await;
+
+        assert!(result.is_ok());
+    }
+
+    // --- LimitsPlugin::subgraph_service ---
+
+    #[tokio::test]
+    async fn subgraph_service_sets_limit_on_context() {
+        let plugin: PluginTestHarness<LimitsPlugin> = PluginTestHarness::builder()
+            .config("limits:\n  subgraph:\n    all:\n      http_max_response_size: 1024b")
+            .build()
+            .await
+            .expect("test harness");
+
+        let result = plugin
+            .subgraph_service(
+                "products",
+                |req: crate::services::SubgraphRequest| async move {
+                    let limit = req
+                        .context
+                        .extensions()
+                        .with_lock(|e| e.get::<SubgraphResponseSizeLimit>().copied());
+                    assert_eq!(
+                        limit.map(|l| l.0),
+                        Some(1024),
+                        "limit should be set on context"
+                    );
+                    Ok(crate::services::SubgraphResponse::fake_builder()
+                        .context(req.context)
+                        .build())
+                },
+            )
+            .call_default()
+            .await;
+
+        assert!(result.is_ok());
+    }
+
+    #[tokio::test]
+    async fn subgraph_service_no_limit_does_not_set_extension() {
+        let plugin: PluginTestHarness<LimitsPlugin> = PluginTestHarness::builder()
+            .config(include_str!("fixtures/content_length_limit.router.yaml"))
+            .build()
+            .await
+            .expect("test harness");
+
+        let result = plugin
+            .subgraph_service(
+                "products",
+                |req: crate::services::SubgraphRequest| async move {
+                    let limit = req
+                        .context
+                        .extensions()
+                        .with_lock(|e| e.get::<SubgraphResponseSizeLimit>().copied());
+                    assert!(limit.is_none(), "no limit should be set on context");
+                    Ok(crate::services::SubgraphResponse::fake_builder()
+                        .context(req.context)
+                        .build())
+                },
+            )
+            .call_default()
+            .await;
+
+        assert!(result.is_ok());
+    }
 }
diff --git a/apollo-router/src/plugins/telemetry/span_factory.rs b/apollo-router/src/plugins/telemetry/span_factory.rs
index 7d7b3e290c..56a83dbc8e 100644
--- a/apollo-router/src/plugins/telemetry/span_factory.rs
+++ b/apollo-router/src/plugins/telemetry/span_factory.rs
@@ -218,6 +218,7 @@ impl SpanMode {
                     "apollo.source.name" = source_name,
                     "otel.kind" = "INTERNAL",
                     "otel.status_code" = ::tracing::field::Empty,
+                    "apollo.connector.response.aborted" = ::tracing::field::Empty,
                 )
             }
             SpanMode::SpecCompliant => {
@@ -225,6 +226,7 @@ impl SpanMode {
                     CONNECT_REQUEST_SPAN_NAME,
                     "otel.kind" = "INTERNAL",
                     "otel.status_code" = ::tracing::field::Empty,
+                    "apollo.connector.response.aborted" = ::tracing::field::Empty,
                 )
             }
         }
diff --git a/apollo-router/src/query_planner/caching_query_planner.rs b/apollo-router/src/query_planner/caching_query_planner.rs
index fda5d82837..557fab83c9 100644
--- a/apollo-router/src/query_planner/caching_query_planner.rs
+++ b/apollo-router/src/query_planner/caching_query_planner.rs
@@ -126,7 +126,7 @@ pub(crate) struct CachingQueryPlanner<T: Clone> {
     enable_authorization_directives: bool,
     config_mode_hash: Arc<ConfigModeHash>,
     cooperative_cancellation: CooperativeCancellation,
-    config_limits: limits::Config,
+    config_limits: limits::RouterLimitsConfig,
 }
 
 fn init_query_plan_from_redis(
@@ -189,7 +189,7 @@ where
             enable_authorization_directives,
             cooperative_cancellation,
             config_mode_hash,
-            config_limits: configuration.limits.clone(),
+            config_limits: configuration.limits.router.clone(),
         })
     }
 
diff --git a/apollo-router/src/query_planner/query_planner_service.rs b/apollo-router/src/query_planner/query_planner_service.rs
index 831fa74527..e8fa9ae806 100644
--- a/apollo-router/src/query_planner/query_planner_service.rs
+++ b/apollo-router/src/query_planner/query_planner_service.rs
@@ -274,7 +274,7 @@ impl QueryPlannerService {
         let executable = &doc.executable;
         crate::spec::operation_limits::check(
             query_metrics_in,
-            &self.configuration.limits,
+            &self.configuration.limits.router,
             &query,
             executable,
             operation_name,
diff --git a/apollo-router/src/services/router/body.rs b/apollo-router/src/services/router/body.rs
index 7aa5819f3a..79f258fac1 100644
--- a/apollo-router/src/services/router/body.rs
+++ b/apollo-router/src/services/router/body.rs
@@ -6,9 +6,11 @@ use http_body::Frame;
 use http_body_util::BodyExt;
 use http_body_util::Empty;
 use http_body_util::Full;
+use http_body_util::Limited;
 use http_body_util::StreamBody;
 use http_body_util::combinators::UnsyncBoxBody;
 use hyper::body::Body as HttpBody;
+use tower::BoxError;
 
 pub type RouterBody = UnsyncBoxBody<Bytes, AxumError>;
 
@@ -45,6 +47,16 @@ where
     ))
 }
 
+/// Like `into_bytes`, but rejects the body if it exceeds `limit` bytes.
+/// Checks size per-frame as data arrives — does not buffer the full body before checking.
+pub(crate) async fn into_bytes_limited<B>(body: B, limit: usize) -> Result<Bytes, BoxError>
+where
+    B: HttpBody,
+    B::Error: Into<BoxError>,
+{
+    Ok(Limited::new(body, limit).collect().await?.to_bytes())
+}
+
 /// Get a body's contents as a utf-8 string for use in test assertions, or return an error.
 pub async fn into_string<B>(input: B) -> Result<String, AxumError>
 where
@@ -60,3 +72,40 @@ where
     let string = String::from_utf8(bytes).map_err(AxumError::new)?;
     Ok(string)
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn into_bytes_limited_under_limit() {
+        let body = from_bytes("hello");
+        let result = into_bytes_limited(body, 10).await;
+        assert!(result.is_ok());
+        assert_eq!(result.unwrap(), "hello");
+    }
+
+    #[tokio::test]
+    async fn into_bytes_limited_at_limit() {
+        let body = from_bytes("hello");
+        let result = into_bytes_limited(body, 5).await;
+        assert!(result.is_ok());
+        assert_eq!(result.unwrap(), "hello");
+    }
+
+    #[tokio::test]
+    async fn into_bytes_limited_over_limit() {
+        use http_body_util::LengthLimitError;
+
+        let body = from_bytes("hello world");
+        let result = into_bytes_limited(body, 5).await;
+        assert!(result.is_err());
+        assert!(
+            result
+                .unwrap_err()
+                .downcast_ref::<LengthLimitError>()
+                .is_some(),
+            "error should be a LengthLimitError"
+        );
+    }
+}
diff --git a/apollo-router/src/services/subgraph_service.rs b/apollo-router/src/services/subgraph_service.rs
index 752b0fb4d1..eecb51b783 100644
--- a/apollo-router/src/services/subgraph_service.rs
+++ b/apollo-router/src/services/subgraph_service.rs
@@ -17,6 +17,7 @@ use http::header::ACCEPT;
 use http::header::CONTENT_TYPE;
 use http::response::Parts;
 use http_body::Body;
+use http_body_util::LengthLimitError;
 use hyper_rustls::ConfigBuilderExt;
 use itertools::Itertools;
 use mediatype::MediaType;
@@ -59,6 +60,7 @@ use crate::layers::DEFAULT_BUFFER_SIZE;
 use crate::layers::unconstrained_buffer::UnconstrainedBuffer;
 use crate::layers::unconstrained_buffer::UnconstrainedBufferLayer;
 use crate::plugins::file_uploads;
+use crate::plugins::limits::SubgraphResponseSizeLimit;
 use crate::plugins::subscription::SubscriptionConfig;
 use crate::plugins::subscription::subgraph::SubscriptionSubgraphLayer;
 use crate::plugins::telemetry::config_new::events::log_event;
@@ -419,7 +421,8 @@ pub(crate) async fn process_batch(
         "http.url" = %schema_uri,
         "net.transport" = "ip_tcp",
         "apollo.subgraph.name" = %&service,
-        "graphql.operation.name" = "batch"
+        "graphql.operation.name" = "batch",
+        "apollo.subgraph.response.aborted" = tracing::field::Empty,
     );
 
     // The graphql spec is lax about what strategy to use for processing responses: https://github.com/graphql/graphql-over-http/blob/main/spec/GraphQLOverHTTP.md#processing-the-response
@@ -850,6 +853,7 @@ pub(crate) async fn call_single_http(
         "net.transport" = "ip_tcp",
         "apollo.subgraph.name" = %service_name,
         "graphql.operation.name" = %operation_name,
+        "apollo.subgraph.response.aborted" = tracing::field::Empty,
     );
 
     // The graphql spec is lax about what strategy to use for processing responses: https://github.com/graphql/graphql-over-http/blob/main/spec/GraphQLOverHTTP.md#processing-the-response
@@ -1084,19 +1088,53 @@ async fn do_fetch(
 
     let content_type = get_graphql_content_type(service_name, &parts);
 
+    let response_size_limit = context
+        .extensions()
+        .with_lock(|e| e.get::<SubgraphResponseSizeLimit>().copied());
+
     let body = if content_type.is_ok() {
-        let body = router::body::into_bytes(body)
-            .instrument(tracing::debug_span!("aggregate_response_data"))
-            .await
-            .map_err(|err| {
-                tracing::error!(fetch_error = ?err);
-                FetchError::SubrequestHttpError {
-                    status_code: Some(parts.status.as_u16()),
-                    service: service_name.to_string(),
-                    reason: err.to_string(),
-                }
-            });
-        Some(body)
+        let body_result = match response_size_limit {
+            Some(SubgraphResponseSizeLimit(limit)) => {
+                router::body::into_bytes_limited(body, limit)
+                    .instrument(tracing::debug_span!("aggregate_response_data"))
+                    .await
+                    .map_err(|err| {
+                        tracing::error!(fetch_error = ?err);
+                        let reason = if err.downcast_ref::<LengthLimitError>().is_some() {
+                            u64_counter!(
+                                "apollo.router.limits.subgraph_response_size.exceeded",
+                                "Number of subgraph responses aborted because they exceeded the configured response size limit",
+                                1,
+                                subgraph.name = service_name.to_string()
+                            );
+                            tracing::Span::current()
+                                .record("apollo.subgraph.response.aborted", "response_size_limit");
+                            format!("subgraph response body exceeded limit of {limit} bytes")
+                        } else {
+                            err.to_string()
+                        };
+                        FetchError::SubrequestHttpError {
+                            status_code: Some(parts.status.as_u16()),
+                            service: service_name.to_string(),
+                            reason,
+                        }
+                    })
+            }
+            None => {
+                router::body::into_bytes(body)
+                    .instrument(tracing::debug_span!("aggregate_response_data"))
+                    .await
+                    .map_err(|err| {
+                        tracing::error!(fetch_error = ?err);
+                        FetchError::SubrequestHttpError {
+                            status_code: Some(parts.status.as_u16()),
+                            service: service_name.to_string(),
+                            reason: err.to_string(),
+                        }
+                    })
+            }
+        };
+        Some(body_result)
     } else {
         None
     };
@@ -1389,6 +1427,21 @@ mod tests {
         serve(listener, handle).await.unwrap();
     }
 
+    // starts a local server emulating a subgraph returning a large JSON response
+    async fn emulate_subgraph_large_response(listener: TcpListener) {
+        async fn handle(_request: http::Request<Body>) -> Result<http::Response<Body>, Infallible> {
+            // 100 bytes of JSON — enough to exceed a small limit in tests
+            let body = format!(r#"{{"data":{{"field":"{}"}}}}"#, "x".repeat(80));
+            Ok(http::Response::builder()
+                .header(CONTENT_TYPE, APPLICATION_JSON.essence_str())
+                .status(StatusCode::OK)
+                .body(body.into())
+                .unwrap())
+        }
+
+        serve(listener, handle).await.unwrap();
+    }
+
     // starts a local server emulating a subgraph returning bad response format
     async fn emulate_subgraph_application_graphql_response(listener: TcpListener) {
         async fn handle(_request: http::Request<Body>) -> Result<http::Response<Body>, Infallible> {
@@ -2178,6 +2231,96 @@ mod tests {
         );
     }
 
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_subgraph_service_response_size_limit_exceeded() {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let socket_addr = listener.local_addr().unwrap();
+        tokio::task::spawn(emulate_subgraph_large_response(listener));
+        let subgraph_service = SubgraphService::new(
+            "test",
+            true,
+            HttpClientServiceFactory::from_config(
+                "test",
+                &Configuration::default(),
+                crate::configuration::shared::Client::default(),
+            ),
+        )
+        .expect("can create a SubgraphService");
+
+        let context = Context::new();
+        context
+            .extensions()
+            .with_lock(|e| e.insert(SubgraphResponseSizeLimit(10)));
+
+        let url = Uri::from_str(&format!("http://{socket_addr}")).unwrap();
+        let response = subgraph_service
+            .oneshot(
+                SubgraphRequest::builder()
+                    .supergraph_request(supergraph_request("query"))
+                    .subgraph_request(subgraph_http_request(url, "query"))
+                    .operation_kind(OperationKind::Query)
+                    .subgraph_name(String::from("test"))
+                    .context(context)
+                    .build(),
+            )
+            .await
+            .unwrap();
+
+        let errors = &response.response.body().errors;
+        assert!(!errors.is_empty(), "expected an error for exceeded limit");
+        assert!(
+            errors[0].message.contains("exceeded limit of 10 bytes"),
+            "unexpected error message: {}",
+            errors[0].message
+        );
+        assert_eq!(
+            errors[0].extensions.get("code").and_then(|v| v.as_str()),
+            Some("SUBREQUEST_HTTP_ERROR")
+        );
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    async fn test_subgraph_service_response_size_limit_under() {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let socket_addr = listener.local_addr().unwrap();
+        tokio::task::spawn(emulate_subgraph_application_json_response(listener));
+        let subgraph_service = SubgraphService::new(
+            "test",
+            true,
+            HttpClientServiceFactory::from_config(
+                "test",
+                &Configuration::default(),
+                crate::configuration::shared::Client::default(),
+            ),
+        )
+        .expect("can create a SubgraphService");
+
+        let context = Context::new();
+        // Limit of 1000 bytes — well above {"data": null} (14 bytes)
+        context
+            .extensions()
+            .with_lock(|e| e.insert(SubgraphResponseSizeLimit(1000)));
+
+        let url = Uri::from_str(&format!("http://{socket_addr}")).unwrap();
+        let response = subgraph_service
+            .oneshot(
+                SubgraphRequest::builder()
+                    .supergraph_request(supergraph_request("query"))
+                    .subgraph_request(subgraph_http_request(url, "query"))
+                    .operation_kind(OperationKind::Query)
+                    .subgraph_name(String::from("test"))
+                    .context(context)
+                    .build(),
+            )
+            .await
+            .unwrap();
+
+        assert!(
+            response.response.body().errors.is_empty(),
+            "expected no errors when response is under the limit"
+        );
+    }
+
     #[tokio::test(flavor = "multi_thread")]
     async fn test_subgraph_invalid_status_invalid_response_application_json() {
         let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
diff --git a/apollo-router/src/spec/operation_limits.rs b/apollo-router/src/spec/operation_limits.rs
index ee2ef37031..2468de7053 100644
--- a/apollo-router/src/spec/operation_limits.rs
+++ b/apollo-router/src/spec/operation_limits.rs
@@ -58,7 +58,7 @@ impl OperationLimits<bool> {
 /// Returns which limits are exceeded by the given query, if any
 pub(crate) fn check(
     query_metrics_in: &mut OperationLimits<u32>,
-    config_limits: &limits::Config,
+    config_limits: &limits::RouterLimitsConfig,
     query: &str,
     document: &ExecutableDocument,
     operation_name: Option<&str>,
@@ -81,7 +81,7 @@ pub(crate) fn check(
 
 pub(crate) fn check_measured(
     query_metrics: &OperationLimits<u32>,
-    config_limits: &limits::Config,
+    config_limits: &limits::RouterLimitsConfig,
     query: &str,
     operation_name: Option<&str>,
 ) -> Result<(), OperationLimits<bool>> {
diff --git a/apollo-router/src/spec/query.rs b/apollo-router/src/spec/query.rs
index ab19799d2a..19548675dd 100644
--- a/apollo-router/src/spec/query.rs
+++ b/apollo-router/src/spec/query.rs
@@ -264,8 +264,8 @@ impl Query {
         configuration: &Configuration,
     ) -> Result<ParsedDocument, SpecError> {
         let parser = &mut apollo_compiler::parser::Parser::new()
-            .recursion_limit(configuration.limits.parser_max_recursion)
-            .token_limit(configuration.limits.parser_max_tokens);
+            .recursion_limit(configuration.limits.router.parser_max_recursion)
+            .token_limit(configuration.limits.router.parser_max_tokens);
         let ast = match parser.parse_ast(query, "query.graphql") {
             Ok(ast) => ast,
             Err(errors) => {
diff --git a/apollo-router/src/state_machine.rs b/apollo-router/src/state_machine.rs
index af5c1ce5e8..3635f0614d 100644
--- a/apollo-router/src/state_machine.rs
+++ b/apollo-router/src/state_machine.rs
@@ -796,9 +796,11 @@ mod tests {
         let mut config = Configuration::builder().build().unwrap();
         config.validated_yaml = Some(json!({
             "limits": {
-                "max_height": 100,
-                "max_aliases": 100,
-                "max_depth": 20
+                "router": {
+                    "max_height": 100,
+                    "max_aliases": 100,
+                    "max_depth": 20
+                }
             }
         }));
         Arc::new(config)
diff --git a/apollo-router/src/uplink/license_enforcement.rs b/apollo-router/src/uplink/license_enforcement.rs
index 4a1b4318ac..b0e8f9ea51 100644
--- a/apollo-router/src/uplink/license_enforcement.rs
+++ b/apollo-router/src/uplink/license_enforcement.rs
@@ -408,19 +408,19 @@ impl LicenseEnforcementReport {
         if !allowed_features.contains(&AllowedFeature::RequestLimits) {
             configuration_restrictions.extend(vec![
                 ConfigurationRestriction::builder()
-                    .path("$.limits.max_depth")
+                    .path("$.limits.router.max_depth")
                     .name("Operation depth limiting")
                     .build(),
                 ConfigurationRestriction::builder()
-                    .path("$.limits.max_height")
+                    .path("$.limits.router.max_height")
                     .name("Operation height limiting")
                     .build(),
                 ConfigurationRestriction::builder()
-                    .path("$.limits.max_root_fields")
+                    .path("$.limits.router.max_root_fields")
                     .name("Operation root fields limiting")
                     .build(),
                 ConfigurationRestriction::builder()
-                    .path("$.limits.max_aliases")
+                    .path("$.limits.router.max_aliases")
                     .name("Operation aliases limiting")
                     .build(),
             ]);
diff --git a/apollo-router/src/uplink/snapshots/apollo_router__uplink__license_enforcement__test__restricted_features_via_config.snap b/apollo-router/src/uplink/snapshots/apollo_router__uplink__license_enforcement__test__restricted_features_via_config.snap
index 2060aa9438..b3e402d2ce 100644
--- a/apollo-router/src/uplink/snapshots/apollo_router__uplink__license_enforcement__test__restricted_features_via_config.snap
+++ b/apollo-router/src/uplink/snapshots/apollo_router__uplink__license_enforcement__test__restricted_features_via_config.snap
@@ -25,16 +25,16 @@ Configuration yaml:
   .subscription.enabled
 
 * Operation depth limiting
-  .limits.max_depth
+  .limits.router.max_depth
 
 * Operation height limiting
-  .limits.max_height
+  .limits.router.max_height
 
 * Operation root fields limiting
-  .limits.max_root_fields
+  .limits.router.max_root_fields
 
 * Operation aliases limiting
-  .limits.max_aliases
+  .limits.router.max_aliases
 
 * Advanced telemetry
   .telemetry..spans.router
diff --git a/apollo-router/src/uplink/snapshots/apollo_router__uplink__license_enforcement__test__restricted_features_via_config_allowed_features_empty.snap b/apollo-router/src/uplink/snapshots/apollo_router__uplink__license_enforcement__test__restricted_features_via_config_allowed_features_empty.snap
index 5e7b7fdafc..d4f25ca8a6 100644
--- a/apollo-router/src/uplink/snapshots/apollo_router__uplink__license_enforcement__test__restricted_features_via_config_allowed_features_empty.snap
+++ b/apollo-router/src/uplink/snapshots/apollo_router__uplink__license_enforcement__test__restricted_features_via_config_allowed_features_empty.snap
@@ -28,13 +28,13 @@ Configuration yaml:
   .plugins.['experimental.restricted'].enabled
 
 * Operation depth limiting
-  .limits.max_depth
+  .limits.router.max_depth
 
 * Operation height limiting
-  .limits.max_height
+  .limits.router.max_height
 
 * Operation root fields limiting
-  .limits.max_root_fields
+  .limits.router.max_root_fields
 
 * Operation aliases limiting
-  .limits.max_aliases
+  .limits.router.max_aliases
diff --git a/apollo-router/src/uplink/snapshots/apollo_router__uplink__license_enforcement__test__restricted_features_via_config_unlicensed.snap b/apollo-router/src/uplink/snapshots/apollo_router__uplink__license_enforcement__test__restricted_features_via_config_unlicensed.snap
index 5e7b7fdafc..d4f25ca8a6 100644
--- a/apollo-router/src/uplink/snapshots/apollo_router__uplink__license_enforcement__test__restricted_features_via_config_unlicensed.snap
+++ b/apollo-router/src/uplink/snapshots/apollo_router__uplink__license_enforcement__test__restricted_features_via_config_unlicensed.snap
@@ -28,13 +28,13 @@ Configuration yaml:
   .plugins.['experimental.restricted'].enabled
 
 * Operation depth limiting
-  .limits.max_depth
+  .limits.router.max_depth
 
 * Operation height limiting
-  .limits.max_height
+  .limits.router.max_height
 
 * Operation root fields limiting
-  .limits.max_root_fields
+  .limits.router.max_root_fields
 
 * Operation aliases limiting
-  .limits.max_aliases
+  .limits.router.max_aliases
diff --git a/apollo-router/src/uplink/snapshots/apollo_router__uplink__license_enforcement__test__restricted_features_via_config_with_allowed_features.snap b/apollo-router/src/uplink/snapshots/apollo_router__uplink__license_enforcement__test__restricted_features_via_config_with_allowed_features.snap
index 4e1a4ec065..e7164841b0 100644
--- a/apollo-router/src/uplink/snapshots/apollo_router__uplink__license_enforcement__test__restricted_features_via_config_with_allowed_features.snap
+++ b/apollo-router/src/uplink/snapshots/apollo_router__uplink__license_enforcement__test__restricted_features_via_config_with_allowed_features.snap
@@ -19,13 +19,13 @@ Configuration yaml:
   .plugins.['experimental.restricted'].enabled
 
 * Operation depth limiting
-  .limits.max_depth
+  .limits.router.max_depth
 
 * Operation height limiting
-  .limits.max_height
+  .limits.router.max_height
 
 * Operation root fields limiting
-  .limits.max_root_fields
+  .limits.router.max_root_fields
 
 * Operation aliases limiting
-  .limits.max_aliases
+  .limits.router.max_aliases
diff --git a/apollo-router/src/uplink/testdata/connectv0_4.router.yaml b/apollo-router/src/uplink/testdata/connectv0_4.router.yaml
index fa7f44ae90..3831773773 100644
--- a/apollo-router/src/uplink/testdata/connectv0_4.router.yaml
+++ b/apollo-router/src/uplink/testdata/connectv0_4.router.yaml
@@ -3,6 +3,7 @@ health_check:
 homepage:
   enabled: false
 limits:
-  parser_max_recursion: 1000
+  router:
+    parser_max_recursion: 1000
 connectors:
   preview_connect_v0_4: true
diff --git a/apollo-router/src/uplink/testdata/oss.router.yaml b/apollo-router/src/uplink/testdata/oss.router.yaml
index b2537e3cad..0b9f75e536 100644
--- a/apollo-router/src/uplink/testdata/oss.router.yaml
+++ b/apollo-router/src/uplink/testdata/oss.router.yaml
@@ -3,4 +3,5 @@ health_check:
 homepage:
   enabled: false
 limits:
-  parser_max_recursion: 1000
+  router:
+    parser_max_recursion: 1000
diff --git a/apollo-router/src/uplink/testdata/restricted.router.yaml b/apollo-router/src/uplink/testdata/restricted.router.yaml
index f39ebfe1a0..92d81951ca 100644
--- a/apollo-router/src/uplink/testdata/restricted.router.yaml
+++ b/apollo-router/src/uplink/testdata/restricted.router.yaml
@@ -22,10 +22,11 @@ supergraph:
         limit: 1000
 
 limits:
-  max_depth: 20
-  max_height: 100
-  max_aliases: 100
-  max_root_fields: 10
+  router:
+    max_depth: 20
+    max_height: 100
+    max_aliases: 100
+    max_root_fields: 10
 
 apq:
   router:
diff --git a/apollo-router/tests/integration/fixtures/coprocessor_body_limit.router.yaml b/apollo-router/tests/integration/fixtures/coprocessor_body_limit.router.yaml
index c59346cea4..4b82fc474a 100644
--- a/apollo-router/tests/integration/fixtures/coprocessor_body_limit.router.yaml
+++ b/apollo-router/tests/integration/fixtures/coprocessor_body_limit.router.yaml
@@ -5,4 +5,5 @@ coprocessor:
     request:
       body: true
 limits:
-  http_max_request_bytes: 100
\ No newline at end of file
+  router:
+    http_max_request_bytes: 100
\ No newline at end of file
diff --git a/apollo-router/tests/integration/fixtures/request_bytes_limit.router.yaml b/apollo-router/tests/integration/fixtures/request_bytes_limit.router.yaml
index 035f1d4c17..f21fdaa116 100644
--- a/apollo-router/tests/integration/fixtures/request_bytes_limit.router.yaml
+++ b/apollo-router/tests/integration/fixtures/request_bytes_limit.router.yaml
@@ -1,5 +1,6 @@
 limits:
-  http_max_request_bytes: 60
+  router:
+    http_max_request_bytes: 60
 coprocessor:
   url: http://localhost:4005
   router:
diff --git a/apollo-router/tests/integration/fixtures/request_bytes_limit_with_coprocessor.router.yaml b/apollo-router/tests/integration/fixtures/request_bytes_limit_with_coprocessor.router.yaml
index 035f1d4c17..f21fdaa116 100644
--- a/apollo-router/tests/integration/fixtures/request_bytes_limit_with_coprocessor.router.yaml
+++ b/apollo-router/tests/integration/fixtures/request_bytes_limit_with_coprocessor.router.yaml
@@ -1,5 +1,6 @@
 limits:
-  http_max_request_bytes: 60
+  router:
+    http_max_request_bytes: 60
 coprocessor:
   url: http://localhost:4005
   router:
diff --git a/apollo-router/tests/integration/telemetry/fixtures/prometheus.router.yaml b/apollo-router/tests/integration/telemetry/fixtures/prometheus.router.yaml
index 486e322c2a..05b4c6a5d3 100644
--- a/apollo-router/tests/integration/telemetry/fixtures/prometheus.router.yaml
+++ b/apollo-router/tests/integration/telemetry/fixtures/prometheus.router.yaml
@@ -1,5 +1,6 @@
 limits:
-  http_max_request_bytes: 200
+  router:
+    http_max_request_bytes: 200
 telemetry:
   instrumentation:
     instruments:
diff --git a/apollo-router/tests/integration/telemetry/fixtures/prometheus_no_reload.router.yaml b/apollo-router/tests/integration/telemetry/fixtures/prometheus_no_reload.router.yaml
index d8a48722fc..251c924241 100644
--- a/apollo-router/tests/integration/telemetry/fixtures/prometheus_no_reload.router.yaml
+++ b/apollo-router/tests/integration/telemetry/fixtures/prometheus_no_reload.router.yaml
@@ -1,5 +1,6 @@
 limits:
-  http_max_request_bytes: 200
+  router:
+    http_max_request_bytes: 200
 telemetry:
   instrumentation:
     instruments:
diff --git a/apollo-router/tests/integration/telemetry/fixtures/prometheus_reload.router.yaml b/apollo-router/tests/integration/telemetry/fixtures/prometheus_reload.router.yaml
index eb6c71cf46..3ed175fdab 100644
--- a/apollo-router/tests/integration/telemetry/fixtures/prometheus_reload.router.yaml
+++ b/apollo-router/tests/integration/telemetry/fixtures/prometheus_reload.router.yaml
@@ -1,5 +1,6 @@
 limits:
-  http_max_request_bytes: 200
+  router:
+    http_max_request_bytes: 200
 telemetry:
   instrumentation:
     instruments:
diff --git a/docs/source/routing/configuration/yaml.mdx b/docs/source/routing/configuration/yaml.mdx
index ff3fb2b6b4..976e219ae2 100644
--- a/docs/source/routing/configuration/yaml.mdx
+++ b/docs/source/routing/configuration/yaml.mdx
@@ -500,7 +500,8 @@ In case the router rejects legitimate queries, you can disable the limit by sett
 supergraph:
   introspection: true
 limits:
-  introspection_max_depth: false
+  router:
+    introspection_max_depth: false
 ```
 
 #### Redacting query validation errors
diff --git a/docs/source/routing/observability/router-telemetry-otel/enabling-telemetry/standard-instruments.mdx b/docs/source/routing/observability/router-telemetry-otel/enabling-telemetry/standard-instruments.mdx
index 1a41e0f3ed..002464b177 100644
--- a/docs/source/routing/observability/router-telemetry-otel/enabling-telemetry/standard-instruments.mdx
+++ b/docs/source/routing/observability/router-telemetry-otel/enabling-telemetry/standard-instruments.mdx
@@ -279,6 +279,14 @@ telemetry:
 - `apollo.router.operations.batching` - A counter of the number of query batches received by the router.
 - `apollo.router.operations.batching.size` - A histogram tracking the number of queries contained within a query batch.
 
+## Limits
+
+- `apollo.router.limits.subgraph_response_size.exceeded`: A counter of subgraph responses aborted because their body exceeded the configured `http_max_response_size` limit under `limits.subgraph`.
+  - `subgraph.name`: The name of the subgraph whose response was aborted.
+
+- `apollo.router.limits.connector_response_size.exceeded` - A counter of connector responses aborted because their body exceeded the configured `http_max_response_size` limit under `limits.connector`.
+  - `connector.source`: The source key (`subgraph_name.source_name`) whose response was aborted.
+
 ## GraphOS Studio
 
 - `apollo.router.telemetry.studio.reports` - The number of reports submitted to GraphOS Studio by the router.
diff --git a/docs/source/routing/security/request-limits.mdx b/docs/source/routing/security/request-limits.mdx
index dd618a86c7..7e531af6a9 100644
--- a/docs/source/routing/security/request-limits.mdx
+++ b/docs/source/routing/security/request-limits.mdx
@@ -1,6 +1,6 @@
 ---
 title: Request Limits
-subtitle: Protect your router from requests exceeding network, parser, and operation-based limits 
+subtitle: Protect your router from requests exceeding network, parser, operation-based, and subgraph response limits
 redirectFrom:
   - /router/configuration/operation-limits/
 ---
@@ -10,23 +10,41 @@ For enhanced security, the GraphOS Router can reject requests that violate any o
 - Operation-based semantic limits
 - Network-based limits
 - Parser-based lexical limits
+- Subgraph response size limits
 
 ```yaml title="router.yaml"
 limits:
-  # Network-based limits
-  http_max_request_bytes: 2000000 # Default value: 2 MB
-  http1_max_request_headers: 200 # Default value: 100
-  http1_max_request_buf_size: 800kb # Default value: 400kib
-
-  # Parser-based limits
-  parser_max_tokens: 15000 # Default value
-  parser_max_recursion: 500 # Default value
-
-  # Operation-based limits (License only)
-  max_depth: 100
-  max_height: 200
-  max_aliases: 30
-  max_root_fields: 20
+  router:
+    # Network-based limits
+    http_max_request_bytes: 2000000 # Default value: 2 MB
+    http1_max_request_headers: 200 # Default value: 100
+    http1_max_request_buf_size: 800kb # Default value: 400kib
+
+    # Parser-based limits
+    parser_max_tokens: 15000 # Default value
+    parser_max_recursion: 500 # Default value
+
+    # Operation-based limits (License only)
+    max_depth: 100
+    max_height: 200
+    max_aliases: 30
+    max_root_fields: 20
+
+  # Subgraph response size limits
+  subgraph:
+    all:
+      http_max_response_size: 10mb # 10 MB for all subgraphs
+    subgraphs:
+      products:
+        http_max_response_size: 20mb # per-subgraph override (20 MB)
+
+  # Connector response size limits
+  connector:
+    all:
+      http_max_response_size: 5mb # 5 MB for all connector sources
+    sources:
+      products.rest:
+        http_max_response_size: 10mb # per-source override (10 MB)
 ```
 
 ## Operation-based limits
@@ -48,13 +66,14 @@ You define operation limits in your router's [YAML config file](/graphos/referen
 
 ```yaml title="router.yaml"
 limits:
-  max_depth: 100
-  max_height: 200
-  max_aliases: 30
-  max_root_fields: 20
-
-  # Uncomment to enable warn_only mode
-  # warn_only: true
+  router:
+    max_depth: 100
+    max_height: 200
+    max_aliases: 30
+    max_root_fields: 20
+
+    # Uncomment to enable warn_only mode
+    # warn_only: true
 ```
 
 Each limit takes an integer value. You can define any combination of [supported limits](#supported-limits).
@@ -153,7 +172,8 @@ You can enable or disable `warn_only` mode in your router's [YAML config file](/
 
 ```yaml title="router.yaml"
 limits:
-  warn_only: true # warn_only mode always enabled
+  router:
+    warn_only: true # warn_only mode always enabled
 ```
 
 ### Response format for exceeded limits
@@ -331,3 +351,95 @@ Note that the router calculates the recursion depth for each operation and fragm
 In versions of the Apollo Router prior to 1.17, this limit was defined via the config option `experimental_parser_recursion_limit`.
 
 </Note>
+
+## Subgraph response size limits
+
+### `http_max_response_size`
+
+Limits the number of bytes the router reads from a subgraph's HTTP response body, to protect against unbounded memory consumption when a subgraph returns an unexpectedly large payload.
+
+This limit is enforced as the response body streams in — the router stops reading and returns an error as soon as the limit is exceeded, without buffering the entire body first.
+
+When the limit is exceeded, the router returns a GraphQL error to the client with extension code `SUBREQUEST_HTTP_ERROR`.
+
+There is no default limit; subgraph responses are unrestricted unless you configure this option.
+
+You can set a global default that applies to all subgraphs, and optionally override it per subgraph:
+
+```yaml title="router.yaml"
+limits:
+  subgraph:
+    all:
+      http_max_response_size: 10485760 # 10 MB for all subgraphs
+    subgraphs:
+      products:
+        http_max_response_size: 20971520 # 20 MB override for 'products'
+```
+
+The per-subgraph entry under `subgraphs` takes precedence over `all`. If only `all` is set, it applies to every subgraph. If only a named entry is set, only that subgraph is limited.
+
+### Choosing a limit value
+
+To determine the appropriate limit, measure the actual response sizes your subgraphs return. Use the built-in `http.client.response.body.size` histogram to collect a distribution of subgraph response sizes:
+
+```yaml title="router.yaml"
+telemetry:
+  instrumentation:
+    instruments:
+      subgraph:
+        http.client.response.body.size: true
+```
+
+This histogram is disabled by default. Once enabled, use it to observe the 95th or 99th percentile of response sizes across your subgraphs, then set `http_max_response_size` above that value with enough headroom for legitimate variation (for example, large page sizes in paginated queries).
+
+### Monitoring exceeded limits
+
+After a subgraph response is aborted because it exceeds the configured limit:
+
+- The router increments the `apollo.router.limits.subgraph_response_size.exceeded` counter with a `subgraph.name` attribute identifying the affected subgraph.
+- The subgraph request span gets an `apollo.subgraph.response.aborted` attribute set to `response_size_limit`.
+
+Monitor the counter to detect misconfigured limits or misbehaving subgraphs. Use the span attribute to filter limit-triggered aborts in your tracing backend.
+
+## Connector response size limits
+
+### `http_max_response_size`
+
+Limits the number of bytes the router reads from a connector's HTTP response body, to protect against unbounded memory consumption when a connector source returns an unexpectedly large payload.
+
+This limit is enforced as the response body streams in — the router stops reading and returns a GraphQL error as soon as the limit is exceeded.
+
+There is no default limit; connector responses are unrestricted unless you configure this option.
+
+Identify sources by `subgraph_name.source_name` (matching the key format used in connector configuration). You can set a global default that applies to all sources, and optionally override it per source:
+
+```yaml title="router.yaml"
+limits:
+  connector:
+    all:
+      http_max_response_size: 5242880 # 5 MB for all connector sources
+    sources:
+      products.rest:
+        http_max_response_size: 10485760 # 10 MB override for 'products.rest'
+```
+
+The per-source entry under `sources` takes precedence over `all`.
+
+### Choosing a limit value
+
+The built-in `http.client.response.body.size` histogram measures actual connector response sizes:
+
+```yaml title="router.yaml"
+telemetry:
+  instrumentation:
+    instruments:
+      connector:
+        http.client.response.body.size: true
+```
+
+### Monitoring exceeded limits
+
+When a connector response is aborted because it exceeds the configured limit:
+
+- The router increments the `apollo.router.limits.connector_response_size.exceeded` counter with a `connector.source` attribute identifying the affected source.
+- The connector request span gets an `apollo.connector.response.aborted` attribute set to `response_size_limit`.