deerflow-code/offline-backend-20260512/backend/packages/harness/deerflow/models/patched_openai.py

"""Patched ChatOpenAI that preserves thought_signature for Gemini thinking models.

When using Gemini with thinking enabled via an OpenAI-compatible gateway (e.g.
Vertex AI, Google AI Studio, or any proxy), the API requires that the
``thought_signature`` field on tool-call objects is echoed back verbatim in
every subsequent request.

The OpenAI-compatible gateway stores the raw tool-call dicts (including
``thought_signature``) in ``additional_kwargs["tool_calls"]``, but standard
``langchain_openai.ChatOpenAI`` only serialises the standard fields (``id``,
``type``, ``function``) into the outgoing payload, silently dropping the
signature.  That causes an HTTP 400 ``INVALID_ARGUMENT`` error:

    Unable to submit request because function call `<tool>` in the N. content
    block is missing a `thought_signature`.

This module fixes the problem by overriding ``_get_request_payload`` to
re-inject thinking metadata back into the outgoing payload for any assistant
message that originally carried it:

- ``thought_signature`` for Gemini-compatible tool calls.
- ``reasoning_content`` and the newer ``reasoning`` spelling used by vLLM and
  other OpenAI-compatible providers.  Both are normalized to
  ``additional_kwargs.reasoning_content`` for DeerFlow's existing clients.

Some OpenAI-compatible gateways (notably intranet DeepSeek) re-send the entire
``function.arguments`` JSON on every streamed chunk instead of a true delta.
LangChain concatenates those snapshots, so ``write_file`` JSON becomes invalid
until the final message and the sandbox can only refresh once.  This adapter
rewrites growing snapshots into suffix deltas before LangChain merges them.

For gateways that buffer ``write_file.content`` completely, the opt-in
``write_file_transport=buffered_gateway`` adapter keeps the original
model-facing and graph-facing ``write_file`` contract. It first obtains only
the target metadata, then requests the body over an isolated ordinary-content
stream and exposes it downstream as standard incremental ``write_file``
argument chunks. Provider reasoning remains separate from file bytes; no extra
model-authored JSON or delimiter is required. The earlier
``artifact_renderer`` handshake and ``text_bridge`` mode remain available for
configuration compatibility. Unconfigured models stay on the native path.
"""

from __future__ import annotations

import asyncio
import contextvars
import json
import logging
import re
import time
import uuid
import warnings
from collections.abc import AsyncIterator, Iterator, Mapping
from dataclasses import dataclass, field
from typing import Any, Literal

from langchain_core.language_models import LanguageModelInput
from langchain_core.messages import AIMessage, AIMessageChunk, ToolMessage
from langchain_core.outputs import ChatGeneration, ChatGenerationChunk, ChatResult
from langchain_openai import ChatOpenAI

from deerflow.models.chat_stream_state import ChatStreamState, needs_qwen_flash_next_integrity
from deerflow.models.dsml_tool_calls import DSMLToolCallRecovery
from deerflow.models.message_ordering import normalize_chat_payload_messages
from deerflow.models.model_exchange_log import (
    bind_exchange,
    finish_exchange,
    finish_owned_exchange,
    open_exchange,
    remember_final_message,
    remember_provider_chunk,
    remember_raw_response,
    remember_request,
    unbind_exchange,
)
from deerflow.utils.stream_text import InlineThinkTagFilter

logger = logging.getLogger(__name__)


_SAFE_FINISH_REASONS = frozenset(
    {
        "stop",
        "length",
        "tool_calls",
        "function_call",
        "content_filter",
        "end_turn",
    }
)

_TOOL_SIGNATURES_KEY = "_deerflow_tool_call_signatures"
_PYDANTIC_SERIALIZER_WARNING = r"^Pydantic serializer warnings:"
_WRITE_FILE_TOOL_NAME = "write_file"
_BEGIN_WRITE_FILE_TOOL_NAME = "begin_write_file"
_WRITE_FILE_BRIDGE_CALL_PREFIX = "call_deerflow_wfbridge_"
_WRITE_FILE_BRIDGE_INCOMPLETE = "__deerflow_stream_not_committed__"
_ARTIFACT_RENDERER_PREREQUISITE_TOOLS = frozenset(
    {
        # These tools can gather source material without becoming an alternate
        # file-writing path.  If a renderer discovers that evidence is still
        # missing, the current body transaction is deferred and LangGraph runs
        # the ordinary tool turn before a fresh prepare/body transaction.
        "glob",
        "grep",
        "ls",
        "read_file",
    }
)

_TEXT_BRIDGE_SELECTION_INSTRUCTION = """<write_file_text_bridge>
This model uses DeerFlow's compatibility transport for file creation.
The real `write_file` function is intentionally hidden from this request.
Whenever an instruction requires `write_file`, call `begin_write_file` with only
the final absolute path, a short description, and the same `append` intent you
would pass to `write_file`. Do not put file content in that tool call. Call it
only when a file really must be created, overwritten, or appended; ordinary
conversational answers must remain ordinary assistant text. Call
`begin_write_file` by itself, never in parallel with another tool. After it
succeeds, the next model invocation will request the exact file body.
</write_file_text_bridge>"""

_TEXT_BRIDGE_BODY_INSTRUCTION = """<write_file_text_bridge_body>
The previous `begin_write_file` call was accepted. Output the complete file body
for the prepared path now. Output the file bytes as text only: no preface, no
explanation, and no surrounding Markdown fence unless that fence is itself part
of the file. Do not call any tool. Preserve all intended leading/trailing
whitespace and finish the whole file in this response. Keep private reasoning in
the provider's reasoning channel, never in the file body. If this provider can
only emit inline reasoning, exactly one leading `<think>...</think>` envelope is
treated as reasoning before file bytes begin.
</write_file_text_bridge_body>"""

_ARTIFACT_RENDERER_SELECTION_INSTRUCTION = """<write_file_artifact_renderer>
This model uses DeerFlow's compatibility renderer for file creation.
The provider-facing `write_file` in this request is a metadata-only preparation
control: its schema intentionally has no `content` field. Whenever an instruction
requires a file, call it with only the final absolute path, a short description,
and the intended `append` value. DeerFlow translates that call into an internal
prepare operation and requests the body separately. Call it only when a file
really must be created, overwritten, or appended; ordinary conversational
answers must remain ordinary assistant text. Call it by itself, never in
parallel with another tool.
</write_file_artifact_renderer>"""

_BUFFERED_GATEWAY_SELECTION_INSTRUCTION = """<write_file_buffered_gateway>
This deployment keeps the standard DeerFlow `write_file` capability but uses a
gateway that buffers long tool arguments. In this selection request, the
advertised `write_file` is intentionally metadata-only. When a file is needed,
call it with the final absolute path, a short description, and the intended
append value. DeerFlow will obtain the body through an isolated ordinary-text
stream and submit one complete standard `write_file` call. Do not emit file
content in this request. Call `write_file` only when a file is genuinely needed
and never in parallel with another tool; ordinary answers remain chat text.
</write_file_buffered_gateway>"""

_ARTIFACT_RENDERER_SYSTEM_INSTRUCTION = """You are DeerFlow's internal artifact renderer.
This is an isolated file-body transaction, not a chat response. Use the preceding
conversation and tool evidence as source material for the prepared artifact. The
ordinary assistant content channel is the exact file body: emit the complete file
bytes directly, with no JSON envelope, preface, explanation, or completion
summary. Markdown and code fences that belong in the requested file must be
emitted normally as part of the body. Keep private reasoning in the provider
reasoning channel. Complete the entire file in this response. If essential source
material is still missing, call exactly one of the explicitly advertised
prerequisite tools instead and emit no file-body text; DeerFlow will resume the
normal agent loop and start a fresh prepared transaction afterward."""

WriteFileTransport = Literal[
    "native",
    "buffered_gateway",
    "artifact_renderer",
    "text_bridge",
]

_WRITE_FILE_BRIDGE_TRANSPORTS = frozenset(
    {"buffered_gateway", "artifact_renderer", "text_bridge"}
)
_ARTIFACT_BODY_TRANSPORTS = frozenset(
    {"buffered_gateway", "artifact_renderer"}
)


class ArtifactRendererTransientProtocolError(RuntimeError):
    """A configured renderer stream ended before a safe tool/file commit.

    No sandbox mutation has happened when this is raised, so the LLM retry
    middleware may safely re-run the provider call.  Deterministic invariant
    violations (different path, conflicting append intent, parallel side
    effects) continue to use ordinary ``RuntimeError`` and are not retried.
    """


# OpenAI's generated Pydantic model may accept non-standard structured tool
# arguments through its no-validation response parser, then include the entire
# argument object in a serializer warning before our converter can normalize
# it. Install one process-wide, narrowly scoped rule rather than holding a
# ``warnings.catch_warnings`` context across network awaits (which is not
# concurrency-safe). ``structured_arg_chunks`` remains the safe diagnostic.
def _install_structured_arg_warning_filter() -> None:
    warnings.filterwarnings(
        "ignore",
        message=_PYDANTIC_SERIALIZER_WARNING,
        category=UserWarning,
        module=r"pydantic\.main",
    )


_install_structured_arg_warning_filter()


def _extract_tool_call_signature_records(raw_tool_calls: Any) -> list[dict[str, Any]]:
    """Extract only opaque tool signatures plus bounded association metadata.

    LangChain drops provider extension fields while converting OpenAI response
    objects into ``AIMessage`` instances. Keep the minimum metadata needed to
    echo Gemini-compatible thought signatures on the next request. Tool names,
    arguments, response content, and other raw provider fields are intentionally
    not copied into this private checkpoint field.
    """
    if not isinstance(raw_tool_calls, list):
        return []

    records: list[dict[str, Any]] = []
    for position, raw_tool_call in enumerate(raw_tool_calls):
        if not isinstance(raw_tool_call, Mapping):
            continue

        candidates: list[tuple[str, Any]] = [
            ("top_snake", raw_tool_call.get("thought_signature")),
            ("top_camel", raw_tool_call.get("thoughtSignature")),
        ]
        for extra_key, extra_label in (
            ("extra_content", "extra_snake"),
            ("extraContent", "extra_camel"),
        ):
            extra_content = raw_tool_call.get(extra_key)
            google = extra_content.get("google") if isinstance(extra_content, Mapping) else None
            if isinstance(google, Mapping):
                candidates.extend(
                    [
                        (f"{extra_label}_google_snake", google.get("thought_signature")),
                        (f"{extra_label}_google_camel", google.get("thoughtSignature")),
                    ]
                )

        location = ""
        signature = None
        for candidate_location, candidate_signature in candidates:
            if isinstance(candidate_signature, str) and candidate_signature:
                location = candidate_location
                signature = candidate_signature
                break
        if signature is None:
            continue

        raw_index = raw_tool_call.get("index")
        try:
            index = int(raw_index) if raw_index is not None else position
        except (TypeError, ValueError):
            index = position
        record: dict[str, Any] = {
            # Do not name this field ``index``: LangChain's chunk merger treats
            # list items with that key as stream fragments and concatenates
            # their string values, corrupting repeated opaque signatures.
            "tool_index": index,
            "location": location,
            "signature": signature,
        }
        call_id = raw_tool_call.get("id")
        if isinstance(call_id, str) and call_id:
            record["id"] = call_id
        records.append(record)
    return records


def _append_tool_call_signature_metadata(
    message: AIMessage | AIMessageChunk,
    records: list[dict[str, Any]],
) -> AIMessage | AIMessageChunk:
    if not records:
        return message
    additional_kwargs = dict(message.additional_kwargs)
    existing = additional_kwargs.get(_TOOL_SIGNATURES_KEY)
    combined = [dict(record) for record in existing if isinstance(record, Mapping)] if isinstance(existing, list) else []
    combined.extend(records)
    additional_kwargs[_TOOL_SIGNATURES_KEY] = combined
    return message.model_copy(update={"additional_kwargs": additional_kwargs})


def _restore_signature_at_location(payload_tool_call: dict[str, Any], record: Mapping[str, Any]) -> None:
    signature = record.get("signature")
    location = record.get("location")
    if not isinstance(signature, str) or not signature:
        return
    if location == "top_snake":
        payload_tool_call["thought_signature"] = signature
        return
    if location == "top_camel":
        payload_tool_call["thoughtSignature"] = signature
        return

    nested_locations = {
        "extra_snake_google_snake": ("extra_content", "thought_signature"),
        "extra_snake_google_camel": ("extra_content", "thoughtSignature"),
        "extra_camel_google_snake": ("extraContent", "thought_signature"),
        "extra_camel_google_camel": ("extraContent", "thoughtSignature"),
    }
    nested_path = nested_locations.get(location)
    if nested_path is None:
        return
    extra_key, signature_key = nested_path
    extra_content = payload_tool_call.get(extra_key)
    if not isinstance(extra_content, dict):
        extra_content = {}
        payload_tool_call[extra_key] = extra_content
    google = extra_content.get("google")
    if not isinstance(google, dict):
        google = {}
        extra_content["google"] = google
    google[signature_key] = signature


def _safe_metric_name(value: Any, *, fallback: str, limit: int = 80) -> str:
    """Return a bounded, low-cardinality value suitable for structured logs."""
    text = str(value or "").strip()
    if not text:
        return fallback
    if not re.fullmatch(r"[A-Za-z0-9_.:/-]+", text):
        return fallback
    return text[:limit]


def _safe_completion_id(value: Any) -> str:
    """Return a provider completion id only when it is an opaque ASCII token.

    Completion ids are useful for joining our compatibility summary to an
    upstream gateway log.  Unlike the more general metric sanitizer, this
    deliberately rejects URL separators and overlong values instead of
    truncating them, so a malformed provider cannot turn the field into a
    partial URL, response body, or log payload.
    """
    if not isinstance(value, str):
        return "none"
    if value != value.strip() or not re.fullmatch(
        r"[A-Za-z0-9][A-Za-z0-9_.-]{0,79}",
        value,
    ):
        return "none"
    return value


def _safe_finish_reason(value: Any) -> str:
    reason = str(value or "").strip().lower()
    return reason if reason in _SAFE_FINISH_REASONS else ("none" if not reason else "other")


def _safe_upstream_code(value: Any) -> str:
    """Expose only conventional numeric provider codes in exception strings."""
    text = str(value or "").strip()
    return text if text.isascii() and text.isdigit() and len(text) <= 10 else "other"


@dataclass(frozen=True, slots=True)
class _PendingWriteFileBridge:
    """Validated, one-shot hand-off from ``begin_write_file`` to body streaming."""

    begin_tool_call_id: str
    write_tool_call_id: str
    path: str
    description: str
    append: bool
    allow_implicit_think_close: bool


@dataclass(slots=True)
class _SelectionAliasCallState:
    """Buffered provider alias until metadata-only vs real writer is known."""

    index: int
    call_id: str = ""
    call_type: str = "function"
    name: str = ""
    arguments: str = ""
    decision: Literal["pending", "begin", "write"] = "pending"
    emitted: bool = False


class _IncrementalArgumentAccumulator:
    """Accumulate delta or growing-snapshot arguments without quadratic copies."""

    __slots__ = ("_length", "_parts")

    def __init__(self) -> None:
        self._parts: list[str] = []
        self._length = 0

    def _is_prefix_of(self, candidate: str) -> bool:
        if len(candidate) < self._length:
            return False
        offset = 0
        for part in self._parts:
            end = offset + len(part)
            if candidate[offset:end] != part:
                return False
            offset = end
        return True

    def merge(self, fragment: str) -> str:
        """Return only the newly added logical suffix."""

        if not fragment:
            return ""
        if not self._parts:
            self._parts.append(fragment)
            self._length = len(fragment)
            return fragment
        if self._is_prefix_of(fragment):
            suffix = fragment[self._length :]
            # A growing snapshot already contains every prior part. Collapse
            # the rope so materialization remains one final join.
            self._parts = [fragment]
            self._length = len(fragment)
            return suffix
        self._parts.append(fragment)
        self._length += len(fragment)
        return fragment

    def materialize(self) -> str:
        return "".join(self._parts)


class _IncrementalJsonStringDecoder:
    """Decode one JSON string across arbitrary fragment boundaries."""

    __slots__ = ("_capture", "_digits", "_mode", "_parts", "complete")

    _ESCAPES = {
        '"': '"',
        "\\": "\\",
        "/": "/",
        "b": "\b",
        "f": "\f",
        "n": "\n",
        "r": "\r",
        "t": "\t",
    }

    def __init__(self, *, capture: bool) -> None:
        self._capture = capture
        self._digits = ""
        self._mode: Literal[
            "normal",
            "escape",
            "unicode",
            "low_backslash",
            "low_u",
            "low_unicode",
        ] = "normal"
        self._parts: list[str] = []
        self.complete = False

    def _emit(self, value: str, emitted: list[str]) -> None:
        if self._capture:
            self._parts.append(value)
            emitted.append(value)

    def feed_char(self, char: str) -> str:
        if self.complete:
            raise ValueError("characters after JSON string")
        emitted: list[str] = []
        if self._mode == "normal":
            if char == '"':
                self.complete = True
            elif char == "\\":
                self._mode = "escape"
            elif ord(char) < 0x20 or 0xD800 <= ord(char) <= 0xDFFF:
                raise ValueError("invalid JSON string character")
            else:
                self._emit(char, emitted)
        elif self._mode == "escape":
            if char in self._ESCAPES:
                self._emit(self._ESCAPES[char], emitted)
                self._mode = "normal"
            elif char == "u":
                self._digits = ""
                self._mode = "unicode"
            else:
                raise ValueError("invalid JSON escape")
        elif self._mode in {"unicode", "low_unicode"}:
            if char not in "0123456789abcdefABCDEF":
                raise ValueError("invalid JSON unicode escape")
            self._digits += char
            if len(self._digits) == 4:
                codepoint = int(self._digits, 16)
                if self._mode == "unicode" and 0xD800 <= codepoint <= 0xDBFF:
                    # Never expose half of a UTF-16 surrogate pair.
                    self._digits = f"{codepoint:04x}"
                    self._mode = "low_backslash"
                elif self._mode == "low_unicode":
                    high = int(self._parts.pop(), 16)
                    if not 0xDC00 <= codepoint <= 0xDFFF:
                        raise ValueError("invalid JSON low surrogate")
                    combined = 0x10000 + ((high - 0xD800) << 10) + (
                        codepoint - 0xDC00
                    )
                    self._emit(chr(combined), emitted)
                    self._digits = ""
                    self._mode = "normal"
                elif 0xDC00 <= codepoint <= 0xDFFF:
                    raise ValueError("unexpected JSON low surrogate")
                else:
                    self._emit(chr(codepoint), emitted)
                    self._digits = ""
                    self._mode = "normal"
        elif self._mode == "low_backslash":
            if char != "\\":
                raise ValueError("missing JSON low surrogate")
            # Store the high surrogate as an internal marker. It is removed
            # before the combined Unicode scalar is emitted.
            self._parts.append(self._digits)
            self._digits = ""
            self._mode = "low_u"
        elif self._mode == "low_u":
            if char != "u":
                raise ValueError("missing JSON low surrogate")
            self._digits = ""
            self._mode = "low_unicode"
        return "".join(emitted)

    def value(self) -> str:
        if not self.complete:
            raise ValueError("incomplete JSON string")
        return "".join(self._parts)


