fix(channels): keep Gateway calls off the environment proxy (#7877)

* fix(channels): keep Gateway calls off the environment proxy

* fix(channels): keep the aio_sandbox policy import lazy

Importing deerflow.community.aio_sandbox.backend at module scope executes
the package __init__, which eagerly imports AioSandbox, AioSandboxProvider
and both backends. That makes every ChannelManager import load the entire
sandbox provider stack even when a different sandbox is configured.

Mirror app/gateway/health.py::_make_provisioner_client: move the import into
a small _make_gateway_client helper and route both Gateway call sites through
it, so the URL policy stays lazy.

* fix(channels): keep the run client off the environment proxy

langgraph_sdk.get_client() builds httpx.AsyncHTTPTransport(retries=5) on an
httpx.AsyncClient that leaves trust_env at its default True and exposes no
transport/trust_env override, so ChannelManager._get_client() still sent
runs.wait submissions to a loopback Gateway through HTTP_PROXY. That is the hop
#7876 reports: the control-plane calls are fixed, the run submission is not.

Route the SDK client through _make_langgraph_client: loopback and private
Gateways get an explicit trust_env=False httpx.AsyncClient carrying the SDK's
own retries, timeout and User-Agent and handed to LangGraphClient, while
external Gateways keep the stock get_client() path.

Also move the URL policy out of the aio_sandbox package into
deerflow.utils.http_policy (control_plane_trust_env). Keeping the import lazy
did not help on its own: aio_sandbox/__init__ eagerly imports the provider
stack, so importing aio_sandbox.backend still loaded all of it. The neutral
module has no httpx or sandbox dependency, so manager.py and gateway/health.py
import it at module scope, and aio_sandbox.backend re-exports it as
sandbox_http_trust_env for the existing sandbox call sites.

* fix(channels): keep AGENTS.md under budget and align legacy CSRF test with new client

* style(sandbox): remove unused urlparse import

---------

Co-authored-by: jack12178 <jack12178@users.noreply.github.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

backend/app/channels/manager.py

@@ -69,6 +69,7 @@ from deerflow.skills.storage.skill_storage import SkillStorage
 from deerflow.trace_context import ensure_trace_context
 from deerflow.uploads.manager import apply_upload_sandbox_permits
 from deerflow.utils.file_io import await_drained
+from deerflow.utils.http_policy import control_plane_trust_env
 from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY
 
 logger = logging.getLogger(__name__)
@@ -301,6 +302,59 @@ _WECOM_MEDIA_COS_HOST_RE = re.compile(r"^ww-aibot-img-(?P<appid>\d+)\.cos\.[a-z0
 _METADATA_DROP_KEYS = frozenset({"raw_message", "ref_msg"})
 
 
+def _make_gateway_client(gateway_url: str) -> httpx.AsyncClient:
+    """Build the short-lived client for Gateway control-plane calls.
+
+    A loopback Gateway must bypass ``HTTP_PROXY``, while an external one keeps
+    the environment proxy. The policy itself lives in
+    :mod:`deerflow.utils.http_policy` so this module never imports the sandbox
+    provider stack.
+    """
+    return httpx.AsyncClient(trust_env=control_plane_trust_env(gateway_url))
+
+
+# Mirrors the client ``langgraph_sdk.get_client()`` builds, so the loopback path
+# below differs from it only in ``trust_env``.
+_LANGGRAPH_SDK_RETRIES = 5
+_LANGGRAPH_SDK_TIMEOUT = httpx.Timeout(connect=5, read=300, write=300, pool=5)
+
+
+def _make_langgraph_client(base_url: str, headers: Mapping[str, str]) -> Any:
+    """Build the ``langgraph_sdk`` async client used for channel run submission.
+
+    ``get_client()`` always builds ``httpx.AsyncHTTPTransport(retries=5)`` on an
+    ``httpx.AsyncClient`` that keeps ``trust_env`` at its default ``True`` and
+    exposes neither ``transport`` nor ``trust_env``, so a loopback Gateway would
+    still be sent through ``HTTP_PROXY``: control-plane calls are fixed but
+    ``runs.wait`` keeps hitting the proxy-generated error of #7876. Loopback and
+    private Gateways therefore get an explicit ``trust_env=False`` client handed
+    to the SDK; external Gateways keep the stock ``get_client()`` path.
+    """
+    from langgraph_sdk import __version__ as sdk_version
+    from langgraph_sdk import get_client
+
+    if control_plane_trust_env(base_url):
+        return get_client(url=base_url, headers=dict(headers))
+
+    try:  # langgraph-sdk keeps the async client in a private package since 0.3
+        from langgraph_sdk._async import LangGraphClient
+    except ImportError:  # older releases expose it directly on ``langgraph_sdk.client``
+        from langgraph_sdk.client import LangGraphClient
+
+    # ``get_client`` also folds an ``LANGGRAPH_API_KEY`` env var into the SDK
+    # headers; a control-plane Gateway authenticates with the internal auth
+    # headers above, so only those are forwarded here.
+    return LangGraphClient(
+        httpx.AsyncClient(
+            base_url=base_url,
+            transport=httpx.AsyncHTTPTransport(retries=_LANGGRAPH_SDK_RETRIES),
+            timeout=_LANGGRAPH_SDK_TIMEOUT,
+            headers={"User-Agent": f"langgraph-sdk-py/{sdk_version}", **headers},
+            trust_env=False,
+        )
+    )
+
+
 def _slim_metadata(meta: dict[str, Any]) -> dict[str, Any]:
     """Return a shallow copy of *meta* with known-large keys removed."""
     return {k: v for k, v in meta.items() if k not in _METADATA_DROP_KEYS}
@@ -2318,11 +2372,9 @@ class ChannelManager:
     def _get_client(self):
         """Return the ``langgraph_sdk`` async client, creating it on first use."""
         if self._client is None:
-            from langgraph_sdk import get_client
-
-            self._client = get_client(
-                url=self._langgraph_url,
-                headers={**create_internal_auth_headers(), **self._csrf_headers()},
+            self._client = _make_langgraph_client(
+                self._langgraph_url,
+                {**create_internal_auth_headers(), **self._csrf_headers()},
             )
         return self._client
 
@@ -4056,7 +4108,7 @@ class ChannelManager:
         headers: dict[str, str],
         json: dict[str, Any] | None = None,
     ) -> dict[str, Any]:
-        async with httpx.AsyncClient() as http:
+        async with _make_gateway_client(self._gateway_url) as http:
             request = getattr(http, method.lower())
             kwargs: dict[str, Any] = {"timeout": 10, "headers": {**headers, **self._csrf_headers()}}
             if json is not None:
@@ -4152,7 +4204,7 @@ class ChannelManager:
         replies should catch and degrade (see ``_fetch_gateway``).
         """
         headers = _owner_headers(msg) if msg is not None else None
-        async with httpx.AsyncClient() as http:
+        async with _make_gateway_client(self._gateway_url) as http:
             resp = await http.get(
                 f"{self._gateway_url}{path}",
                 timeout=10,

backend/app/gateway/health.py

@@ -51,6 +51,7 @@ from sqlalchemy import text
 
 from deerflow.persistence.engine import get_engine
 from deerflow.utils.file_io import await_drained
+from deerflow.utils.http_policy import control_plane_trust_env
 
 if TYPE_CHECKING:
     from deerflow.config.app_config import AppConfig
@@ -313,10 +314,7 @@ def _make_provisioner_client(provisioner_url: str) -> httpx.AsyncClient:
     ``NO_PROXY``, so without this a private provisioner behind a proxy-only
     egress would be reported unreachable on every probe.
     """
-    # Lazy: importing the aio_sandbox package pulls in the whole provider stack.
-    from deerflow.community.aio_sandbox.backend import sandbox_http_trust_env
-
-    return httpx.AsyncClient(timeout=_PROBE_TIMEOUT_SECONDS, trust_env=sandbox_http_trust_env(provisioner_url))
+    return httpx.AsyncClient(timeout=_PROBE_TIMEOUT_SECONDS, trust_env=control_plane_trust_env(provisioner_url))
 
 
 async def _probe_provisioner(provisioner_url: str | None) -> str:

backend/packages/harness/deerflow/community/aio_sandbox/backend.py

@@ -3,44 +3,26 @@
 from __future__ import annotations
 
 import asyncio
-import ipaddress
 import logging
 import time
 from abc import ABC, abstractmethod
 from collections.abc import Mapping
 from typing import Any
-from urllib.parse import urlparse
 
 import httpx
 import requests
 
+from deerflow.utils.http_policy import control_plane_trust_env
+
 from .sandbox_info import SandboxInfo
 
 logger = logging.getLogger(__name__)
 
-
-def sandbox_http_trust_env(sandbox_url: str) -> bool:
-    """Whether HTTP clients for *sandbox_url* should inherit proxy settings.
-
-    Local Docker, DooD, and Kubernetes sandbox endpoints are control-plane
-    connections, not internet traffic. Sending them through ``HTTP_PROXY`` can
-    produce a misleading proxy-generated 502 even though the sandbox container
-    is healthy (#3441). External fully-qualified hosts retain normal environment
-    proxy behavior.
-    """
-    try:
-        hostname = (urlparse(sandbox_url).hostname or "").rstrip(".").lower()
-    except ValueError:
-        return True
-    if not hostname:
-        return True
-    if hostname == "localhost" or hostname.endswith(".localhost") or hostname.endswith(".docker.internal") or hostname.endswith(".containers.internal"):
-        return False
-    try:
-        address = ipaddress.ip_address(hostname)
-    except ValueError:
-        return "." in hostname
-    return not (address.is_loopback or address.is_private or address.is_link_local)
+# The policy lives in ``deerflow.utils.http_policy`` so Gateway-side callers can
+# apply it without importing this package (whose ``__init__`` loads the whole
+# sandbox provider stack). Kept as a module-level name for the existing sandbox
+# call sites and for tests that patch it here.
+sandbox_http_trust_env = control_plane_trust_env
 
 
 # The readiness deadline the local-container provider paths (sync and async)

backend/packages/harness/deerflow/utils/AGENTS.md

@@ -76,6 +76,20 @@ when all remaining ports are reserved or occupied, without probing invalid
 TCP ports. Keep the last valid port allocatable and preserve `max_range`.
 Regression coverage lives in `tests/test_port_allocator_bounds.py`.
 
+### Control-Plane Proxy Policy
+
+`http_policy.py::control_plane_trust_env` is the single answer to "should this
+client inherit `HTTP_PROXY`?" for DeerFlow's own hops. Loopback, private,
+link-local, `.localhost`, Docker/Podman-internal and cluster-local hostnames are
+control-plane traffic and must bypass the environment proxy; external FQDNs and
+public IPs keep it. The module stays dependency-free (no `httpx`, no sandbox
+imports) precisely so `app/channels/manager.py` and `app/gateway/health.py` can
+import it at module scope without loading the sandbox provider stack.
+`community/aio_sandbox/backend.py` re-exports it as `sandbox_http_trust_env` for
+existing sandbox call sites. Policy parity with the original sandbox-local
+implementation is covered by `tests/test_aio_sandbox_readiness.py`; the
+import-graph guard lives in `tests/test_channels.py`.
+
 ### Goal Objective Validation
 
 `goal_objective.py` owns the dependency-free normalized 4000-character objective
@@ -238,3 +252,14 @@ retries; the helper remains responsible for error classification and cadence.
 `parse_retry_after_date_ms(raw, now=...)` owns HTTP-date parsing for Claude, Codex
 and LLM error middleware. Interpret timezone-less asctime as UTC and apply the
 shared delay bound. Callers retain numeric/header precedence and local backoff.
+
+### Control-plane HTTP proxy policy
+
+`http_policy.control_plane_trust_env(url)` decides whether a client inherits
+`HTTP_PROXY`: loopback, private and link-local hosts are control-plane traffic
+and bypass it; external hosts keep the environment proxy. `app.channels.manager`
+uses it for both `_make_gateway_client` and `_make_langgraph_client`.
+`langgraph_sdk.get_client()` hard-codes `trust_env=True` and exposes no transport
+override, so the `runs.wait` client for local Gateways is assembled on our own
+`httpx.AsyncClient` with `trust_env=False`; otherwise #7876 keeps reproducing.
+Importing `ChannelManager` must not pull in `deerflow.community.aio_sandbox`.

backend/packages/harness/deerflow/utils/http_policy.py

@@ -0,0 +1,40 @@
+"""Proxy policy shared by every control-plane HTTP client.
+
+DeerFlow reaches its own Gateway, provisioner and sandboxes over the deployment
+network, not the public internet. When operators export ``HTTP_PROXY`` for
+outbound model/tool traffic, those hops must not be sent through the proxy: the
+proxy answers with its own 502/504 while the target is perfectly healthy
+(#3441, #7876).
+
+This module is deliberately dependency-free (no ``httpx``, no sandbox imports)
+so that ``app.channels.manager``, ``app.gateway.health`` and the sandbox
+backends can all import the policy without dragging in the sandbox provider
+stack.
+"""
+
+from __future__ import annotations
+
+import ipaddress
+from urllib.parse import urlparse
+
+
+def control_plane_trust_env(url: str) -> bool:
+    """Whether HTTP clients for *url* should inherit environment proxy settings.
+
+    Local Docker, DooD, Kubernetes and co-located Gateway endpoints are
+    control-plane connections, not internet traffic. External fully-qualified
+    hosts retain normal environment proxy behavior.
+    """
+    try:
+        hostname = (urlparse(url).hostname or "").rstrip(".").lower()
+    except ValueError:
+        return True
+    if not hostname:
+        return True
+    if hostname == "localhost" or hostname.endswith(".localhost") or hostname.endswith(".docker.internal") or hostname.endswith(".containers.internal"):
+        return False
+    try:
+        address = ipaddress.ip_address(hostname)
+    except ValueError:
+        return "." in hostname
+    return not (address.is_loopback or address.is_private or address.is_link_local)

backend/tests/test_channels.py

@@ -5,6 +5,8 @@ from __future__ import annotations
 import asyncio
 import json
 import logging
+import subprocess
+import sys
 import tempfile
 import threading
 from concurrent.futures import Future
@@ -702,7 +704,9 @@ class TestChannelManager:
 
         bus = MessageBus()
         store = JsonChannelStore(path=Path(tempfile.mkdtemp()) / "store.json")
-        manager = ChannelManager(bus=bus, store=store, langgraph_url="http://localhost:8001")
+        # An external Gateway keeps langgraph_sdk.get_client(); loopback hosts
+        # build a trust_env=False client and are covered by the run-client tests.
+        manager = ChannelManager(bus=bus, store=store, langgraph_url="http://gateway.example.com/api")
 
         with patch("langgraph_sdk.get_client") as get_client:
             get_client.return_value = object()
@@ -711,7 +715,7 @@ class TestChannelManager:
 
         get_client.assert_called_once()
         kwargs = get_client.call_args.kwargs
-        assert kwargs["url"] == "http://localhost:8001"
+        assert kwargs["url"] == "http://gateway.example.com/api"
         headers = kwargs["headers"]
         csrf_token = headers["X-CSRF-Token"]
         assert csrf_token
@@ -14425,3 +14429,167 @@ class TestHumanInputResponseTranslation:
         message = _human_input_message_with_channel_response("hello", metadata={})
         assert message["content"] == "hello"
         assert "additional_kwargs" not in message
+
+
+@pytest.mark.parametrize(
+    ("gateway_url", "expected_trust_env"),
+    [
+        ("http://localhost:8001", False),
+        ("http://127.0.0.1:8001", False),
+        ("http://gateway.example.com", True),
+    ],
+)
+@pytest.mark.asyncio
+async def test_gateway_clients_do_not_proxy_local_gateway(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    gateway_url: str,
+    expected_trust_env: bool,
+) -> None:
+    """Gateway calls are control-plane traffic and must bypass ``HTTP_PROXY``.
+
+    The Gateway runs next to the channel manager (``http://localhost:8001`` by
+    default). Inheriting the environment proxy for that hop produces a
+    proxy-generated error even though the Gateway is healthy -- the same
+    failure ``sandbox_http_trust_env`` already guards against for sandbox
+    control-plane calls.
+    """
+    from app.channels import manager as manager_module
+    from app.channels.manager import ChannelManager
+
+    seen_kwargs: list[dict[str, object]] = []
+
+    class _JsonResponse:
+        def raise_for_status(self) -> None:
+            return None
+
+        def json(self) -> dict[str, object]:
+            return {"ok": True}
+
+    class RecordingClient:
+        def __init__(self, *args: object, **kwargs: object) -> None:
+            seen_kwargs.append(kwargs)
+
+        async def __aenter__(self) -> RecordingClient:
+            return self
+
+        async def __aexit__(self, *exc_info: object) -> None:
+            return None
+
+        async def get(self, url: str, **kwargs: object) -> _JsonResponse:
+            return _JsonResponse()
+
+        async def put(self, url: str, **kwargs: object) -> _JsonResponse:
+            return _JsonResponse()
+
+    monkeypatch.setattr(manager_module.httpx, "AsyncClient", RecordingClient)
+
+    manager = ChannelManager(
+        bus=MessageBus(),
+        store=JsonChannelStore(path=tmp_path / "store.json"),
+        gateway_url=gateway_url,
+    )
+
+    await manager._fetch_gateway_json("/api/models")
+    await manager._goal_request("put", "thread-1", headers={}, json={"objective": "x"})
+
+    assert len(seen_kwargs) == 2, "both Gateway call sites should build a client"
+    assert all(kwargs.get("trust_env") is expected_trust_env for kwargs in seen_kwargs)
+
+
+@pytest.mark.parametrize("langgraph_url", ["http://localhost:8001/api", "http://127.0.0.1:8001/api"])
+def test_run_client_bypasses_env_proxy_for_local_gateway(
+    monkeypatch: pytest.MonkeyPatch,
+    langgraph_url: str,
+) -> None:
+    """``runs.wait`` submissions are control-plane traffic too.
+
+    ``langgraph_sdk.get_client()`` builds its own ``httpx.AsyncClient`` with
+    ``trust_env`` left at the default ``True`` and exposes no override, so a
+    loopback Gateway was still reached through ``HTTP_PROXY`` -- the exact hop
+    #7876 reports, and one the Gateway control-plane clients do not cover.
+    """
+    from app.channels import manager as manager_module
+
+    seen_kwargs: list[dict[str, object]] = []
+
+    class RecordingClient:
+        def __init__(self, *args: object, **kwargs: object) -> None:
+            seen_kwargs.append(kwargs)
+
+    monkeypatch.setattr(manager_module.httpx, "AsyncClient", RecordingClient)
+
+    client = manager_module._make_langgraph_client(langgraph_url, {"X-Internal": "1"})
+
+    assert len(seen_kwargs) == 1, "the SDK client must wrap the client we build"
+    client_kwargs = seen_kwargs[0]
+    assert client_kwargs["trust_env"] is False
+    assert client_kwargs["base_url"] == langgraph_url
+    headers = client_kwargs["headers"]
+    assert isinstance(headers, dict)
+    assert headers["X-Internal"] == "1"
+    assert headers["User-Agent"].startswith("langgraph-sdk-py/")
+    assert isinstance(client.http.client, RecordingClient)
+
+
+def test_run_client_keeps_stock_sdk_client_for_external_gateway(monkeypatch: pytest.MonkeyPatch) -> None:
+    """An external Gateway keeps the SDK's own client, proxy settings included."""
+    import langgraph_sdk
+
+    from app.channels import manager as manager_module
+
+    calls: list[dict[str, object]] = []
+    sentinel = object()
+
+    def fake_get_client(**kwargs: object) -> object:
+        calls.append(kwargs)
+        return sentinel
+
+    monkeypatch.setattr(langgraph_sdk, "get_client", fake_get_client)
+
+    client = manager_module._make_langgraph_client("http://gateway.example.com/api", {"X-Internal": "1"})
+
+    assert client is sentinel
+    assert calls == [{"url": "http://gateway.example.com/api", "headers": {"X-Internal": "1"}}]
+
+
+def test_channel_manager_builds_run_client_with_internal_auth_headers(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The run client carries the same internal auth and CSRF pair as control calls."""
+    from app.channels import manager as manager_module
+    from app.channels.manager import ChannelManager
+
+    calls: list[tuple[str, dict[str, str]]] = []
+
+    def fake_make_langgraph_client(base_url: str, headers: dict[str, str]) -> object:
+        calls.append((base_url, dict(headers)))
+        return object()
+
+    monkeypatch.setattr(manager_module, "_make_langgraph_client", fake_make_langgraph_client)
+
+    manager = ChannelManager(
+        bus=MessageBus(),
+        store=JsonChannelStore(path=tmp_path / "store.json"),
+        gateway_url="http://localhost:8001",
+    )
+    manager._get_client()
+
+    assert len(calls) == 1
+    base_url, headers = calls[0]
+    assert base_url == manager_module.DEFAULT_LANGGRAPH_URL
+    assert manager_module.CSRF_HEADER_NAME in headers
+
+
+def test_importing_channel_manager_loads_no_sandbox_provider_stack() -> None:
+    """The proxy policy must not drag the sandbox package into every Gateway boot."""
+    code = "import sys\nimport app.channels.manager\nprint(','.join(sorted(name for name in sys.modules if name.startswith('deerflow.community.aio_sandbox'))))\n"
+    result = subprocess.run(
+        [sys.executable, "-c", code],
+        cwd=Path(__file__).resolve().parents[1],
+        capture_output=True,
+        text=True,
+        check=True,
+    )
+    assert result.stdout.strip() == "", "ChannelManager import pulled in the sandbox provider stack"