"""Monkey-patches and utilities for third-party LangChain provider quirks. All patches follow the same pattern: wrap an existing method/function to fix upstream bugs, applied at import time or on first use. Patches: - _patch_anthropic_proxy_compat: ccproxy dict→Pydantic model mismatch - _patch_openai_compat_content: list content→string for strict APIs - _patch_ccproxy_codex_compat: ccproxy model fixes + langchain None guard - _patch_ccproxy_system_to_developer: system→developer role for ccproxy - _patch_openai_capture_reasoning_content: capture provider reasoning_content into AIMessage.additional_kwargs (module-level, applied at import) - _patch_openrouter_strip_responses_reasoning: drop OpenAI-Responses encrypted reasoning items (rs_* id) from outgoing OpenRouter messages (store=false → "Item with id rs_... not found") - _patch_langgraph_goto_none_crash: handle Command(goto=None) defensively in langgraph and langgraph-api (langgraph 1.2.9 workaround) Utilities: - _is_ccproxy_codex: detect ccproxy Codex OAuth adapter - _flatten_message_content: convert content blocks to plain string """ from __future__ import annotations import dataclasses import functools import hashlib import os import sys from collections.abc import ( AsyncIterator, Awaitable, Callable, Iterator, Mapping, Sequence, ) from contextvars import ContextVar from typing import Any, cast from langchain_core.messages import AIMessage, BaseMessage # --------------------------------------------------------------------------- # Patch: langchain-anthropic (>=1.3.4) calls .model_dump() on # context_management / container objects returned by the Anthropic SDK. # Proxies like ccproxy may return plain dicts which lack that method. # We wrap the class method to pre-convert dicts before the original runs. # --------------------------------------------------------------------------- def _patch_anthropic_proxy_compat() -> None: try: import types as _types from langchain_anthropic.chat_models import ChatAnthropic as _CA _orig = _CA._make_message_chunk_from_anthropic_event def _safe(self: Any, event: Any, *args: Any, **kwargs: Any) -> Any: for obj, attrs in [ (event, ("context_management",)), (getattr(event, "delta", None), ("container",)), ]: if obj is None: continue for attr in attrs: val = getattr(obj, attr, None) if isinstance(val, dict): d = val.copy() setattr( obj, attr, _types.SimpleNamespace(model_dump=lambda d=d, **kw: d), ) return _orig(self, event, *args, **kwargs) _CA._make_message_chunk_from_anthropic_event = _safe except Exception: pass _patch_anthropic_proxy_compat() # --------------------------------------------------------------------------- # Patch: ccproxy-api 0.2.7 Codex compatibility. # # 1) ResponseObject.output is required but upstream may omit it → 502. # Fix: make output default to []. # 2) CodexMessage.role only allows "user"/"assistant" → 400 on system msgs. # Fix: widen to also accept "system" and "developer". # 3) langchain-openai iterates response.output which can be None after the # proxy strips it. Fix: guard in _construct_lc_result_from_responses_api. # --------------------------------------------------------------------------- def _patch_ccproxy_codex_compat() -> None: """Patch ccproxy-api models for Responses API compatibility.""" # 1) Make ResponseObject.output optional (default=[]) try: import ccproxy.llms.models.openai as _oai_mod _OrigResponse = _oai_mod.ResponseObject from pydantic import Field as _PydanticField class _PatchedResponseObject(_OrigResponse): # type: ignore[misc] output: list = _PydanticField(default_factory=list) # type: ignore[assignment] model_config = _OrigResponse.model_config.copy() _PatchedResponseObject.__name__ = "ResponseObject" _PatchedResponseObject.__qualname__ = "ResponseObject" _oai_mod.ResponseObject = _PatchedResponseObject # type: ignore[misc] # Also patch modules that import ResponseObject directly for _mod_path in ( "ccproxy.llms.formatters.openai_to_openai.responses", "ccproxy.llms.formatters.anthropic_to_openai.responses", ): try: import importlib _mod = importlib.import_module(_mod_path) if hasattr(_mod, "ResponseObject"): _mod.ResponseObject = _PatchedResponseObject # type: ignore[attr-defined] except Exception: pass except Exception: pass # 2) Widen CodexMessage.role to accept system/developer try: from typing import Annotated, Literal import ccproxy.plugins.codex.models as _codex_mod _OrigMessage = _codex_mod.CodexMessage from pydantic import Field as _Field class _PatchedCodexMessage(_OrigMessage): # type: ignore[misc] role: Annotated[ # type: ignore[assignment] Literal["user", "assistant", "system", "developer"], _Field(description="Message role"), ] _PatchedCodexMessage.__name__ = "CodexMessage" _PatchedCodexMessage.__qualname__ = "CodexMessage" _codex_mod.CodexMessage = _PatchedCodexMessage # type: ignore[misc] except Exception: pass # 3) Fix StreamingBufferService returning response.completed event # whose output is None/empty, instead of using accumulated outputs. try: from ccproxy.llms.streaming.accumulators import ResponsesAccumulator _orig_get = ResponsesAccumulator.get_completed_response def _patched_get(self: Any) -> dict | None: result = _orig_get(self) if result is not None: output = result.get("output") if output is None: # output field lost — force rebuild from accumulated items return None return result ResponsesAccumulator.get_completed_response = _patched_get # type: ignore[assignment] except Exception: pass # 4) Guard langchain-openai against None output (final safety net) try: import langchain_openai.chat_models.base as _base _orig_construct = _base._construct_lc_result_from_responses_api def _safe(response: Any, *args: Any, **kwargs: Any) -> Any: if response.output is None: response.output = [] return _orig_construct(response, *args, **kwargs) _base._construct_lc_result_from_responses_api = _safe except Exception: pass _patch_ccproxy_codex_compat() # --------------------------------------------------------------------------- # Utility: detect ccproxy's Codex adapter (as opposed to generic localhost). # --------------------------------------------------------------------------- def _is_ccproxy_codex( base_url: str | None = None, api_key: str | None = None, ) -> bool: """Return True if the OpenAI endpoint is ccproxy's Codex adapter. Checks for the ccproxy-specific markers set by ``setup_codex_env()`` in ``ccproxy_manager.py``: the sentinel API key and the ``/codex/v1`` path. Plain localhost endpoints (vLLM, Ollama, etc.) are not affected. ``base_url`` / ``api_key`` override the environment when supplied, so a resolved invocation plan can be classified without mutating the process environment (Ai4Sci routes models through a compiled plan). """ base_url = os.environ.get("OPENAI_BASE_URL", "") if base_url is None else base_url api_key = os.environ.get("OPENAI_API_KEY", "") if api_key is None else api_key return ( ("127.0.0.1" in base_url or "localhost" in base_url) and api_key == "ccproxy-oauth" and "/codex/" in base_url ) # --------------------------------------------------------------------------- # Utility + Patch: Flatten list content to strings for OpenAI-compatible APIs. # DeepSeek, SiliconFlow, etc. reject assistant messages whose content is a # list rather than a string. Image and file (PDF/document) blocks are # preserved, not flattened away. # --------------------------------------------------------------------------- _SKIP_CONTENT_TYPES = frozenset({"thinking", "reasoning", "reasoning_content"}) # Media block types preserved when flattening (positive allowlist; # thinking/reasoning still dropped). Images + files (PDF/documents): both # serialize on OpenAI-compatible APIs and capable models read them. `video` is # deliberately EXCLUDED (langchain-openai raises ValueError on it); `audio` is # omitted (almost no model support). is_data_content_block is unreliable # (False for OpenAI image_url / Anthropic image-source). Kept as separate # image/file sets so the no-media fallback can gate per modality. _IMAGE_CONTENT_TYPES = frozenset({"image", "image_url", "input_image"}) _FILE_CONTENT_TYPES = frozenset({"file", "input_file", "document"}) _MEDIA_CONTENT_TYPES = _IMAGE_CONTENT_TYPES | _FILE_CONTENT_TYPES def _flatten_message_content(content: Any) -> str | list[Any] | Any: """Convert list-of-blocks content to a string, preserving media blocks. Thinking/reasoning blocks are dropped. When a media block (image or file) is present, returns a list in the ORIGINAL order — consecutive text is joined into a single text block, media blocks kept as-is — so captions stay next to the attachment they describe. Otherwise returns a plain string. Non-list input is returned unchanged. Args: content: Message content — a string, a list of content blocks, or another type. Returns: A plain string for text-only content, a list of blocks when media is present, or the input unchanged for non-list input. """ if isinstance(content, str): return content if not isinstance(content, list): return content parts: list[str] = [] ordered_blocks: list[Any] = [] saw_media = False def _flush_text() -> None: if parts: ordered_blocks.append({"type": "text", "text": "\n\n".join(parts)}) parts.clear() for block in content: if isinstance(block, dict): btype = block.get("type") if btype in _MEDIA_CONTENT_TYPES: # Keep media as-is (never mutate; upstream copy.copy is shallow) # and preserve its position relative to surrounding text. _flush_text() ordered_blocks.append(block) saw_media = True continue if btype in _SKIP_CONTENT_TYPES: continue text = block.get("text") if text: parts.append(text) elif isinstance(block, str): parts.append(block) if saw_media: _flush_text() return ordered_blocks return "\n\n".join(parts) if parts else "" def _sanitize_messages( messages: list[Any], hoist_tool_media: bool = True, *, drop_reasoning_metadata: bool = False, ) -> list[Any]: """Flatten list content for OpenAI-compatible APIs, preserving media. Text/reasoning content is flattened to a string; image blocks are preserved. Tool messages cannot carry media on OpenAI-compatible APIs (content must be a string), so when ``hoist_tool_media`` is set the media in a tool result is moved into a HumanMessage emitted after that turn's (possibly parallel) tool messages. Anthropic-routed providers accept tool-result media natively and pass ``hoist_tool_media=False`` to keep it inline. """ import copy from langchain_core.messages import HumanMessage sanitize_tool_history = _has_assistant_tool_protocol(messages) if sanitize_tool_history: messages = _sanitize_openai_tool_history(messages) out: list[Any] = [] pending_media: list[Any] = [] # media hoisted out of a run of tool messages def _flush() -> None: if pending_media: out.append(HumanMessage(content=list(pending_media))) pending_media.clear() for msg in messages: if drop_reasoning_metadata: additional_kwargs = getattr(msg, "additional_kwargs", None) or {} if set(additional_kwargs) & _NONPORTABLE_REASONING_METADATA: msg = copy.copy(msg) msg.additional_kwargs = { key: value for key, value in additional_kwargs.items() if key not in _NONPORTABLE_REASONING_METADATA } is_tool = getattr(msg, "type", None) == "tool" if not is_tool: _flush() # emit hoisted media before any non-tool message if not isinstance(msg.content, list): out.append(msg) continue flat = _flatten_message_content(msg.content) if hoist_tool_media and is_tool and isinstance(flat, list): text_blocks = [ b for b in flat if isinstance(b, dict) and b.get("type") == "text" ] media_blocks = [ b for b in flat if not (isinstance(b, dict) and b.get("type") == "text") ] tool_msg = copy.copy(msg) # Join ALL text runs (interleaved content can yield more than one) # so no text is lost; tool content must be a string on OpenAI-compat. tool_msg.content = ( "\n\n".join(b["text"] for b in text_blocks) if text_blocks else "[media content provided in the following message]" ) out.append(tool_msg) pending_media.extend(media_blocks) else: msg = copy.copy(msg) msg.content = flat out.append(msg) _flush() # conversation may end with tool messages if sanitize_tool_history: _validate_openai_tool_history(out) return out # Fallback for models that reject some media input (no vision / no file # support): replace blocks of the unsupported types with a placeholder so the # conversation can keep going instead of erroring every turn that re-sends them. _UNSUPPORTED_MEDIA_PLACEHOLDER = ( "[attachment omitted: this model does not support this input type]" ) # Markers that identify WHICH modality an error rejects, so only the implicated # type is remembered (not all media). Generic markers map to all media. _IMAGE_ERROR_MARKERS = ("image_url", "image input", "support image") _FILE_ERROR_MARKERS = ("file input", "pdf input", "document input") # `expected \`text\`` keeps the backticks — the bare "expected text" phrase is # too broad (matches non-media schema errors like "... expected text"). _GENERIC_MEDIA_MARKERS = ("multimodal", "expected `text`") def _media_types_in(messages: list[BaseMessage]) -> set[str]: """Set of preserved-media block types present across the messages.""" found: set[str] = set() for msg in messages: content = getattr(msg, "content", None) if isinstance(content, list): for b in content: if isinstance(b, dict) and b.get("type") in _MEDIA_CONTENT_TYPES: found.add(b["type"]) return found def _media_error_types(exc: Exception) -> set[str]: """Media modalities the error text implicates (empty if not media-specific). Used to remember ONLY the rejected modality, so e.g. a PDF rejection never disables images. """ text = str(exc).lower() types: set[str] = set() if any(m in text for m in _IMAGE_ERROR_MARKERS): types |= _IMAGE_CONTENT_TYPES if any(m in text for m in _FILE_ERROR_MARKERS): types |= _FILE_CONTENT_TYPES if any(m in text for m in _GENERIC_MEDIA_MARKERS): types |= _MEDIA_CONTENT_TYPES return types def _is_http_400(exc: Exception) -> bool: """True if the error carries an HTTP 400 status (bad request).""" status = getattr(exc, "status_code", None) if status is None: status = getattr(getattr(exc, "response", None), "status_code", None) return status == 400 def _strip_media_types( messages: list[BaseMessage], types: set[str] ) -> list[BaseMessage]: """Replace blocks of the given types with a placeholder text block. Each stripped block is replaced IN PLACE (consecutive ones collapse into one placeholder), so surrounding text/media keep their original positions and a model that rejects only one modality still receives the others in order. """ import copy out: list[BaseMessage] = [] for msg in messages: content = getattr(msg, "content", None) if not isinstance(content, list) or not any( isinstance(b, dict) and b.get("type") in types for b in content ): out.append(msg) continue kept: list[Any] = [] last_was_placeholder = False for b in content: if isinstance(b, dict) and b.get("type") in types: if not last_was_placeholder: kept.append( {"type": "text", "text": _UNSUPPORTED_MEDIA_PLACEHOLDER} ) last_was_placeholder = True continue kept.append(b) last_was_placeholder = False msg = copy.copy(msg) msg.content = kept out.append(msg) return out class _OpenAICompatContent: """Apply OpenAI-compatible content normalization without owning a model.""" def __init__( self, profile: Mapping[str, object] | None, hoist_tool_media: bool, *, drop_reasoning_metadata: bool = False, ) -> None: self.hoist_tool_media = hoist_tool_media self.drop_reasoning_metadata = drop_reasoning_metadata self.blocked: set[str] = set() if profile is not None: if profile.get("image_inputs") is False: self.blocked |= _IMAGE_CONTENT_TYPES if profile.get("pdf_inputs") is False: self.blocked |= _FILE_CONTENT_TYPES def _prepare(self, messages: list[BaseMessage]) -> list[BaseMessage]: prepared = ( _strip_media_types(messages, self.blocked) if self.blocked else messages ) return _sanitize_messages( prepared, self.hoist_tool_media, drop_reasoning_metadata=self.drop_reasoning_metadata, ) def _stripped( self, messages: list[BaseMessage], suspects: set[str], ) -> list[BaseMessage]: return _sanitize_messages( _strip_media_types(messages, self.blocked | suspects), self.hoist_tool_media, drop_reasoning_metadata=self.drop_reasoning_metadata, ) def invoke( self, call: Callable[..., Any], messages: list[BaseMessage], *args, **kwargs, ) -> Any: suspects = _media_types_in(messages) - self.blocked prepared = self._prepare(messages) if not suspects: return call(prepared, *args, **kwargs) try: return call(prepared, *args, **kwargs) except Exception as exc: culprit = _media_error_types(exc) & suspects if not culprit and not _is_http_400(exc): raise try: result = call(self._stripped(messages, suspects), *args, **kwargs) except Exception: raise exc from None self.blocked.update(culprit) return result async def ainvoke( self, call: Callable[..., Awaitable[Any]], messages: list[BaseMessage], *args, **kwargs, ) -> Any: suspects = _media_types_in(messages) - self.blocked prepared = self._prepare(messages) if not suspects: return await call(prepared, *args, **kwargs) try: return await call(prepared, *args, **kwargs) except Exception as exc: culprit = _media_error_types(exc) & suspects if not culprit and not _is_http_400(exc): raise try: result = await call(self._stripped(messages, suspects), *args, **kwargs) except Exception: # stripping didn't help — surface original, don't cache raise exc from None self.blocked.update(culprit) return result def stream( self, call: Callable[..., Iterator[Any]], messages: list[BaseMessage], *args, **kwargs, ) -> Iterator[Any]: suspects = _media_types_in(messages) - self.blocked prepared = self._prepare(messages) if not suspects: yield from call(prepared, *args, **kwargs) return started = False try: for chunk in call(prepared, *args, **kwargs): started = True yield chunk return except Exception as exc: culprit = _media_error_types(exc) & suspects if started or (not culprit and not _is_http_400(exc)): raise media_exc = exc retry_started = False try: for chunk in call(self._stripped(messages, suspects), *args, **kwargs): if not retry_started: retry_started = True self.blocked.update(culprit) yield chunk except Exception: if retry_started: raise raise media_exc from None if not retry_started: raise media_exc from None async def astream( self, call: Callable[..., AsyncIterator[Any]], messages: list[BaseMessage], *args, **kwargs, ) -> AsyncIterator[Any]: suspects = _media_types_in(messages) - self.blocked prepared = self._prepare(messages) if not suspects: async for chunk in call(prepared, *args, **kwargs): yield chunk return started = False try: async for chunk in call(prepared, *args, **kwargs): started = True yield chunk return except Exception as exc: culprit = _media_error_types(exc) & suspects if started or (not culprit and not _is_http_400(exc)): raise media_exc = exc retry_started = False try: async for chunk in call( self._stripped(messages, suspects), *args, **kwargs ): if not retry_started: retry_started = True self.blocked.update(culprit) yield chunk except Exception: if retry_started: raise raise media_exc from None if not retry_started: raise media_exc from None def _patch_openai_compat_content( model: Any, hoist_tool_media: bool = True, *, drop_reasoning_metadata: bool = False, ) -> None: """Normalize content for OpenAI-compatible models lacking a native adapter. Ai4Sci extension: ``drop_reasoning_metadata`` strips hidden provider reasoning traces so a generic proxy never receives another provider's private chain-of-thought. """ import functools profile = getattr(model, "profile", None) compat = _OpenAICompatContent( profile if isinstance(profile, Mapping) else None, hoist_tool_media, drop_reasoning_metadata=drop_reasoning_metadata, ) orig_generate = getattr(model, "_generate", None) if orig_generate is None: return @functools.wraps(orig_generate) def _patched_generate( messages: list[BaseMessage], *args: Any, **kwargs: Any ) -> Any: return compat.invoke(orig_generate, messages, *args, **kwargs) model._generate = _patched_generate orig_agenerate = getattr(model, "_agenerate", None) if orig_agenerate is not None: @functools.wraps(orig_agenerate) async def _patched_agenerate( messages: list[BaseMessage], *args: Any, **kwargs: Any ) -> Any: return await compat.ainvoke(orig_agenerate, messages, *args, **kwargs) model._agenerate = _patched_agenerate # Also patch streaming paths — CLI/agent uses _stream/_astream, so without # these the content flattening is bypassed during normal streaming calls. orig_stream = getattr(model, "_stream", None) if orig_stream is not None: @functools.wraps(orig_stream) def _patched_stream( messages: list[BaseMessage], *args: Any, **kwargs: Any ) -> Any: yield from compat.stream(orig_stream, messages, *args, **kwargs) model._stream = _patched_stream orig_astream = getattr(model, "_astream", None) if orig_astream is not None: @functools.wraps(orig_astream) async def _patched_astream( messages: list[BaseMessage], *args: Any, **kwargs: Any ) -> Any: async for chunk in compat.astream(orig_astream, messages, *args, **kwargs): yield chunk model._astream = _patched_astream # --------------------------------------------------------------------------- # Patch: ccproxy Codex Responses API rejects "system" role messages. # Convert SystemMessage to use "developer" role via langchain-openai's # __openai_role__ mechanism. # --------------------------------------------------------------------------- def _patch_ccproxy_system_to_developer(model: Any) -> None: """Convert SystemMessage role from 'system' to 'developer' for ccproxy. ccproxy's Responses API endpoint rejects system role messages with 400 "System messages are not allowed". LangChain's ``langchain_openai`` checks ``additional_kwargs["__openai_role__"]`` and uses that value as the message role when serializing to the API. Args: model: A LangChain chat model instance to patch in-place. """ import copy import functools from langchain_core.messages import SystemMessage def _system_to_developer(messages: list[BaseMessage]) -> list[BaseMessage]: out: list[BaseMessage] = [] for msg in messages: if isinstance(msg, SystemMessage): if msg.additional_kwargs.get("__openai_role__") != "developer": msg = copy.copy(msg) msg.additional_kwargs = { **msg.additional_kwargs, "__openai_role__": "developer", } out.append(msg) return out orig_generate = getattr(model, "_generate", None) if orig_generate is None: return @functools.wraps(orig_generate) def _patched_generate( messages: list[BaseMessage], *args: Any, **kwargs: Any ) -> Any: return orig_generate(_system_to_developer(messages), *args, **kwargs) model._generate = _patched_generate orig_agenerate = getattr(model, "_agenerate", None) if orig_agenerate is not None: @functools.wraps(orig_agenerate) async def _patched_agenerate( messages: list[BaseMessage], *args: Any, **kwargs: Any ) -> Any: return await orig_agenerate(_system_to_developer(messages), *args, **kwargs) model._agenerate = _patched_agenerate orig_stream = getattr(model, "_stream", None) if orig_stream is not None: @functools.wraps(orig_stream) def _patched_stream( messages: list[BaseMessage], *args: Any, **kwargs: Any ) -> Any: return orig_stream(_system_to_developer(messages), *args, **kwargs) model._stream = _patched_stream orig_astream = getattr(model, "_astream", None) if orig_astream is not None: @functools.wraps(orig_astream) async def _patched_astream( messages: list[BaseMessage], *args: Any, **kwargs: Any ) -> Any: async for chunk in orig_astream( _system_to_developer(messages), *args, **kwargs ): yield chunk model._astream = _patched_astream # --------------------------------------------------------------------------- # Patch (module-level): langchain-openai's _convert_dict_to_message and # _convert_delta_to_message_chunk discard provider-specific fields like # `reasoning_content`. We monkey-patch them to capture reasoning_content # into AIMessage.additional_kwargs so downstream code (incl. our passback # patch) can find it. Benign for non-DeepSeek providers — they just don't # return this field, so the patch is a no-op for them. # --------------------------------------------------------------------------- _openai_capture_patched = False def _patch_openai_capture_reasoning_content() -> None: global _openai_capture_patched if _openai_capture_patched: return try: import langchain_openai.chat_models.base as _base _orig_dict_to_msg = _base._convert_dict_to_message _orig_delta_to_chunk = _base._convert_delta_to_message_chunk def _patched_dict_to_msg(_dict, *args, **kwargs): msg = _orig_dict_to_msg(_dict, *args, **kwargs) rc = _dict.get("reasoning_content") if isinstance(_dict, dict) else None if isinstance(rc, str) and rc and hasattr(msg, "additional_kwargs"): msg.additional_kwargs["reasoning_content"] = rc return msg def _patched_delta_to_chunk(_dict, *args, **kwargs): chunk = _orig_delta_to_chunk(_dict, *args, **kwargs) rc = _dict.get("reasoning_content") if isinstance(_dict, dict) else None if isinstance(rc, str) and rc and hasattr(chunk, "additional_kwargs"): # Per-chunk: stash this delta's reasoning_content on the chunk. # Cross-chunk accumulation is handled by AIMessageChunk.__add__ # via merge_dicts (string values in additional_kwargs concatenate). chunk.additional_kwargs["reasoning_content"] = ( chunk.additional_kwargs.get("reasoning_content", "") + rc ) return chunk _base._convert_dict_to_message = _patched_dict_to_msg _base._convert_delta_to_message_chunk = _patched_delta_to_chunk _openai_capture_patched = True except Exception: pass _patch_openai_capture_reasoning_content() # --------------------------------------------------------------------------- # Patch (module-level): silence langgraph_api's OpenAPI schema-generation # warnings for endpoints whose docstrings aren't valid YAML. # # Upstream ``langgraph_api.utils.SchemaGenerator.get_schema`` calls # ``parse_docstring`` (inherited from Starlette's ``BaseSchemaGenerator``) # on every registered endpoint. When the docstring is prose with stray # ``:`` characters, ``yaml.safe_load`` raises and upstream logs the # failure + full traceback at WARNING level. It then falls back to # ``{"description": docstring}`` — the endpoint still ends up in the # schema with its prose as the description, just without structured # ``parameters``/``responses``/``tags`` fields. # # The fallback path is fine; the warning + traceback is just noise. And # it's only triggered for our deploy because mounting any custom Starlette # app (``EvoScientist/langgraph_dev/http.py``) makes upstream call # ``update_openapi_spec`` at startup — which iterates EVERY route, # including upstream's own endpoints whose prose docstrings predate the # YAML convention. # # Fix: wrap ``parse_docstring`` itself and absorb ``yaml.YAMLError`` by # returning the same fallback shape upstream's except branch produces. # Non-YAML exceptions are deliberately left to propagate — upstream's # ``get_schema`` already catches them and logs WARNING + traceback, so # unexpected failures remain debuggable. Patching ``parse_docstring`` (a # small, stable method) instead of ``get_schema`` (the larger loop body) # minimizes our exposure to upstream churn. # --------------------------------------------------------------------------- _langgraph_schema_silenced_patched = False def _patch_langgraph_schema_generator_silence_warnings() -> None: global _langgraph_schema_silenced_patched if _langgraph_schema_silenced_patched: return try: import langgraph_api.utils as _lgapi_utils import yaml _SchemaGenerator = _lgapi_utils.SchemaGenerator _orig_parse_docstring = _SchemaGenerator.parse_docstring def _patched_parse_docstring(self: Any, func: Any) -> dict[str, Any]: try: return _orig_parse_docstring(self, func) except yaml.YAMLError: return {"description": getattr(func, "__doc__", None) or ""} _SchemaGenerator.parse_docstring = _patched_parse_docstring _langgraph_schema_silenced_patched = True except Exception: # Patches are loader-safe: never crash the import. Silent failure # here just leaves the upstream warnings visible in deploy logs, # which is a benign fallback. pass _patch_langgraph_schema_generator_silence_warnings() # --------------------------------------------------------------------------- # Patch (lazy, Anthropic protocol): strip foreign reasoning blocks from # outgoing assistant messages. # # History produced on OpenAI-compatible providers (Qwen/DashScope, DeepSeek, …) # can carry `reasoning_content` content blocks in AI messages. When a session # switches to an Anthropic-protocol endpoint (native Anthropic, custom-anthropic, # kimi-coding, minimax), langchain-anthropic's `_format_messages` passes unknown # block types through verbatim (`else: content.append(block)`), and strict # Anthropic-compatible backends (e.g. Moonshot's Kimi For Coding endpoint at # api.kimi.com) reject the request with an opaque HTTP 400. Upstream already # skips the OpenAI-Responses `reasoning` / `function_call` types the same way; # `reasoning_content` is simply missing from that list. These blocks are display # metadata — no Anthropic-protocol block type has this name — so stripping is # always safe. Anthropic `thinking` blocks are preserved (Kimi backends require # them on tool-call turns, even unsigned), but a missing `signature` is # defaulted to "": Anthropic-compatible backends stream thinking without a # signature_delta, so the aggregated block lacks the key, and strict edges # (e.g. OpenRouter's Anthropic schema) reject replay with 400 # ("signature: expected string, received undefined"). Native Anthropic thinking # always carries a signature string, so the default never fires there. # --------------------------------------------------------------------------- _anthropic_foreign_reasoning_patched = False _FOREIGN_REASONING_BLOCK_TYPES = frozenset({"reasoning_content"}) def _normalize_anthropic_replay_messages( messages: Sequence[BaseMessage], ) -> Sequence[BaseMessage]: """Sanitize AI-message list content for Anthropic-protocol replay.""" cleaned: list[BaseMessage] = [] changed = False for message in messages: if isinstance(message, AIMessage) and isinstance(message.content, list): blocks: list[Any] = [] block_changed = False for block in message.content: if isinstance(block, dict): btype = block.get("type") if btype in _FOREIGN_REASONING_BLOCK_TYPES: block_changed = True continue if btype == "thinking" and not isinstance( block.get("signature"), str ): block = {**block, "signature": ""} block_changed = True blocks.append(block) if block_changed: message = message.model_copy(update={"content": blocks}) changed = True cleaned.append(message) return cleaned if changed else messages def _patch_anthropic_strip_foreign_reasoning() -> None: global _anthropic_foreign_reasoning_patched if _anthropic_foreign_reasoning_patched: return try: import langchain_anthropic.chat_models as _mod _orig = _mod._format_messages @functools.wraps(_orig) def _patched(messages: Sequence[BaseMessage]) -> Any: return _orig(_normalize_anthropic_replay_messages(messages)) _mod._format_messages = _patched _anthropic_foreign_reasoning_patched = True except Exception: pass # --------------------------------------------------------------------------- # Patch (lazy, OpenRouter only): strip OpenAI-Responses encrypted reasoning # items from outgoing assistant messages. # # OpenRouter's `/responses` beta is stateless — it does not propagate # `store=true` / `previous_response_id` upstream. So when a prior turn's # assistant message carries a reasoning item with an `rs_*` id (the encrypted # Responses reasoning block), replaying it on the next turn fails with HTTP 400: # "Item with id 'rs_...' not found. Items are not persisted when `store` is # set to false. ... remove this item from your input." # (observed against both Azure and OpenAI upstreams — so deepagents' # azure-ignore is not sufficient). The only robust fix is to drop these items # on passback, exactly as the upstream error instructs. Reasoning DISPLAY is # unaffected: it happens when the item is generated, not on passback. # Related: langchain-ai/langchain#37777. # --------------------------------------------------------------------------- _openrouter_reasoning_strip_patched = False def _is_responses_reasoning_item(entry: Any) -> bool: """True for an OpenAI-Responses encrypted reasoning item (`rs_` id / data).""" if not isinstance(entry, dict): return False return str(entry.get("id") or "").startswith("rs_") or bool(entry.get("data")) def _patch_openrouter_strip_responses_reasoning() -> None: global _openrouter_reasoning_strip_patched if _openrouter_reasoning_strip_patched: return try: import langchain_openrouter.chat_models as _mod _orig = _mod._convert_message_to_dict def _patched(message: Any) -> Any: result = _orig(message) details = ( result.get("reasoning_details") if isinstance(result, dict) else None ) if isinstance(details, list): kept = [e for e in details if not _is_responses_reasoning_item(e)] if kept: result["reasoning_details"] = kept else: result.pop("reasoning_details", None) return result _mod._convert_message_to_dict = _patched _openrouter_reasoning_strip_patched = True except Exception: pass # --------------------------------------------------------------------------- # Patch (lazy, OpenRouter only): default structured output to json_schema for # Moonshot's always-thinking models (kimi-k3). # # Moonshot rejects a forced tool choice while thinking is enabled with HTTP 400: # "tool_choice 'specified' is incompatible with thinking enabled" # and kimi-k3's thinking cannot be disabled, so the function_calling default of # ChatOpenRouter.with_structured_output fails every structured-output call # (LLMToolSelectorMiddleware included). These endpoints support # response_format json_schema, which needs no tool_choice — route them there. # Moonshot-specific: other mandatory-reasoning models keep function_calling. # Gated on the instance's model_name, so copies behave correctly and other # models keep the function_calling default. # --------------------------------------------------------------------------- _openrouter_structured_output_patched = False def _patch_openrouter_structured_output() -> None: global _openrouter_structured_output_patched if _openrouter_structured_output_patched: return try: from langchain_openrouter import ChatOpenRouter _orig = ChatOpenRouter.with_structured_output def _patched( self: Any, schema: Any = None, *, method: str = "function_calling", **kwargs: Any, ) -> Any: if method == "function_calling": from .models import _OPENROUTER_JSON_SCHEMA_STRUCTURED_OUTPUT_MODELS if self.model_name in _OPENROUTER_JSON_SCHEMA_STRUCTURED_OUTPUT_MODELS: method = "json_schema" return _orig(self, schema, method=method, **kwargs) ChatOpenRouter.with_structured_output = _patched _openrouter_structured_output_patched = True except Exception: pass # --------------------------------------------------------------------------- # Patch (lazy, Anthropic protocol): default structured output to json_schema # for mandatory-thinking Kimi models (K3 / Kimi For Coding). # # These endpoints run with thinking always on and reject a forced tool choice # with HTTP 400 ("tool_choice 'specified' is incompatible with thinking # enabled") — ChatAnthropic's function_calling default forces tool_choice, so # every structured-output call (LLMToolSelectorMiddleware included) fails. # With thinking declared, langchain-anthropic falls back to tool_choice auto, # but the model then calls the tool only sometimes (OutputParserException). # These endpoints accept output_config.format (verified live via OpenRouter's # Anthropic-compatible edge), which needs no tool_choice — route them there. # Gated on the instance's model, so copies behave correctly and Claude models # keep the function_calling default. # --------------------------------------------------------------------------- _anthropic_structured_output_patched = False def _patch_anthropic_structured_output() -> None: global _anthropic_structured_output_patched if _anthropic_structured_output_patched: return try: from langchain_anthropic import ChatAnthropic _orig = ChatAnthropic.with_structured_output def _patched( self: Any, schema: Any = None, *, method: str = "function_calling", **kwargs: Any, ) -> Any: if method == "function_calling": from .models import _is_mandatory_thinking_kimi if _is_mandatory_thinking_kimi(self.model): method = "json_schema" return _orig(self, schema, method=method, **kwargs) ChatAnthropic.with_structured_output = _patched _anthropic_structured_output_patched = True except Exception: pass # --------------------------------------------------------------------------- # Patch: forward CLI's live (model, model_provider) into deepagents' # start_async_task / update_async_task tool calls so the deployed graph # (running in a separate ``langgraph dev`` subprocess) re-resolves the # chat model per run. # # Without this, async sub-agents stay on the model their graph was compiled # with at langgraph dev boot — `/model` switches in the CLI never reach # them because they live in another process. # # Mechanism: wrap deepagents' ``_build_start_tool`` and ``_build_update_tool`` # factories. Each wrapped factory calls the original with a proxied client # cache that intercepts ``runs.create(...)`` calls only and injects # ``config={"configurable": {"model": , "model_provider": }}``. # All other client methods (``threads.create``, ``runs.get``, ``runs.cancel``, # ``runs.join_stream``) pass through unchanged. The deployed graph picks up # ``configurable.model`` via ``ConfigurableModelMiddleware``. # # Reads ``_ensure_config()`` at tool-call time (not patch time) so a # ``/model`` switch in the CLI is reflected on the very next async tool # call without an agent rebuild. # # Upstream PR opportunity: passing ``config`` through ``client.runs.create`` # is generic functionality; worth contributing back to ``langchain-ai/deepagents`` # so this patch can be retired. # --------------------------------------------------------------------------- _model_passthrough_patched = False # The launching run's per-run (model, provider), set by the async-task tools # from ``runtime.config`` for the duration of a single ``runs.create`` call. # ``_merge_runs_config_kwargs`` reads it as the highest-precedence source. # # Why a ContextVar rather than passing ``config=`` through each tool: the fix # has to hold at every ``runs.create`` site (``start_async_task`` and # ``update_async_task``), but only the tool functions can see the caller's # per-run model (via ``runtime.config``, the config langgraph's ToolNode # injects into tool calls). Threading it through a ContextVar # lets the single merge point below inject it, so ``update_async_task`` (whose # body we delegate to upstream unchanged) is covered without reimplementing it. _caller_configurable: ContextVar[dict[str, str] | None] = ContextVar( "_evo_caller_configurable", default=None ) def _extract_caller_configurable(config: Any) -> dict[str, str]: """Pull ``(model, model_provider)`` out of a launching run's ``config``. ``config`` is the tool's ``runtime.config`` (a ``RunnableConfig``-shaped dict). Returns a dict suitable for ``configurable``; empty when the caller carries no model override, so the config-default fallback still applies. ``model_provider`` is only forwarded alongside a model — a bare provider without a model is meaningless to the deployed graph's resolver. """ configurable = (config or {}).get("configurable") or {} model = configurable.get("model") provider = configurable.get("model_provider") out: dict[str, str] = {} if isinstance(model, str) and model: out["model"] = model if isinstance(provider, str) and provider: out["model_provider"] = provider return out def _read_cfg_configurable() -> dict[str, str]: """Read live ``(model, provider)`` from EvoScientist config. Returns a dict suitable for inserting under ``RunnableConfig.configurable``. Empty dict on any failure (so the patch degrades to a no-op rather than breaking async tool calls). """ try: from EvoScientist.EvoScientist import _ensure_config cfg = _ensure_config() except Exception: return {} out: dict[str, str] = {} model = getattr(cfg, "model", None) provider = getattr(cfg, "provider", None) if isinstance(model, str) and model: out["model"] = model if isinstance(provider, str) and provider: out["model_provider"] = provider return out def _merge_runs_config_kwargs(kwargs: dict) -> dict: """Merge the live model override into ``kwargs`` for ``runs.create``. Model source, in increasing precedence: 1. ``_read_cfg_configurable()`` — the effective EvoScientist config. On the in-process backend this is the CLI's live per-run model; on the ``langgraph_server`` backend this proxy runs inside the dev-server process, where it reports the *server's* config-default model, not the CLI's per-run choice. 2. ``_caller_configurable`` — the launching run's actual model, forwarded by the async-task tools from ``runtime.config``. This wins so that a sub-agent launched (or continued) while the caller runs on, say, a free model does not silently fall back to a billed config-default. Any unrelated caller-supplied ``config.configurable`` keys are preserved. """ overrides = {**_read_cfg_configurable(), **(_caller_configurable.get() or {})} if not overrides: return kwargs existing = kwargs.get("config") if not isinstance(existing, dict): existing = {} existing_configurable = existing.get("configurable") if not isinstance(existing_configurable, dict): existing_configurable = {} merged_configurable = {**existing_configurable, **overrides} kwargs = dict(kwargs) kwargs["config"] = {**existing, "configurable": merged_configurable} return kwargs class _SyncRunsProxy: """Wraps a sync ``RunsClient`` and injects config into ``create`` only.""" def __init__(self, real: Any) -> None: object.__setattr__(self, "_real", real) def __getattr__(self, name: str) -> Any: return getattr(self._real, name) def create(self, **kwargs: Any) -> Any: return self._real.create(**_merge_runs_config_kwargs(kwargs)) class _AsyncRunsProxy: """Wraps an async ``RunsClient`` and injects config into ``create`` only.""" def __init__(self, real: Any) -> None: object.__setattr__(self, "_real", real) def __getattr__(self, name: str) -> Any: return getattr(self._real, name) async def create(self, **kwargs: Any) -> Any: return await self._real.create(**_merge_runs_config_kwargs(kwargs)) class _ClientProxy: """Lightweight proxy that swaps ``client.runs`` for a runs proxy. Read-only forwarding: only ``__getattr__`` is overridden. Attribute *writes* on the proxy land on the proxy itself, NOT on the wrapped real client. Current deepagents only reads ``.threads`` / ``.runs``, so this is safe — but if a future caller tries ``client.foo = bar`` through the proxy, the write will be silently lost. ``object.__setattr__`` in ``__init__`` is used solely to avoid infinite recursion when seeding the internal slots. ``client.runs`` is a stable attribute set in ``langgraph_sdk.client.LangGraphClient.__init__`` (``self.runs = RunsClient(...)``) — not a property or lazy initializer — so the wrapped runs proxy is built once at ``__init__`` time and reused on every ``proxy.runs`` access. This avoids the previous behavior of selecting sync/async and instantiating a new ``_RunsProxy`` per attribute access. """ def __init__(self, real: Any, *, is_async: bool) -> None: runs_proxy_cls = _AsyncRunsProxy if is_async else _SyncRunsProxy object.__setattr__(self, "_real", real) object.__setattr__(self, "_runs_proxy", runs_proxy_cls(real.runs)) def __getattr__(self, name: str) -> Any: if name == "runs": return self._runs_proxy return getattr(self._real, name) class _ClientCacheProxy: """Proxy a ``_ClientCache`` so callers receive config-injecting clients.""" def __init__(self, real: Any) -> None: self._real = real def get_sync(self, name: str) -> Any: return _ClientProxy(self._real.get_sync(name), is_async=False) def get_async(self, name: str) -> Any: return _ClientProxy(self._real.get_async(name), is_async=True) def _patch_deepagents_model_passthrough() -> None: """Wrap deepagents' async-launch tool factories to inject CLI model. Idempotent: re-invocation is a no-op once the patch is active. Safe to call from ``_maybe_swap_async_subagents`` on every CLI startup; both that hook and this patch turn on together when async sub-agents are enabled. """ global _model_passthrough_patched if _model_passthrough_patched: return try: from deepagents.middleware import async_subagents as ds_mod except ImportError: return # Defensive ``getattr`` lookups mirror the rest of this file (lines 254, # 266, 279, 290, 339, 351, 362, 373, 473): a deepagents update that # renames or removes either private helper degrades to a no-op instead # of raising ``AttributeError`` at CLI startup. orig_build_start = getattr(ds_mod, "_build_start_tool", None) orig_build_update = getattr(ds_mod, "_build_update_tool", None) if orig_build_start is None or orig_build_update is None: return def _patched_build_start( agent_map: Any, clients: Any, tool_description: str ) -> Any: return orig_build_start(agent_map, _ClientCacheProxy(clients), tool_description) def _patched_build_update(agent_map: Any, clients: Any) -> Any: return orig_build_update(agent_map, _ClientCacheProxy(clients)) ds_mod._build_start_tool = _patched_build_start ds_mod._build_update_tool = _patched_build_update _model_passthrough_patched = True # --------------------------------------------------------------------------- # Patch: LangGraph 1.2.9 _control_branch crashes with TypeError when # Command.goto is None (e.g. from a resume command). We patch both the # generator (map_cmd) and the consumer (_control_branch). # --------------------------------------------------------------------------- def _patch_langgraph_goto_none_crash() -> None: """Handle Command(goto=None) defensively in langgraph and langgraph-api. Either patch alone prevents the production crash; both together cover all code paths (a Command(goto=None) could in principle be created by code other than map_cmd). Matches upstream issue langchain-ai/langgraph#5656. """ # ── 1. map_cmd — prevent goto=None from being created ──────────────── try: import langgraph_api.command as _cmd_mod from langgraph.types import Command as _Command _orig_map_cmd = getattr(_cmd_mod, "map_cmd", None) if _orig_map_cmd and not hasattr(_orig_map_cmd, "_evosci_goto_none_fix"): @functools.wraps(_orig_map_cmd) def _patched_map_cmd(cmd: Any) -> _Command: result = _orig_map_cmd(cmd) if result.goto is None: return dataclasses.replace(result, goto=[]) return result _patched_map_cmd._evosci_goto_none_fix = True # type: ignore[attr-defined] _cmd_mod.map_cmd = _patched_map_cmd # Rebind on already-loaded consumers that grabbed map_cmd via # ``from langgraph_api.command import map_cmd`` — those hold the # original reference and wouldn't see the module-level rebind. for _name in ("langgraph_api.stream", "langgraph_api.grpc.ops.threads"): _mod = sys.modules.get(_name) if _mod is not None and getattr(_mod, "map_cmd", None) is _orig_map_cmd: _mod.map_cmd = _patched_map_cmd except ImportError: pass # ── 2. _control_branch — handle goto=None defensively ─────────────── # Wrap (not reimplement) the upstream function so only the goto=None→[] # normalization is ours; all other routing logic delegates to the # original, avoiding drift as upstream evolves. try: import langgraph.graph.state as _state_mod from langgraph.types import Command as _Command _orig_ctrl = getattr(_state_mod, "_control_branch", None) if _orig_ctrl and not hasattr(_orig_ctrl, "_evosci_goto_none_fix"): def _fix_goto(cmd: Any) -> Any: if isinstance(cmd, _Command) and cmd.goto is None: return dataclasses.replace(cmd, goto=[]) return cmd @functools.wraps(_orig_ctrl) def _patched_ctrl(value: Any) -> Sequence[tuple[str, Any]]: if isinstance(value, (list, tuple)): value = [_fix_goto(v) for v in value] else: value = _fix_goto(value) return _orig_ctrl(value) _patched_ctrl._evosci_goto_none_fix = True # type: ignore[attr-defined] _state_mod._control_branch = _patched_ctrl except (ImportError, AttributeError): pass _patch_langgraph_goto_none_crash() _NONPORTABLE_REASONING_METADATA = frozenset( {"reasoning_content", "reasoning_details"} ) def _stable_tool_call_id(message: Any, message_index: int, call_index: int) -> str: seed = ":".join( ( str(getattr(message, "id", "") or "message"), str(message_index), str(call_index), ) ) return "call_" + hashlib.sha256(seed.encode("utf-8")).hexdigest()[:24] def _tool_message_match_index( tool_messages: list[Any], used_indexes: set[int], *, call_id: str, call_name: str, ) -> int | None: """Find the best unused result for one assistant tool call.""" def _matches(index: int, *, require_id: bool, require_name: bool) -> bool: if index in used_indexes: return False message = tool_messages[index] result_id = str(getattr(message, "tool_call_id", "") or "") result_name = str(getattr(message, "name", "") or "") if require_id and result_id != call_id: return False if not require_id and result_id: return False return not require_name or not result_name or result_name == call_name if call_id: for require_name in (True, False): for index in range(len(tool_messages)): if _matches(index, require_id=True, require_name=require_name): return index for require_name in (True, False): for index in range(len(tool_messages)): if _matches(index, require_id=False, require_name=require_name): return index return None # A result-side identifier is more authoritative than a generated fallback. for require_name in (True, False): for index, message in enumerate(tool_messages): if index in used_indexes: continue result_id = str(getattr(message, "tool_call_id", "") or "") result_name = str(getattr(message, "name", "") or "") if result_id and ( not require_name or not result_name or result_name == call_name ): return index for require_name in (True, False): for index in range(len(tool_messages)): if _matches(index, require_id=False, require_name=require_name): return index return None def _copy_ai_message_with_tool_pairs( message: Any, message_index: int, tool_messages: list[Any], ) -> tuple[Any | None, list[Any]]: """Return a replay-safe assistant message and its matched tool results.""" import copy copied = copy.copy(message) additional_kwargs = dict(getattr(message, "additional_kwargs", None) or {}) # Parsed tool_calls are canonical. Raw copies can otherwise re-introduce an # invalid call after invalid_tool_calls has been cleared. additional_kwargs.pop("tool_calls", None) copied.additional_kwargs = additional_kwargs copied.invalid_tool_calls = [] original_calls = list(getattr(message, "tool_calls", None) or []) used_results: set[int] = set() matched_calls: list[dict[str, Any]] = [] matched_result_indexes: list[int] = [] original_to_matched_call: dict[int, tuple[str, str]] = {} for call_index, original_call in enumerate(original_calls): call = dict(original_call) call_id = str(call.get("id") or "") call_name = str(call.get("name") or "").strip() # A missing name is structurally unreplayable. Never infer it from # arguments or retain its paired ToolMessage in provider history. if not call_name: continue call["name"] = call_name result_index = _tool_message_match_index( tool_messages, used_results, call_id=call_id, call_name=call_name, ) # A historical client-side function call is only replayable together # with its result. Incomplete calls are discarded instead of asking the # provider to continue a broken tool turn. if result_index is None: continue if not call_id: result_id = str( getattr(tool_messages[result_index], "tool_call_id", "") or "" ) call_id = result_id or _stable_tool_call_id( message, message_index, call_index ) call["id"] = call_id matched_calls.append(call) matched_result_indexes.append(result_index) original_to_matched_call[call_index] = (call_id, call_name) used_results.add(result_index) copied.tool_calls = matched_calls if isinstance(copied.content, list): original_call_index = 0 blocks: list[Any] = [] for original_block in copied.content: if not isinstance(original_block, dict): blocks.append(original_block) continue block = dict(original_block) if block.get("type") in {"tool_call", "function_call"}: matched_call = original_to_matched_call.get(original_call_index) original_call_index += 1 if matched_call is None: continue call_id, call_name = matched_call # LangChain content blocks use id; the Responses converter later # maps it to call_id. block["id"] = call_id if "call_id" in block: block["call_id"] = call_id block["name"] = call_name if isinstance(block.get("function"), dict): block["function"] = {**block["function"], "name": call_name} blocks.append(block) copied.content = blocks matched_results: list[Any] = [] result_to_call_id = { result_index: matched_calls[index]["id"] for index, result_index in enumerate(matched_result_indexes) } for result_index, result in enumerate(tool_messages): call_id = result_to_call_id.get(result_index) if call_id is None: continue copied_result = copy.copy(result) copied_result.tool_call_id = call_id matched_results.append(copied_result) had_tool_protocol = bool(original_calls) or bool( getattr(message, "invalid_tool_calls", None) ) if not matched_calls and had_tool_protocol: replayable_content = _flatten_message_content(copied.content) if not replayable_content: return None, matched_results return copied, matched_results def _sanitize_openai_tool_history(messages: list[Any]) -> list[Any]: """Copy history while retaining only complete, replayable tool turns.""" normalized: list[Any] = [] index = 0 while index < len(messages): message = messages[index] message_type = getattr(message, "type", None) if message_type == "tool": # A tool result without its immediately preceding assistant call is # invalid for both Chat Completions and Responses APIs. index += 1 continue if message_type != "ai": normalized.append(message) index += 1 continue next_index = index + 1 tool_messages: list[Any] = [] while ( next_index < len(messages) and getattr(messages[next_index], "type", None) == "tool" ): tool_messages.append(messages[next_index]) next_index += 1 copied, matched_results = _copy_ai_message_with_tool_pairs( message, index, tool_messages, ) if copied is not None: normalized.append(copied) normalized.extend(matched_results) index = next_index return normalized def _ensure_openai_tool_call_ids(messages: list[Any]) -> list[Any]: """Backward-compatible alias for replay-safe tool history normalization.""" return _sanitize_openai_tool_history(messages) def _has_assistant_tool_protocol(messages: list[Any]) -> bool: """Return whether history contains assistant-side tool protocol state.""" for message in messages: if getattr(message, "type", None) != "ai": continue if getattr(message, "tool_calls", None) or getattr( message, "invalid_tool_calls", None ): return True additional_kwargs = getattr(message, "additional_kwargs", None) or {} if additional_kwargs.get("tool_calls"): return True content = getattr(message, "content", None) if isinstance(content, list) and any( isinstance(block, dict) and block.get("type") in {"tool_call", "function_call"} for block in content ): return True return False def _validate_openai_tool_history(messages: list[Any]) -> None: """Raise when sanitized history still contains an invalid tool protocol.""" available_call_ids: set[str] = set() for message in messages: message_type = getattr(message, "type", None) if message_type == "ai": if getattr(message, "invalid_tool_calls", None): raise ValueError("invalid_tool_calls must not be replayed") response_call_ids: set[str] = set() response_calls: dict[str, str] = {} for call in getattr(message, "tool_calls", None) or []: call_name = str(call.get("name") or "").strip() if not call_name: raise ValueError("assistant tool call is missing a name") call_id = str(call.get("id") or "").strip() if not call_id: raise ValueError("assistant tool call is missing an id") if call_id in response_call_ids or call_id in available_call_ids: raise ValueError( "assistant tool call id is duplicated while outstanding" ) response_call_ids.add(call_id) available_call_ids.add(call_id) response_calls[call_id] = call_name content = getattr(message, "content", None) content_call_ids: set[str] = set() if isinstance(content, list): for block in content: if not isinstance(block, dict) or block.get("type") not in { "tool_call", "function_call", }: continue block_id = str( block.get("call_id") or block.get("id") or "" ).strip() block_name = block.get("name") or block.get("tool_name") function = block.get("function") if not block_name and isinstance(function, dict): block_name = function.get("name") block_name = str(block_name or "").strip() if ( not block_id or block_id in content_call_ids or response_calls.get(block_id) != block_name ): raise ValueError( "assistant content block does not match parsed tool call" ) content_call_ids.add(block_id) elif message_type == "tool": call_id = str(getattr(message, "tool_call_id", "") or "") if not call_id or call_id not in available_call_ids: raise ValueError("tool result does not match a prior tool call") available_call_ids.remove(call_id) if available_call_ids: raise ValueError("assistant tool call is missing its tool result") _openai_empty_sse_keepalive_patched = False def _is_blank_sse_keepalive(event: Any) -> bool: """Return whether an SSE event has no JSON payload to parse.""" data = getattr(event, "data", None) return data is None or (isinstance(data, str) and not data.strip()) def _patch_openai_empty_sse_keepalive() -> None: global _openai_empty_sse_keepalive_patched if _openai_empty_sse_keepalive_patched: return try: import functools from openai._streaming import AsyncStream as _AsyncStream from openai._streaming import Stream as _Stream original_async = _AsyncStream._iter_events if not getattr(original_async, "_evoscientist_skips_blank_sse", False): @functools.wraps(original_async) async def _filtered_async_events(self: Any) -> Any: async for event in original_async(self): if not _is_blank_sse_keepalive(event): yield event _filtered_async_events._evoscientist_skips_blank_sse = True # type: ignore[attr-defined] _AsyncStream._iter_events = _filtered_async_events original_sync = _Stream._iter_events if not getattr(original_sync, "_evoscientist_skips_blank_sse", False): @functools.wraps(original_sync) def _filtered_sync_events(self: Any) -> Any: for event in original_sync(self): if not _is_blank_sse_keepalive(event): yield event _filtered_sync_events._evoscientist_skips_blank_sse = True # type: ignore[attr-defined] _Stream._iter_events = _filtered_sync_events _openai_empty_sse_keepalive_patched = True except Exception: # The patch is only needed when the optional OpenAI SDK is available. pass _patch_openai_empty_sse_keepalive() _deepagents_extracted_document_text_patched = False def _patch_deepagents_extracted_document_text() -> None: global _deepagents_extracted_document_text_patched if _deepagents_extracted_document_text_patched: return try: from pathlib import Path as _Path import deepagents.middleware.filesystem as _filesystem from EvoScientist.document_extract import DOCUMENT_EXTENSIONS namespace = cast(dict[str, Any], vars(_filesystem)) original = namespace["_get_file_type"] if getattr(original, "_evoscientist_document_text", False): _deepagents_extracted_document_text_patched = True return def _document_text_type(path: str) -> str: if _Path(path).suffix.lower() in DOCUMENT_EXTENSIONS: return "text" return original(path) _document_text_type._evoscientist_document_text = True # type: ignore[attr-defined] namespace["_get_file_type"] = _document_text_type _deepagents_extracted_document_text_patched = True except Exception: pass _patch_deepagents_extracted_document_text()