Dify模型反代插件

import json

import time

from collections.abc import Generator, Mapping

from typing import Any, cast



from dify_plugin import Endpoint

from dify_plugin.core.entities.invocation import InvokeType

from dify_plugin.entities.model.llm import LLMResult, LLMResultChunk, LLMUsage

from dify_plugin.entities.model.message import AssistantPromptMessage

from werkzeug import Request, Response



from endpoints.auth import BaseAuth

from endpoints.openai_protocol import (

    ProtocolError,

    build_chat_completion_response,

    build_host_llm_invoke_payload,

    iter_sse_chat_completion,

    merge_tool_calls,

    new_completion_id,

    normalize_stop,

    openai_error,

    parse_messages,

    parse_tools,

    prepare_llm_config,

    sanitize_host_llm_payload,

    PLUGIN_VERSION,

    PLUGIN_RUNTIME_FINGERPRINT,

)

from endpoints.openai_protocol import _normalize_stream_content



def _is_unittest_mock(obj: object) -> bool:

    module = type(obj).__module__

    return module == "unittest.mock" or module.startswith("unittest.mock")



class OaicompatDifyModelEndpoint(Endpoint, BaseAuth):

    def _runtime_headers(self) -> dict[str, str]:

        return {

            "X-Oaicompat-Plugin-Version": PLUGIN_VERSION,

            "X-Oaicompat-Build": PLUGIN_RUNTIME_FINGERPRINT,

            "X-Oaicompat-Invoke": "host-safe-backwards",

        }



    def _error(self, status: int, message: str, type_: str) -> Response:

        return Response(

            json.dumps(openai_error(message, type_=type_)),

            status=status,

            content_type="application/json",

            headers=self._runtime_headers(),

        )



    def _invoke(self, r: Request, values: Mapping, settings: Mapping) -> Response:

        if not self.verify(r, settings):

            return self._error(401, "Unauthorized", "invalid_request_error")



        try:

            return self._invoke_validated(r, settings)

        except ProtocolError as exc:

            return self._error(400, str(exc), "invalid_request_error")

        except ValueError as exc:

            return self._error(400, str(exc), "invalid_request_error")

        except Exception as exc:  # noqa: BLE001 - map provider failures to OpenAI-ish 500

            # Werkzeug BadRequest (invalid JSON) is not a ValueError subclass.

            status = getattr(exc, "code", None)

            if status == 400 or type(exc).__name__ == "BadRequest":

                return self._error(

                    400,

                    str(exc).strip() or "invalid request body",

                    "invalid_request_error",

                )

            message = str(exc).strip() or type(exc).__name__

            # Surface host validation fingerprint clearly for Codex/stream debugging.

            if "ModelInvokeLLMRequest" in message or "model_parameters" in message:

                message = (

                    f"{message} | plugin={PLUGIN_VERSION}/{PLUGIN_RUNTIME_FINGERPRINT} "

                    "host-safe payload should never omit model_parameters or roles; "

                    "if you see model_parameters=None, capture request with "

                    "last user content __oaicompat_debug_payload__"

                )

            return self._error(500, message, "server_error")



    def _build_host_payload(

        self,

        *,

        llm_cfg: dict[str, Any],

        prompt_messages: list,

        tools: list | None,

        stop: list[str] | None,

        stream: bool,

    ) -> dict[str, Any]:

        payload = build_host_llm_invoke_payload(

            llm_cfg=llm_cfg,

            prompt_messages=prompt_messages,

            tools=tools,

            stop=stop,

            stream=stream,

        )

        payload = sanitize_host_llm_payload(payload)

        # Hard asserts: these are exactly the host failure fields.

        if not isinstance(payload.get("model_parameters"), dict):

            raise RuntimeError(

                "host payload missing model_parameters dict: "

                f"{type(payload.get('model_parameters')).__name__}"

            )

        roles = [m.get("role") for m in payload.get("prompt_messages", [])]

        if not roles or any(not isinstance(r, str) or not r for r in roles):

            raise RuntimeError(f"host payload has invalid roles: {roles!r}")

        return payload



    def _invoke_llm(

        self,

        *,

        llm_cfg: dict[str, Any],

        prompt_messages: list,

        tools: list | None,

        stop: list[str] | None,

        stream: bool,

    ) -> Generator[LLMResultChunk, None, None] | LLMResult:

        """Invoke host LLM with a host-safe reverse-invoke payload.



        NEVER use session.model.llm.invoke() in production: SDK rebuilds

        LLMModelConfig and drops model_parameters, and re-dumps messages with

        Enum roles. Host ModelInvokeLLMRequest then fails.

        """

        payload = self._build_host_payload(

            llm_cfg=llm_cfg,

            prompt_messages=prompt_messages,

            tools=tools,

            stop=stop,

            stream=stream,

        )



        invoker = getattr(

            getattr(getattr(self, "session", None), "model", None), "llm", None

        )

        if invoker is None:

            raise RuntimeError("LLM invoker is not available")



        # Production MUST use host-safe _backwards_invoke with prebuilt payload.

        # Never call SDK invoke(): it rebuilds LLMModelConfig and drops model_parameters.

        if _is_unittest_mock(invoker) and callable(getattr(invoker, "invoke", None)):

            return invoker.invoke(

                model_config=payload,

                prompt_messages=prompt_messages,

                tools=tools,

                stop=stop,

                stream=stream,

            )



        backwards = getattr(invoker, "_backwards_invoke", None)

        if not callable(backwards):

            raise RuntimeError(

                "LLM invoker missing _backwards_invoke; refusing SDK invoke() fallback"

            )



        # Copy payload defensively so nothing can mutate after validation.

        payload_sent = json.loads(json.dumps(payload, ensure_ascii=False))



        chunks = invoker._backwards_invoke(  # noqa: SLF001 - host-safe path

            InvokeType.LLM,

            LLMResultChunk,

            payload_sent,

        )



        if stream:

            # Eagerly open the reverse-invoke so host ValidationError becomes a

            # normal HTTP error BEFORE we start SSE. Otherwise Codex sees

            # "stream disconnected" after 200 headers.

            chunk_iter = iter(cast(Generator[LLMResultChunk, None, None], chunks))



            def gen() -> Generator[LLMResultChunk, None, None]:

                try:

                    first = next(chunk_iter)

                except StopIteration:

                    return

                yield first

                yield from chunk_iter



            # Force first next() now (may block until host accepts request).

            primed = gen()

            try:

                first_chunk = next(primed)

            except StopIteration:

                def empty() -> Generator[LLMResultChunk, None, None]:

                    if False:

                        yield None  # type: ignore[misc]

                    return

                    yield  # pragma: no cover



                return empty()



            def rest() -> Generator[LLMResultChunk, None, None]:

                yield first_chunk

                yield from primed



            return rest()



        result = LLMResult(

            model=str(llm_cfg.get("model") or ""),

            message=AssistantPromptMessage(content=""),

            usage=LLMUsage.empty_usage(),

        )

        result.message.content = cast("str", result.message.content or "")

        for llm_result in chunks:

            content = llm_result.delta.message.content

            text = _normalize_stream_content(content)

            if text:

                result.message.content = cast("str", result.message.content) + text

            if llm_result.delta.message.tool_calls:

                result.message.tool_calls = merge_tool_calls(

                    result.message.tool_calls,

                    llm_result.delta.message.tool_calls,

                )

            if llm_result.delta.usage:

                usage = llm_result.delta.usage

                result.usage.prompt_tokens += usage.prompt_tokens

                result.usage.completion_tokens += usage.completion_tokens

                result.usage.total_tokens += usage.total_tokens

                result.usage.completion_price = usage.completion_price

                result.usage.prompt_price = usage.prompt_price

                result.usage.total_price = usage.total_price

                result.usage.currency = usage.currency

                result.usage.latency = usage.latency

        return result



    def _invoke_validated(self, r: Request, settings: Mapping) -> Response:

        settings_llm = settings.get("llm")

        if not settings_llm:

            raise ProtocolError("LLM is not set")

        if not isinstance(settings_llm, dict):

            raise ProtocolError("LLM config is invalid")



        try:

            data = r.get_json(force=True)

        except Exception as exc:  # noqa: BLE001 - werkzeug BadRequest on bad JSON

            raise ProtocolError(f"invalid JSON body: {exc}") from exc

        if not data or not isinstance(data, dict):

            raise ProtocolError("Request body is empty")



        # Runtime fingerprint / debug helpers — no reverse-invoke.

        messages = data.get("messages")

        if isinstance(messages, list) and messages:

            first = messages[0] if isinstance(messages[0], dict) else {}

            first_content = first.get("content") if isinstance(first, dict) else None

            # Allow debug marker as last user message so Codex-like histories work.

            last_user = None

            for m in reversed(messages):

                if isinstance(m, dict) and m.get("role") == "user":

                    last_user = m

                    break

            last_content = last_user.get("content") if isinstance(last_user, dict) else None



            if first_content == "__oaicompat_ping__" and len(messages) == 1:

                body = {

                    "id": new_completion_id(),

                    "object": "chat.completion",

                    "created": int(time.time()),

                    "model": (settings_llm or {}).get("model") or "oaicompat",

                    "system_fingerprint": (

                        f"oaicompat-{PLUGIN_VERSION}-{PLUGIN_RUNTIME_FINGERPRINT}"

                    ),

                    "choices": [

                        {

                            "index": 0,

                            "message": {

                                "role": "assistant",

                                "content": (

                                    f"oaicompat_dify_model {PLUGIN_VERSION} "

                                    f"build={PLUGIN_RUNTIME_FINGERPRINT} "

                                    "invoke=host-safe-backwards"

                                ),

                            },

                            "finish_reason": "stop",

                        }

                    ],

                    "usage": {

                        "prompt_tokens": 0,

                        "completion_tokens": 0,

                        "total_tokens": 0,

                    },

                }

                return Response(

                    json.dumps(body),

                    status=200,

                    content_type="application/json",

                    headers=self._runtime_headers(),

                )



            if last_content == "__oaicompat_debug_payload__":

                # Build the exact host reverse-invoke payload and return it.

                # Use messages without the debug marker.

                debug_messages = [

                    m

                    for m in messages

                    if not (

                        isinstance(m, dict)

                        and m.get("role") == "user"

                        and m.get("content") == "__oaicompat_debug_payload__"

                    )

                ]

                if not debug_messages:

                    debug_messages = [{"role": "user", "content": "debug"}]

                debug_body = dict(data)

                debug_body["messages"] = debug_messages

                llm_cfg = prepare_llm_config(settings_llm, debug_body)

                prompt_messages = parse_messages(debug_messages)

                tools = parse_tools(debug_body.get("tools"))

                stop = normalize_stop(debug_body.get("stop"))

                stream = bool(debug_body.get("stream", False))

                payload = self._build_host_payload(

                    llm_cfg=llm_cfg,

                    prompt_messages=prompt_messages,

                    tools=tools or None,

                    stop=stop,

                    stream=stream,

                )

                summary = {

                    "plugin_version": PLUGIN_VERSION,

                    "build": PLUGIN_RUNTIME_FINGERPRINT,

                    "stream": stream,

                    "model_parameters_type": type(payload.get("model_parameters")).__name__,

                    "model_parameters": payload.get("model_parameters"),

                    "roles": [m.get("role") for m in payload.get("prompt_messages", [])],

                    "message_count": len(payload.get("prompt_messages", [])),

                    "tools_count": len(payload["tools"]) if payload.get("tools") else 0,

                    "payload": payload,

                }

                return Response(

                    json.dumps(summary, ensure_ascii=False),

                    status=200,

                    content_type="application/json",

                    headers=self._runtime_headers(),

                )



        llm_cfg = prepare_llm_config(settings_llm, data)

        prompt_messages = parse_messages(data.get("messages"))

        tools = parse_tools(data.get("tools"))

        stop = normalize_stop(data.get("stop"))

        stream = bool(data.get("stream", False))



        completion_id = new_completion_id()

        created = int(time.time())

        model_name = llm_cfg.get("model")



        if not stream:

            result = self._invoke_llm(

                llm_cfg=llm_cfg,

                prompt_messages=prompt_messages,

                tools=tools or None,

                stop=stop,

                stream=False,

            )

            message = getattr(result, "message", None)

            body = build_chat_completion_response(

                model=model_name,

                message_content=getattr(message, "content", None),

                tool_calls=getattr(message, "tool_calls", None),

                usage=getattr(result, "usage", None),

                completion_id=completion_id,

                created=created,

            )

            return Response(

                json.dumps(body),

                status=200,

                content_type="application/json",

                headers=self._runtime_headers(),

            )



        chunk_iter = self._invoke_llm(

            llm_cfg=llm_cfg,

            prompt_messages=prompt_messages,

            tools=tools or None,

            stop=stop,

            stream=True,

        )



        def generator():

            yield from iter_sse_chat_completion(

                model=model_name,

                chunk_iter=chunk_iter,

                completion_id=completion_id,

                created=created,

            )



        headers = {

            "Cache-Control": "no-cache",

            "Connection": "keep-alive",

            "X-Accel-Buffering": "no",

        }

        headers.update(self._runtime_headers())

        return Response(

            generator(),

            status=200,

            content_type="text/event-stream",

            headers=headers,

        )

