"""Compat bridge: convert `AIMessageChunk` streams to protocol events.

The bridge trusts `AIMessageChunk.content_blocks` as the single
protocol view of any chunk.  That property runs the three-tier lookup
(`output_version == "v1"` short-circuit, registered translator, or
best-effort parsing) and returns a `list[ContentBlock]` for every
well-formed message — whether the provider is a registered partner, an
unregistered community model, or not tagged at all.

Per-chunk `content_blocks` output is a **delta slice**, not accumulated
state: providers in this ecosystem emit SSE-style chunks that each carry
their own increment.  The bridge therefore forwards each slice straight
through as a `content-block-delta` event, and accumulates per-index
state only so the final `content-block-finish` event can report a
finalized block (e.g. `tool_call_chunk` args parsed to a dict).

Lifecycle::

    message-start
      -> content-block-start   (first time each index is observed)
      -> content-block-delta*  (per chunk, carrying the slice)
      -> content-block-finish  (finalized block)
    -> message-finish

Public API:

- `chunks_to_events` / `achunks_to_events` — for live streams where
  chunks arrive over time.
- `message_to_events` / `amessage_to_events` — for replaying a finalized
  `AIMessage` (cache hit, checkpoint restore, graph-node return value)
  as a synthetic event lifecycle.
"""

from __future__ import annotations

import json
from typing import TYPE_CHECKING, Any, cast

from langchain_protocol.protocol import (
    ContentBlock,
    ContentBlockDeltaData,
    ContentBlockFinishData,
    ContentBlockStartData,
    FinalizedContentBlock,
    InvalidToolCall,
    MessageFinishData,
    MessageMetadata,
    MessagesData,
    MessageStartData,
    ReasoningContentBlock,
    ServerToolCall,
    ServerToolCallChunk,
    TextContentBlock,
    ToolCall,
    ToolCallChunk,
    UsageInfo,
)

from langchain_core.messages import AIMessageChunk, BaseMessage
from langchain_core.utils._merge import merge_dicts

if TYPE_CHECKING:
    from collections.abc import AsyncIterator, Iterator

    from langchain_protocol.protocol import (
        BlockDelta,
        BlockDeltaFields,
        ContentBlockDelta,
        DataDelta,
        ReasoningDelta,
        TextDelta,
    )

    from langchain_core.messages.ai import UsageMetadata
    from langchain_core.outputs import ChatGenerationChunk


CompatBlock = dict[str, Any]
"""Internal working type for a content block.

The bridge works with plain dicts internally because two separate but
structurally similar `ContentBlock` Unions exist — one in
`langchain_core.messages.content` (returned by `msg.content_blocks`),
one in `langchain_protocol.protocol` (the wire/event shape).  They are
not mypy-compatible despite being near-isomorphic.  Passing through
`dict[str, Any]` launders between them.  See `_to_protocol_block` for
the single seam where the laundering cast lives.
"""


# ---------------------------------------------------------------------------
# Type laundering between core and protocol `ContentBlock` unions
# ---------------------------------------------------------------------------


def _to_protocol_block(block: CompatBlock) -> ContentBlock:
    """Narrow an internal working dict to a protocol `ContentBlock`.

    Single seam between the two `ContentBlock` type systems:
    `langchain_core.messages.content` (what `msg.content_blocks`
    returns) and `langchain_protocol.protocol` (what event payloads
    require).  The two Unions overlap structurally but are nominally
    distinct to mypy, so we launder through `dict[str, Any]`.  When the
    Unions are unified, this helper and its finalized counterpart can be
    deleted.
    """
    return cast("ContentBlock", block)


def _to_finalized_block(block: CompatBlock) -> FinalizedContentBlock:
    """Counterpart of `_to_protocol_block` for finalized blocks."""
    return cast("FinalizedContentBlock", block)


def _to_block_delta_fields(block: CompatBlock) -> BlockDeltaFields:
    """Narrow an internal working dict to protocol block-delta fields."""
    return cast("BlockDeltaFields", block)


