[argus] models/factory: per-event-loop httpx client for ChatOpenAI

langchain-openai (as of 1.1.7) caches the async httpx client with a
process-global @lru_cache. LangGraph's worker model spins up a fresh
asyncio loop per task; reusing an httpx connection from a prior loop
after that loop has been torn down raises:

    RuntimeError: Event loop is closed

at stream-cleanup time. The upstream issue
(langchain-ai/langchain#35783) is open as of 2026-04 — none of the
four community-submitted PRs were merged.

Until upstream lands a fix, side-step the broken cache by passing an
explicit http_async_client into ChatOpenAI ourselves, scoped to the
currently-running event loop. Cache keyed on id(loop) with weak refs,
so clients are GC'd when their loop dies — no leak across long-lived
processes that handle thousands of short-lived loops.

Only injected for ChatOpenAI subclasses (the only models hit by this
bug) and only when the caller hasn't already provided
http_async_client themselves.

Tests cover:
  - Sync construction returns None (factory falls back to default)
  - Same-loop calls reuse the cached client
  - A closed cached client is replaced (the post-loop-teardown case)
  - Non-ChatOpenAI subclasses get no http_async_client injection
  - An explicit http_async_client kwarg is preserved

PR-candidate: no
Upstream-issue: langchain-ai/langchain#35783
Reason: This belongs in langchain-openai, not deer-flow. Kept here as
        a load-bearing workaround until #35783 is fixed upstream.

Author: Nichol4s

backend/packages/harness/deerflow/models/factory.py

@@ -1,6 +1,10 @@
+import asyncio
 import logging
+import weakref
 
+import httpx
 from langchain.chat_models import BaseChatModel
+from langchain_openai import ChatOpenAI
 
 from deerflow.config import get_app_config
 from deerflow.reflection import resolve_class
@@ -9,6 +13,53 @@
 logger = logging.getLogger(__name__)
 
 
+# Per-event-loop httpx.AsyncClient cache.
+#
+# langchain-openai (as of 1.1.7) caches the async httpx client with a
+# process-global @lru_cache. LangGraph's worker model spins up a fresh
+# asyncio loop per task; reusing an httpx connection from a prior loop
+# after that loop has been torn down raises:
+#
+#     RuntimeError: Event loop is closed
+#
+# at stream-cleanup time. The upstream issue (langchain-ai/langchain#35783)
+# is open as of 2026-04. Until it's fixed there, we side-step the broken
+# cache by passing an explicit http_async_client into ChatOpenAI ourselves,
+# scoped to the *currently running* event loop.
+#
+# WeakValueDictionary so the client is GC'd when its loop is GC'd; no leak
+# across long-lived processes that handle thousands of short-lived loops.
+_PER_LOOP_HTTPX_CLIENTS: "weakref.WeakValueDictionary[int, httpx.AsyncClient]" = (
+    weakref.WeakValueDictionary()
+)
+
+
+def _httpx_client_for_current_loop() -> httpx.AsyncClient | None:
+    """Return an httpx.AsyncClient bound to the currently-running event loop.
+
+    Returns None when called outside an async context (e.g. sync model
+    construction during config validation), in which case ChatOpenAI falls
+    back to its default — which is fine for non-async paths.
+    """
+    try:
+        loop = asyncio.get_running_loop()
+    except RuntimeError:
+        return None
+
+    loop_id = id(loop)
+    client = _PER_LOOP_HTTPX_CLIENTS.get(loop_id)
+    if client is not None and not client.is_closed:
+        return client
+
+    # Default keepalive is fine because the pool is per-loop now — when the
+    # loop dies the client is GC'd and the connections close cleanly.
+    client = httpx.AsyncClient(
+        timeout=httpx.Timeout(600.0, connect=10.0),
+    )
+    _PER_LOOP_HTTPX_CLIENTS[loop_id] = client
+    return client
+
+
 def _deep_merge_dicts(base: dict | None, override: dict) -> dict:
     """Recursively merge two dictionaries without mutating the inputs."""
     merged = dict(base or {})
@@ -113,7 +164,17 @@ def create_chat_model(name: str | None = None, thinking_enabled: bool = False, *
         elif "reasoning_effort" not in model_settings_from_config:
             model_settings_from_config["reasoning_effort"] = "medium"
 
-    model_instance = model_class(**{**model_settings_from_config, **kwargs})
+    final_kwargs = {**model_settings_from_config, **kwargs}
+
+    # Bypass langchain-openai's process-global httpx client cache (see
+    # _PER_LOOP_HTTPX_CLIENTS docstring above). Only relevant for ChatOpenAI
+    # subclasses; native Anthropic/Google clients have their own pooling.
+    if issubclass(model_class, ChatOpenAI) and "http_async_client" not in final_kwargs:
+        per_loop_client = _httpx_client_for_current_loop()
+        if per_loop_client is not None:
+            final_kwargs["http_async_client"] = per_loop_client
+
+    model_instance = model_class(**final_kwargs)
 
     callbacks = build_tracing_callbacks()
     if callbacks:

backend/tests/test_model_factory.py

@@ -863,3 +863,100 @@ def __init__(self, **kwargs):
 
     # kwargs (runtime) takes precedence: thinking-disabled path sets reasoning_effort=minimal
     assert captured.get("reasoning_effort") == "minimal"
+
+
+# ---------------------------------------------------------------------------
+# Per-event-loop httpx client (langchain-openai cross-loop bypass)
+# ---------------------------------------------------------------------------
+
+
+class TestPerLoopHttpxClient:
+    """The factory injects a per-event-loop httpx.AsyncClient into ChatOpenAI
+    subclasses to bypass langchain-openai's process-global @lru_cache, which
+    breaks under LangGraph's loop-per-task model.
+
+    See module docstring on _PER_LOOP_HTTPX_CLIENTS for the full context.
+    """
+
+    def test_returns_none_outside_async_context(self):
+        """Sync model construction (e.g. config validation) should get None
+        and let ChatOpenAI fall back to its default."""
+        client = factory_module._httpx_client_for_current_loop()
+        assert client is None
+
+    def test_same_loop_returns_same_client(self):
+        """Two calls inside the same event loop must return the same client
+        (otherwise we'd leak connections per call)."""
+        import asyncio as _asyncio
+
+        async def grab_two():
+            a = factory_module._httpx_client_for_current_loop()
+            b = factory_module._httpx_client_for_current_loop()
+            return a, b
+
+        a, b = _asyncio.run(grab_two())
+        assert a is not None
+        assert a is b
+
+    def test_closed_client_is_replaced(self):
+        """A cached client that has been closed must not be reused — the
+        helper must produce a fresh one. This is the key behavior under
+        LangGraph's loop-per-task model: the previous loop's client may be
+        closed but its id() can still be in our cache (because Python
+        recycles loop addresses across sequential asyncio.run calls)."""
+        import asyncio as _asyncio
+
+        captured: dict = {}
+
+        async def grab_first():
+            client = factory_module._httpx_client_for_current_loop()
+            captured["first"] = client
+            await client.aclose()
+
+        async def grab_second():
+            captured["second"] = factory_module._httpx_client_for_current_loop()
+
+        _asyncio.run(grab_first())
+        _asyncio.run(grab_second())
+
+        assert captured["first"] is not None
+        assert captured["second"] is not None
+        assert captured["second"] is not captured["first"]
+        assert not captured["second"].is_closed
+
+    def test_non_chatopenai_does_not_get_http_client_kwarg(self, monkeypatch):
+        """FakeChatModel extends BaseChatModel directly, not ChatOpenAI —
+        the http_async_client injection must not touch it."""
+        cfg = _make_app_config([_make_model("vanilla")])
+        _patch_factory(monkeypatch, cfg)
+
+        FakeChatModel.captured_kwargs = {}
+        factory_module.create_chat_model(name="vanilla")
+
+        assert "http_async_client" not in FakeChatModel.captured_kwargs
+
+    def test_explicit_http_async_client_in_kwargs_is_preserved(self, monkeypatch):
+        """If a caller already provides http_async_client, the factory must
+        not overwrite it. This protects a future code path that wants to
+        inject its own client (e.g. with custom proxies)."""
+        from langchain_openai import ChatOpenAI
+
+        captured: dict = {}
+        sentinel = object()
+
+        class CapturingChatOpenAI(ChatOpenAI):
+            def __init__(self, **kwargs):
+                captured.update(kwargs)
+                # Don't actually try to build a real client; bail early
+                raise _StopFactory()
+
+        class _StopFactory(Exception):
+            pass
+
+        cfg = _make_app_config([_make_model("openai-clone", use="langchain_openai:ChatOpenAI")])
+        _patch_factory(monkeypatch, cfg, model_class=CapturingChatOpenAI)
+
+        with pytest.raises(_StopFactory):
+            factory_module.create_chat_model(name="openai-clone", http_async_client=sentinel)
+
+        assert captured.get("http_async_client") is sentinel