from __future__ import annotations



import copy

import json

import time

import uuid

from collections.abc import Iterable, Iterator

from typing import Any



from dify_plugin.entities.model.message import (

    AssistantPromptMessage,

    PromptMessage,

    PromptMessageTool,

    SystemPromptMessage,

    ToolPromptMessage,

    UserPromptMessage,

)



PLUGIN_VERSION = "0.0.20"

PLUGIN_RUNTIME_FINGERPRINT = "host-safe-v4-nonnull"



COMPLETION_PARAM_KEYS = frozenset(

    {

        "temperature",

        "top_p",

        "max_tokens",

        "presence_penalty",

        "frequency_penalty",

    }

)



class ProtocolError(ValueError):

    """Raised for invalid OpenAI-compatible request payloads."""



def openai_error(

    message: str,

    type_: str = "invalid_request_error",

    *,

    param: str | None = None,

    code: str | None = None,

) -> dict[str, Any]:

    return {

        "error": {

            "message": message,

            "type": type_,

            "param": param,

            "code": code,

        }

    }



def normalize_content(content: Any, *, allow_empty: bool = True) -> str | None:

    if content is None:

        return None if allow_empty else ""

    if isinstance(content, str):

        return content

    if isinstance(content, list):

        parts: list[str] = []

        for item in content:

            if not isinstance(item, dict):

                raise ProtocolError("message content list items must be objects")

            item_type = item.get("type")

            if item_type in (None, "text"):

                text = item.get("text")

                if text is None:

                    text = item.get("data")

                if text is None:

                    continue

                parts.append(str(text))

                continue

            if item_type in ("image", "image_url"):

                raise ProtocolError("image content is not supported")

            raise ProtocolError(f"unsupported content type: {item_type}")

        return "".join(parts)

    raise ProtocolError("message content must be string, list, or null")