def _to_content_delta(block: CompatBlock) -> ContentBlockDelta:
    """Convert a content-block slice/snapshot to an explicit protocol delta."""
    btype = block.get("type")
    if btype == "text":
        return cast("TextDelta", {"type": "text-delta", "text": block.get("text", "")})
    if btype == "reasoning":
        return cast(
            "ReasoningDelta",
            {
                "type": "reasoning-delta",
                "reasoning": block.get("reasoning", ""),
            },
        )
    if "data" in block:
        delta = cast("DataDelta", {"type": "data-delta", "data": block.get("data", "")})
        if block.get("encoding") == "base64":
            delta["encoding"] = "base64"
        return delta
    return cast(
        "BlockDelta",
        {
            "type": "block-delta",
            "fields": _to_block_delta_fields(block),
        },
    )


# ---------------------------------------------------------------------------
# Block iteration
# ---------------------------------------------------------------------------


def _iter_protocol_blocks(msg: BaseMessage) -> list[tuple[Any, CompatBlock]]:
    """Read per-chunk protocol blocks from `msg.content_blocks`.

    Returns `(key, block)` pairs.  The key is the block's stable identifier
    across the stream: the block's `index` field when present (can be an
    int or a string — some providers use string identifiers like
    `"lc_rs_305f30"`), or the positional index within the message as a
    fallback.  Callers are responsible for allocating wire-level `uint`
    indices; this helper only surfaces the source-side identity.

    For finalized `AIMessage`, also surfaces `invalid_tool_calls`
    — which `AIMessage.content_blocks` currently omits from its return
    value even though they are a defined protocol block type.

    The positional fallback is a known fragility: when a provider emits
    blocks without an `index` field (e.g. Anthropic's `_stream` with
    `coerce_content_to_string=True`, where text chunks lose their
    source-side index), every such chunk gets positional key 0 and
    successive chunks merge into one block. This works correctly for
    single-type streams (pure-text responses merge cleanly) because all
    chunks share the same key and the open-block logic collapses them.
    It would miscategorise a stream that mixed indexed structured
    blocks with non-indexed coerced-text blocks, since an indexed
    block with `index == 0` would collide with the anonymous text
    block's positional-0 key.  In the anthropic integration this
    cannot currently occur: coerce-to-string mode is only selected
    when no tools, thinking, or documents are present, and any of
    those flips the stream to structured mode where every block
    carries an integer index.  A native `_stream_chat_model_events`
    hook per provider (or a bridge-level "continue the open block when
    the source has no identity" rule) would close the gap if another
    integration ever emits mixed content.
    """
    try:
        raw = msg.content_blocks
    except Exception:
        return []

    result: list[tuple[Any, CompatBlock]] = []
    for i, block in enumerate(raw):
        if not isinstance(block, dict):
            continue  # type: ignore[unreachable]
        explicit_idx = block.get("index")
        if explicit_idx is None:
            # No source-side identity. Bucket by (sentinel, block type,
            # positional `i`) so two blocks of different types at the
            # same position across chunks (e.g. Gemini emitting a
            # reasoning block in one chunk and a `tool_call` in the
            # next, both at positional 0 because each chunk carries one
            # block) get distinct wire blocks. Without this, the second
            # type's incoming block hits `_accumulate`'s self-contained
            # `else` branch and clobbers the first. Same-type chunks
            # still share the bucket and merge cleanly, which is what
            # streaming text / reasoning relies on.
            key: Any = ("__lc_no_index__", block.get("type"), i)
        else:
            key = explicit_idx
        result.append((key, dict(block)))

    if not isinstance(msg, AIMessageChunk):
        # Finalized AIMessage: pull invalid_tool_calls from the dedicated
        # field — AIMessage.content_blocks does not currently include them.
        for itc in getattr(msg, "invalid_tool_calls", None) or []:
            itc_block: CompatBlock = {"type": "invalid_tool_call"}
            for key_name in ("id", "name", "args", "error"):
                if itc.get(key_name) is not None:
                    itc_block[key_name] = itc[key_name]
            result.append((len(result), itc_block))

    return result


# ---------------------------------------------------------------------------
# Per-block helpers
# ---------------------------------------------------------------------------


# Fields that can carry large payloads (inline base64 media, parsed args,
# arbitrary dicts).  Stripped from `content-block-start` for self-contained
# block types so the payload rides on `content-block-finish` alone instead
# of being serialized twice on the wire.
_HEAVY_FIELDS = frozenset({"args", "data", "output", "transcript", "value"})


