PIIMiddleware(apply_to_output=True) blanks parameters on tool calls, leading to a GraphRecursion Error.

import time

from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware
from langchain.tools import tool
from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.messages import AIMessageChunk
from langchain_core.outputs import ChatGenerationChunk

ARGS = ["", '{"city"', ': "Bos', 'ton"}']  # {"city": "Boston"} split across chunks


class FakeStreamingModel(BaseChatModel):
    """Streams a tool call in chunks, paced like a real provider."""

    idx: int = 0

    @property
    def _llm_type(self) -> str:
        return "fake-streaming-model"

    def bind_tools(self, tools, **kwargs):
        return self

    def _stream(self, messages, stop=None, run_manager=None, **kwargs):
        i = self.idx
        object.__setattr__(self, "idx", i + 1)
        if i:  # second turn: plain text, so the run terminates
            yield ChatGenerationChunk(message=AIMessageChunk(content="Done."))
            return
        for n, part in enumerate(ARGS):
            # The pauses are required. The PII stream transformer runs on the
            # consumer side, so an instantaneous model assembles the whole
            # message before the mux pumps an event and the bug stays hidden.
            time.sleep(0.01)
            yield ChatGenerationChunk(
                message=AIMessageChunk(
                    content="",
                    tool_call_chunks=[
                        {
                            "name": "get_weather" if n == 0 else None,
                            "args": part,
                            "id": "call_1" if n == 0 else None,
                            "index": 0,
                        }
                    ],
                )
            )
        time.sleep(0.05)  # latency to simulate real provider

    def _generate(self, messages, stop=None, run_manager=None, **kwargs):
        raise NotImplementedError("unused: the v3 streaming path calls _stream")


@tool
def get_weather(city: str) -> str:
    """Get weather for a given city."""
    return f"It's always sunny in {city}!"


agent = create_agent(
    model=FakeStreamingModel(),  # replace with real
    tools=[get_weather],
    name="weather_agent",
    middleware=[PIIMiddleware("credit_card", strategy="hash", apply_to_output=True)],
).with_config(recursion_limit=25)

stream = agent.stream_events(
    {"messages": [{"role": "user", "content": "What's the weather in Boston?"}]},
    version="v3",
)
for tool_call in stream.tool_calls:
    print(tool_call.tool_name, tool_call.input)
    assert tool_call.input == {
        "city": "Boston"
    }, f"expected {{'city': 'Boston'}}, got {tool_call.input}"

N/A. This is a deterministic behavior bug rather than an exception path.

Actual behavior:
- tool_calls[*].args is emptied: the model emits {"city": "Boston"}, the agent executes {}
- the blanked args reach graph state, not just the streamed output, so the tool runs
  with no arguments and the ToolMessage comes back "city: Field required"
- this happens with no PII present anywhere in the run, under every strategy
- with a real provider the model cannot act on that error, retries the same call, and
  the run ends in GraphRecursionError