def _normalize_arguments(arguments: Any) -> str:

    if arguments is None:

        return ""

    if isinstance(arguments, str):

        return arguments

    if isinstance(arguments, (dict, list)):

        return json.dumps(arguments, ensure_ascii=False)

    return str(arguments)



def parse_tools(tools: list[dict[str, Any]] | None) -> list[PromptMessageTool]:

    if not tools:

        return []

    parsed: list[PromptMessageTool] = []

    for tool in tools:

        if not isinstance(tool, dict):

            raise ProtocolError("each tool must be an object")

        if tool.get("type") == "function" and isinstance(tool.get("function"), dict):

            fn = tool["function"]

            name = fn.get("name")

            if not name:

                raise ProtocolError("tool function.name is required")

            parsed.append(

                PromptMessageTool(

                    name=name,

                    description=fn.get("description") or "",

                    parameters=fn.get("parameters") or {},

                )

            )

            continue

        if "name" in tool:

            parsed.append(

                PromptMessageTool(

                    name=tool["name"],

                    description=tool.get("description") or "",

                    parameters=tool.get("parameters") or {},

                )

            )

            continue

        raise ProtocolError("invalid tool definition")

    return parsed



def _parse_assistant_tool_calls(

    raw_tool_calls: Any,

) -> list[AssistantPromptMessage.ToolCall]:

    if not raw_tool_calls:

        return []

    if not isinstance(raw_tool_calls, list):

        raise ProtocolError("assistant tool_calls must be a list")

    tool_calls: list[AssistantPromptMessage.ToolCall] = []

    for index, item in enumerate(raw_tool_calls):

        if not isinstance(item, dict):

            raise ProtocolError("tool_call must be an object")

        function = item.get("function") or {}

        if not isinstance(function, dict):

            raise ProtocolError("tool_call.function must be an object")

        name = function.get("name")

        if not name:

            raise ProtocolError("tool_call.function.name is required")

        call_id = item.get("id")

        if not call_id:

            call_id = f"call_{index}"

        tool_calls.append(

            AssistantPromptMessage.ToolCall(

                id=str(call_id),

                type=item.get("type") or "function",

                function=AssistantPromptMessage.ToolCall.ToolCallFunction(

                    name=str(name),

                    arguments=_normalize_arguments(function.get("arguments")),

                ),

            )

        )

    return tool_calls