def _start_skeleton(block: CompatBlock) -> ContentBlock:
    """Empty-content placeholder for the `content-block-start` event.

    Deltaable block types (text, reasoning, the `_chunk` tool variants)
    get an empty payload so the lifecycle's "start" signal is distinct
    from the first incremental delta.  Self-contained types (image,
    audio, video, file, non_standard, finalized tool calls) drop their
    heavy payload fields; those are carried by `content-block-finish`.
    Correlation fields (id, name, toolCallId) and small metadata
    (mime_type, url, status, …) are preserved on the start event.
    """
    btype = block.get("type", "text")
    if btype == "text":
        return TextContentBlock(type="text", text="")
    if btype == "reasoning":
        return ReasoningContentBlock(type="reasoning", reasoning="")
    if btype == "tool_call_chunk":
        return ToolCallChunk(
            type="tool_call_chunk",
            id=block.get("id"),
            name=block.get("name"),
            args="",
        )
    if btype == "server_tool_call_chunk":
        s_skel = ServerToolCallChunk(
            type="server_tool_call_chunk",
            args="",
        )
        if block.get("id") is not None:
            s_skel["id"] = block["id"]
        if block.get("name") is not None:
            s_skel["name"] = block["name"]
        return s_skel

    stripped: CompatBlock = {k: v for k, v in block.items() if k not in _HEAVY_FIELDS}
    # Restore required-but-heavy fields with minimal placeholders so the
    # start event still validates against the CDDL shape of the block type.
    if btype in {"tool_call", "server_tool_call"}:
        stripped["args"] = {}
    elif btype == "non_standard":
        stripped["value"] = {}
    return _to_protocol_block(stripped)


def _should_emit_delta(block: CompatBlock) -> bool:
    """Whether a per-chunk block carries content worth a delta event.

    Deltaable types emit only when they have fresh content.  Self-contained
    / already-finalized types skip the delta entirely — the `finish`
    event carries them.
    """
    btype = block.get("type")
    if btype == "text":
        return bool(block.get("text"))
    if btype == "reasoning":
        return bool(block.get("reasoning"))
    if btype in {"tool_call_chunk", "server_tool_call_chunk"}:
        return bool(
            block.get("args") or block.get("id") or block.get("name"),
        )
    if "data" in block:
        return bool(block.get("data"))
    return False


def _accumulate(state: CompatBlock | None, delta: CompatBlock) -> CompatBlock:
    """Merge a per-chunk delta slice into accumulated per-index state.

    Used only for the finalization pass — live delta events are emitted
    directly from the per-chunk block, without round-tripping through
    accumulated state.
    """
    if state is None:
        return dict(delta)
    btype = state.get("type")
    dtype = delta.get("type")
    if btype == "text" and dtype == "text":
        state["text"] = state.get("text", "") + delta.get("text", "")
        # Providers may send non-text fields (like `id`, or annotations)
        # on later deltas. Merging (not replacing) keeps earlier keys
        # intact while picking up these late-arriving fields.
        for key, value in delta.items():
            if key in {"type", "text"} or value is None:
                continue
            if key == "extras" and isinstance(value, dict):
                state["extras"] = {**(state.get("extras") or {}), **value}
            else:
                state[key] = value
    elif btype == "reasoning" and dtype == "reasoning":
        state["reasoning"] = state.get("reasoning", "") + delta.get("reasoning", "")
        # Providers may ship non-text fields on later deltas. Claude's
        # `signature_delta` arrives after the reasoning text, surfaced
        # as `extras.signature`; merging (not replacing) keeps earlier
        # keys intact.
        for key, value in delta.items():
            if key in {"type", "reasoning"} or value is None:
                continue
            if key == "extras" and isinstance(value, dict):
                state["extras"] = {**(state.get("extras") or {}), **value}
            else:
                state[key] = value
    elif btype in {"tool_call_chunk", "server_tool_call_chunk"} and dtype == btype:
        state["args"] = (state.get("args", "") or "") + (delta.get("args") or "")
        if delta.get("id") is not None:
            state["id"] = delta["id"]
        if delta.get("name") is not None:
            state["name"] = delta["name"]
    elif btype == dtype and "data" in delta:
        state["data"] = (state.get("data", "") or "") + (delta.get("data") or "")
        for key, value in delta.items():
            if key in {"type", "data"} or value is None:
                continue
            if key == "extras" and isinstance(value, dict):
                state["extras"] = {**(state.get("extras") or {}), **value}
            else:
                state[key] = value
    else:
        # Self-contained or already-finalized types: replace wholesale.
        state.clear()
        state.update(delta)
    return state


