RunnableAssign leaves mapper work running after passthrough input validation fails

Submission checklist

• This is a bug, not a usage question.
• I added a clear and descriptive title that summarizes this issue.
• I used the GitHub search to find a similar question and didn't find it.
• I am sure that this is a bug in LangChain rather than my code.
• The bug is not resolved by updating to the latest stable version of LangChain (or the specific integration package).
• This is not related to the langchain-community package.
• I posted a self-contained, minimal, reproducible example. A maintainer can copy it and run it AS IS.

Package (Required)

• langchain
• langchain-openai
• langchain-anthropic
• langchain-classic
• langchain-core
• langchain-model-profiles
• langchain-tests
• langchain-text-splitters
• langchain-chroma
• langchain-deepseek
• langchain-exa
• langchain-fireworks
• langchain-groq
• langchain-huggingface
• langchain-mistralai
• langchain-nomic
• langchain-ollama
• langchain-openrouter
• langchain-perplexity
• langchain-qdrant
• langchain-xai
• Other / not sure / general

Related Issues / PRs

No response

Reproduction Steps / Example Code (Python)

import asyncio

from langchain_core.runnables import RunnableLambda, RunnablePassthrough


async def main() -> None:
    mapper_started = asyncio.Event()
    mapper_cancelled = False

    async def slow_mapper(value):
        nonlocal mapper_cancelled
        mapper_started.set()

        try:
            await asyncio.sleep(60)
            return "finished"
        except asyncio.CancelledError:
            mapper_cancelled = True
            raise

    runnable = RunnablePassthrough.assign(
        mapped=RunnableLambda(slow_mapper),
    )

    async def invalid_input():
        yield {"valid": "first"}
        await mapper_started.wait()

        # RunnablePassthrough.assign() only accepts dictionaries.
        yield "invalid input"

    try:
        async for _ in runnable.atransform(invalid_input()):
            pass
    except ValueError as error:
        print("input error:", error)

    await asyncio.sleep(0)

    current_task = asyncio.current_task()
    pending_tasks = [
        task
        for task in asyncio.all_tasks()
        if task is not current_task and not task.done()
    ]

    print("pending background tasks:", len(pending_tasks))
    print("mapper cancelled:", mapper_cancelled)

    # Cleanup for this reproduction script.
    for task in pending_tasks:
        task.cancel()

    await asyncio.gather(*pending_tasks, return_exceptions=True)


asyncio.run(main())

Error Message and Stack Trace (if applicable)

input error: The input to RunnablePassthrough.assign() must be a dict.
pending background tasks: 4
mapper cancelled: False

Description

RunnableAssign._atransform() starts the mapper in a background task before it finishes validating all passthrough input chunks.

If a later input chunk is not a dictionary, the passthrough loop raises ValueError. The method exits without cancelling and awaiting the already started mapper task.

The relevant control flow is:

first_map_chunk_task = asyncio.create_task(
    anext(map_output, None),
)

async for chunk in for_passthrough:
    if not isinstance(chunk, dict):
        msg = "The input to RunnablePassthrough.assign() must be a dict."
        raise ValueError(msg)

first_map_chunk = await first_map_chunk_task

### System Info

System Information
------------------
> OS:  Windows
> OS Version:  10.0.26200
> Python Version:  3.13.3 (main, May 30 2025, 05:37:00) [MSC v.1943 64 bit (AMD64)]

Package Information
-------------------
> langchain_core: 1.6.3
> langsmith: 0.12.5
> langchain_classic: 1.0.8
> langchain_huggingface: 1.2.2
> langchain_protocol: 0.0.19
> langchain_text_splitters: 1.1.2

Optional packages not installed
-------------------------------
> deepagents
> deepagents-cli

Other Dependencies
------------------
> anyio: 4.15.1
> distro: 1.9.0
> httpx: 0.28.1
> httpx2: 2.13.0
> huggingface-hub: 1.31.0
> jsonpatch: 1.33
> orjson: 3.12.0
> packaging: 26.3
> pydantic: 2.13.5
> pyyaml: 6.0.3
> requests: 2.34.2
> requests-toolbelt: 1.0.0
> sniffio: 1.3.1
> sqlalchemy: 2.0.54
> tenacity: 9.1.4
> tokenizers: 0.23.2
> typing-extensions: 4.16.0
> uuid-utils: 0.17.1
> websockets: 17.1
> xxhash: 4.0.1
> zstandard: 0.25.0

### Social handles (optional)

_No response_