def parse_messages(messages: list[dict[str, Any]] | None) -> list[PromptMessage]:

    if not isinstance(messages, list) or not messages:

        raise ProtocolError("messages must be a non-empty list")

    prompt_messages: list[PromptMessage] = []

    for message in messages:

        if message is None:

            continue

        if not isinstance(message, dict):

            raise ProtocolError("each message must be an object")

        raw_role = message.get("role")

        if not isinstance(raw_role, str) or not raw_role.strip():

            raise ProtocolError(

                "message role is required and must be one of: "

                "system, user, assistant, tool, developer"

            )

        role = raw_role.strip().lower()

        # OpenAI o-series / Codex use developer; map to system for Dify host.

        if role == "developer":

            role = "system"

        if role == "user":

            prompt_messages.append(

                UserPromptMessage(content=normalize_content(message.get("content")) or "")

            )

        elif role == "system":

            prompt_messages.append(

                SystemPromptMessage(

                    content=normalize_content(message.get("content")) or ""

                )

            )

        elif role == "assistant":

            tool_calls = _parse_assistant_tool_calls(message.get("tool_calls"))

            content = normalize_content(message.get("content"))

            if tool_calls and (content is None or content == ""):

                content = None

            prompt_messages.append(

                AssistantPromptMessage(content=content, tool_calls=tool_calls)

            )

        elif role == "tool":

            tool_call_id = message.get("tool_call_id")

            if not tool_call_id:

                raise ProtocolError("tool message requires tool_call_id")

            # ToolPromptMessage.content is str; map null/missing to "".

            tool_content = normalize_content(message.get("content"))

            kwargs: dict[str, Any] = {

                "content": tool_content if tool_content is not None else "",

                "tool_call_id": str(tool_call_id),

            }

            if message.get("name"):

                kwargs["name"] = str(message["name"])

            prompt_messages.append(ToolPromptMessage(**kwargs))

        else:

            raise ProtocolError(f"invalid message role: {raw_role}")

    if not prompt_messages:

        raise ProtocolError("messages must be a non-empty list")

    return prompt_messages



def normalize_stop(stop: Any) -> list[str] | None:

    if stop is None:

        return None

    if isinstance(stop, str):

        return [stop]

    if isinstance(stop, list):

        if not all(isinstance(item, str) for item in stop):

            raise ProtocolError("stop list items must be strings")

        return list(stop)

    raise ProtocolError("stop must be string or list of strings")



def merge_completion_params(

    base: dict[str, Any] | None, request: dict[str, Any]

) -> dict[str, Any]:

    merged = dict(base or {})

    for key in COMPLETION_PARAM_KEYS:

        if key in request and request[key] is not None:

            merged[key] = request[key]

    # Request max_tokens wins; else request max_completion_tokens overrides base.

    if request.get("max_tokens") is not None:

        merged["max_tokens"] = request["max_tokens"]

    elif request.get("max_completion_tokens") is not None:

        merged["max_tokens"] = request["max_completion_tokens"]

    return merged



def prepare_llm_config(

    settings_llm: dict[str, Any], request: dict[str, Any]

) -> dict[str, Any]:

    llm_cfg = copy.deepcopy(settings_llm)

    if "completion_params" not in llm_cfg or llm_cfg["completion_params"] is None:

        llm_cfg["completion_params"] = {}

    params = merge_completion_params(llm_cfg.get("completion_params") or {}, request)

    if not isinstance(params, dict):

        params = {}

    llm_cfg["completion_params"] = params

    # Host ModelInvokeLLMRequest requires model_parameters: dict.

    # Keep both keys so either mapping path works.

    llm_cfg["model_parameters"] = params

    return llm_cfg