def finalize_tool_call_chunk(
    *,
    raw_args: str | None,
    id_: str | None,
    name: str | None,
    extras: dict[str, Any],
    finalized_type: str,
) -> FinalizedContentBlock:
    """Parse accumulated tool-chunk args into a finalized block.

    Shared between the compat bridge's `_finalize_block` and the
    `ChatModelStream` end-of-stream sweep. Parses `raw_args` as JSON:
    on success builds the requested finalized type (`tool_call` or
    `server_tool_call`) with provider-specific fields (`extras`)
    preserved; on failure falls back to `invalid_tool_call` carrying
    the raw string so downstream consumers can still introspect the
    malformed payload.

    Args:
        raw_args: Accumulated partial-JSON string; `None` or empty
            treated as `{}`.
        id_: Tool-call id collected across chunks.
        name: Tool name collected across chunks.
        extras: Provider-specific fields to carry onto the finalized
            block. Callers are responsible for having already dropped
            keys they don't want propagated (notably `type`, `id`,
            `name`, `args`, and `index` on client-side `tool_call`).
        finalized_type: `"tool_call"` or `"server_tool_call"`.

    Returns:
        A `ToolCall`, `ServerToolCall`, or `InvalidToolCall` — the
        latter when `raw_args` is non-empty but not valid JSON.
    """
    raw = raw_args or "{}"
    try:
        parsed = json.loads(raw) if raw else {}
    except (json.JSONDecodeError, TypeError):
        invalid = InvalidToolCall(
            type="invalid_tool_call",
            id=id_,
            name=name,
            args=raw,
            error="Failed to parse tool call arguments as JSON",
        )
        invalid.update(extras)  # type: ignore[typeddict-item]
        return invalid
    if finalized_type == "tool_call":
        finalized_tc = ToolCall(
            type="tool_call",
            id=id_ or "",
            name=name or "",
            args=parsed,
        )
        finalized_tc.update(extras)  # type: ignore[typeddict-item]
        return finalized_tc
    finalized_stc = ServerToolCall(
        type="server_tool_call",
        id=id_ or "",
        name=name or "",
        args=parsed,
    )
    finalized_stc.update(extras)  # type: ignore[typeddict-item]
    return finalized_stc


def _finalize_block(block: CompatBlock) -> FinalizedContentBlock:
    """Promote chunk variants to their finalized form.

    `tool_call_chunk` becomes `tool_call` — or `invalid_tool_call`
    if the accumulated `args` don't parse as JSON.
    `server_tool_call_chunk` becomes `server_tool_call` under the same
    rule.  Everything else passes through: text/reasoning blocks carry
    their accumulated snapshot, and self-contained types are already in
    their terminal shape.
    """
    btype = block.get("type")
    if btype in {"tool_call_chunk", "server_tool_call_chunk"}:
        # Carry provider-specific fields from the accumulated chunk onto
        # the finalized block. Drop the chunk-only keys we rewrite
        # explicitly. `index` is stripped on client-side
        # `tool_call` / `invalid_tool_call` finalizations to match v1
        # (`AIMessage.init_tool_calls` rebuilds tool_call blocks without
        # `index`), preventing `merge_lists` from re-merging further
        # chunks into an already-parsed args dict. `server_tool_call`
        # retains `index` because v1's `init_server_tool_calls`
        # finalizes in-place and preserves it.
        client_tool_call = btype == "tool_call_chunk"
        extras_drop = {"type", "id", "name", "args"}
        if client_tool_call:
            extras_drop |= {"index"}
        extras = {
            k: v for k, v in block.items() if k not in extras_drop and v is not None
        }
        return finalize_tool_call_chunk(
            raw_args=block.get("args"),
            id_=block.get("id"),
            name=block.get("name"),
            extras=extras,
            finalized_type="tool_call" if client_tool_call else "server_tool_call",
        )
    return _to_finalized_block(block)


# ---------------------------------------------------------------------------
# Metadata, usage, finish-reason
# ---------------------------------------------------------------------------