class _IncrementalTopLevelStringExtractor:
    """Single-pass extractor for one root-object JSON string property."""

    __slots__ = (
        "_container_stack",
        "_content_parts",
        "_current_key",
        "_decoder",
        "_expected_key",
        "_phase",
        "_primitive_started",
        "complete",
        "consumed_characters",
        "found",
    )

    def __init__(self, expected_key: str) -> None:
        self._container_stack: list[str] = []
        self._content_parts: list[str] = []
        self._current_key = ""
        self._decoder: _IncrementalJsonStringDecoder | None = None
        self._expected_key = expected_key
        self._phase: Literal[
            "start",
            "key_or_end",
            "key",
            "colon",
            "value",
            "content",
            "skip_string",
            "skip_container",
            "skip_primitive",
            "after_value",
            "done",
        ] = "start"
        self._primitive_started = False
        self.complete = False
        self.consumed_characters = 0
        self.found = False

    def _start_skipped_value(self, char: str) -> None:
        if char == '"':
            self._decoder = _IncrementalJsonStringDecoder(capture=False)
            self._phase = "skip_string"
        elif char in "{[":
            self._container_stack = ["}" if char == "{" else "]"]
            self._decoder = None
            self._phase = "skip_container"
        elif char in "-0123456789tfn":
            self._primitive_started = True
            self._phase = "skip_primitive"
        else:
            raise ValueError("invalid JSON value")

    def feed(self, fragment: str) -> str:
        emitted: list[str] = []
        index = 0
        while index < len(fragment):
            char = fragment[index]
            self.consumed_characters += 1
            reprocess = False
            if self._phase == "start":
                if char.isspace():
                    pass
                elif char == "{":
                    self._phase = "key_or_end"
                else:
                    raise ValueError("tool arguments must be a JSON object")
            elif self._phase == "key_or_end":
                if char.isspace():
                    pass
                elif char == "}":
                    self._phase = "done"
                    self.complete = True
                elif char == '"':
                    self._decoder = _IncrementalJsonStringDecoder(capture=True)
                    self._phase = "key"
                else:
                    raise ValueError("invalid JSON object key")
            elif self._phase == "key":
                assert self._decoder is not None
                self._decoder.feed_char(char)
                if self._decoder.complete:
                    self._current_key = self._decoder.value()
                    self._decoder = None
                    self._phase = "colon"
            elif self._phase == "colon":
                if char.isspace():
                    pass
                elif char == ":":
                    self._phase = "value"
                else:
                    raise ValueError("missing JSON object colon")
            elif self._phase == "value":
                if char.isspace():
                    pass
                elif self._current_key == self._expected_key:
                    if self.found:
                        raise ValueError(f"duplicate {self._expected_key} property")
                    if char != '"':
                        raise ValueError(
                            f"{self._expected_key} must be a JSON string"
                        )
                    self.found = True
                    self._decoder = _IncrementalJsonStringDecoder(capture=True)
                    self._phase = "content"
                else:
                    self._start_skipped_value(char)
            elif self._phase == "content":
                assert self._decoder is not None
                decoded = self._decoder.feed_char(char)
                if decoded:
                    self._content_parts.append(decoded)
                    emitted.append(decoded)
                if self._decoder.complete:
                    self._decoder = None
                    self._phase = "after_value"
            elif self._phase == "skip_string":
                assert self._decoder is not None
                self._decoder.feed_char(char)
                if self._decoder.complete:
                    self._decoder = None
                    self._phase = "after_value"
            elif self._phase == "skip_container":
                if self._decoder is not None:
                    self._decoder.feed_char(char)
                    if self._decoder.complete:
                        self._decoder = None
                elif char == '"':
                    self._decoder = _IncrementalJsonStringDecoder(capture=False)
                elif char in "{[":
                    self._container_stack.append("}" if char == "{" else "]")
                elif char in "}]":
                    if not self._container_stack or char != self._container_stack[-1]:
                        raise ValueError("mismatched JSON container delimiter")
                    self._container_stack.pop()
                    if not self._container_stack:
                        self._phase = "after_value"
            elif self._phase == "skip_primitive":
                if char in ",}" or char.isspace():
                    self._phase = "after_value"
                    reprocess = True
            elif self._phase == "after_value":
                if char.isspace():
                    pass
                elif char == ",":
                    self._phase = "key_or_end"
                elif char == "}":
                    self._phase = "done"
                    self.complete = True
                else:
                    raise ValueError("invalid JSON object delimiter")
            elif self._phase == "done":
                if not char.isspace():
                    raise ValueError("characters after JSON object")
            if reprocess:
                self.consumed_characters -= 1
                continue
            index += 1
        return "".join(emitted)

    def content(self) -> str:
        if not self.found:
            raise ValueError(f"missing {self._expected_key} property")
        return "".join(self._content_parts)


@dataclass(slots=True)
class _WriteFileBridgeStreamState:
    """Invocation-local state; never shared through the model instance."""

    transport: WriteFileTransport
    pending: _PendingWriteFileBridge | None = None
    started: bool = False
    closed: bool = False
    aborted: bool = False
    inline_think_filter: InlineThinkTagFilter = field(
        default_factory=lambda: InlineThinkTagFilter(leading_only=True)
    )
    markdown_fence_projection: _MarkdownFenceProjection | None = None
    selection_write_file_alias: bool = False
    selection_alias_calls: dict[int, _SelectionAliasCallState] = field(
        default_factory=dict
    )
    selection_alias_early_stop: bool = False
    # ``buffered_gateway`` performs the metadata selection and ordinary-text
    # body request inside one ChatModel stream.  LangGraph therefore receives
    # only the final canonical write_file call and never executes an internal
    # preparation tool.
    buffered_gateway_body_request: bool = False
    selection_alias_has_native_reasoning: bool = False
    selection_alias_saw_think_open: bool = False
    selection_alias_implicit_think_close: bool = False
    selection_alias_think_probe_suffix: str = ""
    renderer_prerequisite_tools: frozenset[str] = field(default_factory=frozenset)
    renderer_deferred_tool_names: dict[int, str] = field(default_factory=dict)
    renderer_deferred: bool = False
    renderer_leading_whitespace: str = ""
    renderer_html_reference: str | None = None
    renderer_html_probe: str = ""
    renderer_html_probe_resolved: bool = False
    renderer_html_recovered: bool = False
    renderer_html_suspected_truncated: bool = False
    renderer_html_content: list[str] = field(default_factory=list)
    renderer_html_references: dict[str, str] = field(default_factory=dict)
    renderer_fallback_active: bool = False
    renderer_fallback_index: int | None = None
    renderer_fallback_call_id: str = ""
    renderer_fallback_name: str = ""
    renderer_fallback_arguments: _IncrementalArgumentAccumulator = field(
        default_factory=_IncrementalArgumentAccumulator
    )
    renderer_fallback_content: _IncrementalTopLevelStringExtractor = field(
        default_factory=lambda: _IncrementalTopLevelStringExtractor("content")
    )
    renderer_fallback_control_filter: InlineThinkTagFilter = field(
        default_factory=lambda: InlineThinkTagFilter(leading_only=True)
    )


class _MarkdownFenceProjection:
    """Project an optional response-level Markdown fence out of a body stream.

    Chat-oriented models occasionally wrap an entire ``.md`` artifact in a
    `````markdown`` response fence even though the renderer requested raw file
    bytes.  A leading fence is ambiguous while the response is still running:
    it may instead be an intentional first code block followed by more file
    content.  This projection therefore streams the likely body immediately,
    retains a small terminal guard, and validates the *whole* response at the
    safe finish frame.

    If the response is a complete outer wrapper, only that first and last line
    are removed.  If it is not, ``finish`` returns an exact replacement so the
    bridge can append a duplicate JSON ``content`` key (last-key-wins), restoring
    every original byte before ToolNode sees the call.  Native models, chat
    messages, non-Markdown artifacts, and inner fences never enter this class.
    """

    _OPEN_RE = re.compile(
        r"\A[\ufeff \t\r\n]*```(?:markdown|md)[ \t]*\r?\n",
        re.IGNORECASE,
    )
    _CLOSE_RE = re.compile(
        r"(?P<linebreak>\r?\n)[ \t]*```[ \t]*(?:\r?\n)?[ \t\r\n]*\Z"
    )
    _OPEN_MARKERS = ("```markdown", "```md")
    _PREFIX_LIMIT = 512
    _TERMINAL_GUARD = 64

    def __init__(self) -> None:
        self._mode: Literal["probe", "plain", "candidate"] = "probe"
        self._prefix = ""
        self._raw_candidate = ""
        self._opening_end = 0
        self._tail = ""
        self._emitted_length = 0

    @classmethod
    def _could_still_be_opening(cls, value: str) -> bool:
        candidate = value.lstrip("\ufeff \t\r\n").lower()
        if not candidate:
            return True
        for marker in cls._OPEN_MARKERS:
            if marker.startswith(candidate):
                return True
            if candidate.startswith(marker) and not candidate[len(marker) :].strip(
                " \t\r"
            ):
                return True
        return False

    def push(self, text: str) -> str:
        """Return bytes safe to expose as the current streamed file prefix."""

        if not text:
            return ""
        if self._mode == "plain":
            return text
        if self._mode == "candidate":
            self._raw_candidate += text
            return self._push_candidate_body(text)

        self._prefix += text
        opening = self._OPEN_RE.match(self._prefix)
        if opening is not None:
            self._mode = "candidate"
            self._raw_candidate = self._prefix
            self._opening_end = opening.end()
            body = self._prefix[self._opening_end :]
            self._prefix = ""
            return self._push_candidate_body(body)

        if len(self._prefix) <= self._PREFIX_LIMIT and self._could_still_be_opening(
            self._prefix
        ):
            return ""

        self._mode = "plain"
        confirmed = self._prefix
        self._prefix = ""
        return confirmed

    def _push_candidate_body(self, text: str) -> str:
        self._tail += text
        if len(self._tail) <= self._TERMINAL_GUARD:
            return ""
        confirmed = self._tail[: -self._TERMINAL_GUARD]
        self._tail = self._tail[-self._TERMINAL_GUARD :]
        self._emitted_length += len(confirmed)
        return confirmed

    def finish(self) -> tuple[str, str | None]:
        """Return ``(tail, replacement)`` after whole-response validation.

        ``replacement`` is populated only when speculative bytes have already
        streamed but the final whole-file interpretation differs.  The caller
        then repairs the canonical tool arguments with a duplicate JSON key;
        no sandbox write occurs until that corrected object is complete.
        """

        if self._mode == "probe":
            tail = self._prefix
            self._prefix = ""
            self._mode = "plain"
            return tail, None
        if self._mode == "plain":
            return "", None

        raw = self._raw_candidate
        closing = self._CLOSE_RE.search(raw, self._opening_end)
        if closing is None or not self._has_balanced_inner_fences(
            raw[self._opening_end : closing.start()]
        ):
            final_content = raw
        else:
            final_content = (
                raw[self._opening_end : closing.start()]
                + closing.group("linebreak")
            )

        speculative_prefix = raw[
            self._opening_end : self._opening_end + self._emitted_length
        ]
        self._tail = ""
        if final_content.startswith(speculative_prefix):
            return final_content[self._emitted_length :], None
        return "", final_content

    @staticmethod
    def _has_balanced_inner_fences(body: str) -> bool:
        """Do not mistake separate Markdown code blocks for one outer wrapper.

        A bare closer outside an inner block closes the proposed outer fence
        early. An unclosed inner block would consume the proposed final fence.
        In either case retain the original response instead of deleting bytes.
        """
        active: tuple[str, int] | None = None
        for line in body.splitlines():
            match = re.match(r"^ {0,3}(`{3,}|~{3,})(.*)$", line)
            if match is None:
                continue
            marker, suffix = match.groups()
            if active is not None:
                if marker[0] == active[0] and len(marker) >= active[1] and not suffix.strip():
                    active = None
            elif marker[0] == "`" and not suffix.strip():
                return False
            else:
                active = (marker[0], len(marker))
        return active is None


@dataclass(slots=True)
class _ToolCallStreamDiagnostics:
    """Per-invocation, content-free diagnostics for OpenAI-compatible streams.

    Logs contain only counters, timings, and lengths. Transient parser state is
    scoped to this invocation; prompt text, response content, request URLs, and
    exception messages are never included in diagnostic logs.
    """

    model_name: str
    run_id: str = "none"
    provider_completion_id: str = "none"
    started_at: float = field(default_factory=time.monotonic)
    first_chunk_at: float | None = None
    first_tool_args_at: float | None = None
    first_reasoning_at: float | None = None
    chunk_count: int = 0
    incoming_reasoning_chunks: int = 0
    emitted_reasoning_chunks: int = 0
    request_enable_thinking: str = "unset"
    intentional_early_finish: bool = False
    tool_arg_chunks: int = 0
    initial_chunks: int = 0
    delta_chunks: int = 0
    cumulative_chunks: int = 0
    repeated_chunks: int = 0
    rewind_chunks: int = 0
    empty_arg_chunks: int = 0
    invalid_shapes: int = 0
    message_as_delta_chunks: int = 0
    root_tool_call_chunks: int = 0
    singleton_tool_call_chunks: int = 0
    structured_arg_chunks: int = 0
    tool_id_resets: int = 0
    repeated_tool_names: int = 0
    raw_arg_first_chars: int | None = None
    raw_arg_last_chars: int = 0
    raw_arg_max_chars: int = 0
    raw_arg_growths: int = 0
    raw_arg_flats: int = 0
    raw_arg_shrinks: int = 0
    finish_reason: str = "none"
    strict_stream_integrity: bool = False
    text_state: ChatStreamState = field(default_factory=ChatStreamState)
    snapshots: dict[int, str] = field(default_factory=dict)
    call_ids: dict[int, str] = field(default_factory=dict)
    argument_modes: dict[int, str] = field(default_factory=dict)
    tool_name_snapshots: dict[int, str] = field(default_factory=dict)
    complete_tool_names: set[str] = field(default_factory=set)
    last_raw_lengths: dict[int, int] = field(default_factory=dict)
    seen_indices: set[int] = field(default_factory=set)
    _logged: bool = False

    def observe_chunk(self, chunk: Mapping[str, Any]) -> None:
        self.chunk_count += 1
        if self.first_chunk_at is None:
            self.first_chunk_at = time.monotonic()
            self.provider_completion_id = _safe_completion_id(chunk.get("id"))
        choices = _stream_choices(chunk)
        # The converter consumes the first choice only. Count a frame once,
        # even if the gateway duplicates reasoning across aliases/containers.
        if choices and any(_extract_reasoning_content(choices[0].get(key)) for key in ("delta", "message")):
            self.incoming_reasoning_chunks += 1
            if self.first_reasoning_at is None:
                self.first_reasoning_at = time.monotonic()
        for choice in choices:
            reason = choice.get("finish_reason")
            if reason:
                self.finish_reason = _safe_finish_reason(reason)

    def mark_invalid(self) -> None:
        self.invalid_shapes += 1

    def observe_argument(self, *, index: int, raw_length: int, classification: str) -> None:
        self.tool_arg_chunks += 1
        if self.first_tool_args_at is None:
            self.first_tool_args_at = time.monotonic()
        if self.raw_arg_first_chars is None:
            self.raw_arg_first_chars = raw_length
        self.raw_arg_last_chars = raw_length
        self.raw_arg_max_chars = max(self.raw_arg_max_chars, raw_length)

        previous_length = self.last_raw_lengths.get(index)
        if previous_length is not None:
            if raw_length > previous_length:
                self.raw_arg_growths += 1
            elif raw_length == previous_length:
                self.raw_arg_flats += 1
            else:
                self.raw_arg_shrinks += 1
        self.last_raw_lengths[index] = raw_length

        if classification == "initial":
            self.initial_chunks += 1
        elif classification == "delta":
            self.delta_chunks += 1
        elif classification == "cumulative":
            self.cumulative_chunks += 1
        elif classification == "repeated":
            self.repeated_chunks += 1
        elif classification == "rewind":
            self.rewind_chunks += 1

    def log_summary(self, error: BaseException | None = None) -> None:
        # This runs in the stream's finally block. Malformed diagnostic input
        # or a failing deployment log handler must not replace the original
        # provider exception/cancellation, or turn a successful answer into a
        # failed call that the retry middleware sends to the model again.
        try:
            self._log_summary(error)
        except Exception:
            pass

    def _log_summary(self, error: BaseException | None = None) -> None:
        if self._logged:
            return
        self._logged = True

        # One bounded summary per stream (never per token). Zero reasoning
        # counts are useful too: they distinguish no upstream reasoning from
        # reasoning that was lost during normalization.

        now = time.monotonic()
        total_ms = max(0, int((now - self.started_at) * 1000))
        first_chunk_ms = max(0, int((self.first_chunk_at - self.started_at) * 1000)) if self.first_chunk_at is not None else -1
        first_tool_args_ms = max(0, int((self.first_tool_args_at - self.started_at) * 1000)) if self.first_tool_args_at is not None else -1
        first_reasoning_ms = max(0, int((self.first_reasoning_at - self.started_at) * 1000)) if self.first_reasoning_at is not None else -1
        if error is None:
            status = "success"
            error_type = "none"
        elif isinstance(error, (asyncio.CancelledError, GeneratorExit)):
            status = "cancelled"
            error_type = _safe_metric_name(type(error).__name__, fallback="other")
        else:
            status = "error"
            error_type = _safe_metric_name(type(error).__name__, fallback="other")

        assembled_final_chars = sum(len(value) for value in self.snapshots.values())
        assembled_max_chars = max((len(value) for value in self.snapshots.values()), default=0)
        cancelled = isinstance(error, (asyncio.CancelledError, GeneratorExit))
        valid_final_json = 0
        invalid_final_json = 0
        if not self.intentional_early_finish:
            for assembled_arguments in self.snapshots.values():
                if not assembled_arguments:
                    continue
                try:
                    json.loads(assembled_arguments)
                except (TypeError, ValueError):
                    invalid_final_json += 1
                else:
                    valid_final_json += 1
        self._record_truncation_event(
            status=status,
            error=error,
            error_type=error_type,
            total_ms=total_ms,
            first_chunk_ms=first_chunk_ms,
        )
        log = (
            logger.info
            if cancelled
            else (
                logger.warning
                if error is not None or self.invalid_shapes or invalid_final_json
                else logger.info
            )
        )
        log(
            "OPENAI_TOOL_STREAM_SUMMARY model=%s run_id=%s provider_completion_id=%s "
            "status=%s total_ms=%d "
            "first_chunk_ms=%d first_tool_args_ms=%d chunks=%d tool_arg_chunks=%d "
            "tool_calls=%d initial=%d delta=%d cumulative=%d repeated=%d rewind=%d "
            "empty_args=%d invalid_shapes=%d message_as_delta=%d root_tool_calls=%d "
            "singleton_tool_calls=%d structured_args=%d tool_id_resets=%d raw_arg_first_chars=%d "
            "raw_arg_last_chars=%d raw_arg_max_chars=%d raw_arg_growths=%d "
            "raw_arg_flats=%d raw_arg_shrinks=%d assembled_final_chars=%d "
            "assembled_max_chars=%d valid_final_json=%d invalid_final_json=%d "
            "finish_reason=%s error_type=%s incoming_reasoning_chunks=%d "
            "emitted_reasoning_chunks=%d first_reasoning_ms=%d request_enable_thinking=%s "
            "repeated_tool_names=%d",
            self.model_name,
            self.run_id,
            self.provider_completion_id,
            status,
            total_ms,
            first_chunk_ms,
            first_tool_args_ms,
            self.chunk_count,
            self.tool_arg_chunks,
            len(self.seen_indices) + self.tool_id_resets,
            self.initial_chunks,
            self.delta_chunks,
            self.cumulative_chunks,
            self.repeated_chunks,
            self.rewind_chunks,
            self.empty_arg_chunks,
            self.invalid_shapes,
            self.message_as_delta_chunks,
            self.root_tool_call_chunks,
            self.singleton_tool_call_chunks,
            self.structured_arg_chunks,
            self.tool_id_resets,
            self.raw_arg_first_chars or 0,
            self.raw_arg_last_chars,
            self.raw_arg_max_chars,
            self.raw_arg_growths,
            self.raw_arg_flats,
            self.raw_arg_shrinks,
            assembled_final_chars,
            assembled_max_chars,
            valid_final_json,
            invalid_final_json,
            self.finish_reason,
            error_type,
            self.incoming_reasoning_chunks,
            self.emitted_reasoning_chunks,
            first_reasoning_ms,
            self.request_enable_thinking,
            self.repeated_tool_names,
        )

    def _record_truncation_event(
        self,
        *,
        status: str,
        error: BaseException | None,
        error_type: str,
        total_ms: int,
        first_chunk_ms: int,
    ) -> None:
        """Persist an admin-visible event when a provider stream ended truncated.

        Classification (only genuine truncations; the write_file bridge's
        intentional early finishes and user cancellations are excluded):
        - ``silent_no_terminal``: the stream ended "successfully" without a
          terminal frame after streaming content — the silently committed half
          answer this log exists to surface (non-strict models).
        - ``incomplete_stream_error``: strict integrity mode raised
          ``IncompleteChatStreamError`` — a detected truncation being retried.
        - ``length_truncated`` / ``content_filter``: the provider closed the
          stream with a truncating finish reason.
        - ``midstream_error``: a transport/provider error cut the stream after
          at least one chunk (the error middleware retries afterwards).
        """
        if self.intentional_early_finish or status == "cancelled":
            return
        event_type: str | None = None
        if error is not None:
            if error_type == "IncompleteChatStreamError":
                event_type = "incomplete_stream_error"
            elif self.chunk_count > 0:
                event_type = "midstream_error"
        elif self.finish_reason == "none":
            if self.chunk_count > 0:
                event_type = "silent_no_terminal"
        elif self.finish_reason == "length":
            event_type = "length_truncated"
        elif self.finish_reason == "content_filter":
            event_type = "content_filter"
        if event_type is None:
            return
        try:
            from deerflow.persistence.stream_truncation import record_stream_truncation_event

            record_stream_truncation_event(
                model_name=self.model_name,
                event_type=event_type,
                strict_mode=self.strict_stream_integrity,
                finish_reason=self.finish_reason,
                status=status,
                error_type=error_type if error is not None else None,
                chunk_count=self.chunk_count,
                duration_ms=total_ms,
                first_chunk_ms=first_chunk_ms,
                provider_completion_id=self.provider_completion_id,
                run_id=self.run_id,
            )
        except Exception:
            pass


