LLMAdapter with a LangGraph checkpointer duplicates assistant history on the next turn

Start with the reproduction script's LLMAdapter setup and add_messages reducer path, then compare the IDs of the assistant message written by the graph with the one AgentSession records in session.history. Done means the first assistant reply appears once at each turn's node entry with a checkpointer enabled, matching the expected counts [0, 1, 1] in both stream modes. Confirm the fix against the messages and custom cases, which must both pass.

Problem

A graph using the add_messages reducer and InMemorySaver stores its generated AIMessage.
When the next AgentSession turn starts, LLMAdapter also supplies the transport's copy of that
assistant reply. The copies have different IDs, so add_messages retains both.

The duplicate is in the graph's input history, not duplicate audible playback. AgentSession
history contains one reply per turn, while the graph receives two copies of prior replies.
This may distort subsequent model context and increase input tokens; this reproduction does
not measure production cost or downstream tool effects.

Environment and reproduction

Reproduced on Windows with Python 3.14.7:

• langgraph 1.2.14, langgraph-checkpoint 4.2.0, langchain-core 1.6.9
• livekit-agents 1.8.5, livekit-plugins-langchain 1.8.5

Also reproduced with langgraph 1.2.7 and livekit-agents/livekit-plugins-langchain 1.6.4.

The script uses the official GenericFakeChatModel and a real, text-only AgentSession.
It needs no room, credentials, microphone, TTS provider, or model API calls.
Preemptive generation is not part of this reproduction.

Install the current-release packages listed above in a fresh environment and run this script:

"""Compare transport and checkpoint history using synthetic, local replies."""

import asyncio
import json
from importlib.metadata import version
from typing import Annotated, TypedDict

from langchain_core.language_models.fake_chat_models import GenericFakeChatModel
from langchain_core.messages import AIMessage, BaseMessage
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.config import get_stream_writer
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages
from livekit.agents import Agent, AgentSession
from livekit.plugins.langchain import LLMAdapter


class State(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]


async def run_case(checkpointed: bool, stream_mode: str) -> dict:
    observations = []
    model = GenericFakeChatModel(
        messages=iter(
            AIMessage(content=f"synthetic reply {n}", id=f"graph-reply-{n}") for n in range(1, 4)
        )
    )

    async def reply(state: State) -> dict:
        observations.append(
            [{"role": m.type, "id": m.id, "text": m.content} for m in state["messages"]]
        )
        response = await model.ainvoke(state["messages"])
        if stream_mode == "custom":
            get_stream_writer()(response.content)
        return {"messages": [response]}

    graph = (
        StateGraph(State)
        .add_node("reply", reply)
        .add_edge(START, "reply")
        .add_edge("reply", END)
        .compile(checkpointer=InMemorySaver() if checkpointed else None)
    )
    config = {"configurable": {"thread_id": "synthetic-session"}}
    session = AgentSession(llm=LLMAdapter(graph, config=config, stream_mode=stream_mode))
    await session.start(Agent(instructions=""), record=False)
    try:
        for n in range(1, 4):
            speech = session.generate_reply(user_input=f"synthetic turn {n}")
            await asyncio.wait_for(speech.wait_for_playout(), timeout=15)
        transport = [
            {"role": item.role, "id": item.id, "text": item.text_content}
            for item in session.history.items
            if item.type == "message" and item.role in {"user", "assistant"}
        ]
        return {
            "checkpointed": checkpointed,
            "stream_mode": stream_mode,
            "reply_1_copies_at_node_entry": [
                sum(m["text"] == "synthetic reply 1" for m in messages) for messages in observations
            ],
            "node_inputs": observations,
            "transport_history": transport,
        }
    finally:
        await session.aclose()


async def main() -> None:
    packages = (
        "langgraph",
        "langgraph-checkpoint",
        "langchain-core",
        "livekit-agents",
        "livekit-plugins-langchain",
    )
    print(json.dumps({"versions": {name: version(name) for name in packages}}))
    for mode in ("messages", "custom"):
        for checkpointed in (False, True):
            print(json.dumps(await run_case(checkpointed, mode)))


if __name__ == "__main__":
    asyncio.run(main())

Expected and observed

At entry to turns 1, 2, and 3, the first assistant reply should occur 0, 1, and 1 times.

On the third turn, the second reply is also present twice. The user-message IDs remain aligned.
The script prints both graph inputs and session history, including IDs, so the distinction is
visible without relying on an assertion or a real provider.

Relation to previous work

PR #2767 preserved the IDs of assistant, user, and
system messages when converting transport history. This reproduction is a different remaining
identity problem: the assistant message already written by the graph has a different ID from the
assistant message recorded by AgentSession.

Targeted GitHub issue/PR searches for langchain duplication and checkpointer usage did not find
a report of this exact mechanism. That search is not proof that no related discussion exists.

Could the integration preserve a common assistant-message identity across these two histories,
or document the supported history-ownership pattern when using a checkpointer?