def _extract_start_metadata(response_metadata: dict[str, Any]) -> MessageMetadata:
    """Pull provider/model hints for the `message-start` event."""
    metadata: MessageMetadata = {}
    if "model_provider" in response_metadata:
        metadata["provider"] = response_metadata["model_provider"]
    if "model_name" in response_metadata:
        metadata["model"] = response_metadata["model_name"]
    return metadata


def _accumulate_usage(current: UsageInfo | None, delta: UsageMetadata) -> UsageInfo:
    """Sum usage counts and merge detail dicts across chunks.

    `delta` is a chunk's `usage_metadata`; `current` is the running total.
    Both sides are read and written by literal key so the typed shape is
    preserved end to end — no `dict[str, Any]` detour.
    """
    new: UsageInfo = current if current is not None else {}
    if "input_tokens" in delta:
        new["input_tokens"] = new.get("input_tokens", 0) + delta["input_tokens"]
    if "output_tokens" in delta:
        new["output_tokens"] = new.get("output_tokens", 0) + delta["output_tokens"]
    if "total_tokens" in delta:
        new["total_tokens"] = new.get("total_tokens", 0) + delta["total_tokens"]
    input_details = delta.get("input_token_details")
    if input_details:
        merged_input = new.get("input_token_details", {})
        merged_input.update(input_details)
        new["input_token_details"] = merged_input
    output_details = delta.get("output_token_details")
    if output_details:
        merged_output = new.get("output_token_details", {})
        merged_output.update(output_details)
        new["output_token_details"] = merged_output
    return new


def _isolate_usage(usage: UsageInfo | None) -> UsageInfo | None:
    """Copy usage for the event so consumers can't mutate the source message.

    The replay path (`message_to_events`) feeds the live `msg.usage_metadata`,
    so the emitted event must not share its dicts: copy the top level plus the
    nested `input_token_details` / `output_token_details` to de-alias it. The
    streaming accumulator already owns the dicts it builds, so the copy is a
    harmless no-op on that path.
    """
    if not usage:
        return None
    result: UsageInfo = usage.copy()
    input_details = result.get("input_token_details")
    if input_details is not None:
        result["input_token_details"] = input_details.copy()
    output_details = result.get("output_token_details")
    if output_details is not None:
        result["output_token_details"] = output_details.copy()
    return result


# ---------------------------------------------------------------------------
# Event builders
# ---------------------------------------------------------------------------


def _build_message_start(
    msg: BaseMessage,
    message_id: str | None,
) -> MessageStartData:
    start_data = MessageStartData(event="message-start", role="ai", id="")
    resolved_id = message_id if message_id is not None else getattr(msg, "id", None)
    if resolved_id:
        start_data["id"] = resolved_id
    start_metadata = _extract_start_metadata(msg.response_metadata or {})
    if start_metadata:
        start_data["metadata"] = start_metadata
    return start_data


def _build_message_finish(
    *,
    usage: UsageInfo | None,
    response_metadata: dict[str, Any] | None,
    additional_kwargs: dict[str, Any] | None = None,
) -> MessageFinishData:
    # Protocol 0.0.9 removed the top-level `reason` field from
    # `MessageFinishData`; the provider's raw `finish_reason` /
    # `stop_reason` now rides inside `metadata` alongside other
    # response metadata. Pass it through unchanged.
    finish_data: dict[str, Any] = {"event": "message-finish"}
    usage_info = _isolate_usage(usage)
    if usage_info is not None:
        finish_data["usage"] = usage_info
    if response_metadata:
        finish_data["metadata"] = dict(response_metadata)
    # `additional_kwargs` is an off-spec extension on the message-finish
    # event (parallel to `metadata`, which `MessageFinishData` also doesn't
    # formally declare but the consumer reads). It carries provider-side
    # kwargs that don't map onto a typed protocol field — notably Gemini's
    # `__gemini_function_call_thought_signatures__`, which the model
    # requires on follow-up turns to replay prior thinking. Without this,
    # streaming-assembled messages would silently drop data that
    # `ainvoke` preserves, breaking multi-turn streaming flows.
    if additional_kwargs:
        finish_data["additional_kwargs"] = dict(additional_kwargs)
    return cast("MessageFinishData", finish_data)


