diff --git a/.env.example b/.env.example
index b67c8c13cc..492e1bc1ec 100644
--- a/.env.example
+++ b/.env.example
@@ -64,7 +64,7 @@ CSRF_ROTATE_ON_LOGIN=true
 CSRF_TRUSTED_ORIGINS=["http://localhost:4444","http://localhost:8080"]
 
 # Paths exempt from CSRF protection (JSON array)
-CSRF_EXEMPT_PATHS=["/health","/auth/login","/auth/logout","/auth/refresh","/auth/email/login","/auth/email/register","/auth/email/forgot-password","/auth/email/reset-password","/admin","/oauth/fetch-tools","/docs","/redoc","/openapi.json","/metrics","/mcp/","/sse","/message"]
+CSRF_EXEMPT_PATHS=["/health","/auth/login","/auth/logout","/auth/refresh","/auth/email/login","/auth/email/register","/auth/email/forgot-password","/auth/email/reset-password","/admin","/oauth/fetch-tools","/docs","/redoc","/openapi.json","/metrics","/mcp/","/sse","/message","/_internal/mcp/"]
 
 # Bootstrap admin credentials (email auth)
 # PRODUCTION: Change these values
diff --git a/mcpgateway/auth_context.py b/mcpgateway/auth_context.py
index 07d29872da..4fffbca829 100644
--- a/mcpgateway/auth_context.py
+++ b/mcpgateway/auth_context.py
@@ -310,6 +310,94 @@ def has_valid_internal_mcp_runtime_auth_header(request: Request) -> bool:
     return hmac.compare_digest(provided, _expected_internal_mcp_runtime_auth_header())
 
 