def merge_tool_calls(existing: Any, incoming: Any) -> list[Any]:

    """Merge tool_calls by id (preferred) or index; last non-empty fields win."""

    if not incoming:

        return list(existing or [])

    if not existing:

        return list(incoming)



    merged: list[Any] = list(existing)

    by_id: dict[str, int] = {}

    for i, tc in enumerate(merged):

        tc_id = getattr(tc, "id", None)

        if tc_id is None and isinstance(tc, dict):

            tc_id = tc.get("id")

        if tc_id:

            by_id[str(tc_id)] = i



    for index, tc in enumerate(incoming):

        tc_id = getattr(tc, "id", None)

        if tc_id is None and isinstance(tc, dict):

            tc_id = tc.get("id")

        if tc_id and str(tc_id) in by_id:

            merged[by_id[str(tc_id)]] = tc

            continue

        if index < len(merged) and not tc_id:

            # index-aligned partial update without id

            merged[index] = tc

            continue

        merged.append(tc)

        if tc_id:

            by_id[str(tc_id)] = len(merged) - 1

    return merged



def tool_calls_to_openai(tool_calls: Any) -> list[dict[str, Any]]:

    if not tool_calls:

        return []

    result: list[dict[str, Any]] = []

    for index, tool_call in enumerate(tool_calls):

        function = getattr(tool_call, "function", None)

        name = getattr(function, "name", "") if function is not None else ""

        arguments = getattr(function, "arguments", "") if function is not None else ""

        result.append(

            {

                "index": index,

                "id": getattr(tool_call, "id", "") or f"call_{index}",

                "type": getattr(tool_call, "type", None) or "function",

                "function": {

                    "name": name or "",

                    "arguments": _normalize_arguments(arguments),

                },

            }

        )

    return result



def usage_to_openai(usage: Any) -> dict[str, int]:

    if usage is None:

        return {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}

    return {

        "prompt_tokens": int(getattr(usage, "prompt_tokens", 0) or 0),

        "completion_tokens": int(getattr(usage, "completion_tokens", 0) or 0),

        "total_tokens": int(getattr(usage, "total_tokens", 0) or 0),

    }



def infer_finish_reason(tool_calls: Any) -> str:

    return "tool_calls" if tool_calls else "stop"



def new_completion_id() -> str:

    return "chatcmpl-" + str(uuid.uuid4())



def build_chat_completion_response(

    *,

    model: str | None,

    message_content: Any,

    tool_calls: Any,

    usage: Any,

    completion_id: str,

    created: int | None = None,

) -> dict[str, Any]:

    openai_tool_calls = tool_calls_to_openai(tool_calls)

    finish_reason = infer_finish_reason(openai_tool_calls)

    # Normalize to OpenAI text (or null). Never Python-repr objects/lists.

    if isinstance(message_content, str):

        content: Any = message_content

    elif message_content is None:

        content = None

    else:

        content = _normalize_stream_content(message_content)

    if openai_tool_calls and (content is None or content == ""):

        content = None

    message: dict[str, Any] = {

        "role": "assistant",

        "content": content,

    }

    if openai_tool_calls:

        message["tool_calls"] = [

            {

                "id": item["id"],

                "type": item["type"],

                "function": item["function"],

            }

            for item in openai_tool_calls

        ]

    return {

        "id": completion_id,

        "object": "chat.completion",

        "created": int(created if created is not None else time.time()),

        "model": model,

        "system_fingerprint": f"oaicompat-{PLUGIN_VERSION}-{PLUGIN_RUNTIME_FINGERPRINT}",

        "choices": [

            {

                "index": 0,

                "message": message,

                "finish_reason": finish_reason,

            }

        ],

        "usage": usage_to_openai(usage),

    }



def _sse_line(payload: dict[str, Any]) -> str:

    return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n"



def _chunk_payload(

    *,

    completion_id: str,

    created: int,

    model: str | None,

    delta: dict[str, Any],

    finish_reason: str | None = None,

    usage: dict[str, int] | None = None,

) -> dict[str, Any]:

    body: dict[str, Any] = {

        "id": completion_id,

        "object": "chat.completion.chunk",

        "created": created,

        "model": model,

        "system_fingerprint": f"oaicompat-{PLUGIN_VERSION}-{PLUGIN_RUNTIME_FINGERPRINT}",

        "choices": [

            {

                "index": 0,

                "delta": delta,

                "finish_reason": finish_reason,

            }

        ],

    }

    if usage is not None:

        body["usage"] = usage

    return body



def _normalize_stream_content(content: Any) -> str | None:

    if content is None:

        return None

    if isinstance(content, str):

        return content or None

    if isinstance(content, list):

        parts: list[str] = []

        for item in content:

            if isinstance(item, dict):

                item_type = item.get("type")

                if item_type in (None, "text"):

                    text = item.get("text")

                    if text is None:

                        text = item.get("data")

                    if text is not None:

                        parts.append(str(text))

                continue

            data = getattr(item, "data", None)

            if data is not None:

                parts.append(str(data))

                continue

            text = getattr(item, "text", None)

            if text is not None:

                parts.append(str(text))

        joined = "".join(parts)

        return joined or None

    return str(content) or None