_ACTIVE_TOOL_STREAM: contextvars.ContextVar[_ToolCallStreamDiagnostics | None] = contextvars.ContextVar(
    "deerflow_openai_tool_stream",
    default=None,
)

_ACTIVE_WRITE_FILE_BRIDGE: contextvars.ContextVar[_WriteFileBridgeStreamState | None] = contextvars.ContextVar(
    "deerflow_openai_write_file_bridge",
    default=None,
)

_ACTIVE_DSML_RECOVERY: contextvars.ContextVar[DSMLToolCallRecovery | None] = contextvars.ContextVar(
    "deerflow_openai_dsml_recovery",
    default=None,
)


def _message_content_text(content: Any) -> str:
    if isinstance(content, str):
        return content
    if not isinstance(content, list):
        return ""
    parts: list[str] = []
    for item in content:
        if isinstance(item, str):
            parts.append(item)
        elif isinstance(item, Mapping):
            text = item.get("text")
            if isinstance(text, str):
                parts.append(text)
    return "".join(parts)


def _tool_call_value(tool_call: Any, key: str, default: Any = None) -> Any:
    if isinstance(tool_call, Mapping):
        return tool_call.get(key, default)
    return getattr(tool_call, key, default)


def _tool_call_args(tool_call: Any) -> dict[str, Any]:
    raw = _tool_call_value(tool_call, "args", {})
    return dict(raw) if isinstance(raw, Mapping) else {}


def _pending_write_file_bridge(messages: list[Any]) -> _PendingWriteFileBridge | None:
    """Return a pending bridge only for an immediately preceding successful handshake.

    This deliberately performs no semantic inspection of ordinary assistant
    text. A bridge can start only when the last checkpointed message is the
    successful result of the explicit ``begin_write_file`` tool.
    """

    if not messages or not isinstance(messages[-1], ToolMessage):
        return None
    result = messages[-1]
    if str(getattr(result, "status", "") or "").lower() == "error":
        return None
    result_text = _message_content_text(getattr(result, "content", "")).strip()
    if not result_text or result_text.lower().startswith("error"):
        return None
    try:
        ready = json.loads(result_text)
    except (TypeError, ValueError):
        return None
    if not isinstance(ready, Mapping) or ready.get("status") != "ready":
        return None

    begin_tool_call_id = str(getattr(result, "tool_call_id", "") or "").strip()
    if not begin_tool_call_id:
        return None
    for message in reversed(messages[:-1]):
        if not isinstance(message, AIMessage):
            continue
        tool_calls = message.tool_calls or []
        # A body round is intentionally one-shot. Parallel calls would leave
        # multiple ToolMessages around the control exchange and make it
        # ambiguous which operation the generated body belongs to.
        if len(tool_calls) != 1:
            return None
        for tool_call in reversed(tool_calls):
            if str(_tool_call_value(tool_call, "id", "") or "") != begin_tool_call_id:
                continue
            if str(_tool_call_value(tool_call, "name", "") or "") != _BEGIN_WRITE_FILE_TOOL_NAME:
                return None
            args = _tool_call_args(tool_call)
            path = str(args.get("path") or "").strip()
            description = str(args.get("description") or "").strip()
            append = args.get("append", False)
            if not path or not path.startswith("/") or not isinstance(append, bool):
                return None
            ready_path = str(ready.get("path") or "").strip()
            ready_append = ready.get("append", False)
            if (
                ready_path != path
                or not isinstance(ready_append, bool)
                or ready_append != append
            ):
                return None
            # Provider-generated tool-call ids may be reused on later turns.
            # Include checkpoint message identities so each prepared write gets
            # a stable *and transaction-unique* preview id.
            begin_message_id = str(getattr(message, "id", "") or "")
            result_message_id = str(getattr(result, "id", "") or "")
            transaction_identity = "\x1f".join(
                (begin_message_id, result_message_id)
            )
            # LangChain normally assigns message ids, but direct SDK callers
            # and some restored legacy checkpoints may not have them. A fresh
            # invocation nonce prevents a provider-reused begin id from
            # colliding with an earlier virtual preview in that case.
            transaction_token = (
                uuid.uuid5(
                    uuid.NAMESPACE_OID,
                    f"{begin_tool_call_id}\x1f{transaction_identity}\x1f{path}",
                ).hex[:24]
                if transaction_identity.strip("\x1f")
                else uuid.uuid4().hex[:24]
            )
            return _PendingWriteFileBridge(
                begin_tool_call_id=begin_tool_call_id,
                write_tool_call_id=(
                    f"{_WRITE_FILE_BRIDGE_CALL_PREFIX}{transaction_token}"
                ),
                path=path,
                description=description or "Write the prepared file",
                append=append,
                allow_implicit_think_close=(
                    _message_uses_implicit_think_close(message)
                ),
            )
        # The matching tool call must be on the nearest preceding AI message.
        return None
    return None


def _payload_tool_name(tool: Any) -> str:
    if not isinstance(tool, Mapping):
        return ""
    function = tool.get("function")
    if isinstance(function, Mapping):
        return str(function.get("name") or "")
    return str(tool.get("name") or "")


def _selection_tool(tool: Any, *, provider_name: str | None = None) -> Any:
    """Return a strict provider schema for the internal prepare handshake.

    ``artifact_renderer`` deliberately exposes the metadata-only schema under
    the familiar ``write_file`` name. Weak models often copy that name from the
    lead-agent contract even when the real writer has been removed. The raw
    response is translated back to ``begin_write_file`` before LangGraph sees
    it, so ToolNode still executes only the server-owned preparation control.
    """

    if not isinstance(tool, Mapping) or _payload_tool_name(tool) != _BEGIN_WRITE_FILE_TOOL_NAME:
        return tool
    function = tool.get("function")
    if not isinstance(function, Mapping):
        return tool
    parameters = function.get("parameters")
    if not isinstance(parameters, Mapping):
        return tool
    strict_parameters = dict(parameters)
    # LangChain's injected-argument subset model does not retain the full
    # Pydantic model's ``extra=forbid`` setting. Restore it at the actual
    # provider boundary so a weak model is explicitly told that file bytes are
    # not legal in this small control call.
    strict_parameters["additionalProperties"] = False
    if provider_name:
        # The renderer can hand off as soon as these three small fields have
        # arrived.  Requiring all of them avoids guessing an append intent from
        # an unfinished JSON object and lets us cancel a gateway that ignored
        # the metadata-only schema before it spends the rest of the response
        # generating a hidden ``content`` argument.
        strict_parameters["required"] = ["description", "path", "append"]
    strict_function = dict(function)
    strict_function["parameters"] = strict_parameters
    if provider_name:
        strict_function["name"] = provider_name
    strict_tool = dict(tool)
    strict_tool["function"] = strict_function
    return strict_tool


def _buffered_gateway_selection_tool(tool: Any) -> Any:
    """Expose the original writer as a strict metadata-only selector.

    The provider continues to see the familiar ``write_file`` name.  Only its
    large ``content`` argument is removed for the selection request, avoiding
    the gateway buffer that motivated this compatibility mode.  The final
    LangGraph/ToolNode call is still the untouched canonical writer schema.
    """

    if not isinstance(tool, Mapping) or _payload_tool_name(tool) != _WRITE_FILE_TOOL_NAME:
        return tool
    function = tool.get("function")
    if not isinstance(function, Mapping):
        return tool
    parameters = function.get("parameters")
    if not isinstance(parameters, Mapping):
        return tool
    properties = parameters.get("properties")
    if not isinstance(properties, Mapping):
        return tool
    metadata_properties = {
        key: value
        for key, value in properties.items()
        if key in {"description", "path", "append"}
    }
    if not {"description", "path", "append"}.issubset(metadata_properties):
        raise RuntimeError(
            "buffered_gateway requires write_file description/path/append fields"
        )
    strict_parameters = dict(parameters)
    strict_parameters["properties"] = metadata_properties
    strict_parameters["required"] = ["description", "path", "append"]
    strict_parameters["additionalProperties"] = False
    strict_function = dict(function)
    strict_function["parameters"] = strict_parameters
    strict_tool = dict(tool)
    strict_tool["function"] = strict_function
    return strict_tool


def _message_uses_implicit_think_close(message: Any) -> bool:
    """Whether one model response demonstrated a template-injected open tag."""

    content = _message_content_text(getattr(message, "content", ""))
    lowered = content.lower()
    open_at = lowered.find("<think>")
    close_at = lowered.find("</think>")
    return close_at >= 0 and (open_at < 0 or close_at < open_at)


def _strip_historical_begin_write_file_exchanges(
    messages: list[Any],
) -> list[Any]:
    """Remove only reserved bridge calls/results from provider history.

    A user can switch a thread from a bridge model to a native model, and
    non-stream calls can reuse the same history. ``begin_write_file`` is an
    internal transport control, so it must not be sent to a provider that
    cannot see its schema. Exact tool names and call ids keep ordinary tool
    history intact, including other calls made by the same assistant message.
    """

    cleaned: list[Any] = []
    changed = False
    index = 0
    while index < len(messages):
        message = messages[index]
        if not isinstance(message, Mapping) or message.get("role") != "assistant":
            cleaned.append(message)
            index += 1
            continue
        tool_calls = message.get("tool_calls")
        if not isinstance(tool_calls, list):
            cleaned.append(message)
            index += 1
            continue
        begin_call_ids: set[str] = set()
        for call in tool_calls:
            function = call.get("function") if isinstance(call, Mapping) else None
            call_id = str(call.get("id") or "") if isinstance(call, Mapping) else ""
            if (
                call_id
                and isinstance(function, Mapping)
                and function.get("name") == _BEGIN_WRITE_FILE_TOOL_NAME
            ):
                begin_call_ids.add(call_id)
        if not begin_call_ids:
            cleaned.append(message)
            index += 1
            continue

        changed = True
        visible_calls = [
            call
            for call in tool_calls
            if not (
                isinstance(call, Mapping)
                and str(call.get("id") or "") in begin_call_ids
                and isinstance(call.get("function"), Mapping)
                and call["function"].get("name") == _BEGIN_WRITE_FILE_TOOL_NAME
            )
        ]
        rewritten = dict(message)
        if visible_calls:
            rewritten["tool_calls"] = visible_calls
            cleaned.append(rewritten)
        else:
            # A begin-only turn is transport control even when the provider
            # attached a preface. Dropping it avoids an assistant→assistant
            # sequence after its tool receipt is removed. Mixed real-tool
            # turns retain their content and remaining calls above.
            rewritten.pop("tool_calls", None)

        # Tool call ids are scoped to this assistant turn. Remove matching
        # begin receipts only from its immediately following result block;
        # later turns are allowed to reuse the same provider-generated id.
        index += 1
        while index < len(messages):
            result = messages[index]
            if not isinstance(result, Mapping) or result.get("role") != "tool":
                break
            if str(result.get("tool_call_id") or "") not in begin_call_ids:
                cleaned.append(result)
            index += 1
    return cleaned if changed else messages


def _hide_begin_write_file_for_native_request(payload: dict[str, Any]) -> None:
    """Keep the internal handshake out of native/non-stream provider calls."""

    messages = payload.get("messages")
    if isinstance(messages, list):
        cleaned_messages = _strip_historical_begin_write_file_exchanges(messages)
        if cleaned_messages is not messages:
            payload["messages"] = cleaned_messages

    tools = payload.get("tools")
    has_write_file = False
    if isinstance(tools, list):
        visible_tools = [
            tool for tool in tools if _payload_tool_name(tool) != _BEGIN_WRITE_FILE_TOOL_NAME
        ]
        has_write_file = any(
            _payload_tool_name(tool) == _WRITE_FILE_TOOL_NAME for tool in visible_tools
        )
        if len(visible_tools) != len(tools):
            payload["tools"] = visible_tools
    choice = payload.get("tool_choice")
    if choice == _BEGIN_WRITE_FILE_TOOL_NAME:
        if has_write_file:
            payload["tool_choice"] = _WRITE_FILE_TOOL_NAME
        else:
            payload.pop("tool_choice", None)
        return
    if not isinstance(choice, dict):
        return
    function = choice.get("function")
    if isinstance(function, dict) and function.get("name") == _BEGIN_WRITE_FILE_TOOL_NAME:
        if has_write_file:
            rewritten_function = dict(function)
            rewritten_function["name"] = _WRITE_FILE_TOOL_NAME
            rewritten_choice = dict(choice)
            rewritten_choice["function"] = rewritten_function
            payload["tool_choice"] = rewritten_choice
        else:
            payload.pop("tool_choice", None)


def _append_system_instruction(messages: list[Any], instruction: str) -> None:
    """Append transient bridge context for later normalization at the boundary."""

    messages.append({"role": "system", "content": instruction})


def _payload_tool_arguments(call: Any) -> tuple[dict[str, Any], str]:
    if not isinstance(call, Mapping):
        return {}, ""
    function = call.get("function")
    if not isinstance(function, Mapping):
        return {}, ""
    raw = function.get("arguments")
    if isinstance(raw, Mapping):
        return dict(raw), json.dumps(raw, ensure_ascii=False, separators=(",", ":"))
    if not isinstance(raw, str):
        return {}, ""
    try:
        parsed = json.loads(raw)
    except (TypeError, ValueError):
        return {}, raw
    return (dict(parsed) if isinstance(parsed, Mapping) else {}), raw


def _successful_read_file_contents(messages: list[Any]) -> dict[str, str]:
    """Map each path to its newest successful complete ``read_file`` result.

    Calls are consumed in message order. Reused provider ids overwrite the
    pending path and are popped with their result, so a later read cannot attach
    its bytes to an older same-id request.
    """

    pending_calls: dict[str, str] = {}
    latest: dict[str, str] = {}
    for message in messages:
        if isinstance(message, AIMessage):
            for tool_call in message.tool_calls or []:
                if str(_tool_call_value(tool_call, "name", "") or "") != "read_file":
                    continue
                args = _tool_call_args(tool_call)
                path = str(args.get("path") or "")
                call_id = str(_tool_call_value(tool_call, "id", "") or "")
                if call_id and path:
                    pending_calls[call_id] = path
            continue
        if not isinstance(message, ToolMessage):
            continue
        call_id = str(getattr(message, "tool_call_id", "") or "")
        path = pending_calls.pop(call_id, None)
        if path is None:
            continue
        if str(getattr(message, "status", "") or "").lower() == "error":
            continue
        content = _message_content_text(getattr(message, "content", ""))
        if content:
            if re.search(r"\.(?:html?|xhtml)\Z", path, re.IGNORECASE) and not (
                _is_complete_html_document(content)
            ):
                # A later ranged/truncated read must not replace an earlier
                # complete template that is still the only safe prefix source.
                continue
            latest[path] = content
    return latest


def _is_overwrite_html_artifact(pending: _PendingWriteFileBridge) -> bool:
    return not pending.append and bool(
        re.search(r"\.(?:html?|xhtml)\Z", pending.path, re.IGNORECASE)
    )


def _is_complete_html_document(content: str) -> bool:
    """Validate the recovered standalone envelope, not general HTML validity."""

    stripped = content.lstrip("\ufeff \t\r\n")
    if re.match(r"<!doctype\s+html(?:\s+[^>]*)?>", stripped, re.IGNORECASE) is None:
        return False
    patterns = (
        r"<html(?:\s|>)",
        r"<head(?:\s|>)",
        r"</head\s*>",
        r"<body(?:\s|>)",
        r"</body\s*>",
        r"</html\s*>\s*\Z",
    )
    matches = [re.search(pattern, stripped, re.IGNORECASE) for pattern in patterns]
    if any(match is None for match in matches):
        return False
    positions = [match.start() for match in matches if match is not None]
    return positions == sorted(positions)


def _probe_html_prefix_recovery(
    content: str,
    reference: str | None,
    *,
    final: bool,
) -> tuple[str, bool, bool, bool]:
    """Resolve a chunk-safe, evidence-backed HTML prefix projection.

    Returns ``(projected, resolved, recovered, suspected_truncation)``. A prefix
    is copied only when at least 48 exact leading bytes occur once in the latest
    complete same-target read. Merely sharing a tag name is never sufficient.
    """

    stripped = content.lstrip("\ufeff \t\r\n")
    lowered = stripped.lower()
    complete_starts = ("<!doctype", "<html")
    internal_starts = ("<head", "<style", "<body", "<main", "<script")
    complete_match = re.match(
        r"(?:<!doctype\s+html(?:\s+[^>]*)?>|<html(?:\s|>))",
        stripped,
        re.IGNORECASE,
    )
    if complete_match is not None:
        # Detection may use the stripped view, but complete documents retain
        # their byte-exact BOM and leading whitespace.
        return content, True, False, False
    possible_complete = bool(lowered) and any(
        start.startswith(lowered) for start in complete_starts
    )
    if possible_complete:
        if not final:
            return content, False, False, False
        return content, True, False, True

    possible_internal = bool(lowered) and any(
        start.startswith(lowered) for start in internal_starts
    )
    internal_match = re.match(
        r"<(head|style|body|main|script)(?:\s|>)",
        stripped,
        re.IGNORECASE,
    )
    if internal_match is None:
        if possible_internal:
            if not final:
                return content, False, False, False
            return content, True, False, True
        return content, True, False, False
    if not reference or not _is_complete_html_document(reference):
        return content, True, False, True
    if len(stripped) < 48:
        return content, final, False, True

    reference_stripped = reference.lstrip("\ufeff \t\r\n")
    anchor = internal_match.group(1).lower()
    head_open = re.search(r"<head(?:\s|>)", reference_stripped, re.IGNORECASE)
    head_close = re.search(r"</head\s*>", reference_stripped, re.IGNORECASE)
    body_open = re.search(r"<body(?:\s|>)", reference_stripped, re.IGNORECASE)
    max_overlap = min(256, len(stripped))
    min_overlap = 48
    for overlap in range(max_overlap, min_overlap - 1, -1):
        needle = stripped[:overlap]
        first = reference_stripped.find(needle)
        if first <= 0:
            continue
        if reference_stripped.find(needle, first + 1) >= 0:
            continue
        if anchor in {"style", "script"} and not (
            head_open is not None
            and head_close is not None
            and head_open.start() < first < head_close.start()
        ):
            continue
        if anchor == "head" and (
            head_open is None or first != head_open.start()
        ):
            continue
        if anchor == "body" and (
            body_open is None or first != body_open.start()
        ):
            continue
        if anchor == "main" and (
            body_open is None or first <= body_open.start()
        ):
            continue
        return (
            f"{reference_stripped[:first]}{stripped}",
            True,
            True,
            True,
        )
    if not final and len(stripped) < 256 and reference_stripped.count(
        stripped[:min_overlap]
    ) > 1:
        return content, False, False, True
    return content, True, False, True


