diff --git a/headroom/proxy/helpers.py b/headroom/proxy/helpers.py
index cd62c22f9..ef6f1e105 100644
--- a/headroom/proxy/helpers.py
+++ b/headroom/proxy/helpers.py
@@ -24,6 +24,7 @@ from typing import TYPE_CHECKING, Any, Literal, cast
 
 from headroom import paths as _paths
 from headroom._subprocess import run
+from headroom.proxy import request_limit_policy
 from headroom.proxy.body_forwarding import (
     BodyMutationTracker as BodyMutationTracker,  # noqa: F401 - compatibility export
 )
@@ -518,8 +519,8 @@ MAX_SSE_BUFFER_SIZE = 10 * 1024 * 1024
 # HEADROOM_SSE_BUFFER_MAX_BYTES. Guards against pathological huge events
 # (a single event > 1 MB by default is treated as an upstream protocol bug
 # and surfaces loudly rather than silently growing the buffer).
-_SSE_EVENT_MAX_BYTES_ENV = "HEADROOM_SSE_BUFFER_MAX_BYTES"
-_SSE_EVENT_MAX_BYTES_DEFAULT = 1 * 1024 * 1024  # 1 MB
+_SSE_EVENT_MAX_BYTES_ENV = request_limit_policy.SSE_EVENT_MAX_BYTES_ENV
+_SSE_EVENT_MAX_BYTES_DEFAULT = request_limit_policy.SSE_EVENT_MAX_BYTES_DEFAULT
 
 
 def get_sse_event_max_bytes() -> int:
@@ -528,37 +529,23 @@ def get_sse_event_max_bytes() -> int:
     Read at request time so operators can flip the env var without a
     restart. Negative values are rejected loudly (no silent fallback).
     """
-    raw = os.environ.get(_SSE_EVENT_MAX_BYTES_ENV)
-    if raw is None or raw == "":
-        return _SSE_EVENT_MAX_BYTES_DEFAULT
-    try:
-        value = int(raw)
-    except ValueError as exc:
-        raise ValueError(f"{_SSE_EVENT_MAX_BYTES_ENV} must be an integer, got {raw!r}") from exc
-    if value <= 0:
-        raise ValueError(f"{_SSE_EVENT_MAX_BYTES_ENV} must be positive, got {value}")
-    return value
+    return request_limit_policy.resolve_sse_event_max_bytes(
+        os.environ.get(_SSE_EVENT_MAX_BYTES_ENV)
+    )
 
 
 # Body-too-large status code (PR-A8 / P5-59). Default 413 (RFC 7231 §6.5.11).
 # Configurable via HEADROOM_PROXY_BODY_TOO_LARGE_STATUS for operators who need
 # to override (no expected production use; documentation knob).
-_BODY_TOO_LARGE_STATUS_ENV = "HEADROOM_PROXY_BODY_TOO_LARGE_STATUS"
-_BODY_TOO_LARGE_STATUS_DEFAULT = 413
+_BODY_TOO_LARGE_STATUS_ENV = request_limit_policy.BODY_TOO_LARGE_STATUS_ENV
+_BODY_TOO_LARGE_STATUS_DEFAULT = request_limit_policy.BODY_TOO_LARGE_STATUS_DEFAULT
 
 
 def get_body_too_large_status() -> int:
     """Return the HTTP status code for body-too-large rejections."""
-    raw = os.environ.get(_BODY_TOO_LARGE_STATUS_ENV)
-    if raw is None or raw == "":
-        return _BODY_TOO_LARGE_STATUS_DEFAULT
-    try:
-        value = int(raw)
-    except ValueError as exc:
-        raise ValueError(f"{_BODY_TOO_LARGE_STATUS_ENV} must be an integer, got {raw!r}") from exc
-    if not 400 <= value < 600:
-        raise ValueError(f"{_BODY_TOO_LARGE_STATUS_ENV} must be a 4xx/5xx status, got {value}")
-    return value
+    return request_limit_policy.resolve_body_too_large_status(
+        os.environ.get(_BODY_TOO_LARGE_STATUS_ENV)
+    )
 
 
 # SSE byte-buffer helper supports LF and CRLF event separators. Per the SSE
diff --git a/headroom/proxy/request_limit_policy.py b/headroom/proxy/request_limit_policy.py
new file mode 100644
index 000000000..0ceb8821a
--- /dev/null
+++ b/headroom/proxy/request_limit_policy.py
@@ -0,0 +1,35 @@
+"""Validation policy for proxy request and stream limits."""
+
+from __future__ import annotations
+
+SSE_EVENT_MAX_BYTES_ENV = "HEADROOM_SSE_BUFFER_MAX_BYTES"
+SSE_EVENT_MAX_BYTES_DEFAULT = 1 * 1024 * 1024
+
+BODY_TOO_LARGE_STATUS_ENV = "HEADROOM_PROXY_BODY_TOO_LARGE_STATUS"
+BODY_TOO_LARGE_STATUS_DEFAULT = 413
+
+
+def resolve_sse_event_max_bytes(raw: str | None) -> int:
+    """Resolve the per-event SSE size cap from an optional env string."""
+    if raw is None or raw == "":
+        return SSE_EVENT_MAX_BYTES_DEFAULT
+    try:
+        value = int(raw)
+    except ValueError as exc:
+        raise ValueError(f"{SSE_EVENT_MAX_BYTES_ENV} must be an integer, got {raw!r}") from exc
+    if value <= 0:
+        raise ValueError(f"{SSE_EVENT_MAX_BYTES_ENV} must be positive, got {value}")
+    return value
+
+
+def resolve_body_too_large_status(raw: str | None) -> int:
+    """Resolve the HTTP status code for body-too-large rejections."""
+    if raw is None or raw == "":
+        return BODY_TOO_LARGE_STATUS_DEFAULT
+    try:
+        value = int(raw)
+    except ValueError as exc:
+        raise ValueError(f"{BODY_TOO_LARGE_STATUS_ENV} must be an integer, got {raw!r}") from exc
+    if not 400 <= value < 600:
+        raise ValueError(f"{BODY_TOO_LARGE_STATUS_ENV} must be a 4xx/5xx status, got {value}")
+    return value
diff --git a/tests/test_request_limit_policy.py b/tests/test_request_limit_policy.py
new file mode 100644
index 000000000..63fc96a76
--- /dev/null
+++ b/tests/test_request_limit_policy.py
@@ -0,0 +1,41 @@
+from __future__ import annotations
+
+import pytest
+
+from headroom.proxy.request_limit_policy import (
+    BODY_TOO_LARGE_STATUS_DEFAULT,
+    SSE_EVENT_MAX_BYTES_DEFAULT,
+    resolve_body_too_large_status,
+    resolve_sse_event_max_bytes,
+)
+
+
+def test_resolve_sse_event_max_bytes_uses_default_for_missing_value() -> None:
+    assert resolve_sse_event_max_bytes(None) == SSE_EVENT_MAX_BYTES_DEFAULT
+    assert resolve_sse_event_max_bytes("") == SSE_EVENT_MAX_BYTES_DEFAULT
+
+
+def test_resolve_sse_event_max_bytes_accepts_positive_integer() -> None:
+    assert resolve_sse_event_max_bytes("2048") == 2048
+
+
+@pytest.mark.parametrize("raw", ["0", "-1", "not-int"])
+def test_resolve_sse_event_max_bytes_rejects_invalid_values(raw: str) -> None:
+    with pytest.raises(ValueError):
+        resolve_sse_event_max_bytes(raw)
+
+
+def test_resolve_body_too_large_status_uses_default_for_missing_value() -> None:
+    assert resolve_body_too_large_status(None) == BODY_TOO_LARGE_STATUS_DEFAULT
+    assert resolve_body_too_large_status("") == BODY_TOO_LARGE_STATUS_DEFAULT
+
+
+def test_resolve_body_too_large_status_accepts_4xx_or_5xx_status() -> None:
+    assert resolve_body_too_large_status("413") == 413
+    assert resolve_body_too_large_status("529") == 529
+
+
+@pytest.mark.parametrize("raw", ["399", "600", "not-int"])
+def test_resolve_body_too_large_status_rejects_invalid_values(raw: str) -> None:
+    with pytest.raises(ValueError):
+        resolve_body_too_large_status(raw)