+# Internal-dispatch trust gate, defined once and shared by all callers.
+INTERNAL_MCP_PATH_PREFIX = "/_internal/mcp"
+INTERNAL_A2A_PATH_PREFIX = "/_internal/a2a"
+INTERNAL_RUNTIME_MARKER_HEADER = "x-contextforge-mcp-runtime"
+INTERNAL_AUTH_CONTEXT_HEADER = "x-contextforge-auth-context"
+TRUSTED_INTERNAL_RUNTIME_MARKERS = frozenset({"rust", "affinity"})
+
+
+def _internal_path_requires_auth_context(path: str) -> bool:
+    """Whether an internal route requires an auth context.
+
+    Only ``*/authenticate`` is exempt, since it creates the context; every
+    other internal route must carry one.
+
+    Args:
+        path: The request path to classify.
+
+    Returns:
+        ``False`` for ``*/authenticate``, otherwise ``True``.
+    """
+    return not path.rstrip("/").endswith("/authenticate")
+
+
+def is_trusted_internal_runtime_request(
+    request: Request,
+    *,
+    allowed_prefixes: tuple[str, ...],
+    require_auth_context: bool,
+    path: Optional[str] = None,
+) -> bool:
+    """Return whether a request is a trusted in-process internal-runtime hop.
+
+    The trust boundary is the HMAC header and, when required, the encoded
+    ``x-contextforge-auth-context``. The loopback check is defense in depth,
+    not an independent gate: ProxyHeaders(trusted_hosts="*") lets a direct
+    external caller influence ``request.client.host`` (the replay is hardened
+    separately by stripping forwarded / client-IP headers).
+
+    Args:
+        request: Incoming request to inspect.
+        allowed_prefixes: Internal path prefixes this caller trusts.
+        require_auth_context: Require a non-empty ``x-contextforge-auth-context``.
+        path: Explicit path override for callers that strip a ``root_path``;
+            defaults to ``request.url.path``.
+
+    Returns:
+        ``True`` only when prefix, runtime marker, HMAC, optional auth context,
+        and loopback all hold; otherwise ``False``.
+    """
+    p = path if path is not None else (getattr(getattr(request, "url", None), "path", "") or "")
+    if not any(p == prefix or p.startswith(f"{prefix}/") for prefix in allowed_prefixes):
+        return False
+    if request.headers.get(INTERNAL_RUNTIME_MARKER_HEADER) not in TRUSTED_INTERNAL_RUNTIME_MARKERS:
+        return False
+    if not has_valid_internal_mcp_runtime_auth_header(request):
+        return False
+    if require_auth_context and not request.headers.get(INTERNAL_AUTH_CONTEXT_HEADER):
+        return False
+    client_host = getattr(getattr(request, "client", None), "host", None)
+    return client_host in ("127.0.0.1", "::1")
+
+
+def is_trusted_internal_mcp_request(request: Request, *, path: Optional[str] = None) -> bool:
+    """MCP + A2A internal trust gate with a path-aware auth-context requirement.
+
+    ``*/authenticate`` routes are exempt from the auth-context requirement;
+    every other internal route requires it. ``/_internal/a2a/*`` is trusted
+    only when the A2A feature is enabled.
+
+    Args:
+        request: Incoming request to inspect.
+        path: Explicit path override (e.g. a ``root_path``-stripped path);
+            defaults to ``request.url.path``.
+
+    Returns:
+        ``True`` when the request is a trusted internal MCP/A2A hop.
+    """
+    p = path if path is not None else (getattr(getattr(request, "url", None), "path", "") or "")
+    if p.startswith(f"{INTERNAL_A2A_PATH_PREFIX}/") and not settings.mcpgateway_a2a_enabled:
+        return False
+    return is_trusted_internal_runtime_request(
+        request,
+        allowed_prefixes=(INTERNAL_MCP_PATH_PREFIX, INTERNAL_A2A_PATH_PREFIX),
+        require_auth_context=_internal_path_requires_auth_context(p),
+        path=p,
+    )
+
+
 def get_token_teams_from_request(request: Request) -> Optional[List[str]]:
     """Extract and normalize teams from verified JWT token.
 
diff --git a/mcpgateway/main.py b/mcpgateway/main.py
index 2c4204bf7c..7a229575a4 100644
--- a/mcpgateway/main.py
+++ b/mcpgateway/main.py
@@ -81,8 +81,8 @@
     get_scoped_resource_access_context,
     get_token_teams_from_request,
     get_user_email,
-    has_valid_internal_mcp_runtime_auth_header,
     INTERNAL_MCP_SESSION_VALIDATED_HEADER,
+    is_trusted_internal_mcp_request,
 )
 from mcpgateway.cache import ResourceCache, SessionRegistry
 from mcpgateway.common.models import InitializeResult
@@ -296,17 +296,7 @@ def _is_trusted_internal_mcp_runtime_request(request: Request) -> bool:
         ``True`` when the request carries a trusted internal-runtime marker
         from loopback, otherwise ``False``.
     """
-    runtime_marker = request.headers.get("x-contextforge-mcp-runtime")
-    client_host = getattr(getattr(request, "client", None), "host", None)
-    if runtime_marker not in ("rust", "affinity") or not has_valid_internal_mcp_runtime_auth_header(request) or client_host not in ("127.0.0.1", "::1"):
-        return False
-    # Defense-in-depth: /_internal/a2a/* endpoints must refuse requests when
-    # A2A support is disabled, even from an otherwise-trusted local sidecar.
-    # A legitimate sidecar should not be running when the feature is off.
-    path = getattr(getattr(request, "url", None), "path", "") or ""
-    if path.startswith("/_internal/a2a/") and not settings.mcpgateway_a2a_enabled:
-        return False
-    return True
+    return is_trusted_internal_mcp_request(request)
 
 
 def _is_jwt_token(token: str) -> bool:
diff --git a/mcpgateway/middleware/rate_limit_middleware.py b/mcpgateway/middleware/rate_limit_middleware.py
index 53c3fee936..ee1a336eca 100644
--- a/mcpgateway/middleware/rate_limit_middleware.py
+++ b/mcpgateway/middleware/rate_limit_middleware.py
@@ -38,6 +38,7 @@
 
 # First-Party
 from mcpgateway import auth