def _project_renderer_visible_text(
    state: _WriteFileBridgeStreamState,
    pending: _PendingWriteFileBridge,
    text: str,
    *,
    final: bool = False,
) -> str:
    """Delay ambiguous prefaces and project the first HTML fragment safely."""

    # An inline ``<think>...</think>`` chunk is consumed entirely by the
    # reasoning filter before it reaches this projector.  Treat that empty
    # projection as "no file evidence yet".  Resolving the HTML prefix probe on
    # it would make the following harmless separator newlines look like file
    # bytes; if the provider then falls back to a complete ``write_file`` call,
    # the bridge would incorrectly reject the transition as mixed output.
    if not text and not final:
        return ""

    if not state.started and not state.renderer_html_probe_resolved:
        candidate = f"{state.renderer_leading_whitespace}{text}"
        if candidate and not candidate.strip():
            state.renderer_leading_whitespace = candidate
            return ""
        state.renderer_leading_whitespace = ""
        if _is_overwrite_html_artifact(pending):
            state.renderer_html_probe = f"{state.renderer_html_probe}{candidate}"
            projected, resolved, recovered, suspected = _probe_html_prefix_recovery(
                state.renderer_html_probe,
                state.renderer_html_reference,
                final=final,
            )
            state.renderer_html_recovered = recovered
            state.renderer_html_suspected_truncated = suspected
            if not resolved:
                return ""
            state.renderer_html_probe_resolved = True
            state.renderer_html_probe = ""
            text = projected
        else:
            text = candidate
    if text and _is_overwrite_html_artifact(pending):
        state.renderer_html_content.append(text)
    return text


def _validate_renderer_artifact_complete(
    state: _WriteFileBridgeStreamState,
    pending: _PendingWriteFileBridge,
) -> None:
    if not _is_overwrite_html_artifact(pending):
        return
    content = "".join(state.renderer_html_content)
    stripped = content.lstrip("\ufeff \t\r\n").lower()
    requires_complete_document = (
        state.renderer_html_recovered
        or state.renderer_html_suspected_truncated
        or stripped.startswith(("<!doctype", "<html"))
    )
    if not requires_complete_document or _is_complete_html_document(content):
        return
    state.aborted = True
    logger.warning(
        "ARTIFACT_RENDERER_PROTOCOL_REJECTED "
        "reason=incomplete_html_document finish_reason=stop"
    )
    raise ArtifactRendererTransientProtocolError(
        "Artifact renderer returned an incomplete HTML document"
    )


def _normalize_complete_renderer_html(
    content: str,
    reference: str | None,
) -> str:
    projected, _resolved, recovered, suspected = _probe_html_prefix_recovery(
        content,
        reference,
        final=True,
    )
    stripped = projected.lstrip("\ufeff \t\r\n").lower()
    requires_complete_document = (
        recovered
        or suspected
        or stripped.startswith(("<!doctype", "<html"))
    )
    if not requires_complete_document or _is_complete_html_document(projected):
        return projected
    raise ArtifactRendererTransientProtocolError(
        "Artifact renderer returned an incomplete HTML document"
    )


def _normalize_complete_renderer_content(
    content: str,
    *,
    path: str,
    append: bool,
    html_reference: str | None,
    preserve_leading_think_literal: bool,
) -> tuple[str, str]:
    """Apply the body renderer's reasoning and envelope rules atomically."""

    think_filter = InlineThinkTagFilter(leading_only=True)
    if preserve_leading_think_literal:
        think_filter.confirm_visible_channel()
    visible, reasoning = think_filter.push_parts(content)
    unclosed_thinking = think_filter.has_unclosed_thinking
    tail_visible, tail_reasoning = think_filter.finish_parts()
    if unclosed_thinking:
        raise ArtifactRendererTransientProtocolError(
            "Artifact renderer ended inside a reasoning envelope"
        )
    visible = f"{visible}{tail_visible}"
    reasoning = f"{reasoning}{tail_reasoning}"

    if not append and re.search(r"\.(?:md|markdown)\Z", path, re.IGNORECASE):
        projection = _MarkdownFenceProjection()
        projected = projection.push(visible)
        projected_tail, replacement = projection.finish()
        visible = replacement if replacement is not None else f"{projected}{projected_tail}"
    if not append and re.search(r"\.(?:html?|xhtml)\Z", path, re.IGNORECASE):
        visible = _normalize_complete_renderer_html(visible, html_reference)
    if not visible:
        raise RuntimeError("Artifact renderer returned no file body")
    return visible, reasoning


def _json_object_has_top_level_key(raw: str, expected_key: str) -> bool:
    """Recognize a top-level JSON object key in a possibly incomplete stream."""

    stack: list[str] = []
    index = 0
    length = len(raw)
    while index < length:
        char = raw[index]
        if char in " \t\r\n":
            index += 1
            continue
        if char == '"':
            index += 1
            token: list[str] = []
            escaped = False
            while index < length:
                char = raw[index]
                if escaped:
                    token.append(char)
                    escaped = False
                    index += 1
                    continue
                if char == "\\":
                    escaped = True
                    index += 1
                    continue
                if char == '"':
                    break
                token.append(char)
                index += 1
            if index >= length:
                return False
            index += 1
            lookahead = index
            while lookahead < length and raw[lookahead] in " \t\r\n":
                lookahead += 1
            if (
                stack == ["{"]
                and "".join(token) == expected_key
                and lookahead < length
                and raw[lookahead] == ":"
            ):
                return True
            continue
        if char == "{":
            stack.append("{")
        elif char == "[":
            stack.append("[")
        elif char == "}" and stack and stack[-1] == "{":
            stack.pop()
        elif char == "]" and stack and stack[-1] == "[":
            stack.pop()
        index += 1
    return False


def _decode_json_string_prefix(
    raw: str,
    start: int,
) -> tuple[str, int, bool]:
    """Decode one JSON string, including its currently complete prefix.

    OpenAI-compatible gateways may split an escape sequence (including a
    ``\\uXXXX`` pair) across arbitrary SSE frames.  The returned ``end`` is
    the next input offset for a complete string and ``len(raw)`` otherwise.
    Incomplete escapes are retained by the caller for the next cumulative
    parse and are never exposed as file bytes.
    """

    if start >= len(raw) or raw[start] != '"':
        raise ValueError("expected a JSON string")
    index = start + 1
    safe_end = index
    while index < len(raw):
        char = raw[index]
        if char == '"':
            token = raw[start : index + 1]
            try:
                return json.loads(token), index + 1, True
            except (TypeError, ValueError) as exc:
                raise ValueError("invalid JSON string") from exc
        if char == "\\":
            if index + 1 >= len(raw):
                break
            escape = raw[index + 1]
            if escape == "u":
                if index + 6 > len(raw):
                    break
                digits = raw[index + 2 : index + 6]
                if not re.fullmatch(r"[0-9a-fA-F]{4}", digits):
                    raise ValueError("invalid JSON unicode escape")
                index += 6
                safe_end = index
                continue
            if escape not in {'"', "\\", "/", "b", "f", "n", "r", "t"}:
                raise ValueError("invalid JSON escape")
            index += 2
            safe_end = index
            continue
        if ord(char) < 0x20:
            raise ValueError("unescaped JSON control character")
        index += 1
        safe_end = index

    inner = raw[start + 1 : safe_end]
    # Do not expose half of an escaped UTF-16 surrogate pair.  ``json.loads``
    # accepts a lone high surrogate, but serializing that transient value into
    # the canonical UTF-8 write_file stream is not safe.  The next SSE frame
    # will make the pair complete and the cumulative parse will include it.
    if re.search(r"\\u[dD][89aAbB][0-9a-fA-F]{2}\Z", inner):
        inner = inner[:-6]
    try:
        decoded = json.loads(f'"{inner}"')
    except (TypeError, ValueError) as exc:
        raise ValueError("invalid partial JSON string") from exc
    return decoded, len(raw), False


def _partial_top_level_json_string(
    raw: str,
    expected_key: str,
) -> tuple[str, bool, bool]:
    """Return a top-level JSON string value before its object is complete.

    The scanner is deliberately schema-small: it locates only a direct key of
    the root object while still tracking nested objects/arrays and escaped
    strings.  This lets the artifact renderer project ``content`` incrementally
    without trusting provider-authored ``path`` or ``append`` metadata.
    """

    index = 0
    while index < len(raw) and raw[index] in " \t\r\n":
        index += 1
    if index >= len(raw):
        return "", False, False
    if raw[index] != "{":
        raise ValueError("tool arguments must be a JSON object")
    stack = ["{"]
    index += 1
    while index < len(raw):
        char = raw[index]
        if char in " \t\r\n":
            index += 1
            continue
        if char == '"':
            token, end, complete = _decode_json_string_prefix(raw, index)
            if not complete:
                return "", False, False
            lookahead = end
            while lookahead < len(raw) and raw[lookahead] in " \t\r\n":
                lookahead += 1
            if (
                stack == ["{"]
                and lookahead < len(raw)
                and raw[lookahead] == ":"
                and token == expected_key
            ):
                value_start = lookahead + 1
                while value_start < len(raw) and raw[value_start] in " \t\r\n":
                    value_start += 1
                if value_start >= len(raw):
                    return "", True, False
                if raw[value_start] != '"':
                    raise ValueError(f"{expected_key} must be a JSON string")
                value, _value_end, value_complete = _decode_json_string_prefix(
                    raw,
                    value_start,
                )
                return value, True, value_complete
            index = end
            continue
        if char == "{":
            stack.append("{")
        elif char == "[":
            stack.append("[")
        elif char == "}":
            if not stack or stack[-1] != "{":
                raise ValueError("mismatched JSON object delimiter")
            stack.pop()
        elif char == "]":
            if not stack or stack[-1] != "[":
                raise ValueError("mismatched JSON array delimiter")
            stack.pop()
        index += 1
    return "", False, False


def _partial_top_level_json_boolean(
    raw: str,
    expected_key: str,
) -> tuple[bool | None, bool, bool]:
    """Return one complete top-level boolean from an unfinished JSON object.

    Compatible gateways sometimes serialize booleans as the canonical strings
    ``"true"``/``"false"``.  Accept those just like the terminal validator,
    but never infer a value from a prefix such as ``fal``.
    """

    index = 0
    while index < len(raw) and raw[index] in " \t\r\n":
        index += 1
    if index >= len(raw):
        return None, False, False
    if raw[index] != "{":
        raise ValueError("tool arguments must be a JSON object")
    stack = ["{"]
    index += 1
    while index < len(raw):
        char = raw[index]
        if char in " \t\r\n":
            index += 1
            continue
        if char == '"':
            token, end, complete = _decode_json_string_prefix(raw, index)
            if not complete:
                return None, False, False
            lookahead = end
            while lookahead < len(raw) and raw[lookahead] in " \t\r\n":
                lookahead += 1
            if (
                stack == ["{"]
                and lookahead < len(raw)
                and raw[lookahead] == ":"
                and token == expected_key
            ):
                value_start = lookahead + 1
                while value_start < len(raw) and raw[value_start] in " \t\r\n":
                    value_start += 1
                if value_start >= len(raw):
                    return None, True, False
                if raw[value_start] == '"':
                    value, value_end, value_complete = _decode_json_string_prefix(
                        raw,
                        value_start,
                    )
                    if not value_complete:
                        return None, True, False
                    normalized = _compatible_boolean(value)
                    if normalized is None:
                        raise ValueError(f"{expected_key} must be a JSON boolean")
                    while value_end < len(raw) and raw[value_end] in " \t\r\n":
                        value_end += 1
                    if value_end >= len(raw):
                        return None, True, False
                    if raw[value_end] not in ",}":
                        raise ValueError(f"{expected_key} must be a JSON boolean")
                    return normalized, True, True

                remainder = raw[value_start:]
                for literal, value in (("true", True), ("false", False)):
                    if remainder.startswith(literal):
                        value_end = value_start + len(literal)
                        while value_end < len(raw) and raw[value_end] in " \t\r\n":
                            value_end += 1
                        if value_end >= len(raw):
                            return None, True, False
                        if raw[value_end] not in ",}":
                            raise ValueError(
                                f"{expected_key} must be a JSON boolean"
                            )
                        return value, True, True
                    if len(remainder) < len(literal) and literal.startswith(remainder):
                        return None, True, False
                raise ValueError(f"{expected_key} must be a JSON boolean")
            index = end
            continue
        if char == "{":
            stack.append("{")
        elif char == "[":
            stack.append("[")
        elif char == "}":
            if not stack or stack[-1] != "{":
                raise ValueError("mismatched JSON object delimiter")
            stack.pop()
        elif char == "]":
            if not stack or stack[-1] != "[":
                raise ValueError("mismatched JSON array delimiter")
            stack.pop()
        index += 1
    return None, False, False


def _top_level_json_string_is_delimited(raw: str, expected_key: str) -> bool:
    """Confirm a complete root string is followed by a JSON delimiter."""

    index = 0
    while index < len(raw) and raw[index] in " \t\r\n":
        index += 1
    if index >= len(raw) or raw[index] != "{":
        return False
    stack = ["{"]
    index += 1
    while index < len(raw):
        char = raw[index]
        if char in " \t\r\n":
            index += 1
            continue
        if char == '"':
            token, end, complete = _decode_json_string_prefix(raw, index)
            if not complete:
                return False
            lookahead = end
            while lookahead < len(raw) and raw[lookahead] in " \t\r\n":
                lookahead += 1
            if (
                stack == ["{"]
                and lookahead < len(raw)
                and raw[lookahead] == ":"
                and token == expected_key
            ):
                value_start = lookahead + 1
                while value_start < len(raw) and raw[value_start] in " \t\r\n":
                    value_start += 1
                if value_start >= len(raw) or raw[value_start] != '"':
                    return False
                _value, value_end, value_complete = _decode_json_string_prefix(
                    raw,
                    value_start,
                )
                if not value_complete:
                    return False
                while value_end < len(raw) and raw[value_end] in " \t\r\n":
                    value_end += 1
                return value_end < len(raw) and raw[value_end] in ",}"
            index = end
            continue
        if char == "{":
            stack.append("{")
        elif char == "[":
            stack.append("[")
        elif char == "}":
            if not stack or stack[-1] != "{":
                return False
            stack.pop()
        elif char == "]":
            if not stack or stack[-1] != "[":
                return False
            stack.pop()
        index += 1
    return False


def _complete_selection_alias_metadata(raw: str) -> dict[str, Any] | None:
    """Extract a safe prepare hand-off before a provider starts file bytes."""

    try:
        description, description_found, description_complete = (
            _partial_top_level_json_string(raw, "description")
        )
        path, path_found, path_complete = _partial_top_level_json_string(
            raw,
            "path",
        )
        append, append_found, append_complete = _partial_top_level_json_boolean(
            raw,
            "append",
        )
    except ValueError:
        # The terminal path reports a precise protocol failure.  While the JSON
        # is still arriving, an invalid-looking suffix may merely be split at a
        # token boundary, so do not abort or guess here.
        return None

    if not all(
        (
            description_found,
            description_complete,
            path_found,
            path_complete,
            append_found,
            append_complete,
        )
    ):
        return None
    if not _top_level_json_string_is_delimited(
        raw,
        "description",
    ) or not _top_level_json_string_is_delimited(raw, "path"):
        return None
    description = description.strip() or "Write the prepared file"
    path = path.strip()
    if not path.startswith("/") or append is None:
        return None
    return {
        "description": description,
        "path": path,
        "append": append,
    }


def _pending_from_buffered_gateway_selection(
    buffered: _SelectionAliasCallState,
    metadata: Mapping[str, Any],
    *,
    allow_implicit_think_close: bool,
) -> _PendingWriteFileBridge:
    """Create a server-owned body transaction without a graph tool round."""

    transaction_token = uuid.uuid5(
        uuid.NAMESPACE_OID,
        "\x1f".join(
            (
                buffered.call_id,
                str(metadata["path"]),
                uuid.uuid4().hex,
            )
        ),
    ).hex[:24]
    return _PendingWriteFileBridge(
        # Retained for the legacy handshake data shape only.  This id is never
        # emitted to LangGraph or executed by ToolNode in buffered_gateway mode.
        begin_tool_call_id=buffered.call_id,
        write_tool_call_id=f"{_WRITE_FILE_BRIDGE_CALL_PREFIX}{transaction_token}",
        path=str(metadata["path"]),
        description=str(metadata["description"]),
        append=bool(metadata["append"]),
        # Enable close-only probing only when the selection response actually
        # demonstrated that provider convention. Enabling it unconditionally
        # would buffer an ordinary file body until the terminal frame and
        # defeat the live preview this transport exists to restore.
        allow_implicit_think_close=allow_implicit_think_close,
    )


def _failed_tool_result(message: Any) -> bool:
    if not isinstance(message, Mapping) or message.get("role") != "tool":
        return False
    if str(message.get("status") or "").lower() == "error":
        return True
    content = _message_content_text(message.get("content")).strip().lower()
    return content.startswith("error:") or content.startswith("command blocked:")


def _strip_failed_artifact_write_attempts(
    messages: list[Any],
    pending: _PendingWriteFileBridge,
) -> list[Any]:
    """Remove failed same-target writer attempts from renderer source context.

    A weak model may first emit metadata-only ``write_file`` and then place an
    entire artifact in a failed ``bash`` heredoc. Replaying those calls into the
    isolated renderer both wastes the context window and teaches it to repeat
    the same orchestration mistake. Keep successful files and all unrelated
    tool evidence; remove only failed ``write_file``/``bash`` pairs that target
    the already prepared path.
    """

    failed_results = {
        str(message.get("tool_call_id") or "")
        for message in messages
        if _failed_tool_result(message)
    }
    failed_results.discard("")
    if not failed_results:
        return messages

    removable_ids: set[str] = set()
    for message in messages:
        if not isinstance(message, Mapping) or message.get("role") != "assistant":
            continue
        tool_calls = message.get("tool_calls")
        if not isinstance(tool_calls, list):
            continue
        for call in tool_calls:
            if not isinstance(call, Mapping):
                continue
            call_id = str(call.get("id") or "")
            if call_id not in failed_results:
                continue
            name = _payload_tool_name(call)
            args, raw_args = _payload_tool_arguments(call)
            if name == _WRITE_FILE_TOOL_NAME and str(args.get("path") or "") == pending.path:
                removable_ids.add(call_id)
                continue
            if name == "bash":
                command = args.get("command")
                command_text = command if isinstance(command, str) else raw_args
                if pending.path in command_text:
                    removable_ids.add(call_id)

    if not removable_ids:
        return messages

    cleaned: list[Any] = []
    for message in messages:
        if not isinstance(message, Mapping):
            cleaned.append(message)
            continue
        if (
            message.get("role") == "tool"
            and str(message.get("tool_call_id") or "") in removable_ids
        ):
            continue
        if message.get("role") != "assistant":
            cleaned.append(message)
            continue
        tool_calls = message.get("tool_calls")
        if not isinstance(tool_calls, list):
            cleaned.append(message)
            continue
        visible_calls = [
            call
            for call in tool_calls
            if not (
                isinstance(call, Mapping)
                and str(call.get("id") or "") in removable_ids
            )
        ]
        if not visible_calls and len(visible_calls) != len(tool_calls):
            # Its preface/reasoning described the failed orchestration rather
            # than artifact content; do not retain a misleading assistant turn.
            continue
        if len(visible_calls) != len(tool_calls):
            rewritten = dict(message)
            rewritten["tool_calls"] = visible_calls
            cleaned.append(rewritten)
        else:
            cleaned.append(message)
    return cleaned


def _artifact_renderer_messages(
    messages: list[Any],
    pending: _PendingWriteFileBridge,
    *,
    has_control_exchange: bool = True,
) -> list[Any]:
    """Build a renderer-only request without the lead-agent delivery prompt.

    Non-system conversation turns and completed tool evidence remain available
    as source material.  The current ``begin_write_file`` assistant/tool pair
    is transport control and is replaced by one server-owned renderer request;
    in particular, any conversational preface on that assistant turn cannot
    become file bytes or steer the renderer into another tool call. The server
    owns the target metadata, while the provider's ordinary content channel is
    the exact file-body channel.
    """

    if has_control_exchange:
        cleaned = _strip_older_begin_write_file_exchanges(messages)
        historical = cleaned[:-2]
    else:
        # The buffered-gateway path never publishes or executes a preparation
        # call.  Reuse the original conversation directly while still removing
        # any historical controls left by an older compatibility model.
        historical = _strip_historical_begin_write_file_exchanges(messages)
    historical_messages = _strip_failed_artifact_write_attempts(
        historical,
        pending,
    )
    source_messages = [
        dict(message)
        for message in historical_messages
        if isinstance(message, Mapping) and message.get("role") != "system"
    ]
    agent_system_context = [
        _message_content_text(message.get("content"))
        for message in historical_messages
        if isinstance(message, Mapping)
        and message.get("role") == "system"
        and _message_content_text(message.get("content"))
    ]
    renderer_system = _ARTIFACT_RENDERER_SYSTEM_INSTRUCTION
    if agent_system_context:
        # Preserve the content/style/safety contract of custom agents without
        # restoring their orchestration instructions as active chat turns.
        # The server-owned transaction remains the trust boundary; this context
        # can improve artifact fidelity but cannot change the prepared path,
        # append intent, tool catalogue, or reasoning/content channel mapping.
        renderer_system = (
            f"{renderer_system}\n\n"
            "The following trusted original agent system context remains "
            "applicable only to artifact content, style, and safety. Its "
            "tool-selection, delivery, and user-facing response directions "
            "are superseded by the renderer transaction above:\n"
            + json.dumps(
                agent_system_context,
                ensure_ascii=False,
                separators=(",", ":"),
            )
            + "\nThe server-owned file-body transaction remains controlling."
        )
    prepared = json.dumps(
        {
            "path": pending.path,
            "description": pending.description,
            "append": pending.append,
        },
        ensure_ascii=False,
        separators=(",", ":"),
    )
    return [
        {"role": "system", "content": renderer_system},
        *source_messages,
        {
            "role": "user",
            "content": (
                "Render the prepared artifact now. The server owns the target "
                f"metadata and will perform the final write: {prepared}"
            ),
        },
    ]