def _finalize_and_build_finish(
    wire_idx: int,
    block: CompatBlock,
) -> MessagesData:
    """Finalize a block and wrap it in a `content-block-finish` event."""
    return ContentBlockFinishData(
        event="content-block-finish",
        index=wire_idx,
        content=_finalize_block(block),
    )


# ---------------------------------------------------------------------------
# Main generators
# ---------------------------------------------------------------------------


def chunks_to_events(
    chunks: Iterator[ChatGenerationChunk],
    *,
    message_id: str | None = None,
) -> Iterator[MessagesData]:
    """Convert a stream of `ChatGenerationChunk` to protocol events.

    Blocks are tracked independently by source-side identifier. Providers
    such as Anthropic can interleave parallel tool-call chunks by index, so
    each first-seen block gets a `content-block-start`, deltas keep their
    stable wire index, and all open blocks are finalized at message end.
    Source-side identifiers (from the block's `index` field, which may be
    int or string) are translated to sequential `uint` wire indices.

    Args:
        chunks: Iterator of `ChatGenerationChunk` from `_stream()`.
        message_id: Optional stable message ID.

    Yields:
        `MessagesData` lifecycle events.
    """
    started = False
    blocks: dict[Any, tuple[int, CompatBlock]] = {}
    next_wire_idx = 0
    usage: UsageInfo | None = None
    response_metadata: dict[str, Any] = {}
    additional_kwargs: dict[str, Any] = {}

    for chunk in chunks:
        msg = chunk.message
        if not isinstance(msg, AIMessageChunk):
            continue

        # The v1 `stream()` wrapper merges `generation_info` into
        # `response_metadata` before yielding (`chat_models.py` via
        # `_gen_info_and_msg_metadata`). We bypass that wrapper by reading
        # `_stream` directly, so reproduce the merge here with the same
        # priority: `generation_info` first, then `message.response_metadata`
        # overlays. This is how provider fields like `model_name`,
        # `system_fingerprint`, and `finish_reason` reach the bridge when
        # a provider emits them via `generation_info` instead of the
        # message's `response_metadata`.
        merged_rm: dict[str, Any] = {
            **(chunk.generation_info or {}),
            **(msg.response_metadata or {}),
        }
        if merged_rm:
            response_metadata.update(merged_rm)

        # Carry chunks' `additional_kwargs` through to the assembled
        # message. Provider-side fields that don't map onto a typed
        # protocol block (e.g. Gemini's per-tool-call thought signatures)
        # live here on non-streaming `ainvoke` results; dropping them on
        # the streaming path silently diverges multi-turn behavior. Use
        # `merge_dicts` because the same key can arrive in pieces across
        # chunks (e.g. an accumulating `function_call`), matching how
        # `AIMessageChunk` merges itself.
        if msg.additional_kwargs:
            additional_kwargs = merge_dicts(additional_kwargs, msg.additional_kwargs)

        if not started:
            started = True
            yield _build_message_start(msg, message_id)

        for key, block in _iter_protocol_blocks(msg):
            if key not in blocks:
                wire_idx = next_wire_idx
                next_wire_idx += 1
                blocks[key] = (wire_idx, dict(block))
                yield ContentBlockStartData(
                    event="content-block-start",
                    index=wire_idx,
                    content=_start_skeleton(block),
                )
            else:
                wire_idx, existing = blocks[key]
                blocks[key] = (wire_idx, _accumulate(existing, block))
            if _should_emit_delta(block):
                wire_idx, current = blocks[key]
                is_block_delta = block.get("type") in {
                    "tool_call_chunk",
                    "server_tool_call_chunk",
                }
                delta_source = current if is_block_delta else block
                yield ContentBlockDeltaData(
                    event="content-block-delta",
                    index=wire_idx,
                    delta=_to_content_delta(delta_source or block),
                )

        if msg.usage_metadata:
            usage = _accumulate_usage(usage, msg.usage_metadata)

    if not started:
        return

    for wire_idx, block in blocks.values():
        yield _finalize_and_build_finish(wire_idx, block)

    yield _build_message_finish(
        usage=usage,
        response_metadata=response_metadata,
        additional_kwargs=additional_kwargs,
    )