def iter_sse_chat_completion(

    *,

    model: str | None,

    chunk_iter: Iterable[Any],

    completion_id: str,

    created: int | None = None,

) -> Iterator[str]:

    """Emit OpenAI-compatible SSE frames.



    Always ends with a finish_reason frame + `data: [DONE]`, even if the

    upstream chunk iterator raises. Clients like Codex abort when a stream

    ends without finish_reason.

    """

    created_ts = int(created if created is not None else time.time())

    yield _sse_line(

        _chunk_payload(

            completion_id=completion_id,

            created=created_ts,

            model=model,

            delta={"role": "assistant", "content": ""},

            finish_reason=None,

        )

    )



    last_tool_calls: Any = None

    last_finish: str | None = None

    last_usage: Any = None

    stream_error: Exception | None = None

    finish = "stop"

    openai_tool_calls: list[dict[str, Any]] = []



    try:

        for chunk in chunk_iter:

            delta = getattr(chunk, "delta", None)

            if delta is None:

                continue

            message = getattr(delta, "message", None)

            content = (

                getattr(message, "content", None) if message is not None else None

            )

            text = _normalize_stream_content(content)

            if text:

                yield _sse_line(

                    _chunk_payload(

                        completion_id=completion_id,

                        created=created_ts,

                        model=model,

                        delta={"content": text},

                        finish_reason=None,

                    )

                )

            tool_calls = (

                getattr(message, "tool_calls", None) if message is not None else None

            )

            if tool_calls:

                last_tool_calls = merge_tool_calls(last_tool_calls, tool_calls)

            finish_reason = getattr(delta, "finish_reason", None)

            if finish_reason:

                last_finish = finish_reason

            usage = getattr(delta, "usage", None)

            if usage is not None:

                last_usage = usage



        try:

            openai_tool_calls = tool_calls_to_openai(last_tool_calls)

        except Exception as exc:  # noqa: BLE001

            stream_error = exc

            openai_tool_calls = []



        if openai_tool_calls:

            yield _sse_line(

                _chunk_payload(

                    completion_id=completion_id,

                    created=created_ts,

                    model=model,

                    delta={"tool_calls": openai_tool_calls},

                    finish_reason=None,

                )

            )

            finish = "tool_calls"

        elif last_finish in {"stop", "length", "content_filter", "tool_calls"}:

            finish = last_finish

        else:

            finish = "stop"

    except Exception as exc:  # noqa: BLE001 - must still close SSE for clients

        stream_error = exc

        # Prefer tool_calls finish if we already observed tools before the error.

        finish = "tool_calls" if last_tool_calls else "stop"



    # If we collected tool_calls but the success path did not emit them (exception

    # mid-loop), emit one final tool_calls frame before finish.

    if last_tool_calls and not openai_tool_calls:

        try:

            openai_tool_calls = tool_calls_to_openai(last_tool_calls)

        except Exception as exc:  # noqa: BLE001

            stream_error = stream_error or exc

            openai_tool_calls = []

        if openai_tool_calls:

            yield _sse_line(

                _chunk_payload(

                    completion_id=completion_id,

                    created=created_ts,

                    model=model,

                    delta={"tool_calls": openai_tool_calls},

                    finish_reason=None,

                )

            )

            finish = "tool_calls"



    finish_payload = _chunk_payload(

        completion_id=completion_id,

        created=created_ts,

        model=model,

        delta={},

        finish_reason=finish,

        usage=usage_to_openai(last_usage),

    )

    if stream_error is not None:

        # Keep OpenAI shape; surface upstream failure without dropping finish_reason.

        finish_payload["error"] = openai_error(

            str(stream_error) or "upstream stream error",

            type_="server_error",

        )["error"]

    yield _sse_line(finish_payload)

    yield "data: [DONE]\n\n"



def _role_value(role: Any) -> str:

    if role is None:

        return ""

    value = getattr(role, "value", role)

    if not isinstance(value, str):

        value = str(value)

    return value.strip().lower()



def serialize_prompt_message(message: PromptMessage) -> dict[str, Any]:

    """Serialize PromptMessage for host ModelInvokeLLMRequest.



    Always emit plain JSON-safe values. Never leave Enum objects or empty roles.

    """

    role = _role_value(getattr(message, "role", None))

    if role not in {"system", "user", "assistant", "tool", "developer"}:

        raise ProtocolError(

            f"invalid message role after parse: {getattr(message, 'role', None)!r}"

        )



    # Dify host (Go daemon) rejects null content; use "" so assistant messages

    # with tool_calls still parse. See sanitize_host_llm_payload.

    content = getattr(message, "content", None)

    if content is None:

        content = ""

    elif not isinstance(content, (str, list)):

        content = str(content)



    payload: dict[str, Any] = {

        "role": role,

        "content": content,

    }

    name = getattr(message, "name", None)

    if name:

        payload["name"] = str(name)



    if role == "assistant":

        # Host RequestInvokeLLM requires tool_calls to be a list, never null.

        raw_tool_calls = getattr(message, "tool_calls", None) or []

        serialized_calls: list[dict[str, Any]] = []

        for index, tool_call in enumerate(raw_tool_calls):

            function = getattr(tool_call, "function", None)

            serialized_calls.append(

                {

                    "id": str(getattr(tool_call, "id", "") or f"call_{index}"),

                    "type": str(getattr(tool_call, "type", None) or "function"),

                    "function": {

                        "name": str(getattr(function, "name", "") or ""),

                        "arguments": _normalize_arguments(

                            getattr(function, "arguments", "") if function else ""

                        ),

                    },

                }

            )

        payload["tool_calls"] = serialized_calls

    elif role == "tool":

        tool_call_id = getattr(message, "tool_call_id", None)

        if not tool_call_id:

            raise ProtocolError("tool message requires tool_call_id")

        payload["tool_call_id"] = str(tool_call_id)



    return payload