def _has_pending_begin_exchange(
    messages: list[Any],
    pending: _PendingWriteFileBridge,
) -> bool:
    """Validate the immediately preceding internal prepare exchange.

    The accepted assistant/tool pair is intentionally retained in the body
    request. It is the causal hand-off that tells the provider the internal
    prepare call succeeded and that the next response is the prepared body.
    Dropping it leaves the lead-agent system prompt asking for ``write_file``
    while the body request advertises no tools; compatible gateways may then
    serialize a tool call as ordinary ``<write_file>{...}`` text.
    """

    if len(messages) < 2:
        return False
    tool_result = messages[-1]
    assistant = messages[-2]
    if not isinstance(tool_result, dict) or tool_result.get("role") != "tool":
        return False
    if str(tool_result.get("tool_call_id") or "") != pending.begin_tool_call_id:
        return False
    if not isinstance(assistant, dict) or assistant.get("role") != "assistant":
        return False
    tool_calls = assistant.get("tool_calls")
    if not isinstance(tool_calls, list) or len(tool_calls) != 1:
        return False
    call = tool_calls[0]
    function = call.get("function") if isinstance(call, Mapping) else None
    if (
        not isinstance(call, Mapping)
        or str(call.get("id") or "") != pending.begin_tool_call_id
        or not isinstance(function, Mapping)
        or function.get("name") != _BEGIN_WRITE_FILE_TOOL_NAME
    ):
        return False
    return True


def _strip_older_begin_write_file_exchanges(
    messages: list[Any],
) -> list[Any]:
    """Hide historical bridge controls while preserving the current hand-off.

    The last assistant/tool pair has already been validated by
    :func:`_has_pending_begin_exchange`. Older bridge exchanges are transport
    implementation details and must not accumulate in provider context.
    """

    historical = _strip_historical_begin_write_file_exchanges(messages[:-2])
    return [*historical, *messages[-2:]]


def _rewrite_write_file_tool_choice(payload: dict[str, Any]) -> None:
    """Map an explicitly forced writer to the bridge's prepare tool."""

    choice = payload.get("tool_choice")
    if choice == _WRITE_FILE_TOOL_NAME:
        payload["tool_choice"] = _BEGIN_WRITE_FILE_TOOL_NAME
        return
    if not isinstance(choice, dict):
        return
    function = choice.get("function")
    if isinstance(function, dict) and function.get("name") == _WRITE_FILE_TOOL_NAME:
        rewritten_function = dict(function)
        rewritten_function["name"] = _BEGIN_WRITE_FILE_TOOL_NAME
        rewritten_choice = dict(choice)
        rewritten_choice["function"] = rewritten_function
        payload["tool_choice"] = rewritten_choice


def _rewrite_artifact_renderer_tool_choice(payload: dict[str, Any]) -> None:
    """Keep forced choices aligned with the renderer's provider-facing alias."""

    choice = payload.get("tool_choice")
    if choice == _BEGIN_WRITE_FILE_TOOL_NAME:
        payload["tool_choice"] = _WRITE_FILE_TOOL_NAME
        return
    if not isinstance(choice, dict):
        return
    function = choice.get("function")
    if isinstance(function, dict) and function.get("name") == _BEGIN_WRITE_FILE_TOOL_NAME:
        rewritten_function = dict(function)
        rewritten_function["name"] = _WRITE_FILE_TOOL_NAME
        rewritten_choice = dict(choice)
        rewritten_choice["function"] = rewritten_function
        payload["tool_choice"] = rewritten_choice


def _configure_write_file_bridge_payload(
    payload: dict[str, Any],
    original_messages: list[Any],
    state: _WriteFileBridgeStreamState,
) -> None:
    """Apply the model-scoped compatibility transport to one provider request."""

    if state.transport not in _WRITE_FILE_BRIDGE_TRANSPORTS:
        # ``begin_write_file`` is an internal compatibility control tool. Even
        # if a deployment accidentally registers it globally, native models see
        # the same tool catalogue they had before this feature existed.
        _hide_begin_write_file_for_native_request(payload)
        return
    payload_messages = payload.get("messages")
    if not isinstance(payload_messages, list):
        state.pending = None
        # Responses-style payloads use a different input container. This
        # bridge targets chat-completions messages, so degrade safely to the
        # native writer and never expose an unusable handshake.
        _hide_begin_write_file_for_native_request(payload)
        return

    tools = payload.get("tools")
    tool_names = {_payload_tool_name(tool) for tool in tools} if isinstance(tools, list) else set()
    if state.transport in _ARTIFACT_BODY_TRANSPORTS:
        state.renderer_html_references = {
            path: content
            for path, content in _successful_read_file_contents(
                original_messages
            ).items()
            if _is_complete_html_document(content)
        }
    direct_body_request = (
        state.transport == "buffered_gateway"
        and state.buffered_gateway_body_request
        and state.pending is not None
    )
    pending = (
        state.pending
        if direct_body_request
        else _pending_write_file_bridge(original_messages)
    )
    if pending is not None and (
        direct_body_request
        or _has_pending_begin_exchange(payload_messages, pending)
    ):
        state.pending = pending
        state.inline_think_filter = InlineThinkTagFilter(
            allow_implicit_close=pending.allow_implicit_think_close,
            leading_only=True,
        )
        state.markdown_fence_projection = (
            _MarkdownFenceProjection()
            if state.transport in _ARTIFACT_BODY_TRANSPORTS
            and not pending.append
            and re.search(r"\.(?:md|markdown)\Z", pending.path, re.IGNORECASE)
            else None
        )
        if state.transport in _ARTIFACT_BODY_TRANSPORTS:
            state.renderer_html_reference = state.renderer_html_references.get(
                pending.path
            )
            payload["messages"] = _artifact_renderer_messages(
                payload_messages,
                pending,
                has_control_exchange=not direct_body_request,
            )
        else:
            payload_messages = _strip_older_begin_write_file_exchanges(
                payload_messages
            )
            payload["messages"] = payload_messages
            prepared = json.dumps(
                {
                    "path": pending.path,
                    "description": pending.description,
                    "append": pending.append,
                },
                ensure_ascii=False,
                separators=(",", ":"),
            )
            _append_system_instruction(
                payload_messages,
                f"{_TEXT_BRIDGE_BODY_INSTRUCTION}\nPrepared target: {prepared}",
            )
        # The graph still owns and executes the canonical write_file assembled
        # from the content stream. Artifact renderers retain only bounded
        # prerequisite tools, so a model can defer instead of hallucinating a
        # read call into an otherwise tool-less body round. Mutation, shell and
        # writer tools remain unavailable. Legacy text_bridge stays text-only.
        prerequisite_tools = (
            [
                tool
                for tool in tools
                if _payload_tool_name(tool)
                in _ARTIFACT_RENDERER_PREREQUISITE_TOOLS
            ]
            if state.transport in _ARTIFACT_BODY_TRANSPORTS
            and isinstance(tools, list)
            else []
        )
        state.renderer_prerequisite_tools = frozenset(
            _payload_tool_name(tool) for tool in prerequisite_tools
        )
        if prerequisite_tools:
            payload["tools"] = prerequisite_tools
            payload["parallel_tool_calls"] = False
        else:
            payload.pop("tools", None)
            payload.pop("parallel_tool_calls", None)
        for key in ("tool_choice", "functions", "function_call"):
            payload.pop(key, None)
        return
    state.pending = None

    historical_messages = _strip_historical_begin_write_file_exchanges(
        payload_messages
    )
    if historical_messages is not payload_messages:
        payload["messages"] = historical_messages
        payload_messages = historical_messages

    if not isinstance(tools, list):
        return
    if state.transport == "buffered_gateway":
        if _WRITE_FILE_TOOL_NAME not in tool_names:
            # No writer means there is nothing to adapt. Keep every other tool
            # untouched and ensure a stale internal control cannot leak out.
            _hide_begin_write_file_for_native_request(payload)
            return
        state.selection_write_file_alias = True
        payload["tools"] = [
            _buffered_gateway_selection_tool(tool)
            for tool in tools
            if _payload_tool_name(tool) != _BEGIN_WRITE_FILE_TOOL_NAME
        ]
        payload["parallel_tool_calls"] = False
        _rewrite_artifact_renderer_tool_choice(payload)
        _append_system_instruction(
            payload_messages,
            _BUFFERED_GATEWAY_SELECTION_INSTRUCTION,
        )
        return
    if not {
        _WRITE_FILE_TOOL_NAME,
        _BEGIN_WRITE_FILE_TOOL_NAME,
    }.issubset(tool_names):
        # A partial registration cannot complete the handshake. Preserve the
        # original native writer and hide the internal companion.
        _hide_begin_write_file_for_native_request(payload)
        return
    state.selection_write_file_alias = state.transport == "artifact_renderer"
    payload["tools"] = [
        _selection_tool(
            tool,
            provider_name=(
                _WRITE_FILE_TOOL_NAME
                if state.selection_write_file_alias
                else None
            ),
        )
        for tool in tools
        if _payload_tool_name(tool) != _WRITE_FILE_TOOL_NAME
    ]
    if state.selection_write_file_alias:
        payload["parallel_tool_calls"] = False
        _rewrite_artifact_renderer_tool_choice(payload)
    else:
        _rewrite_write_file_tool_choice(payload)
    _append_system_instruction(
        payload_messages,
        _ARTIFACT_RENDERER_SELECTION_INSTRUCTION
        if state.transport == "artifact_renderer"
        else _TEXT_BRIDGE_SELECTION_INSTRUCTION,
    )


def _json_string_fragment(value: str) -> str:
    return json.dumps(value, ensure_ascii=False)[1:-1]


def _compatible_boolean(value: Any) -> bool | None:
    """Normalize only booleans and their canonical JSON text spellings."""

    if isinstance(value, bool):
        return value
    if isinstance(value, str) and value.strip().lower() in {"true", "false"}:
        return value.strip().lower() == "true"
    return None


def _rewrite_selection_write_file_alias(
    chunk: dict[str, Any],
    state: _WriteFileBridgeStreamState,
) -> None:
    """Turn the provider-facing alias into the metadata-only prepare control.

    Some gateways ignore the advertised schema and start generating a large
    ``content`` property anyway.  Keeping that direct writer as a fallback
    recreates the original problem: the frontend receives no canonical
    ``write_file`` until the whole JSON argument has finished.  As soon as all
    metadata is complete, synthesize a terminal ``begin_write_file`` and close
    this upstream stream.  LangGraph then runs the normal isolated body round,
    whose text is projected incrementally as canonical ``write_file`` chunks.

    If the gateway returns the whole call atomically, its content is discarded
    at the real terminal frame and only the prepare metadata is exposed.  This
    keeps one invariant for every opted-in renderer model: file bytes always
    travel through the streamable body round, never through a hidden bypass.
    """

    if (
        state.transport not in _ARTIFACT_BODY_TRANSPORTS
        or state.pending is not None
        or not state.selection_write_file_alias
    ):
        return
    for choice in _stream_choices(chunk):
        delta = choice.get("delta")
        if not isinstance(delta, dict):
            continue
        if _extract_reasoning_content(delta):
            state.selection_alias_has_native_reasoning = True
        selection_text = _message_content_text(delta.get("content"))
        if selection_text:
            probe = (
                f"{state.selection_alias_think_probe_suffix}{selection_text}"
            ).lower()
            open_at = probe.find("<think>")
            close_at = probe.find("</think>")
            if open_at >= 0 and (close_at < 0 or open_at < close_at):
                state.selection_alias_saw_think_open = True
            if close_at >= 0 and not state.selection_alias_saw_think_open:
                state.selection_alias_implicit_think_close = True
            state.selection_alias_think_probe_suffix = probe[-8:]
        tool_calls = delta.get("tool_calls")
        visible_calls: list[dict[str, Any]] = []
        saw_unrelated_call = False
        if isinstance(tool_calls, list):
            for position, tool_call in enumerate(tool_calls):
                if not isinstance(tool_call, dict):
                    continue
                try:
                    index = int(tool_call.get("index", position))
                except (TypeError, ValueError):
                    index = position
                function = tool_call.get("function")
                if not isinstance(function, dict):
                    visible_calls.append(tool_call)
                    saw_unrelated_call = True
                    continue
                name_fragment = function.get("name")
                name_fragment = (
                    name_fragment if isinstance(name_fragment, str) else ""
                )
                buffered = state.selection_alias_calls.get(index)
                if buffered is None:
                    if not name_fragment or not _WRITE_FILE_TOOL_NAME.startswith(
                        name_fragment
                    ):
                        visible_calls.append(tool_call)
                        saw_unrelated_call = True
                        continue
                    buffered = _SelectionAliasCallState(index=index)
                    state.selection_alias_calls[index] = buffered

                call_id = str(tool_call.get("id") or "")
                if call_id and not buffered.call_id:
                    buffered.call_id = call_id
                call_type = str(tool_call.get("type") or "")
                if call_type:
                    buffered.call_type = call_type
                buffered.name = _merge_compatible_fragment(
                    buffered.name,
                    name_fragment,
                )
                if not _WRITE_FILE_TOOL_NAME.startswith(buffered.name):
                    raise RuntimeError(
                        "Artifact renderer returned a malformed write_file alias"
                    )
                arguments_fragment = function.get("arguments")
                arguments_fragment = (
                    arguments_fragment
                    if isinstance(arguments_fragment, str)
                    else ""
                )
                buffered.arguments = _merge_compatible_fragment(
                    buffered.arguments,
                    arguments_fragment,
                )
        if len(state.selection_alias_calls) > 1:
            raise RuntimeError(
                "Artifact renderer emitted more than one write_file call"
            )
        if state.selection_alias_calls and (saw_unrelated_call or visible_calls):
            raise RuntimeError(
                "Artifact renderer emitted write_file in parallel with another tool"
            )

        if visible_calls:
            delta["tool_calls"] = visible_calls
        else:
            delta.pop("tool_calls", None)

        finish_reason = str(choice.get("finish_reason") or "")
        if not finish_reason and len(state.selection_alias_calls) == 1:
            buffered = next(iter(state.selection_alias_calls.values()))
            metadata = (
                _complete_selection_alias_metadata(buffered.arguments)
                if buffered.name == _WRITE_FILE_TOOL_NAME and buffered.call_id
                else None
            )
            if metadata is not None:
                try:
                    _content, content_found, content_complete = (
                        _partial_top_level_json_string(
                            buffered.arguments,
                            "content",
                        )
                    )
                except ValueError:
                    content_found = content_complete = False
                # A complete content value in the same frame has already paid
                # the generation cost. Let the terminal path preserve any
                # leading reasoning before it discards those bytes. An absent
                # or still-open content value is the useful cancellation point.
                if not (content_found and content_complete):
                    buffered.decision = "begin"
                    buffered.emitted = True
                    if state.transport == "buffered_gateway":
                        state.pending = _pending_from_buffered_gateway_selection(
                            buffered,
                            metadata,
                            allow_implicit_think_close=(
                                state.selection_alias_implicit_think_close
                                and not state.selection_alias_has_native_reasoning
                            ),
                        )
                        # Keep this as one logical assistant stream. The
                        # adapter closes the buffered tool stream and starts
                        # the ordinary-text body request before yielding a
                        # terminal frame to LangGraph.
                        delta.pop("tool_calls", None)
                        choice["finish_reason"] = None
                    else:
                        delta["tool_calls"] = [
                            {
                                "index": buffered.index,
                                "id": buffered.call_id,
                                "type": buffered.call_type,
                                "function": {
                                    "name": _BEGIN_WRITE_FILE_TOOL_NAME,
                                    "arguments": json.dumps(
                                        metadata,
                                        ensure_ascii=False,
                                        separators=(",", ":"),
                                    ),
                                },
                            }
                        ]
                        choice["finish_reason"] = "tool_calls"
                    state.selection_alias_early_stop = True
                    diagnostics = _ACTIVE_TOOL_STREAM.get()
                    if diagnostics is not None:
                        diagnostics.finish_reason = (
                            "none"
                            if state.transport == "buffered_gateway"
                            else "tool_calls"
                        )
                        diagnostics.intentional_early_finish = True
                    return
        if not finish_reason:
            continue
        if state.selection_alias_calls and finish_reason not in {
            "tool_calls",
            "function_call",
            "stop",
            "end_turn",
        }:
            state.aborted = True
            raise ArtifactRendererTransientProtocolError(
                "Artifact renderer did not complete the write_file alias"
            )
        terminal_calls: list[dict[str, Any]] = list(
            delta.get("tool_calls") or []
        )
        for index, buffered in sorted(state.selection_alias_calls.items()):
            if buffered.emitted:
                continue
            if buffered.name != _WRITE_FILE_TOOL_NAME:
                raise ArtifactRendererTransientProtocolError(
                    "Artifact renderer returned an incomplete write_file alias"
                )
            terminal_arguments = buffered.arguments
            try:
                parsed_arguments = json.loads(terminal_arguments)
            except (TypeError, ValueError) as exc:
                raise ArtifactRendererTransientProtocolError(
                    "Artifact renderer returned invalid write_file arguments"
                ) from exc
            if not isinstance(parsed_arguments, Mapping):
                raise RuntimeError(
                    "Artifact renderer returned invalid write_file arguments"
                )
            path_value = parsed_arguments.get("path")
            description_value = parsed_arguments.get("description")
            path = path_value.strip() if isinstance(path_value, str) else ""
            description = (
                description_value.strip()
                if isinstance(description_value, str)
                else ""
            )
            append = _compatible_boolean(parsed_arguments.get("append", False))
            if not path.startswith("/") or append is None:
                raise RuntimeError(
                    "Artifact renderer returned invalid write_file arguments"
                )
            content = parsed_arguments.get("content")
            if "content" in parsed_arguments:
                if not isinstance(content, str):
                    raise RuntimeError(
                        "Artifact renderer returned invalid write_file arguments"
                    )
                think_filter = InlineThinkTagFilter(leading_only=True)
                if state.selection_alias_has_native_reasoning:
                    think_filter.confirm_visible_channel()
                _visible, inline_reasoning = think_filter.push_parts(content)
                _tail_visible, tail_reasoning = think_filter.finish_parts()
                _append_bridge_reasoning_delta(
                    delta,
                    f"{inline_reasoning}{tail_reasoning}",
                )
            buffered.decision = "begin"
            metadata = {
                "description": description or "Write the prepared file",
                "path": path,
                "append": append,
            }
            if state.transport == "buffered_gateway":
                state.pending = _pending_from_buffered_gateway_selection(
                    buffered,
                    metadata,
                    allow_implicit_think_close=(
                        state.selection_alias_implicit_think_close
                        and not state.selection_alias_has_native_reasoning
                    ),
                )
                choice["finish_reason"] = None
                state.selection_alias_early_stop = True
            else:
                terminal_arguments = json.dumps(
                    metadata,
                    ensure_ascii=False,
                    separators=(",", ":"),
                )
                terminal_calls.append(
                    {
                        "index": index,
                        "id": buffered.call_id,
                        "type": buffered.call_type,
                        "function": {
                            "name": _BEGIN_WRITE_FILE_TOOL_NAME,
                            "arguments": terminal_arguments,
                        },
                    }
                )
            buffered.emitted = True
        if terminal_calls:
            delta["tool_calls"] = terminal_calls