+from mcpgateway.auth_context import is_trusted_internal_mcp_request
 from mcpgateway.config import settings
 from mcpgateway.services.security_logger import SecurityEventType, SecurityLogger, SecuritySeverity
 
@@ -199,6 +200,10 @@ async def dispatch(self, request: Request, call_next):
         if not self.enabled:
             return await call_next(request)
 
+        # Skip rate limiting for the trusted-internal dispatch; the edge request was already counted.
+        if is_trusted_internal_mcp_request(request):
+            return await call_next(request)
+
         tier = self.get_endpoint_tier(request.url.path)
         dimensions = self._get_client_dimensions(request)
 
diff --git a/mcpgateway/middleware/token_scoping.py b/mcpgateway/middleware/token_scoping.py
index b53d53ea2c..7b638f1fa4 100644
--- a/mcpgateway/middleware/token_scoping.py
+++ b/mcpgateway/middleware/token_scoping.py
@@ -13,8 +13,6 @@
 # Standard
 from datetime import datetime, timedelta, timezone
 from functools import lru_cache
-import hashlib
-import hmac
 import ipaddress
 import re
 from typing import List, Optional, Pattern, Tuple
@@ -64,11 +62,6 @@
     (re.compile(r"/gateways/?([a-f0-9\-]+)"), "gateway"),
 ]
 _AUTH_COOKIE_NAMES = ("jwt_token", "access_token")
-_INTERNAL_MCP_PATH_PREFIX = "/_internal/mcp"
-_INTERNAL_MCP_RUNTIME_HEADER = "x-contextforge-mcp-runtime"
-_INTERNAL_MCP_AUTH_CONTEXT_HEADER = "x-contextforge-auth-context"
-_INTERNAL_MCP_RUNTIME_AUTH_HEADER = "x-contextforge-mcp-runtime-auth"
-_INTERNAL_MCP_RUNTIME_AUTH_CONTEXT = "contextforge-internal-mcp-runtime-v1"
 
 # Permission map with precompiled patterns
 # Maps (HTTP method, path pattern) to required permission
@@ -1350,70 +1343,24 @@ async def __call__(self, request: Request, call_next):
             )
 
     def _is_trusted_internal_mcp_runtime_request(self, request: Request, normalized_path: str) -> bool:
-        """Return whether the request is a trusted loopback Rust MCP sidecar hop.
+        """Return whether the request is a trusted internal MCP/A2A runtime hop.
+
+        Delegates to the shared gate in ``auth_context`` so the trust decision is
+        defined in one place. Unlike the previous local check, this trusts both
+        the ``rust`` and ``affinity`` runtime markers — closing the gap where the
+        in-process session-affinity dispatch was still token-scoped.
 
         Args:
             request: Incoming HTTP request.
             normalized_path: Canonicalized request path used for route matching.
 
         Returns:
-            ``True`` when the request originated from the local Rust MCP runtime and
-            includes the expected trusted headers.
-        """
-        if normalized_path != _INTERNAL_MCP_PATH_PREFIX and not normalized_path.startswith(f"{_INTERNAL_MCP_PATH_PREFIX}/"):
-            return False
-
-        if request.headers.get(_INTERNAL_MCP_RUNTIME_HEADER) != "rust":
-            return False
-
-        provided_auth = request.headers.get(_INTERNAL_MCP_RUNTIME_AUTH_HEADER)
-        if not provided_auth:
-            return False
-
-        expected_auth = self._expected_internal_mcp_runtime_auth_header()
-        if not hmac.compare_digest(provided_auth, expected_auth):
-            return False
-
-        if not request.headers.get(_INTERNAL_MCP_AUTH_CONTEXT_HEADER):
-            return False
-
-        client_host = getattr(getattr(request, "client", None), "host", None)
-        return client_host in ("127.0.0.1", "::1")
-
-    @staticmethod
-    def _auth_encryption_secret_value() -> str:
-        """Return the configured auth-encryption secret as a plain string.
-
-        Returns:
-            The auth-encryption secret, normalized to a regular string.
-        """
-        secret = settings.auth_encryption_secret
-        if hasattr(secret, "get_secret_value"):
-            return secret.get_secret_value()
-        return str(secret)
-
-    @staticmethod
-    @lru_cache(maxsize=8)
-    def _expected_internal_mcp_runtime_auth_header_for_secret(secret: str) -> str:
-        """Return the expected shared internal-auth header for a specific secret.
-
-        Args:
-            secret: Auth-encryption secret to derive the trust header from.
-
-        Returns:
-            Hex-encoded SHA-256 digest derived from the provided auth secret.
+            ``True`` when the request is a trusted internal MCP/A2A hop.
         """
-        material = f"{secret}:{_INTERNAL_MCP_RUNTIME_AUTH_CONTEXT}".encode("utf-8")
-        return hashlib.sha256(material).hexdigest()
-
-    @staticmethod
-    def _expected_internal_mcp_runtime_auth_header() -> str:
-        """Return the expected shared internal-auth header for Rust MCP hops.
+        # First-Party
+        from mcpgateway.auth_context import is_trusted_internal_mcp_request  # pylint: disable=import-outside-toplevel
 
