test(sdk-py): add factory-graph integration test exercising the server factory path (#7978)

.github/workflows/ci.yml

@@ -50,8 +50,10 @@ jobs:
               - '**/uv.lock'
             sdk_py:
               - 'libs/sdk-py/**'
-              - 'libs/langgraph/langgraph/pregel/remote.py'
-              - 'libs/langgraph/langgraph/pregel/_remote_run_stream.py'
+              # The integration suite runs the local langgraph core inside the
+              # server (see libs/sdk-py/integration/Dockerfile), so any core
+              # change is now exercised end-to-end and should trigger it.
+              - 'libs/langgraph/**'
 
   lint:
     needs: changes

libs/langgraph/langgraph/pregel/main.py

@@ -2835,7 +2835,9 @@ class Pregel(
                 config[CONF][CONFIG_KEY_DURABILITY] = durability_
 
             # build server_info from metadata + parent runtime
-            parent_runtime = config[CONF].get(CONFIG_KEY_RUNTIME, DEFAULT_RUNTIME)
+            parent_runtime = _coerce_parent_runtime(
+                config[CONF].get(CONFIG_KEY_RUNTIME, DEFAULT_RUNTIME)
+            )
             server_info = _build_server_info(config, parent_runtime)
 
             runtime = Runtime(
@@ -3276,7 +3278,9 @@ class Pregel(
                 config[CONF][CONFIG_KEY_DURABILITY] = durability_
 
             # build server_info from metadata + parent runtime
-            parent_runtime = config[CONF].get(CONFIG_KEY_RUNTIME, DEFAULT_RUNTIME)
+            parent_runtime = _coerce_parent_runtime(
+                config[CONF].get(CONFIG_KEY_RUNTIME, DEFAULT_RUNTIME)
+            )
             server_info = _build_server_info(config, parent_runtime)
 
             runtime = Runtime(
@@ -4253,6 +4257,23 @@ def _resolve_parent_ns(
     return tuple(ns.split(NS_SEP))
 
 
+def _coerce_parent_runtime(value: Any) -> Runtime[Any]:
+    """Normalize the value stored under `CONFIG_KEY_RUNTIME` into a `Runtime`.
+
+    During a graph run this is always a `Runtime` the framework created and
+    published for child tasks to inherit. Layers outside the run (for example a
+    server's graph-factory path) may instead seed an object that only carries
+    `context`/`store`. Adopt its `context` so context set at the graph level
+    plumbs through (`merge` lets the run's own `context` take precedence when
+    one is provided). `store` is resolved separately (passed to the graph
+    directly), so it is not read off here. `merge` then combines this with the
+    run's own runtime, including the framework's `control`.
+    """
+    if isinstance(value, Runtime):
+        return value
+    return Runtime(context=getattr(value, "context", None))
+
+
 def _build_server_info(
     config: RunnableConfig, parent_runtime: Runtime[Any]
 ) -> ServerInfo | None:

libs/langgraph/tests/test_runtime.py

@@ -1164,3 +1164,57 @@ def test_execution_info_inherited_by_subgraph() -> None:
     # task_id appears in its own namespace segment
     assert main_info.task_id in main_info.checkpoint_ns
     assert sub_info.task_id in sub_info.checkpoint_ns
+
+
+def test_foreign_object_in_runtime_slot_is_coerced() -> None:
+    """A non-`Runtime` under `CONFIG_KEY_RUNTIME` is adopted, not crashed on,
+    and the `context` it carries is plumbed through.
+
+    Layers outside a run (e.g. a server's graph-factory path, like LangGraph
+    API) seed an object carrying `context`/`store` into the runtime slot. The
+    run must still execute (regression for `AttributeError: '...' object has no
+    attribute 'control'`), and `context` set at that level must reach nodes via
+    `merge` when no per-run `context` is provided. `store` is resolved
+    separately, so it is not read off the foreign object in the coercion.
+    """
+    from langgraph.store.memory import InMemoryStore
+
+    from langgraph._internal._constants import CONFIG_KEY_RUNTIME
+
+    store = InMemoryStore()
+    graph_level_context = {"source": "graph-level"}
+
+    class _ServerLikeRuntime:
+        """Carries context/store but is not a `Runtime` (no control/merge)."""
+
+        def __init__(self) -> None:
+            self.context = graph_level_context
+            self.store = store
+
+    seen: dict[str, Any] = {}
+
+    class State(TypedDict, total=False):
+        text: str
+
+    def echo(state: State, runtime: Runtime) -> State:
+        seen["context"] = runtime.context
+        seen["store"] = runtime.store
+        return {"text": (state.get("text") or "") + " echoed"}
+
+    builder = StateGraph(State)
+    builder.add_node("echo", echo)
+    builder.add_edge(START, "echo")
+    builder.add_edge("echo", END)
+    graph = builder.compile()
+
+    # No per-run `context=`: the graph-level context on the slot must plumb through.
+    result = graph.invoke(
+        {"text": "hi"},
+        {"configurable": {CONFIG_KEY_RUNTIME: _ServerLikeRuntime()}},
+    )
+
+    assert result == {"text": "hi echoed"}
+    # context set at the graph level is plumbed through to the node via merge
+    assert seen["context"] == graph_level_context
+    # store still reaches the node (resolved separately, not via the coercion)
+    assert seen["store"] is store

libs/sdk-py/integration/Dockerfile

@@ -22,6 +22,19 @@ RUN pip install --no-cache-dir \
     "langchain>=1.3.0" \
     "deepagents>=0.6.2"
 
+# Swap the published langgraph *core* for this monorepo's local copy, so the
+# server executes the langgraph under test rather than the latest release.
+# We keep the rest of the base image (langgraph-api, runtime, Go core server)
+# on `latest` — that still surfaces upstream regressions — but the core the
+# server runs is now the PR's, which is what lets this suite catch core
+# regressions (e.g. ensure_config / runtime changes) before they're published.
+# `--no-deps` keeps the base image's already-compatible
+# checkpoint/prebuilt/sdk; we only replace core. The local source comes from
+# the `langgraph_src` additional build context (see docker-compose.yml).
+COPY --from=langgraph_src pyproject.toml README.md LICENSE /opt/langgraph-src/
+COPY --from=langgraph_src langgraph /opt/langgraph-src/langgraph/
+RUN pip install --no-cache-dir --force-reinstall --no-deps /opt/langgraph-src
+
 # Project graphs + registration config.
 COPY graph/ /app/graph/
 COPY langgraph.json /app/langgraph.json

libs/sdk-py/integration/docker-compose.yml

@@ -36,11 +36,18 @@ services:
   api:
     # Thin layer on top of langchain/langgraph-api:latest-py3.12 —
     # see ./Dockerfile. The base image bundles langgraph-api +
-    # langgraph_runtime_postgres + langgraph_license + the Go core-server,
-    # so we only add graph deps (deepagents) and the graph files on top.
+    # langgraph_runtime_postgres + langgraph_license + the Go core-server;
+    # on top we add graph deps (deepagents), the graph files, and this
+    # monorepo's local langgraph *core* (so the server runs the code under
+    # test, not the released langgraph).
     build:
       context: .
       dockerfile: Dockerfile
+      additional_contexts:
+        # The monorepo's local langgraph core (libs/langgraph), installed over
+        # the base image so the server runs the langgraph under test. See the
+        # Dockerfile for why.
+        langgraph_src: ../../langgraph
     image: langgraph-v3-integration-api:local
     depends_on:
       postgres:
@@ -70,7 +77,7 @@ services:
       # Tell langgraph-api which graphs to register. The langgraph CLI sets
       # this from langgraph.json; we're bypassing the CLI (running uvicorn
       # directly) so we set it manually.
-      LANGSERVE_GRAPHS: '{"agent":"/app/graph/streaming_graph.py:graph","tools_agent":"/app/graph/tools_agent.py:graph","deep_agent":"/app/graph/deep_agent.py:graph"}'
+      LANGSERVE_GRAPHS: '{"agent":"/app/graph/streaming_graph.py:graph","tools_agent":"/app/graph/tools_agent.py:graph","deep_agent":"/app/graph/deep_agent.py:graph","factory_agent":"/app/graph/factory_graph.py:make_graph"}'
 
       # The published langgraph-api image is the `licensed` variant and
       # requires a real LANGSMITH_API_KEY (or LANGGRAPH_CLOUD_LICENSE_KEY)

libs/sdk-py/integration/graph/factory_graph.py

@@ -0,0 +1,70 @@
+"""Integration fixture: a graph *factory* (not a pre-compiled graph).
+
+Registered as `factory_agent`. The other integration graphs are all
+pre-compiled objects, so this is the only fixture that drives the server's
+graph-factory code path on a run. That path is where langgraph-api seeds a
+`ServerRuntime` into the run config; executing a run against this graph is the
+end-to-end regression guard for the langgraph 1.2.3 `ensure_config`
+configurable-merge surfacing a leaked `__pregel_runtime`
+(`AttributeError: '_ExecutionRuntime' object has no attribute 'control'`).
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from langgraph.graph.state import END, CompiledStateGraph, StateGraph
+from langgraph.store.base import BaseStore
+from typing_extensions import TypedDict
+
+if TYPE_CHECKING:
+    from langgraph.types import RunnableConfig
+
+    from langgraph_sdk.runtime import ServerRuntime
+
+
+class State(TypedDict, total=False):
+    text: str
+    access_context: str
+    is_for_execution: bool
+
+
+async def make_graph(
+    config: RunnableConfig, runtime: ServerRuntime
+) -> CompiledStateGraph:
+    """2-arg factory (config + runtime) returning a compiled echo graph.
+
+    Validates that the server injected a real `ServerRuntime`, then builds a
+    one-node graph whose output echoes the input and reports the runtime's
+    access context (so the test can confirm the factory ran under execution).
+    """
+    from langgraph_sdk.runtime import (
+        _ExecutionRuntime,
+        _ReadRuntime,
+    )
+
+    # `if/raise` rather than `assert` so the contract holds under `python -O`.
+    if not isinstance(runtime, (_ExecutionRuntime, _ReadRuntime)):
+        raise TypeError(
+            f"factory must receive a ServerRuntime, got {type(runtime).__name__}"
+        )
+    if not isinstance(runtime.store, BaseStore):
+        raise TypeError(
+            f"runtime.store must be a BaseStore, got {type(runtime.store).__name__}"
+        )
+
+    access_context = runtime.access_context
+    is_for_execution = runtime.execution_runtime is not None
+
+    def echo(state: State) -> State:
+        return {
+            "text": (state.get("text") or "") + " echoed",
+            "access_context": access_context,
+            "is_for_execution": is_for_execution,
+        }
+
+    workflow = StateGraph(State)
+    workflow.add_node("echo", echo)
+    workflow.set_entry_point("echo")
+    workflow.add_edge("echo", END)
+    return workflow.compile()

libs/sdk-py/integration/langgraph.json

@@ -4,7 +4,8 @@
   "graphs": {
     "agent": "./graph/streaming_graph.py:graph",
     "tools_agent": "./graph/tools_agent.py:graph",
-    "deep_agent": "./graph/deep_agent.py:graph"
+    "deep_agent": "./graph/deep_agent.py:graph",
+    "factory_agent": "./graph/factory_graph.py:make_graph"
   },
   "env": {}
 }

libs/sdk-py/tests/integration/conftest.py

@@ -22,6 +22,7 @@ BASE_URL = os.environ.get("LANGGRAPH_INTEGRATION_URL", "http://localhost:2024")
 ASSISTANT_ID = "agent"
 TOOLS_ASSISTANT_ID = "tools_agent"
 DEEP_AGENT_ASSISTANT_ID = "deep_agent"
+FACTORY_ASSISTANT_ID = "factory_agent"
 
 EXPECTED_TERMINAL_ITEMS = ["streamed", "tool", "asked", "sub"]
 

libs/sdk-py/tests/integration/test_factory_graph.py

@@ -0,0 +1,65 @@
+"""Factory-graph execution regression test.
+
+Unlike the other integration graphs (all pre-compiled), `factory_agent` is a
+graph *factory*, so executing a run against it drives the server's graph-factory
+code path. That path regressed in langgraph 1.2.3: a leaked `__pregel_runtime`
+(an SDK `_ExecutionRuntime`) survived `ensure_config`'s configurable-merge into
+`astream`, which then raised
+`AttributeError: '_ExecutionRuntime' object has no attribute 'control'`. A
+successful `runs.wait` here proves the factory path executes end to end.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from .conftest import FACTORY_ASSISTANT_ID
+
+pytestmark = pytest.mark.integration
+
+
+def _async_runs(raw):
+    from langgraph_sdk._async.http import HttpClient
+    from langgraph_sdk._async.runs import RunsClient
+
+    return RunsClient(HttpClient(raw))
+
+
+def _sync_runs(raw):
+    from langgraph_sdk._sync.http import SyncHttpClient
+    from langgraph_sdk._sync.runs import SyncRunsClient
+
+    return SyncRunsClient(SyncHttpClient(raw))
+
+
+async def test_factory_graph_executes_async(async_threads) -> None:
+    """A run against a factory graph completes and echoes the input."""
+    threads, raw = async_threads
+    runs = _async_runs(raw)
+    thread = await threads.create(
+        metadata={"suite": "integration", "label": "factory-async"}
+    )
+    tid = thread["thread_id"]
+    try:
+        result = await runs.wait(tid, FACTORY_ASSISTANT_ID, input={"text": "hi"})
+        assert isinstance(result, dict), result
+        assert result.get("text") == "hi echoed"
+        assert result.get("access_context") == "threads.create_run"
+        assert result.get("is_for_execution") is True
+    finally:
+        await threads.delete(tid)
+
+
+def test_factory_graph_executes_sync(sync_threads) -> None:
+    threads, raw = sync_threads
+    runs = _sync_runs(raw)
+    thread = threads.create(metadata={"suite": "integration", "label": "factory-sync"})
+    tid = thread["thread_id"]
+    try:
+        result = runs.wait(tid, FACTORY_ASSISTANT_ID, input={"text": "hi"})
+        assert isinstance(result, dict), result
+        assert result.get("text") == "hi echoed"
+        assert result.get("access_context") == "threads.create_run"
+        assert result.get("is_for_execution") is True
+    finally:
+        threads.delete(tid)