def _append_bridge_reasoning_delta(delta: dict[str, Any], reasoning: str) -> None:
    """Route inline provider thinking through the normal reasoning channel.

    The bridge operates before LangChain converts a raw provider chunk. Putting
    the separated trace on ``reasoning_content`` lets the existing converter
    and frontend render it exactly like native provider reasoning, while the
    synthetic tool arguments receive only confirmed file bytes.
    """

    if not reasoning:
        return
    existing = _extract_reasoning_content(delta)
    if existing:
        delta["reasoning_content"] = (
            existing if reasoning in existing else f"{existing}{reasoning}"
        )
    else:
        delta["reasoning_content"] = reasoning


def _write_file_argument_prefix(pending: _PendingWriteFileBridge) -> str:
    metadata = json.dumps(
        {
            "description": pending.description,
            "path": pending.path,
            # LangChain best-effort parses unfinished JSON. Keep every
            # intermediate call schema-invalid; the normal finishing frame
            # appends a duplicate key with the boolean commit value (JSON
            # last-key-wins).
            "append": _WRITE_FILE_BRIDGE_INCOMPLETE,
        },
        ensure_ascii=False,
        separators=(",", ":"),
    )
    return f'{metadata[:-1]},"content":"'


def _synthetic_write_tool_delta(
    pending: _PendingWriteFileBridge,
    arguments: str,
    *,
    first: bool,
) -> dict[str, Any]:
    function: dict[str, Any] = {"arguments": arguments}
    tool_call: dict[str, Any] = {"index": 0, "function": function}
    if first:
        tool_call.update({"id": pending.write_tool_call_id, "type": "function"})
        function["name"] = _WRITE_FILE_TOOL_NAME
    return tool_call


def _merge_compatible_fragment(previous: str, current: str) -> str:
    """Merge either a true delta or a growing compatible-gateway snapshot."""

    if not current:
        return previous
    if not previous:
        return current
    if current == previous:
        return previous
    if current.startswith(previous):
        return current
    return f"{previous}{current}"


def _capture_renderer_fallback_write(
    tool_calls: Any,
    state: _WriteFileBridgeStreamState,
) -> str:
    """Capture one provider-authored writer call inside a renderer body turn.

    The ordinary content channel remains the preferred streaming path. This is
    a compatibility fallback for weak models that ignore the no-tools renderer
    contract and emit ``write_file`` anyway. The cumulative argument buffer is
    retained for terminal validation against the server-owned target, while a
    separate parser may project only the top-level ``content`` string into the
    canonical synthetic writer stream. No sandbox write executes before the
    complete call passes validation.
    """

    if not isinstance(tool_calls, list) or len(tool_calls) != 1:
        raise RuntimeError(
            "Artifact renderer emitted an unexpected provider tool call"
        )
    tool_call = tool_calls[0]
    if not isinstance(tool_call, Mapping):
        raise RuntimeError(
            "Artifact renderer emitted an unexpected provider tool call"
        )
    try:
        index = int(tool_call.get("index", 0))
    except (TypeError, ValueError):
        index = 0
    call_id = str(tool_call.get("id") or "")
    function = tool_call.get("function")
    if not isinstance(function, Mapping):
        raise RuntimeError(
            "Artifact renderer emitted an unexpected provider tool call"
        )

    name_fragment = function.get("name")
    name_fragment = name_fragment if isinstance(name_fragment, str) else ""
    arguments_fragment = function.get("arguments")
    arguments_fragment = (
        arguments_fragment if isinstance(arguments_fragment, str) else ""
    )

    if not state.renderer_fallback_active:
        if not name_fragment or not _WRITE_FILE_TOOL_NAME.startswith(name_fragment):
            raise RuntimeError(
                "Artifact renderer emitted an unexpected provider tool call"
            )
        state.renderer_fallback_active = True
        state.renderer_fallback_index = index
        state.renderer_fallback_call_id = call_id
    elif state.renderer_fallback_index != index:
        raise RuntimeError(
            "Artifact renderer emitted an unexpected provider tool call"
        )
    elif (
        call_id
        and state.renderer_fallback_call_id
        and call_id != state.renderer_fallback_call_id
    ):
        raise RuntimeError(
            "Artifact renderer emitted an unexpected provider tool call"
        )
    elif call_id and not state.renderer_fallback_call_id:
        state.renderer_fallback_call_id = call_id

    state.renderer_fallback_name = _merge_compatible_fragment(
        state.renderer_fallback_name,
        name_fragment,
    )
    if not _WRITE_FILE_TOOL_NAME.startswith(state.renderer_fallback_name):
        raise RuntimeError(
            "Artifact renderer emitted an unexpected provider tool call"
        )
    new_arguments = state.renderer_fallback_arguments.merge(arguments_fragment)
    try:
        return state.renderer_fallback_content.feed(new_arguments)
    except ValueError as exc:
        raise ArtifactRendererTransientProtocolError(
            "Artifact renderer returned invalid write_file arguments"
        ) from exc


def _validated_renderer_fallback_content(
    state: _WriteFileBridgeStreamState,
    pending: _PendingWriteFileBridge,
) -> str:
    if state.renderer_fallback_name != _WRITE_FILE_TOOL_NAME:
        raise ArtifactRendererTransientProtocolError(
            "Artifact renderer returned an incomplete write_file call"
        )
    try:
        raw_arguments = state.renderer_fallback_arguments.materialize()
        arguments = json.loads(raw_arguments)
    except (TypeError, ValueError) as exc:
        raise ArtifactRendererTransientProtocolError(
            "Artifact renderer returned invalid write_file arguments"
        ) from exc
    if not isinstance(arguments, Mapping):
        raise RuntimeError("Artifact renderer returned invalid write_file arguments")
    if str(arguments.get("path") or "") != pending.path:
        raise RuntimeError("Artifact renderer attempted a different file path")
    # Several OpenAI-compatible tool parsers preserve a JSON-schema boolean as
    # its textual spelling. Arbitrary truthy strings remain a protocol error.
    append = _compatible_boolean(arguments.get("append", pending.append))
    if append is None or append != pending.append:
        raise RuntimeError("Artifact renderer changed the prepared append intent")
    content = arguments.get("content")
    if not isinstance(content, str) or not content:
        raise RuntimeError("Artifact renderer returned no file body")
    try:
        streamed_content = state.renderer_fallback_content.content()
    except ValueError as exc:
        raise ArtifactRendererTransientProtocolError(
            "Artifact renderer returned invalid write_file arguments"
        ) from exc
    if (
        not state.renderer_fallback_content.complete
        or streamed_content != content
    ):
        raise ArtifactRendererTransientProtocolError(
            "Artifact renderer returned an incomplete write_file call"
        )
    return content


def _capture_renderer_prerequisite_calls(
    tool_calls: Any,
    state: _WriteFileBridgeStreamState,
) -> bool:
    """Return true when every provider call is an advertised prerequisite."""

    if not isinstance(tool_calls, list) or not tool_calls:
        return False
    if len(tool_calls) != 1:
        raise RuntimeError(
            "Artifact renderer emitted parallel prerequisite tool calls"
        )
    captured: dict[int, str] = dict(state.renderer_deferred_tool_names)
    for position, tool_call in enumerate(tool_calls):
        if not isinstance(tool_call, Mapping):
            return False
        try:
            index = int(tool_call.get("index", position))
        except (TypeError, ValueError):
            index = position
        if captured and index not in captured:
            raise RuntimeError(
                "Artifact renderer emitted parallel prerequisite tool calls"
            )
        function = tool_call.get("function")
        if not isinstance(function, Mapping):
            return False
        fragment = function.get("name")
        fragment = fragment if isinstance(fragment, str) else ""
        previous = captured.get(index, "")
        candidate = _merge_compatible_fragment(previous, fragment)
        if not candidate:
            return False
        if not any(
            allowed.startswith(candidate)
            for allowed in state.renderer_prerequisite_tools
        ):
            return False
        captured[index] = candidate
    state.renderer_deferred_tool_names = captured
    state.renderer_deferred = True
    return True


def _route_renderer_deferred_content(
    delta: dict[str, Any],
    state: _WriteFileBridgeStreamState,
) -> None:
    """Keep reasoning visible but reject body bytes mixed with a deferred tool."""

    control_text = _message_content_text(delta.get("content"))
    if control_text:
        if _extract_reasoning_content(delta):
            state.inline_think_filter.confirm_visible_channel()
        visible, inline_reasoning = state.inline_think_filter.push_parts(control_text)
        _append_bridge_reasoning_delta(delta, inline_reasoning)
        if visible.strip():
            state.aborted = True
            raise RuntimeError(
                "Artifact renderer mixed file content with a prerequisite tool call"
            )
    state.renderer_leading_whitespace = ""
    delta["content"] = ""


def _bridge_streamed_file_body(chunk: dict[str, Any], state: _WriteFileBridgeStreamState) -> None:
    """Convert one prepared ordinary-text response into write_file deltas.

    The explicit ``begin_write_file`` receipt is the deterministic boundary:
    only the immediately following invocation enters this path. One optional
    leading inline ``<think>`` envelope is routed to the normal reasoning
    channel; provider-native reasoning stays on that channel as well. The first
    ordinary content byte becomes a standard tool-call chunk immediately,
    matching native sandbox preview behavior without asking the model to obey a
    second JSON or delimiter protocol.
    """

    pending = state.pending
    if pending is None:
        return
    choices = _stream_choices(chunk)
    if not choices:
        return
    if state.closed or state.aborted:
        # A conforming provider emits no more body/tool bytes after its terminal
        # frame (usage-only frames are fine). A previously yielded valid tool
        # chunk cannot be retracted safely by appending malformed arguments:
        # LangChain may retain the earlier parsed call. Abort the whole model
        # node instead, so LangGraph never advances to ToolNode/sandbox IO.
        trailing_payload = False
        for trailing_choice in choices:
            trailing_delta = trailing_choice.get("delta")
            if isinstance(trailing_delta, dict):
                trailing_payload = trailing_payload or bool(
                    _message_content_text(trailing_delta.get("content"))
                    or trailing_delta.get("tool_calls")
                )
                trailing_delta["content"] = ""
                trailing_delta.pop("tool_calls", None)
        if trailing_payload and state.closed and not state.aborted:
            state.aborted = True
            logger.warning(
                "ARTIFACT_RENDERER_PROTOCOL_REJECTED "
                "reason=post_terminal_payload finish_reason=tool_calls"
            )
            raise RuntimeError(
                "Artifact renderer protocol rejected after terminal frame"
            )
        return
    choice = choices[0]
    delta = choice.get("delta")
    if not isinstance(delta, dict):
        delta = {}
        choice["delta"] = delta

    existing_tool_calls = delta.get("tool_calls")
    fallback_file_text = ""
    if existing_tool_calls and state.transport in _ARTIFACT_BODY_TRANSPORTS:
        if _capture_renderer_prerequisite_calls(existing_tool_calls, state):
            if state.started or state.renderer_fallback_active:
                state.aborted = True
                raise RuntimeError(
                    "Artifact renderer requested a prerequisite after file bytes began"
                )
        else:
            fallback_was_active = state.renderer_fallback_active
            if not fallback_was_active and state.started:
                state.aborted = True
                raise ArtifactRendererTransientProtocolError(
                    "Artifact renderer switched to write_file after file bytes began"
                )
            fallback_file_text = _capture_renderer_fallback_write(
                existing_tool_calls,
                state,
            )
            if not fallback_was_active:
                # A provider may emit harmless whitespace alongside native
                # reasoning before switching to a fallback tool call.  None of
                # that speculative ordinary-content state belongs to the
                # tool's content value; restart only the file-body projectors.
                state.inline_think_filter = InlineThinkTagFilter(
                    allow_implicit_close=pending.allow_implicit_think_close,
                    leading_only=True,
                )
                state.markdown_fence_projection = (
                    _MarkdownFenceProjection()
                    if not pending.append and re.search(
                        r"\.(?:md|markdown)\Z",
                        pending.path,
                        re.IGNORECASE,
                    )
                    else None
                )
                state.renderer_leading_whitespace = ""
                state.renderer_html_probe = ""
                state.renderer_html_probe_resolved = False
                state.renderer_html_recovered = False
                state.renderer_html_suspected_truncated = False
                state.renderer_html_content.clear()
            delta.pop("tool_calls", None)

    if state.renderer_deferred:
        _route_renderer_deferred_content(delta, state)
        finish_reason = str(choice.get("finish_reason") or "")
        if finish_reason:
            if finish_reason not in {"tool_calls", "function_call"}:
                state.aborted = True
                raise RuntimeError(
                    "Artifact renderer did not complete its prerequisite tool call"
                )
            tail_visible, tail_reasoning = state.inline_think_filter.finish_parts()
            _append_bridge_reasoning_delta(delta, tail_reasoning)
            if tail_visible.strip():
                state.aborted = True
                raise RuntimeError(
                    "Artifact renderer mixed file content with a prerequisite tool call"
                )
            names = tuple(state.renderer_deferred_tool_names.values())
            if (
                len(names) != 1
                or names[0] not in state.renderer_prerequisite_tools
            ):
                state.aborted = True
                raise RuntimeError(
                    "Artifact renderer returned an incomplete prerequisite tool call"
                )
            state.closed = True
        return

    if state.renderer_fallback_active:
        # Preserve provider/inline reasoning for the message list.  Extract
        # only the fallback call's top-level content string and immediately
        # project newly decoded bytes into the server-owned canonical writer.
        # Provider path/append bytes are never forwarded to LangGraph.
        control_text = _message_content_text(delta.get("content"))
        if control_text:
            control_visible, inline_reasoning = (
                state.renderer_fallback_control_filter.push_parts(control_text)
            )
            _append_bridge_reasoning_delta(
                delta,
                f"{inline_reasoning}{control_visible}",
            )
        delta["content"] = ""
        arguments = ""
        first = not state.started
        file_text = fallback_file_text
        if _extract_reasoning_content(delta):
            confirmed = state.inline_think_filter.confirm_visible_channel()
            if confirmed:
                file_text = f"{confirmed}{file_text}"
        if file_text:
            visible_text, inline_reasoning = state.inline_think_filter.push_parts(
                file_text
            )
            _append_bridge_reasoning_delta(delta, inline_reasoning)
            if state.markdown_fence_projection is not None:
                visible_text = state.markdown_fence_projection.push(visible_text)
            visible_text = _project_renderer_visible_text(
                state,
                pending,
                visible_text,
            )
            if visible_text:
                if first:
                    arguments += _write_file_argument_prefix(pending)
                arguments += _json_string_fragment(visible_text)
                state.started = True

        finish_reason = str(choice.get("finish_reason") or "")
        if finish_reason:
            if finish_reason not in {
                "stop",
                "end_turn",
                "tool_calls",
                "function_call",
            }:
                state.aborted = True
                raise RuntimeError(
                    "Artifact renderer did not complete the fallback write_file call"
                )
            _validated_renderer_fallback_content(state, pending)
            control_visible, control_reasoning = (
                state.renderer_fallback_control_filter.finish_parts()
            )
            _append_bridge_reasoning_delta(
                delta,
                f"{control_reasoning}{control_visible}",
            )
            unclosed_thinking = state.inline_think_filter.has_unclosed_thinking
            tail_visible, tail_reasoning = state.inline_think_filter.finish_parts()
            _append_bridge_reasoning_delta(delta, tail_reasoning)
            if unclosed_thinking:
                state.aborted = True
                logger.warning(
                    "ARTIFACT_RENDERER_PROTOCOL_REJECTED "
                    "reason=unclosed_reasoning finish_reason=%s",
                    _safe_finish_reason(finish_reason),
                )
                raise RuntimeError(
                    "Artifact renderer ended inside a reasoning envelope"
                )
            replacement_content: str | None = None
            if state.markdown_fence_projection is not None:
                projected_tail = state.markdown_fence_projection.push(tail_visible)
                final_tail, replacement_content = (
                    state.markdown_fence_projection.finish()
                )
                tail_visible = f"{projected_tail}{final_tail}"
            tail_visible = _project_renderer_visible_text(
                state,
                pending,
                tail_visible,
                final=True,
            )
            if tail_visible:
                if not state.started:
                    first = True
                    arguments += _write_file_argument_prefix(pending)
                arguments += _json_string_fragment(tail_visible)
                state.started = True
            if replacement_content is not None:
                if not state.started:
                    first = True
                    arguments += _write_file_argument_prefix(pending)
                else:
                    arguments += '","content":"'
                arguments += _json_string_fragment(replacement_content)
                state.started = True
            if not state.started:
                state.aborted = True
                logger.warning(
                    "ARTIFACT_RENDERER_PROTOCOL_REJECTED "
                    "reason=empty_body finish_reason=%s",
                    _safe_finish_reason(finish_reason),
                )
                raise RuntimeError("Artifact renderer returned no file body")
            _validate_renderer_artifact_complete(state, pending)
            arguments += f'","append":{str(pending.append).lower()}}}'
            state.closed = True
            state.renderer_fallback_active = False
            choice["finish_reason"] = "tool_calls"
        if arguments:
            delta["tool_calls"] = [
                _synthetic_write_tool_delta(
                    pending,
                    arguments,
                    first=first,
                )
            ]
        return

    if existing_tool_calls:
        # The body request advertises no tools. Treat an unexpected provider
        # tool call as a protocol failure, suppress it, and never let ToolNode
        # execute a hallucinated operation.
        delta["content"] = ""
        delta.pop("tool_calls", None)
        state.aborted = True
        if state.transport in _ARTIFACT_BODY_TRANSPORTS:
            raise RuntimeError(
                "Artifact renderer emitted an unexpected provider tool call"
            )
        return

    content = delta.get("content")
    text = _message_content_text(content)
    arguments = ""
    first = not state.started
    # A provider-native reasoning field is authoritative: ordinary content is
    # already the answer/file channel, so stop probing it for inline tags. This
    # also preserves a file that legitimately starts with literal <think> text.
    if _extract_reasoning_content(delta):
        confirmed = state.inline_think_filter.confirm_visible_channel()
        if confirmed:
            text = f"{confirmed}{text}"
    if text:
        visible_text, inline_reasoning = state.inline_think_filter.push_parts(text)
        _append_bridge_reasoning_delta(delta, inline_reasoning)
        if state.markdown_fence_projection is not None:
            visible_text = state.markdown_fence_projection.push(visible_text)
        if state.transport in _ARTIFACT_BODY_TRANSPORTS:
            visible_text = _project_renderer_visible_text(
                state,
                pending,
                visible_text,
            )
        # File bytes are represented only as synthetic tool arguments.
        delta["content"] = ""
        if visible_text:
            if first:
                arguments += _write_file_argument_prefix(pending)
            arguments += _json_string_fragment(visible_text)
            state.started = True

    finish_reason = str(choice.get("finish_reason") or "")
    if finish_reason:
        if finish_reason in {"stop", "end_turn"}:
            unclosed_thinking = state.inline_think_filter.has_unclosed_thinking
            tail_visible, tail_reasoning = state.inline_think_filter.finish_parts()
            _append_bridge_reasoning_delta(delta, tail_reasoning)
            if unclosed_thinking:
                # A provider that never leaves its reasoning envelope has not
                # delivered any confirmed file body. Do not manufacture a
                # successful write from private thought text.
                state.aborted = True
                if state.transport in _ARTIFACT_BODY_TRANSPORTS:
                    logger.warning(
                        "ARTIFACT_RENDERER_PROTOCOL_REJECTED "
                        "reason=unclosed_reasoning finish_reason=%s",
                        _safe_finish_reason(finish_reason),
                    )
                    raise RuntimeError(
                        "Artifact renderer ended inside a reasoning envelope"
                    )
            else:
                replacement_content: str | None = None
                if state.markdown_fence_projection is not None:
                    projected_tail = state.markdown_fence_projection.push(
                        tail_visible
                    )
                    final_tail, replacement_content = (
                        state.markdown_fence_projection.finish()
                    )
                    tail_visible = f"{projected_tail}{final_tail}"
                if state.transport in _ARTIFACT_BODY_TRANSPORTS:
                    tail_visible = _project_renderer_visible_text(
                        state,
                        pending,
                        tail_visible,
                        final=True,
                    )
                if tail_visible:
                    if not state.started:
                        first = True
                        arguments += _write_file_argument_prefix(pending)
                    arguments += _json_string_fragment(tail_visible)
                    state.started = True
                if replacement_content is not None:
                    if not state.started:
                        first = True
                        arguments += _write_file_argument_prefix(pending)
                    else:
                        # The first content value was a speculative streaming
                        # projection. A duplicate key repairs an ambiguous
                        # leading fence before JSON validation/tool execution;
                        # parsers use the final value, just as for ``append``.
                        arguments += '","content":"'
                    arguments += _json_string_fragment(replacement_content)
                    state.started = True
                if not state.started:
                    if state.transport in _ARTIFACT_BODY_TRANSPORTS:
                        state.aborted = True
                        logger.warning(
                            "ARTIFACT_RENDERER_PROTOCOL_REJECTED "
                            "reason=empty_body finish_reason=%s",
                            _safe_finish_reason(finish_reason),
                        )
                        raise RuntimeError(
                            "Artifact renderer returned no file body"
                        )
                    # Legacy text_bridge retains its existing validation path.
                    arguments += _write_file_argument_prefix(pending)
                    state.started = True
                    first = True
                if state.transport in _ARTIFACT_BODY_TRANSPORTS:
                    _validate_renderer_artifact_complete(state, pending)
                arguments += f'","append":{str(pending.append).lower()}}}'
                state.closed = True
                choice["finish_reason"] = "tool_calls"
        else:
            # length/content_filter/error/tool_calls: the accumulated call
            # retains its invalid commit sentinel, so ToolNode validation
            # stops it before any file IO.
            state.aborted = True
            if state.transport in _ARTIFACT_BODY_TRANSPORTS:
                logger.warning(
                    "ARTIFACT_RENDERER_PROTOCOL_REJECTED "
                    "reason=unsafe_finish finish_reason=%s",
                    _safe_finish_reason(finish_reason),
                )
                raise RuntimeError(
                    "Artifact renderer did not complete the file body"
                )

    if arguments:
        delta["tool_calls"] = [
            _synthetic_write_tool_delta(
                pending,
                arguments,
                first=first,
            )
        ]