-        Returns:
-            Shared secret-derived digest expected on trusted internal Rust MCP calls.
-        """
-        return TokenScopingMiddleware._expected_internal_mcp_runtime_auth_header_for_secret(TokenScopingMiddleware._auth_encryption_secret_value())
+        return is_trusted_internal_mcp_request(request, path=normalized_path)
 
 
 # Create middleware instance
diff --git a/mcpgateway/middleware/token_usage_middleware.py b/mcpgateway/middleware/token_usage_middleware.py
index 214f8a9086..605ca2870f 100644
--- a/mcpgateway/middleware/token_usage_middleware.py
+++ b/mcpgateway/middleware/token_usage_middleware.py
@@ -30,6 +30,7 @@
 from starlette.types import ASGIApp, Receive, Scope, Send
 
 # First-Party
+from mcpgateway.auth_context import is_trusted_internal_mcp_request
 from mcpgateway.db import fresh_db_session
 from mcpgateway.middleware.path_filter import should_skip_auth_context
 from mcpgateway.services.token_catalog_service import TokenCatalogService
@@ -87,6 +88,11 @@ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
             await self.app(scope, receive, send)
             return
 
+        # Skip usage logging for the trusted-internal dispatch; the edge request is already recorded.
+        if is_trusted_internal_mcp_request(Request(scope), path=path):
+            await self.app(scope, receive, send)
+            return
+
         # Record start time
         start_time = time.time()
 
diff --git a/mcpgateway/utils/passthrough_headers.py b/mcpgateway/utils/passthrough_headers.py
index 82a8c5cf03..2434db886b 100644
--- a/mcpgateway/utils/passthrough_headers.py
+++ b/mcpgateway/utils/passthrough_headers.py
@@ -573,6 +573,17 @@ async def set_global_passthrough_headers(db: Session) -> None:
         "x-contextforge-mcp-runtime",
         "x-contextforge-mcp-runtime-auth",
         "x-contextforge-session-validated",
+        # Forwarded / client-IP headers: strip so a caller cannot spoof the loopback
+        # client address that ProxyHeaders(trusted_hosts="*") would otherwise trust.
+        "forwarded",
+        "x-forwarded-for",
+        "x-forwarded-host",
+        "x-forwarded-proto",
+        "x-forwarded-port",
+        "x-forwarded-prefix",
+        "x-real-ip",
+        "cf-connecting-ip",
+        "true-client-ip",
     }
 )
 
diff --git a/tests/unit/mcpgateway/middleware/test_rate_limit_middleware.py b/tests/unit/mcpgateway/middleware/test_rate_limit_middleware.py
index ee404fe9a1..cd2413462c 100644
--- a/tests/unit/mcpgateway/middleware/test_rate_limit_middleware.py
+++ b/tests/unit/mcpgateway/middleware/test_rate_limit_middleware.py
@@ -16,6 +16,30 @@
 import pytest
 
 
+def _trusted_internal_request(path, *, marker="rust", hmac_value=None, ctx="ctx", client="127.0.0.1"):
+    """Build a real loopback internal request for trust-gate exemption tests."""
+    from starlette.requests import Request
+
+    from mcpgateway.auth_context import _expected_internal_mcp_runtime_auth_header
+
+    headers = [
+        (b"x-contextforge-mcp-runtime", marker.encode()),
+        (b"x-contextforge-mcp-runtime-auth", (hmac_value or _expected_internal_mcp_runtime_auth_header()).encode()),
+    ]
+    if ctx is not None:
+        headers.append((b"x-contextforge-auth-context", ctx.encode()))
+    scope = {
+        "type": "http",
+        "method": "POST",
+        "path": path,
+        "raw_path": path.encode(),
+        "query_string": b"",
+        "headers": headers,
+        "client": (client, 12345),
+    }
+    return Request(scope)
+
+
 class TestRateLimitMiddleware:
     """Tests for RateLimitMiddleware."""
 
@@ -58,6 +82,29 @@ def mock_request(self):
         request.scope = {"client": ("192.168.1.100", 12345)}
         return request
 
+    @pytest.mark.asyncio
+    async def test_trusted_internal_dispatch_skips_rate_limiting(self, middleware):
+        """A trusted-internal hop is not rate-limited; the edge request already was."""
+        request = _trusted_internal_request("/_internal/mcp/rpc")
+        call_next = AsyncMock(return_value="passthrough")
+        # If the exemption fires, tier computation is never reached.
+        middleware.get_endpoint_tier = MagicMock(side_effect=AssertionError("trusted internal must not be rate-limited"))
+
+        result = await middleware.dispatch(request, call_next)
+
+        assert result == "passthrough"
+        call_next.assert_awaited_once_with(request)
+
+    @pytest.mark.asyncio
+    async def test_forged_internal_request_is_still_rate_limited(self, middleware):
+        """A forged HMAC fails the trust gate, so the rate-limit path runs."""
+        request = _trusted_internal_request("/_internal/mcp/rpc", hmac_value="forged")
+        call_next = AsyncMock(return_value="passthrough")
+        middleware.get_endpoint_tier = MagicMock(side_effect=RuntimeError("reached rate-limit path"))
+
+        with pytest.raises(RuntimeError, match="reached rate-limit path"):
+            await middleware.dispatch(request, call_next)
+
     def test_endpoint_tier_critical(self, middleware):
         """Test CRITICAL tier detection for auth endpoints."""
         tier = middleware.get_endpoint_tier("/auth/email/login")
diff --git a/tests/unit/mcpgateway/middleware/test_token_usage_middleware.py b/tests/unit/mcpgateway/middleware/test_token_usage_middleware.py
index df289251de..f38370f2a4 100644
--- a/tests/unit/mcpgateway/middleware/test_token_usage_middleware.py
+++ b/tests/unit/mcpgateway/middleware/test_token_usage_middleware.py
@@ -27,6 +27,60 @@ async def _make_asgi_call(middleware, scope, receive=None, send=None):
     return send
 
 
+def _trusted_internal_scope(path, *, marker="rust", hmac_value=None, ctx="ctx", client="127.0.0.1"):
+    """Build a loopback internal ASGI scope for trust-gate exemption tests."""
+    # First-Party
+    from mcpgateway.auth_context import _expected_internal_mcp_runtime_auth_header
+
+    headers = [
+        (b"x-contextforge-mcp-runtime", marker.encode()),
+        (b"x-contextforge-mcp-runtime-auth", (hmac_value or _expected_internal_mcp_runtime_auth_header()).encode()),
+    ]
+    if ctx is not None:
+        headers.append((b"x-contextforge-auth-context", ctx.encode()))
+    return {
+        "type": "http",
+        "method": "POST",
+        "path": path,
+        "raw_path": path.encode(),
+        "query_string": b"",
+        "headers": headers,
+        "client": (client, 12345),
+    }
+
+
+@pytest.mark.asyncio
+async def test_trusted_internal_dispatch_skips_usage_logging():
+    """A trusted-internal hop is not recorded; the edge request already was.
+
+    The skip path forwards the original ``send`` unwrapped (no status capture, no
+    logging), which distinguishes it from the normal metered path.
+    """
+    app = AsyncMock()
+    middleware = TokenUsageMiddleware(app=app)
+    scope = _trusted_internal_scope("/_internal/mcp/rpc")
+    receive, send = AsyncMock(), AsyncMock()
+
+    await middleware(scope, receive, send)
+
+    app.assert_awaited_once_with(scope, receive, send)
+
+
+@pytest.mark.asyncio
+async def test_forged_internal_request_is_not_skipped_for_usage():
+    """A forged HMAC fails the trust gate, so the usage path runs (wrapped send)."""
+    app = AsyncMock()
+    middleware = TokenUsageMiddleware(app=app)
+    scope = _trusted_internal_scope("/_internal/mcp/rpc", hmac_value="forged")
+    receive, send = AsyncMock(), AsyncMock()
+
+    await middleware(scope, receive, send)
+
+    assert app.await_count == 1
+    forwarded_send = app.await_args.args[2]
+    assert forwarded_send is not send
+
+
 @pytest.mark.asyncio
 async def test_skips_health_check_paths():
     """Middleware should skip health check and static paths."""
diff --git a/tests/unit/mcpgateway/test_internal_a2a_endpoints.py b/tests/unit/mcpgateway/test_internal_a2a_endpoints.py
index 446b5e8f9d..06effd1b12 100644
--- a/tests/unit/mcpgateway/test_internal_a2a_endpoints.py
+++ b/tests/unit/mcpgateway/test_internal_a2a_endpoints.py
@@ -1162,7 +1162,7 @@ def test_endpoints_reject_when_a2a_disabled(self, url, client, monkeypatch):
             "x-contextforge-mcp-runtime": "rust",
             "x-contextforge-mcp-runtime-auth": "stub",  # pragma: allowlist secret
         }
-        with patch("mcpgateway.main.has_valid_internal_mcp_runtime_auth_header", return_value=True):
+        with patch("mcpgateway.auth_context.has_valid_internal_mcp_runtime_auth_header", return_value=True):
             resp = client.post(url, json={}, headers=headers)
         assert resp.status_code == 403
 
@@ -1173,7 +1173,7 @@ def test_agent_endpoints_reject_when_a2a_disabled(self, url, client, monkeypatch
             "x-contextforge-mcp-runtime": "rust",
             "x-contextforge-mcp-runtime-auth": "stub",  # pragma: allowlist secret
         }
-        with patch("mcpgateway.main.has_valid_internal_mcp_runtime_auth_header", return_value=True):
+        with patch("mcpgateway.auth_context.has_valid_internal_mcp_runtime_auth_header", return_value=True):
             resp = client.post(url, json={}, headers=headers)
         assert resp.status_code == 403
 
@@ -1210,11 +1210,11 @@ def test_mcp_runtime_endpoints_still_trusted_when_a2a_disabled(self, client, mon
         # First-Party
         from mcpgateway.main import _is_trusted_internal_mcp_runtime_request
 
-        with patch("mcpgateway.main.has_valid_internal_mcp_runtime_auth_header", return_value=True):
+        with patch("mcpgateway.auth_context.has_valid_internal_mcp_runtime_auth_header", return_value=True):
             assert _is_trusted_internal_mcp_runtime_request(request) is True
 
         # And the equivalent /_internal/a2a/ path with the same headers
         # MUST still be rejected (the feature flag narrows correctly).
         a2a_scope = {**scope, "path": "/_internal/a2a/authenticate", "raw_path": b"/_internal/a2a/authenticate"}
-        with patch("mcpgateway.main.has_valid_internal_mcp_runtime_auth_header", return_value=True):
+        with patch("mcpgateway.auth_context.has_valid_internal_mcp_runtime_auth_header", return_value=True):
             assert _is_trusted_internal_mcp_runtime_request(Request(a2a_scope)) is False
diff --git a/tests/unit/mcpgateway/test_internal_runtime_trust.py b/tests/unit/mcpgateway/test_internal_runtime_trust.py
new file mode 100644
index 0000000000..d95dc6070c
--- /dev/null
+++ b/tests/unit/mcpgateway/test_internal_runtime_trust.py
@@ -0,0 +1,142 @@
+# -*- coding: utf-8 -*-
+"""Unit tests for the consolidated internal-runtime trust gate.
+
+Covers the shared helper in ``auth_context`` (``is_trusted_internal_runtime_request`` /
+``is_trusted_internal_mcp_request``), the path-aware auth-context rule, the A2A
+feature guard, the affinity marker, the HMAC-is-the-trust-boundary property, and
+that the forwarded/client-IP headers are stripped from loopback passthrough.
+"""
+
+# Third-Party
+import pytest
+from starlette.requests import Request
+
+# First-Party
+from mcpgateway.auth_context import (
+    _expected_internal_mcp_runtime_auth_header,
+    _internal_path_requires_auth_context,
+    is_trusted_internal_mcp_request,
+    is_trusted_internal_runtime_request,
+)
+
+VALID_HMAC = _expected_internal_mcp_runtime_auth_header()
+
+
+def _req(path, *, marker="rust", hmac="valid", ctx="ctx", client="127.0.0.1", extra=None):
+    """Build a synthetic internal request."""
+    headers = []
+    if marker is not None:
+        headers.append((b"x-contextforge-mcp-runtime", marker.encode()))
+    if hmac is not None:
+        headers.append((b"x-contextforge-mcp-runtime-auth", (VALID_HMAC if hmac == "valid" else hmac).encode()))
+    if ctx is not None:
+        headers.append((b"x-contextforge-auth-context", ctx.encode()))
+    for k, v in (extra or []):
+        headers.append((k, v))
+    scope = {"type": "http", "method": "POST", "path": path, "raw_path": path.encode(),
+             "query_string": b"", "headers": headers, "client": (client, 12345) if client else None}
+    return Request(scope)
+
+
+# --- path-aware auth-context rule ------------------------------------------
+
+@pytest.mark.parametrize("path,expected", [
+    ("/_internal/mcp/authenticate", False),
+    ("/_internal/mcp/authenticate/", False),
+    ("/_internal/a2a/authenticate", False),
+    ("/_internal/mcp/rpc", True),
+    ("/_internal/mcp/initialize", True),
+    ("/_internal/a2a/tasks/get", True),
+])
+def test_path_requires_auth_context(path, expected):
+    assert _internal_path_requires_auth_context(path) is expected
+
+
+# --- MCP gate: allow --------------------------------------------------------
+
+def test_rpc_trusted_with_rust_marker_hmac_and_ctx():
+    assert is_trusted_internal_mcp_request(_req("/_internal/mcp/rpc")) is True
+
+
+def test_rpc_trusted_with_affinity_marker():
+    assert is_trusted_internal_mcp_request(_req("/_internal/mcp/rpc", marker="affinity")) is True
+
+
+def test_authenticate_trusted_without_ctx():
+    # /authenticate creates the context, so no ctx required.
+    assert is_trusted_internal_mcp_request(_req("/_internal/mcp/authenticate", ctx=None)) is True
+
+
+# --- MCP gate: deny ---------------------------------------------------------
+
+def test_deny_bad_hmac_even_on_loopback_with_ctx():
+    # The HMAC is the trust boundary: loopback + marker + ctx but a bad HMAC is denied.
+    assert is_trusted_internal_mcp_request(_req("/_internal/mcp/rpc", hmac="not-the-real-hmac")) is False
+
+
+def test_deny_wrong_marker():
+    assert is_trusted_internal_mcp_request(_req("/_internal/mcp/rpc", marker="bogus")) is False
+
+
+def test_deny_non_loopback_client():
+    assert is_trusted_internal_mcp_request(_req("/_internal/mcp/rpc", client="10.0.0.9")) is False
+
+
+def test_deny_rpc_missing_ctx():
+    assert is_trusted_internal_mcp_request(_req("/_internal/mcp/rpc", ctx=None)) is False
+
+
+def test_deny_non_internal_prefix():
+    assert is_trusted_internal_mcp_request(_req("/servers/x/mcp")) is False
+
+
+def test_xff_cannot_make_untrusted_request_trusted():
+    # Even with a spoofed X-Forwarded-For and a loopback client, a bad HMAC is denied:
+    # loopback is defense-in-depth, the HMAC is the boundary.
+    req = _req("/_internal/mcp/rpc", hmac="forged", extra=[(b"x-forwarded-for", b"127.0.0.1")])
+    assert is_trusted_internal_mcp_request(req) is False
+
+
+# --- A2A feature guard ------------------------------------------------------
+
+def test_a2a_trusted_when_enabled(monkeypatch):
+    monkeypatch.setattr("mcpgateway.auth_context.settings.mcpgateway_a2a_enabled", True)
+    assert is_trusted_internal_mcp_request(_req("/_internal/a2a/tasks/get")) is True
+
+
+def test_a2a_denied_when_disabled(monkeypatch):
+    monkeypatch.setattr("mcpgateway.auth_context.settings.mcpgateway_a2a_enabled", False)
+    assert is_trusted_internal_mcp_request(_req("/_internal/a2a/tasks/get")) is False
+
+
+# --- generic helper: explicit require_auth_context + prefixes ---------------
+
+def test_generic_helper_respects_prefix_allowlist():
+    req = _req("/_internal/mcp/rpc")
+    assert is_trusted_internal_runtime_request(req, allowed_prefixes=("/_internal/a2a",), require_auth_context=True) is False
+    assert is_trusted_internal_runtime_request(req, allowed_prefixes=("/_internal/mcp",), require_auth_context=True) is True
+
+
+# --- token_scoping now trusts the affinity marker ---------------------------
+
+def test_token_scoping_trusts_affinity_hop():
+    # First-Party
+    from mcpgateway.middleware.token_scoping import token_scoping_middleware
+
+    req = _req("/_internal/mcp/rpc", marker="affinity")
+    assert token_scoping_middleware._is_trusted_internal_mcp_runtime_request(req, "/_internal/mcp/rpc") is True
+
+
+# --- forwarded / client-IP headers are stripped from loopback passthrough ---
+
+@pytest.mark.parametrize("header", [
+    "forwarded", "x-forwarded-for", "x-forwarded-host", "x-forwarded-proto",
+    "x-forwarded-port", "x-forwarded-prefix", "x-real-ip", "cf-connecting-ip", "true-client-ip",
+])
+def test_loopback_passthrough_strips_forwarded_headers(header):
+    # First-Party
+    from mcpgateway.utils.passthrough_headers import filter_loopback_skip_headers
+
+    out = filter_loopback_skip_headers({header: "1.2.3.4", "x-keep": "ok"})
+    assert header not in {k.lower() for k in out}
+    assert out.get("x-keep") == "ok"