async def achunks_to_events(
    chunks: AsyncIterator[ChatGenerationChunk],
    *,
    message_id: str | None = None,
) -> AsyncIterator[MessagesData]:
    """Async variant of `chunks_to_events`."""
    started = False
    blocks: dict[Any, tuple[int, CompatBlock]] = {}
    next_wire_idx = 0
    usage: UsageInfo | None = None
    response_metadata: dict[str, Any] = {}
    additional_kwargs: dict[str, Any] = {}

    async for chunk in chunks:
        msg = chunk.message
        if not isinstance(msg, AIMessageChunk):
            continue

        # See sync twin for rationale: merge `generation_info` into the
        # accumulated `response_metadata` with the same priority as the
        # v1 `stream()` wrapper.
        merged_rm: dict[str, Any] = {
            **(chunk.generation_info or {}),
            **(msg.response_metadata or {}),
        }
        if merged_rm:
            response_metadata.update(merged_rm)

        # See sync twin: carry chunk `additional_kwargs` through so
        # provider-specific data (e.g. Gemini thought signatures) reaches
        # the assembled message instead of being dropped.
        if msg.additional_kwargs:
            additional_kwargs = merge_dicts(additional_kwargs, msg.additional_kwargs)

        if not started:
            started = True
            yield _build_message_start(msg, message_id)

        for key, block in _iter_protocol_blocks(msg):
            if key not in blocks:
                wire_idx = next_wire_idx
                next_wire_idx += 1
                blocks[key] = (wire_idx, dict(block))
                yield ContentBlockStartData(
                    event="content-block-start",
                    index=wire_idx,
                    content=_start_skeleton(block),
                )
            else:
                wire_idx, existing = blocks[key]
                blocks[key] = (wire_idx, _accumulate(existing, block))
            if _should_emit_delta(block):
                wire_idx, current = blocks[key]
                is_block_delta = block.get("type") in {
                    "tool_call_chunk",
                    "server_tool_call_chunk",
                }
                delta_source = current if is_block_delta else block
                yield ContentBlockDeltaData(
                    event="content-block-delta",
                    index=wire_idx,
                    delta=_to_content_delta(delta_source or block),
                )

        if msg.usage_metadata:
            usage = _accumulate_usage(usage, msg.usage_metadata)

    if not started:
        return

    for wire_idx, block in blocks.values():
        yield _finalize_and_build_finish(wire_idx, block)

    yield _build_message_finish(
        usage=usage,
        response_metadata=response_metadata,
        additional_kwargs=additional_kwargs,
    )


def message_to_events(
    msg: BaseMessage,
    *,
    message_id: str | None = None,
) -> Iterator[MessagesData]:
    """Replay a finalized message as a synthetic event lifecycle.

    For a message returned whole (from a graph node, checkpoint, or
    cache), produce the same `message-start` / per-block /
    `message-finish` event stream a live call would produce.  Consumers
    downstream see a uniform event shape regardless of source.

    Text and reasoning blocks emit a single `content-block-delta` with
    the full accumulated content.  Already-finalized blocks (tool_call,
    server_tool_call, image, etc.) skip the delta and rely on the
    `content-block-finish` event alone.

    Args:
        msg: The finalized message — typically an `AIMessage`.
        message_id: Optional stable message ID; falls back to `msg.id`.

    Yields:
        `MessagesData` lifecycle events.
    """
    response_metadata = msg.response_metadata or {}
    yield _build_message_start(msg, message_id)

    for wire_idx, (_key, block) in enumerate(_iter_protocol_blocks(msg)):
        yield ContentBlockStartData(
            event="content-block-start",
            index=wire_idx,
            content=_start_skeleton(block),
        )
        if _should_emit_delta(block):
            yield ContentBlockDeltaData(
                event="content-block-delta",
                index=wire_idx,
                delta=_to_content_delta(block),
            )
        yield ContentBlockFinishData(
            event="content-block-finish",
            index=wire_idx,
            content=_finalize_block(block),
        )

    yield _build_message_finish(
        usage=getattr(msg, "usage_metadata", None),
        response_metadata=response_metadata,
    )


async def amessage_to_events(
    msg: BaseMessage,
    *,
    message_id: str | None = None,
) -> AsyncIterator[MessagesData]:
    """Async variant of `message_to_events`."""
    for event in message_to_events(msg, message_id=message_id):
        yield event


__all__ = [
    "CompatBlock",
    "achunks_to_events",
    "amessage_to_events",
    "chunks_to_events",
    "finalize_tool_call_chunk",
    "message_to_events",
]