def _ensure_write_file_bridge_finished(state: _WriteFileBridgeStreamState) -> None:
    """Reject a body stream that ended without an explicit safe finish frame."""

    if state.transport not in _ARTIFACT_BODY_TRANSPORTS or state.aborted:
        return
    incomplete_alias = any(
        not call.emitted for call in state.selection_alias_calls.values()
    )
    if state.pending is None and not incomplete_alias:
        return
    if state.pending is not None and state.closed:
        return
    state.aborted = True
    logger.warning(
        "ARTIFACT_RENDERER_PROTOCOL_REJECTED "
        "reason=missing_terminal_frame finish_reason=none"
    )
    raise ArtifactRendererTransientProtocolError(
        "Artifact renderer stream ended without a terminal frame"
    )


def _log_stream_cleanup_failure(error: BaseException, *, asynchronous: bool) -> None:
    try:
        logger.debug(
            "OpenAI-compatible %s stream cleanup failed error_type=%s",
            "async" if asynchronous else "sync",
            _safe_metric_name(type(error).__name__, fallback="other"),
        )
    except Exception:
        # Preserve the stream/cleanup exception even if logging itself fails.
        pass


class OpenAICompatibleUpstreamError(RuntimeError):
    """An OpenAI-compatible gateway reported an error inside a HTTP 200 body.

    Several compatible gateways do not use an HTTP error status for quota,
    concurrency, or inference failures. Instead they return ``choices=null``
    (or an empty streamed chunk) together with a vendor ``base_resp`` object.
    LangChain raises a generic ``TypeError`` for the non-streaming shape and
    treats the streaming shape as an empty successful response. Raising one
    stable exception here lets DeerFlow's model middleware retry both paths.
    """

    def __init__(self, *, code: Any, message: str | None = None) -> None:
        self.code = code
        # Retain bounded provider text only for internal retry classification.
        # Exception strings often reach framework logs implicitly, so never put
        # the upstream response body there.
        self._classification_detail = _bounded_error_message(message)
        display_code = _safe_upstream_code(code)
        super().__init__(f"OpenAI-compatible upstream application error (code={display_code})")


def _bounded_error_message(value: Any, *, limit: int = 500) -> str:
    """Return a bounded one-line provider message for in-process classification."""
    if not isinstance(value, str):
        return ""
    normalized = " ".join(value.split())
    return normalized[:limit]


def _application_error(payload: Mapping[str, Any]) -> tuple[Any, str | None] | None:
    """Extract a non-success ``base_resp`` error from a compatible payload."""
    top_level_error = payload.get("error")
    # Compatible gateways may emit an empty error object on successful SSE
    # frames. The SDK treats that as no error; do not turn its first reasoning
    # delta into a retry before the UI can receive it. Still inspect base_resp
    # below, since a placeholder must not hide a real vendor failure.
    if isinstance(top_level_error, Mapping) and top_level_error:
        code = top_level_error.get("code")
        if code is None:
            code = top_level_error.get("type")
        if code is None:
            code = top_level_error.get("status_code")
        message = top_level_error.get("message")
        return code if code is not None else "unknown", message if isinstance(message, str) else None
    if isinstance(top_level_error, str) and top_level_error:
        return "unknown", top_level_error

    raw = payload.get("base_resp") or payload.get("baseResp")
    if not isinstance(raw, Mapping):
        return None

    code = raw.get("status_code")
    if code is None:
        code = raw.get("statusCode")
    if code is None:
        code = raw.get("code")

    # MiniMax-compatible responses use 0 for success. Be conservative for
    # gateways that put informational metadata in ``base_resp``.
    normalized_code = str(code).strip().lower() if code is not None else ""
    if normalized_code in {"", "0", "200", "ok", "success"}:
        return None

    message = raw.get("status_msg")
    if message is None:
        message = raw.get("statusMsg")
    if message is None:
        message = raw.get("message")
    return code, message if isinstance(message, str) else None


def _raise_for_application_error(payload: Mapping[str, Any]) -> None:
    error = _application_error(payload)
    if error is not None:
        code, message = error
        raise OpenAICompatibleUpstreamError(code=code, message=message)


def _stream_choices(chunk: Mapping[str, Any]) -> list[dict[str, Any]]:
    choices = chunk.get("choices")
    if not isinstance(choices, list) or not choices:
        nested = chunk.get("chunk")
        if isinstance(nested, Mapping):
            choices = nested.get("choices")
    if not isinstance(choices, list):
        return []
    return [choice for choice in choices if isinstance(choice, dict)]


def _normalize_compatible_stream_shapes(
    chunk: dict[str, Any],
    diagnostics: _ToolCallStreamDiagnostics | None,
) -> None:
    """Normalize bounded OpenAI-compatible stream deviations in-place.

    A few gateways return a full ``message`` in a streamed choice, put
    ``tool_calls`` directly on the choice, or encode ``function.arguments``
    as an object instead of a JSON string. All three shapes carry the same
    information as the OpenAI delta schema and can be repaired without
    inspecting or logging their content.
    """
    choices = _stream_choices(chunk)
    if diagnostics is not None and diagnostics.strict_stream_integrity:
        diagnostics.text_state.observe(choices)
    for position, choice in enumerate(choices):
        delta = choice.get("delta")
        original_delta = dict(delta) if isinstance(delta, dict) else None
        if not isinstance(delta, dict):
            message = choice.get("message")
            if isinstance(message, Mapping):
                delta = dict(message)
                choice["delta"] = delta
                if diagnostics is not None:
                    diagnostics.message_as_delta_chunks += 1
            elif isinstance(choice.get("tool_calls"), (list, Mapping)):
                delta = {"tool_calls": choice["tool_calls"]}
                choice["delta"] = delta
                if diagnostics is not None:
                    diagnostics.root_tool_call_chunks += 1
            elif delta is not None and diagnostics is not None:
                diagnostics.mark_invalid()
                continue
            elif choice.get("finish_reason") and diagnostics is not None and diagnostics.strict_stream_integrity:
                # Compatible endpoints sometimes send a null/absent delta on
                # the terminal frame. Keep that frame's finish/usage metadata.
                delta = {}
                choice["delta"] = delta
            else:
                continue

        message = choice.get("message")
        if diagnostics is not None and diagnostics.strict_stream_integrity:
            # Explicit message snapshots can contain the missing answer tail,
            # or repeat text already delivered through standard deltas. Keep
            # the accumulator on this invocation, never on the shared model.
            diagnostics.text_state.normalize_text(
                index=position,
                delta=delta,
                original_delta=original_delta,
                message=message if isinstance(message, Mapping) else None,
            )
        elif isinstance(message, Mapping):
            # Preserve existing reasoning compatibility for other models.
            if _extract_reasoning_content(delta) is None:
                for field_name in ("reasoning_content", "reasoning"):
                    value = message.get(field_name)
                    if isinstance(value, str) and value:
                        delta[field_name] = value

        # Some gateways wrap a useful full-message/root tool call beside an
        # otherwise empty ``delta``. Fill only a missing tool_calls field; a
        # standard delta always wins and is never overwritten.
        if delta.get("tool_calls") is None:
            message = choice.get("message")
            message_tool_calls = message.get("tool_calls") if isinstance(message, Mapping) else None
            root_tool_calls = choice.get("tool_calls")
            if isinstance(message_tool_calls, (list, Mapping)):
                delta["tool_calls"] = message_tool_calls
                if diagnostics is not None:
                    diagnostics.message_as_delta_chunks += 1
            elif isinstance(root_tool_calls, (list, Mapping)):
                delta["tool_calls"] = root_tool_calls
                if diagnostics is not None:
                    diagnostics.root_tool_call_chunks += 1

        tool_calls = delta.get("tool_calls")
        if tool_calls is None:
            continue
        if isinstance(tool_calls, Mapping):
            tool_calls = [dict(tool_calls)]
            delta["tool_calls"] = tool_calls
            if diagnostics is not None:
                diagnostics.singleton_tool_call_chunks += 1
        elif not isinstance(tool_calls, list):
            if diagnostics is not None:
                diagnostics.mark_invalid()
            continue

        for tool_call in tool_calls:
            if not isinstance(tool_call, dict):
                if diagnostics is not None:
                    diagnostics.mark_invalid()
                continue
            function = tool_call.get("function")
            if function is None:
                continue
            if not isinstance(function, dict):
                if diagnostics is not None:
                    diagnostics.mark_invalid()
                continue
            arguments = function.get("arguments")
            if isinstance(arguments, (Mapping, list)):
                function["arguments"] = json.dumps(
                    arguments,
                    ensure_ascii=False,
                    separators=(",", ":"),
                )
                if diagnostics is not None:
                    diagnostics.structured_arg_chunks += 1
            elif arguments is not None and not isinstance(arguments, str):
                if diagnostics is not None:
                    diagnostics.mark_invalid()


def _normalize_compatible_nonstream_shapes(response: dict[str, Any]) -> bool:
    """Normalize structured tool arguments before LangChain parses messages."""
    changed = False
    choices = response.get("choices")
    if not isinstance(choices, list):
        return changed
    for choice in choices:
        if not isinstance(choice, dict):
            continue
        message = choice.get("message")
        if not isinstance(message, dict):
            continue
        tool_calls = message.get("tool_calls")
        if isinstance(tool_calls, Mapping):
            tool_calls = [dict(tool_calls)]
            message["tool_calls"] = tool_calls
            changed = True
        if not isinstance(tool_calls, list):
            continue
        for tool_call in tool_calls:
            if not isinstance(tool_call, dict):
                continue
            function = tool_call.get("function")
            if not isinstance(function, dict):
                continue
            arguments = function.get("arguments")
            if isinstance(arguments, (Mapping, list)):
                function["arguments"] = json.dumps(
                    arguments,
                    ensure_ascii=False,
                    separators=(",", ":"),
                )
                changed = True
    return changed


def rewrite_cumulative_tool_call_arguments(
    chunk: dict[str, Any],
    snapshots: dict[int, str],
    *,
    call_ids: dict[int, str] | None = None,
    argument_modes: dict[int, str] | None = None,
    tool_name_snapshots: dict[int, str] | None = None,
    complete_tool_names: set[str] | None = None,
    diagnostics: _ToolCallStreamDiagnostics | None = None,
) -> dict[int, str]:
    """Rewrite snapshot-style tool-call arguments into true deltas in-place.

    Growing snapshots (each chunk repeats the JSON seen so far) become the
    newly appended suffix. True OpenAI deltas are left unchanged. Duplicate
    snapshots emit an empty string so LangChain does not concatenate twice.
    """
    for choice in _stream_choices(chunk):
        delta = choice.get("delta")
        if not isinstance(delta, dict):
            continue
        tool_calls = delta.get("tool_calls")
        if not isinstance(tool_calls, list):
            continue
        for position, tool_call in enumerate(tool_calls):
            if not isinstance(tool_call, dict):
                if diagnostics is not None:
                    diagnostics.mark_invalid()
                continue
            raw_call_id = tool_call.get("id")
            raw_index = tool_call.get("index")
            invalid_index = raw_index is None
            if invalid_index:
                index = position
            else:
                try:
                    index = int(raw_index)
                except (TypeError, ValueError):
                    index = position
                    invalid_index = True
            if invalid_index:
                if call_ids is not None and isinstance(raw_call_id, str) and raw_call_id:
                    bounded_call_id = raw_call_id[:256]
                    for known_index, known_id in call_ids.items():
                        if known_id == bounded_call_id:
                            index = known_index
                            break
                    else:
                        # Parallel compatible gateways may emit one id-bearing
                        # tool call per chunk while omitting every index. Each
                        # item then has list position zero; allocate a free
                        # canonical slot instead of resetting the first call.
                        if index in call_ids and call_ids[index] != bounded_call_id:
                            index = 0
                            while index in call_ids:
                                index += 1
                if diagnostics is not None:
                    diagnostics.mark_invalid()
            # LangChain's OpenAI delta converter indexes every tool call with
            # ``rtc["index"]``. Compatible gateways sometimes omit it (or send
            # it as a numeric string); normalizing in-place keeps the call from
            # being dropped after our snapshot repair.
            tool_call["index"] = index

            if diagnostics is not None:
                diagnostics.seen_indices.add(index)
            if call_ids is not None and isinstance(raw_call_id, str) and raw_call_id:
                bounded_call_id = raw_call_id[:256]
                previous_call_id = call_ids.get(index)
                if previous_call_id is not None and previous_call_id != bounded_call_id:
                    # A new explicit call id reusing an index belongs to a new
                    # tool call. Keeping the old snapshot would concatenate two
                    # independent JSON documents and can prevent tool execution.
                    snapshots.pop(index, None)
                    if argument_modes is not None:
                        argument_modes.pop(index, None)
                    if tool_name_snapshots is not None:
                        tool_name_snapshots.pop(index, None)
                    if diagnostics is not None:
                        diagnostics.last_raw_lengths.pop(index, None)
                        diagnostics.tool_id_resets += 1
                call_ids[index] = bounded_call_id

            function = tool_call.get("function")
            if not isinstance(function, dict):
                if function is not None and diagnostics is not None:
                    diagnostics.mark_invalid()
                continue
            name = function.get("name")
            if tool_name_snapshots is not None and isinstance(name, str) and name:
                previous_name = tool_name_snapshots.get(index, "")
                if previous_name == name and (snapshots.get(index) or name in (complete_tool_names or ())):
                    # Some gateways repeat the complete function name beside
                    # every argument snapshot/delta. LangChain concatenates
                    # names too, producing e.g. write_filewrite_file, which is
                    # no longer executable or recognizable by the live preview.
                    # Suppress only after arguments have started or a request
                    # tool schema proves this is the complete name. Legitimate
                    # name fragments (write_ + file, or foo + foo) stay intact.
                    function["name"] = None
                    if diagnostics is not None:
                        diagnostics.repeated_tool_names += 1
                else:
                    tool_name_snapshots[index] = f"{previous_name}{name}"
            arguments = function.get("arguments")
            if not isinstance(arguments, str):
                if arguments is not None and diagnostics is not None:
                    diagnostics.mark_invalid()
                continue
            if arguments == "":
                if diagnostics is not None:
                    diagnostics.empty_arg_chunks += 1
                continue
            previous = snapshots.get(index, "")
            mode = argument_modes.get(index, "unknown") if argument_modes is not None else "unknown"
            if mode == "delta":
                snapshots[index] = f"{previous}{arguments}" if previous else arguments
                if diagnostics is not None:
                    diagnostics.observe_argument(
                        index=index,
                        raw_length=len(arguments),
                        classification="delta" if previous else "initial",
                    )
                continue
            if previous and arguments == previous:
                function["arguments"] = ""
                if argument_modes is not None:
                    argument_modes[index] = "cumulative"
                if diagnostics is not None:
                    diagnostics.observe_argument(
                        index=index,
                        raw_length=len(arguments),
                        classification="repeated",
                    )
                continue
            if previous and arguments.startswith(previous):
                function["arguments"] = arguments[len(previous) :]
                snapshots[index] = arguments
                if argument_modes is not None:
                    argument_modes[index] = "cumulative"
                if diagnostics is not None:
                    diagnostics.observe_argument(
                        index=index,
                        raw_length=len(arguments),
                        classification="cumulative",
                    )
                continue
            # A shorter fragment that happens to be a prefix of the assembled
            # JSON is ambiguous and is valid in ordinary delta streams (for
            # example a nested ``{`` split). Never discard it as a rewind.
            # Once a non-snapshot fragment appears, lock this call to delta
            # mode so later chunks are always appended verbatim.
            if previous and argument_modes is not None:
                argument_modes[index] = "delta"
            snapshots[index] = f"{previous}{arguments}" if previous else arguments
            if diagnostics is not None:
                diagnostics.observe_argument(
                    index=index,
                    raw_length=len(arguments),
                    classification="delta" if previous else "initial",
                )
    return snapshots