def serialize_tool(tool: PromptMessageTool | dict[str, Any]) -> dict[str, Any]:

    if isinstance(tool, dict):

        name = tool.get("name")

        if not name:

            raise ProtocolError("tool.name is required")

        return {

            "name": str(name),

            "description": str(tool.get("description") or ""),

            "parameters": tool.get("parameters") or {},

        }

    return {

        "name": str(tool.name),

        "description": str(tool.description or ""),

        "parameters": tool.parameters or {},

    }



def build_host_llm_invoke_payload(

    *,

    llm_cfg: dict[str, Any],

    prompt_messages: list[PromptMessage],

    tools: list[PromptMessageTool] | None,

    stop: list[str] | None,

    stream: bool,

) -> dict[str, Any]:

    """Build reverse-invoke payload that passes host ModelInvokeLLMRequest.



    Why this exists:

    - SDK LLMInvocation does `**model_config.model_dump()` + message.model_dump().

    - Host validates `model_parameters: dict` (not completion_params).

    - Host rejects empty role strings.

    - Enum objects in nested dumps can serialize poorly depending on path.



    This builder emits only JSON-safe primitives with dual-written params.

    """

    if not isinstance(llm_cfg, dict):

        raise ProtocolError("LLM config is invalid")



    provider = llm_cfg.get("provider")

    model = llm_cfg.get("model")

    mode = llm_cfg.get("mode") or "chat"

    if not provider or not model:

        raise ProtocolError("LLM provider/model is required")



    params = llm_cfg.get("model_parameters")

    if not isinstance(params, dict):

        params = llm_cfg.get("completion_params")

    if not isinstance(params, dict):

        params = {}



    model_type = llm_cfg.get("model_type") or "llm"

    if hasattr(model_type, "value"):

        model_type = model_type.value

    model_type = str(model_type)



    serialized_messages = [serialize_prompt_message(m) for m in prompt_messages]

    if not serialized_messages:

        raise ProtocolError("messages must be a non-empty list")

    for item in serialized_messages:

        if not item.get("role"):

            raise ProtocolError("serialized message role is empty")



    serialized_tools = None

    if tools:

        serialized_tools = [serialize_tool(t) for t in tools]



    # Dual-write both keys. Host currently requires model_parameters.

    payload = {

        "provider": str(provider),

        "model": str(model),

        "model_type": model_type,

        "mode": str(mode),

        "completion_params": params,

        "model_parameters": params,

        "prompt_messages": serialized_messages,

        "tools": serialized_tools,

        "stop": stop,

        "stream": bool(stream),

    }

    return sanitize_host_llm_payload(payload)



