From 03b2a9fe5ff8adbd1f7c922fd2c5e9ebbfab7d17 Mon Sep 17 00:00:00 2001 From: Nick Hollon Date: Tue, 2 Jun 2026 12:38:33 -0400 Subject: [PATCH] test(sdk-py): add factory-graph integration test exercising the server factory path (#7978) --- .github/workflows/ci.yml | 6 +- libs/langgraph/langgraph/pregel/main.py | 25 ++++++- libs/langgraph/tests/test_runtime.py | 54 ++++++++++++++ libs/sdk-py/integration/Dockerfile | 13 ++++ libs/sdk-py/integration/docker-compose.yml | 13 +++- .../sdk-py/integration/graph/factory_graph.py | 70 +++++++++++++++++++ libs/sdk-py/integration/langgraph.json | 3 +- libs/sdk-py/tests/integration/conftest.py | 1 + .../tests/integration/test_factory_graph.py | 65 +++++++++++++++++ 9 files changed, 242 insertions(+), 8 deletions(-) create mode 100644 libs/sdk-py/integration/graph/factory_graph.py create mode 100644 libs/sdk-py/tests/integration/test_factory_graph.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 2626abb08..602322fdb 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/libs/langgraph/langgraph/pregel/main.py b/libs/langgraph/langgraph/pregel/main.py index 1550e5a92..52d8fe58f 100644 --- a/libs/langgraph/langgraph/pregel/main.py +++ b/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: diff --git a/libs/langgraph/tests/test_runtime.py b/libs/langgraph/tests/test_runtime.py index f1de3ba03..f2f2fcc23 100644 --- a/libs/langgraph/tests/test_runtime.py +++ b/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 diff --git a/libs/sdk-py/integration/Dockerfile b/libs/sdk-py/integration/Dockerfile index 4c5a3d04a..cd830b8e8 100644 --- a/libs/sdk-py/integration/Dockerfile +++ b/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 diff --git a/libs/sdk-py/integration/docker-compose.yml b/libs/sdk-py/integration/docker-compose.yml index 580139ce2..320607342 100644 --- a/libs/sdk-py/integration/docker-compose.yml +++ b/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) diff --git a/libs/sdk-py/integration/graph/factory_graph.py b/libs/sdk-py/integration/graph/factory_graph.py new file mode 100644 index 000000000..845290d00 --- /dev/null +++ b/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() diff --git a/libs/sdk-py/integration/langgraph.json b/libs/sdk-py/integration/langgraph.json index 53a34ee8c..23da8ca57 100644 --- a/libs/sdk-py/integration/langgraph.json +++ b/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": {} } diff --git a/libs/sdk-py/tests/integration/conftest.py b/libs/sdk-py/tests/integration/conftest.py index 283c72d43..29eea134e 100644 --- a/libs/sdk-py/tests/integration/conftest.py +++ b/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"] diff --git a/libs/sdk-py/tests/integration/test_factory_graph.py b/libs/sdk-py/tests/integration/test_factory_graph.py new file mode 100644 index 000000000..5ffd0d957 --- /dev/null +++ b/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)