class PatchedChatOpenAI(ChatOpenAI):
    """ChatOpenAI with ``thought_signature`` preservation for Gemini thinking via OpenAI gateway.

    When using Gemini with thinking enabled via an OpenAI-compatible gateway,
    the API expects ``thought_signature`` to be present on tool-call objects in
    multi-turn conversations.  This patched version restores those signatures
    from ``AIMessage.additional_kwargs["tool_calls"]`` into the serialised
    request payload before it is sent to the API.

    Usage in ``config.yaml``::

        - name: gemini-2.5-pro-thinking
          display_name: Gemini 2.5 Pro (Thinking)
          use: deerflow.models.patched_openai:PatchedChatOpenAI
          model: google/gemini-2.5-pro-preview
          api_key: $GEMINI_API_KEY
          base_url: https://<your-openai-compat-gateway>/v1
          max_tokens: 16384
          supports_thinking: true
          supports_vision: true
          when_thinking_enabled:
            extra_body:
              thinking:
                type: enabled
    """

    # DeerFlow-only transport capability. ``native`` is deliberately the
    # default so every existing/unconfigured model stays on the exact original
    # OpenAI tool-call path.
    write_file_transport: WriteFileTransport = "native"

    # Opt-in fallback for OpenAI-compatible DeepSeek V4/V4.1 endpoints whose
    # serving layer leaves native DSML in assistant content. The factory keeps
    # this local field out of the provider request.
    tool_call_transport: Literal["native", "deepseek_dsml"] = "native"

    # Local compatibility switch, never forwarded to the provider. None uses
    # the affected model id; the factory also recognizes the configured alias.
    strict_stream_integrity: bool | None = None

    def _generate(self, *args: Any, **kwargs: Any) -> ChatResult:
        """Own non-stream log cleanup in the context that opened the request."""
        capture = open_exchange()
        token = bind_exchange(capture)
        dsml_token = _ACTIVE_DSML_RECOVERY.set(
            DSMLToolCallRecovery(enabled=self.tool_call_transport == "deepseek_dsml")
        )
        try:
            return super()._generate(*args, **kwargs)
        finally:
            try:
                finish_exchange(capture)
            finally:
                try:
                    _ACTIVE_DSML_RECOVERY.reset(dsml_token)
                finally:
                    unbind_exchange(token)

    async def _agenerate(self, *args: Any, **kwargs: Any) -> ChatResult:
        # LangChain runs _create_chat_result in a copied executor context. A
        # ContextVar token created by _get_request_payload cannot be reset
        # there: even a correct model response would become a ValueError.
        # Bind here and clean up after awaiting the complete provider call,
        # including errors/cancellation, just as the streaming path does.
        capture = open_exchange()
        token = bind_exchange(capture)
        dsml_token = _ACTIVE_DSML_RECOVERY.set(
            DSMLToolCallRecovery(enabled=self.tool_call_transport == "deepseek_dsml")
        )
        try:
            return await super()._agenerate(*args, **kwargs)
        finally:
            try:
                finish_exchange(capture)
            finally:
                try:
                    _ACTIVE_DSML_RECOVERY.reset(dsml_token)
                finally:
                    unbind_exchange(token)

    def _stream(self, *args: Any, **kwargs: Any) -> Iterator[ChatGenerationChunk]:
        """Wrap one synchronous provider stream in isolated diagnostics state."""
        # Test runners and application bootstrap code may replace warning
        # filters after this module is imported. Reassert the same exact global
        # rule at invocation start; unlike catch_warnings this holds no mutable
        # context across iteration or network waits.
        _install_structured_arg_warning_filter()
        capture = open_exchange()
        diagnostics = self._new_tool_stream_diagnostics(kwargs.get("run_manager"))
        bridge_state = _WriteFileBridgeStreamState(transport=self.write_file_transport)
        dsml_state = DSMLToolCallRecovery(enabled=self.tool_call_transport == "deepseek_dsml")
        error: BaseException | None = None
        parent_stream = super()._stream(*args, **kwargs)
        iterator = iter(parent_stream)
        tail_chunks: list[ChatGenerationChunk] = []
        try:
            while True:
                # Scope both context vars to one provider ``next`` call. They are
                # reset *before* yielding, so closing a suspended generator from
                # another context cannot reset a foreign token.
                exchange_token = bind_exchange(capture)
                token = _ACTIVE_TOOL_STREAM.set(diagnostics)
                bridge_token = _ACTIVE_WRITE_FILE_BRIDGE.set(bridge_state)
                dsml_token = _ACTIVE_DSML_RECOVERY.set(dsml_state)
                try:
                    generation_chunk = next(iterator)
                except StopIteration:
                    tail_chunks = dsml_state.finish()
                    _ensure_write_file_bridge_finished(bridge_state)
                    diagnostics.text_state.ensure_finished()
                    break
                finally:
                    _ACTIVE_DSML_RECOVERY.reset(dsml_token)
                    _ACTIVE_WRITE_FILE_BRIDGE.reset(bridge_token)
                    _ACTIVE_TOOL_STREAM.reset(token)
                    unbind_exchange(exchange_token)
                yield from dsml_state.feed(generation_chunk)
                if bridge_state.selection_alias_early_stop:
                    if (
                        bridge_state.transport == "buffered_gateway"
                        and bridge_state.pending is not None
                    ):
                        close = getattr(iterator, "close", None)
                        if close is not None:
                            close()
                        bridge_state.selection_alias_early_stop = False
                        bridge_state.buffered_gateway_body_request = True
                        # Start a second provider request with the same original
                        # LangChain input. The active bridge state rewrites it
                        # into an isolated ordinary-text body request. Both
                        # responses remain one logical model stream to the
                        # caller, so ToolNode sees only the final write_file.
                        parent_stream = super()._stream(*args, **kwargs)
                        iterator = iter(parent_stream)
                        continue
                    _ensure_write_file_bridge_finished(bridge_state)
                    break
            yield from tail_chunks
        except BaseException as exc:
            error = exc
            raise
        finally:
            close_error: BaseException | None = None
            close = getattr(iterator, "close", None)
            if close is not None:
                try:
                    close()
                except BaseException as exc:
                    close_error = exc
                    _log_stream_cleanup_failure(exc, asynchronous=False)
            diagnostics.log_summary(error or close_error)
            finish_exchange(capture)
            if error is None and close_error is not None:
                raise close_error

    async def _astream(self, *args: Any, **kwargs: Any) -> AsyncIterator[ChatGenerationChunk]:
        """Wrap one async provider stream in isolated diagnostics state.

        A ``ContextVar`` keeps cumulative-argument snapshots invocation-local.
        The former model-instance dictionary could be shared by overlapping
        calls and concatenate one request's tool JSON onto another request.
        """
        _install_structured_arg_warning_filter()
        capture = open_exchange()
        diagnostics = self._new_tool_stream_diagnostics(kwargs.get("run_manager"))
        bridge_state = _WriteFileBridgeStreamState(transport=self.write_file_transport)
        dsml_state = DSMLToolCallRecovery(enabled=self.tool_call_transport == "deepseek_dsml")
        error: BaseException | None = None
        parent_stream = super()._astream(*args, **kwargs)
        iterator = parent_stream.__aiter__()
        tail_chunks: list[ChatGenerationChunk] = []
        try:
            while True:
                # Keep both context vars only around ``__anext__``. The converter
                # runs inside this await, and each token is reset in that same
                # context before control reaches yield.
                exchange_token = bind_exchange(capture)
                token = _ACTIVE_TOOL_STREAM.set(diagnostics)
                bridge_token = _ACTIVE_WRITE_FILE_BRIDGE.set(bridge_state)
                dsml_token = _ACTIVE_DSML_RECOVERY.set(dsml_state)
                try:
                    generation_chunk = await iterator.__anext__()
                except StopAsyncIteration:
                    tail_chunks = dsml_state.finish()
                    _ensure_write_file_bridge_finished(bridge_state)
                    diagnostics.text_state.ensure_finished()
                    break
                finally:
                    _ACTIVE_DSML_RECOVERY.reset(dsml_token)
                    _ACTIVE_WRITE_FILE_BRIDGE.reset(bridge_token)
                    _ACTIVE_TOOL_STREAM.reset(token)
                    unbind_exchange(exchange_token)
                for recovered_chunk in dsml_state.feed(generation_chunk):
                    yield recovered_chunk
                if bridge_state.selection_alias_early_stop:
                    if (
                        bridge_state.transport == "buffered_gateway"
                        and bridge_state.pending is not None
                    ):
                        aclose = getattr(iterator, "aclose", None)
                        if aclose is not None:
                            await aclose()
                        bridge_state.selection_alias_early_stop = False
                        bridge_state.buffered_gateway_body_request = True
                        parent_stream = super()._astream(*args, **kwargs)
                        iterator = parent_stream.__aiter__()
                        continue
                    _ensure_write_file_bridge_finished(bridge_state)
                    break
            for recovered_chunk in tail_chunks:
                yield recovered_chunk
        except BaseException as exc:
            error = exc
            raise
        finally:
            close_error: BaseException | None = None
            aclose = getattr(iterator, "aclose", None)
            if aclose is not None:
                try:
                    await aclose()
                except BaseException as exc:
                    close_error = exc
                    _log_stream_cleanup_failure(exc, asynchronous=True)
            diagnostics.log_summary(error or close_error)
            finish_exchange(capture)
            if error is None and close_error is not None:
                raise close_error

    def _new_tool_stream_diagnostics(self, run_manager: Any = None) -> _ToolCallStreamDiagnostics:
        model_name = _safe_metric_name(
            getattr(self, "model_name", None) or getattr(self, "model", None),
            fallback="unknown",
        )
        run_id = _safe_metric_name(getattr(run_manager, "run_id", None), fallback="none")
        strict = self.strict_stream_integrity
        if strict is None:
            strict = needs_qwen_flash_next_integrity(self.model_name)
        return _ToolCallStreamDiagnostics(model_name=model_name, run_id=run_id, strict_stream_integrity=strict)

    def _get_request_payload(
        self,
        input_: LanguageModelInput,
        *,
        stop: list[str] | None = None,
        **kwargs: Any,
    ) -> dict:
        """Get request payload with ``thought_signature`` preserved on tool-call objects.

        Overrides the parent method to re-inject ``thought_signature`` fields
        on tool-call objects that were stored in
        ``additional_kwargs["tool_calls"]`` by LangChain but dropped during
        serialisation.
        """
        # Capture the original LangChain messages *before* conversion so we can
        # access fields that the serialiser might drop.
        original_messages = self._convert_input(input_).to_messages()

        # Obtain the base payload from the parent implementation.
        payload = super()._get_request_payload(input_, stop=stop, **kwargs)
        payload_messages = payload.get("messages", [])

        # Restore provider metadata while payload and source history still
        # have their original one-to-one shape. The compatibility layer may
        # subsequently remove older internal begin_write_file exchanges;
        # doing this afterward could pair a later assistant turn with a
        # removed control message.
        if len(payload_messages) == len(original_messages):
            for payload_msg, orig_msg in zip(payload_messages, original_messages):
                if payload_msg.get("role") == "assistant" and isinstance(orig_msg, AIMessage):
                    _restore_tool_call_signatures(payload_msg, orig_msg)
                    _restore_reasoning_content(payload_msg, orig_msg)
        else:
            # Fallback: match assistant-role entries positionally against AIMessages.
            ai_messages = [m for m in original_messages if isinstance(m, AIMessage)]
            assistant_payloads = [(i, m) for i, m in enumerate(payload_messages) if m.get("role") == "assistant"]
            for (_, payload_msg), ai_msg in zip(assistant_payloads, ai_messages):
                _restore_tool_call_signatures(payload_msg, ai_msg)
                _restore_reasoning_content(payload_msg, ai_msg)

        bridge_state = _ACTIVE_WRITE_FILE_BRIDGE.get()
        if bridge_state is not None:
            _configure_write_file_bridge_payload(payload, original_messages, bridge_state)
        else:
            # invoke/ainvoke does not pass through our streaming ContextVar.
            # Degrade a configured bridge to the original atomic write_file
            # behavior, and defensively hide the internal tool for native
            # models too if a caller bound it accidentally.
            _hide_begin_write_file_for_native_request(payload)
        diagnostics = _ACTIVE_TOOL_STREAM.get()
        if diagnostics is not None:
            extra_body = payload.get("extra_body")
            value = extra_body.get("enable_thinking", payload.get("enable_thinking")) if isinstance(extra_body, Mapping) else payload.get("enable_thinking")
            # Only an allowlisted boolean label; never stringify provider or
            # user request values, which can contain prompt text or secrets.
            diagnostics.request_enable_thinking = "true" if value is True else "false" if value is False else "unset" if value is None else "other"
            tools = payload.get("tools")
            if isinstance(tools, list):
                for tool in tools:
                    function = tool.get("function") if isinstance(tool, Mapping) else None
                    name = function.get("name") if isinstance(function, Mapping) else None
                    if isinstance(name, str) and name:
                        diagnostics.complete_tool_names.add(name)

        dsml_recovery = _ACTIVE_DSML_RECOVERY.get()
        if dsml_recovery is not None:
            dsml_recovery.configure_from_payload(payload)

        normalize_chat_payload_messages(payload)
        from deerflow.models.prompt_suffix import append_payload_prompt_suffix

        append_payload_prompt_suffix(
            payload,
            getattr(self, "deerflow_prompt_model_name", ""),
            enabled=getattr(self, "deerflow_prompt_suffix_enabled", True),
        )
        remember_request(payload)
        return payload

    def _convert_chunk_to_generation_chunk(
        self,
        chunk: dict,
        default_chunk_class: type,
        base_generation_info: dict | None,
    ) -> ChatGenerationChunk | None:
        """Preserve streamed ``reasoning_content``/``reasoning`` metadata."""
        remember_provider_chunk(chunk)
        diagnostics = _ACTIVE_TOOL_STREAM.get()
        if diagnostics is not None:
            diagnostics.observe_chunk(chunk)
        # Some compatible gateways return a vendor error event followed by a
        # normal-looking usage chunk, all under HTTP 200. Detect the first
        # event before LangChain converts it into an empty AIMessageChunk.
        _raise_for_application_error(chunk)
        _normalize_compatible_stream_shapes(chunk, diagnostics)
        bridge_state = _ACTIVE_WRITE_FILE_BRIDGE.get()
        if bridge_state is not None:
            _bridge_streamed_file_body(chunk, bridge_state)
        if diagnostics is not None:
            snapshots = diagnostics.snapshots
            call_ids = diagnostics.call_ids
            argument_modes = diagnostics.argument_modes
        else:
            snapshots = self._tool_arg_snapshots_for_chunk(chunk)
            call_ids = None
            argument_modes = None
        rewrite_cumulative_tool_call_arguments(
            chunk,
            snapshots,
            call_ids=call_ids,
            argument_modes=argument_modes,
            tool_name_snapshots=diagnostics.tool_name_snapshots if diagnostics is not None else None,
            complete_tool_names=diagnostics.complete_tool_names if diagnostics is not None else None,
            diagnostics=diagnostics,
        )
        if bridge_state is not None:
            _rewrite_selection_write_file_alias(chunk, bridge_state)
        choices = chunk.get("choices", []) or chunk.get("chunk", {}).get("choices", [])
        choice = choices[0] if choices else {}
        delta = choice.get("delta") if isinstance(choice, dict) else None
        signature_records = _extract_tool_call_signature_records(delta.get("tool_calls") if isinstance(delta, dict) else None)
        generation_chunk = super()._convert_chunk_to_generation_chunk(
            chunk,
            default_chunk_class,
            base_generation_info,
        )
        if generation_chunk is None:
            return None

        if signature_records and isinstance(generation_chunk.message, AIMessageChunk):
            generation_chunk.message = _append_tool_call_signature_metadata(
                generation_chunk.message,
                signature_records,
            )
        reasoning_content = _extract_reasoning_content(delta if isinstance(delta, dict) else None)
        if reasoning_content and isinstance(generation_chunk.message, AIMessageChunk):
            generation_chunk.message = _append_reasoning_content(
                generation_chunk.message,
                reasoning_content,
                reasoning=delta.get("reasoning") if isinstance(delta, dict) else None,
            )
        if isinstance(choice, dict) and choice.get("finish_reason"):
            if diagnostics is None:
                self._cumulative_tool_arg_snapshots = {}
        if diagnostics is not None and _extract_reasoning_content(generation_chunk.message.additional_kwargs):
            diagnostics.emitted_reasoning_chunks += 1
        return generation_chunk

    def _tool_arg_snapshots_for_chunk(self, chunk: dict) -> dict[int, str]:
        """Return per-index argument snapshots, reset when a new completion starts."""
        stream_id = chunk.get("id")
        if isinstance(stream_id, str) and stream_id:
            if stream_id != getattr(self, "_tool_arg_stream_id", None):
                self._tool_arg_stream_id = stream_id
                self._cumulative_tool_arg_snapshots = {}
        snapshots = getattr(self, "_cumulative_tool_arg_snapshots", None)
        if snapshots is None:
            snapshots = {}
            self._cumulative_tool_arg_snapshots = snapshots
        return snapshots

    def _create_chat_result(
        self,
        response: dict | Any,
        generation_info: dict | None = None,
    ) -> ChatResult:
        """Preserve non-stream ``reasoning_content``/``reasoning`` metadata."""
        response_dict = response if isinstance(response, dict) else response.model_dump(
            warnings=False,
            exclude={"choices": {"__all__": {"message": {"parsed"}}}},
        )
        remember_raw_response(response_dict)
        _raise_for_application_error(response_dict)
        _normalize_compatible_nonstream_shapes(response_dict)
        dsml_recovery = _ACTIVE_DSML_RECOVERY.get()
        if dsml_recovery is not None:
            dsml_recovery.recover_nonstream_response(response_dict)
        # Reuse the safe dump. Passing the typed response back to LangChain
        # serializes it again with warnings enabled and can print provider data.
        result = super()._create_chat_result(response_dict, generation_info)
        choices = response_dict.get("choices", [])

        generations: list[ChatGeneration] = []
        for index, generation in enumerate(result.generations):
            message = generation.message
            choice = choices[index] if index < len(choices) else {}
            choice_message = choice.get("message", {}) if isinstance(choice, dict) else {}
            signature_records = _extract_tool_call_signature_records(choice_message.get("tool_calls") if isinstance(choice_message, dict) else None)
            if signature_records and isinstance(message, AIMessage):
                message = _append_tool_call_signature_metadata(message, signature_records)
            reasoning_content = _extract_reasoning_content(choice_message if isinstance(choice_message, dict) else None)
            if reasoning_content and isinstance(message, AIMessage):
                message = _append_reasoning_content(
                    message,
                    reasoning_content,
                    reasoning=choice_message.get("reasoning") if isinstance(choice_message, dict) else None,
                )
            if message is not generation.message:
                generation = ChatGeneration(message=message, generation_info=generation.generation_info)
            generations.append(generation)

        # Mirror LangChain's typed structured-output/refusal handling without
        # serializing arbitrary parsed Pydantic objects into the raw response.
        typed_choices = getattr(response, "choices", None)
        if generations and isinstance(typed_choices, list) and typed_choices:
            typed_message = typed_choices[0].message
            for field_name in ("parsed", "refusal"):
                if hasattr(typed_message, field_name):
                    generations[0].message.additional_kwargs[field_name] = getattr(typed_message, field_name)

        if choices and isinstance(choices[0], dict):
            message = choices[0].get("message")
            if isinstance(message, dict):
                remember_final_message(message)
        finish_owned_exchange()
        return ChatResult(generations=generations, llm_output=result.llm_output)


def _restore_tool_call_signatures(payload_msg: dict, orig_msg: AIMessage) -> None:
    """Re-inject ``thought_signature`` onto tool-call objects in *payload_msg*.

    The private metadata is populated before LangChain discards provider
    extension fields. Legacy messages that still retain raw ``tool_calls`` in
    ``additional_kwargs`` remain supported. Matching prefers call id and then
    the canonical tool index, so concurrent tool calls cannot exchange tokens.
    """
    payload_tool_calls: list[dict] = payload_msg.get("tool_calls") or []
    if not payload_tool_calls:
        return

    raw_records = orig_msg.additional_kwargs.get(_TOOL_SIGNATURES_KEY)
    records = [dict(record) for record in raw_records if isinstance(record, Mapping)] if isinstance(raw_records, list) else []
    legacy_tool_calls = orig_msg.additional_kwargs.get("tool_calls")
    records.extend(_extract_tool_call_signature_records(legacy_tool_calls))
    if not records:
        return

    by_id: dict[str, dict[str, Any]] = {}
    by_index: dict[int, dict[str, Any]] = {}
    for record in records:
        call_id = record.get("id")
        if isinstance(call_id, str) and call_id:
            by_id[call_id] = record
        raw_index = record.get("tool_index", record.get("index"))
        try:
            record_index = int(raw_index)
        except (TypeError, ValueError):
            continue
        by_index[record_index] = record

    for idx, payload_tc in enumerate(payload_tool_calls):
        call_id = payload_tc.get("id")
        record = by_id.get(call_id) if isinstance(call_id, str) else None
        if record is None:
            record = by_index.get(idx)
        if record is not None:
            _restore_signature_at_location(payload_tc, record)


def _restore_reasoning_content(payload_msg: dict, orig_msg: AIMessage) -> None:
    """Re-inject provider reasoning onto an assistant history message.

    Some thinking-enabled OpenAI-compatible providers require each historical
    assistant turn to include its original reasoning in subsequent requests.
    Preserve the spelling returned by the provider when available; otherwise
    retain the established ``reasoning_content`` behavior.
    """
    reasoning = orig_msg.additional_kwargs.get("reasoning")
    if reasoning is not None:
        payload_msg["reasoning"] = reasoning
        return

    reasoning_content = orig_msg.additional_kwargs.get("reasoning_content")
    if reasoning_content is not None:
        payload_msg["reasoning_content"] = reasoning_content


def _extract_reasoning_content(payload: dict | None) -> str | None:
    """Extract either supported reasoning spelling from provider payloads."""
    if not isinstance(payload, dict):
        return None
    reasoning_content = payload.get("reasoning_content") or payload.get("reasoning")
    return reasoning_content if isinstance(reasoning_content, str) and reasoning_content else None


def _append_reasoning_content(
    message: AIMessage | AIMessageChunk,
    reasoning_content: str,
    *,
    reasoning: Any = None,
) -> AIMessage | AIMessageChunk:
    """Normalize reasoning for clients while retaining its provider spelling."""
    additional_kwargs = dict(message.additional_kwargs)
    existing = additional_kwargs.get("reasoning_content")
    if isinstance(existing, str) and existing:
        if reasoning_content not in existing:
            additional_kwargs["reasoning_content"] = f"{existing}{reasoning_content}"
    else:
        additional_kwargs["reasoning_content"] = reasoning_content
    if reasoning is not None:
        additional_kwargs["reasoning"] = reasoning
    return message.model_copy(update={"additional_kwargs": additional_kwargs})