def sanitize_host_llm_payload(payload: dict[str, Any]) -> dict[str, Any]:

    """Final hard sanitization before reverse-invoke.



    Guarantees host ModelInvokeLLMRequest / RequestInvokeLLM constraints:

    - model_parameters is a dict (never None)

    - completion_params is a dict

    - every prompt_messages[].role is a non-empty allowed string

    - every assistant message has tool_calls as a list (never null)

    - tools is list or None (not other types)

    - stop is list[str] or None

    """

    if not isinstance(payload, dict):

        raise ProtocolError("invoke payload must be a dict")



    out = dict(payload)



    params = out.get("model_parameters")

    if not isinstance(params, dict):

        params = out.get("completion_params")

    if not isinstance(params, dict):

        params = {}

    out["model_parameters"] = params

    out["completion_params"] = params



    provider = out.get("provider")

    model = out.get("model")

    if not provider or not model:

        raise ProtocolError("LLM provider/model is required")

    out["provider"] = str(provider)

    out["model"] = str(model)

    mode = out.get("mode") or "chat"

    out["mode"] = str(mode)

    model_type = out.get("model_type") or "llm"

    if hasattr(model_type, "value"):

        model_type = model_type.value

    out["model_type"] = str(model_type)



    messages = out.get("prompt_messages")

    if not isinstance(messages, list) or not messages:

        raise ProtocolError("messages must be a non-empty list")



    allowed = {"system", "user", "assistant", "tool", "developer"}

    cleaned_messages: list[dict[str, Any]] = []

    for item in messages:

        if not isinstance(item, dict):

            raise ProtocolError("each prompt message must be an object")

        role_raw = item.get("role")

        if hasattr(role_raw, "value"):

            role_raw = role_raw.value

        if not isinstance(role_raw, str) or not role_raw.strip():

            raise ProtocolError(

                "message role is required and must be one of: system, user, assistant, tool"

            )

        role = role_raw.strip().lower()

        if role not in allowed:

            raise ProtocolError(f"invalid message role: {role_raw}")



        # Dify host (Go daemon) hard-rejects null content: its PromptMessage

        # JSON decoder returns "content field is required" for null, which fails

        # the whole reverse-invoke parse and degrades every field (role='',

        # model_parameters=None) at the model provider. Native Dify uses "".

        raw_content = item.get("content")

        if raw_content is None:

            raw_content = ""

        elif not isinstance(raw_content, (str, list)):

            raw_content = str(raw_content)

        msg: dict[str, Any] = {

            "role": role,

            "content": raw_content,

        }

        name = item.get("name")

        if name:

            msg["name"] = str(name)



        if role == "assistant":

            tool_calls = item.get("tool_calls")

            if tool_calls is None:

                tool_calls = []

            if not isinstance(tool_calls, list):

                raise ProtocolError("assistant tool_calls must be a list")

            fixed_calls: list[dict[str, Any]] = []

            for index, call in enumerate(tool_calls):

                if not isinstance(call, dict):

                    raise ProtocolError("tool_call must be an object")

                function = call.get("function") or {}

                if not isinstance(function, dict):

                    raise ProtocolError("tool_call.function must be an object")

                fixed_calls.append(

                    {

                        "id": str(call.get("id") or f"call_{index}"),

                        "type": str(call.get("type") or "function"),

                        "function": {

                            "name": str(function.get("name") or ""),

                            "arguments": _normalize_arguments(function.get("arguments")),

                        },

                    }

                )

            msg["tool_calls"] = fixed_calls

        elif role == "tool":

            tool_call_id = item.get("tool_call_id")

            if not tool_call_id:

                raise ProtocolError("tool message requires tool_call_id")

            msg["tool_call_id"] = str(tool_call_id)



        cleaned_messages.append(msg)



    out["prompt_messages"] = cleaned_messages



    tools = out.get("tools")

    if tools is None:

        out["tools"] = None

    elif isinstance(tools, list):

        cleaned_tools: list[dict[str, Any]] = []

        for tool in tools:

            if not isinstance(tool, dict):

                raise ProtocolError("each tool must be an object")

            name = tool.get("name")

            if not name:

                raise ProtocolError("tool.name is required")

            cleaned_tools.append(

                {

                    "name": str(name),

                    "description": str(tool.get("description") or ""),

                    "parameters": tool.get("parameters")

                    if isinstance(tool.get("parameters"), dict)

                    else {},

                }

            )

        out["tools"] = cleaned_tools

    else:

        raise ProtocolError("tools must be a list or null")



    stop = out.get("stop")

    if stop is None:

        out["stop"] = None

    elif isinstance(stop, list) and all(isinstance(s, str) for s in stop):

        out["stop"] = list(stop)

    else:

        raise ProtocolError("stop must be string list or null")



    out["stream"] = bool(out.get("stream", False))

    return out

import json

from collections.abc import Mapping

from typing import Optional



from dify_plugin import Endpoint

from dify_plugin.entities.model.text_embedding import TextEmbeddingModelConfig

from werkzeug import Request, Response



from endpoints.auth import BaseAuth

from endpoints.openai_protocol import ProtocolError, openai_error



class OaicompatDifyModelEndpoint(Endpoint, BaseAuth):

    def _error(self, status: int, message: str, type_: str = "invalid_request_error") -> Response:

        return Response(

            json.dumps(openai_error(message, type_=type_)),

            status=status,

            content_type="application/json",

        )



    def _invoke(self, r: Request, values: Mapping, settings: Mapping) -> Response:

        if not self.verify(r, settings):

            return self._error(401, "Unauthorized")



        try:

            model: Optional[dict] = settings.get("text_embedding", None)

            if not model:

                raise ProtocolError("Text embedding model is not set")



            try:

                data = r.get_json(force=True)

            except Exception as exc:  # noqa: BLE001

                raise ProtocolError(f"invalid JSON body: {exc}") from exc

            if not data:

                raise ProtocolError("Request body is empty")



            texts: list[str] = []

            if isinstance(data.get("input"), str):

                texts.append(data.get("input"))

            elif isinstance(data.get("input"), list):

                texts = data.get("input")

            else:

                raise ProtocolError("Invalid input type")



            text_embedding_response = self.session.model.text_embedding.invoke(

                model_config=TextEmbeddingModelConfig(**model),

                texts=texts,

            )



            return Response(

                json.dumps(

                    {

                        "object": "list",

                        "data": [

                            {

                                "object": "embedding",

                                "embedding": embedding,

                                "index": index,

                            }

                            for index, embedding in enumerate(

                                text_embedding_response.embeddings

                            )

                        ],

                        "usage": {

                            "prompt_tokens": text_embedding_response.usage.total_tokens,

                            "total_tokens": text_embedding_response.usage.total_tokens,

                        },

                        "model": text_embedding_response.model,

                    }

                ),

                status=200,

                content_type="application/json",

            )

        except ProtocolError as exc:

            return self._error(400, str(exc))

        except ValueError as exc:

            return self._error(400, str(exc))

        except Exception as exc:  # noqa: BLE001

            return self._error(500, str(exc) or type(exc).__name__, "server_error")