"""OpenAI-compatible routes for the API server adapter.

``OpenAICompatRoutesMixin`` (inherited by ``APIServerAdapter``) carries ``/v1/chat/completions``,
``/v1/responses`` (+ GET/DELETE), their SSE writers and the Responses-transcript helpers.
api_server-internal helpers are imported lazily inside each method: the origin imports this
module (top-level import = cycle), and lazy lookup keeps ``patch("...api_server.X")`` effective.
"""

import asyncio
import copy
import json
import logging
import re
import time
import uuid
from contextlib import suppress
from typing import Any, Dict, List, Optional

from agent.i18n import t

try:
    from aiohttp import web
except ImportError:  # pragma: no cover - mirrors api_server's optional import
    web = None  # type: ignore[assignment]

# Logger parity with the origin module (moved log records keep their name).
logger = logging.getLogger("gateway.platforms.api_server")

async def _iter_stream_items(stream_q, agent_task, response):
    """Yield agent stream items until EOS, writing SSE keepalives while idle.

    Yields the ``None`` sentinel once so callers can run EOS-only work; when ``agent_task``
    is already done the remaining queue is drained and the sentinel swallowed.
    """
    from gateway.platforms.api_server import CHAT_COMPLETIONS_SSE_KEEPALIVE_SECONDS

    last_activity = time.monotonic()
    while True:
        try:
            item = await asyncio.wait_for(stream_q.get(), timeout=0.5)
        except TimeoutError:
            if agent_task.done():
                while True:
                    try:
                        item = stream_q.get_nowait()
                    except asyncio.QueueEmpty:
                        return
                    if item is None:
                        return
                    yield item
                    last_activity = time.monotonic()
            if time.monotonic() - last_activity >= CHAT_COMPLETIONS_SSE_KEEPALIVE_SECONDS:
                await response.write(b": keepalive\n\n")
                last_activity = time.monotonic()
            continue
        if item is None:
            yield None
            return
        yield item
        last_activity = time.monotonic()


def _result_flags(result: Any) -> tuple:
    """``(completed, partial, failed, error)`` from an agent result dict (defaults if not a dict)."""
    if not isinstance(result, dict):
        return True, False, False, None
    return (bool(result.get("completed", True)) and not result.get('interrupted'), bool(result.get("partial")),
            bool(result.get("failed")), result.get("error"))


def _budget_exhausted(result: Any) -> bool:
    """The turn ran out of its iteration budget; its final text is the agent's own wrap-up."""
    reason = str(result.get("turn_exit_reason") or "") if isinstance(result, dict) else ""
    return reason.startswith(("budget_exhausted", "max_iterations_reached("))


def _finish_reason(completed, is_partial, is_failed, err_msg, agent_error=None, *, budget_exhausted=False) -> str:
    """OpenAI ``finish_reason``: "length" for truncation, "error" for failure, else "stop"."""
    # OpenAI uses "length" for truncation, "stop" for normal completion, and downstream SDKs accept "error"
    # / custom codes. See issue #22496.
    if is_partial and err_msg and "truncat" in err_msg.lower():
        return "length"
    if agent_error is not None or is_failed or (not completed and err_msg):
        return "error"
    # "cancelled" means the user interrupted; a budget-capped turn stopped short of the task instead,
    # and a broker reading "cancelled" discards a usable wrap-up answer.
    if budget_exhausted:
        return "length"
    return 'cancelled' if not completed else 'stop'


_RESPONSES_FINGERPRINT_KEYS = (
    "input", "instructions", "previous_response_id", "conversation", "conversation_history", "model",
    "provider", "model_options", "tools",
)


def _settled(record):
    """A stored idempotency record that carries its answer. A claim written without one (an
    older release recorded the claim before the run) has an unknown outcome and never replays."""
    return record is not None and record.get('response') is not None


def _responses_fingerprint_keys(body):
    """``truncation`` joins the fingerprint only when sent, so an exact retry of a request
    recorded before it was fingerprinted keeps matching its stored key."""
    return _RESPONSES_FINGERPRINT_KEYS + (("truncation",) if body.get("truncation") is not None else ())


def _response_identity(store, durable_key):
    if durable_key is not None:
        from gateway.platforms.api_server_response_identity import durable_response_id
        return durable_response_id(durable_key[0]), store.request_created_at(durable_key[0])
    return f'resp_{uuid.uuid4().hex[:28]}', int(time.time())


def _response_status(result):
    if result.get('interrupted'):
        return 'cancelled'
    if result.get('failed') or result.get('error'):
        return 'failed'
    if result.get('completed') is False:
        return 'incomplete'
    return 'completed'


def _hermes_extras(completed, is_partial, is_failed, err_msg, finish_reason: str, *,
                   budget_exhausted: bool = False) -> dict[str, Any]:
    error_code = ("iteration_budget_exhausted" if budget_exhausted and finish_reason == "length"
                  else "output_truncated" if finish_reason == "length" else "agent_error")
    return {"completed": completed, "partial": is_partial, "failed": is_failed, "error": err_msg,
            "error_code": error_code}


def _transformed_notice() -> str:
    return t("platform.api_server.transformed_notice")


def _post_stream_transform(result: Any) -> tuple:
    """``(text, appended)`` a stream still owes after ``transform_llm_output`` rewrote the final
    (the deltas already carried the raw reply): the appended suffix, or the whole rewrite with
    ``appended=False`` when it is not a pure append. ``("", False)`` when nothing was transformed."""
    if not isinstance(result, dict) or not result.get("response_transformed"):
        return "", False
    final = result.get("final_response") or ""
    original = result.get("pre_transform_response") or ""
    if original and final.startswith(original):
        return final[len(original):], True
    return final, False


def _message_item(text: Any) -> dict[str, Any]:
    """Responses ``message`` output item carrying one ``output_text`` part."""
    return {"type": "message", "role": "assistant",
            "content": [{"type": "output_text", "text": text}]}


def _cap_text(text: str, keep: int) -> str:
    """Head of ``text`` plus a marker saying how much was cut (the Responses truncation rule)."""
    return text[:keep] + "...[" + str(len(text) - keep) + " more chars]"


def _cap_history_tool_outputs(history: list[dict[str, Any]], max_chars: int) -> list[dict[str, Any]]:
    """Copy of ``history`` with tool outputs and string tool-call arguments longer than
    ``max_chars`` cut down. Only tool rows and ``tool_calls`` blobs change; user/assistant text
    is left alone, and the agent's own transcript rows are never mutated (rows are copied).
    Opt-in via gateway.api_server.history_tool_output_max_chars: a single stored snapshot
    embeds the full cumulative history, so a few large tool outputs pushed one
    response_store.db write to ~677 KB (#82513)."""
    if max_chars <= 0:
        return history
    out: list[dict[str, Any]] = []
    for msg in history:
        if not isinstance(msg, dict):
            out.append(msg)
            continue
        content = msg.get("content")
        if msg.get("role") == "tool" and isinstance(content, str) and len(content) > max_chars:
            msg = {**msg, "content": _cap_text(content, max_chars)}
        tool_calls = msg.get("tool_calls")
        if msg.get("role") == "assistant" and isinstance(tool_calls, list):
            capped_calls = []
            for call in tool_calls:
                fn = call.get("function") if isinstance(call, dict) else None
                raw = fn.get("arguments") if isinstance(fn, dict) else None
                if isinstance(raw, str) and len(raw) > max_chars:
                    try:
                        args = json.loads(raw)
                    except ValueError:
                        args = None
                    if isinstance(args, dict):
                        for k, v in args.items():
                            if isinstance(v, str) and len(v) > max_chars:
                                args[k] = _cap_text(v, max_chars)
                        call = {**call, "function": {**fn, "arguments": json.dumps(args)}}
                capped_calls.append(call)
            msg = {**msg, "tool_calls": capped_calls}
        out.append(msg)
    return out


def _reasoning_item(text: str) -> dict[str, Any]:
    """Completed Responses ``reasoning`` output item (same shape the SSE writer closes with)."""
    return {"id": f"rs_{uuid.uuid4().hex[:24]}", "type": "reasoning", "status": "completed",
            "summary": [{"type": "summary_text", "text": text}]}


def _is_reasoning_input_item(item: Any) -> bool:
    """Echoed-back ``reasoning`` output item: Responses SDK clients replay a prior response's
    ``output`` list as the next ``input``. It carries no message content, so it must be
    skipped rather than parsed into an empty ``user`` turn (#99552)."""
    return isinstance(item, dict) and item.get("type") == "reasoning"


def _turn_reasoning_text(
        conversation_history: list[dict[str, Any]], user_message: Any, result: dict[str, Any]) -> str:
    """Reasoning the model produced on this turn, joined for a non-streaming
    ``message.reasoning_content``. Read from the assistant messages the agent already
    persisted (``build_assistant_message`` stores the structured reasoning under
    ``reasoning``) rather than re-accumulating callback deltas, so it is exactly what the
    stream would have carried and cannot double-count the post-response fallback."""
    messages = result.get("messages") if isinstance(result, dict) else None
    if not isinstance(messages, list):
        return ""
    start = OpenAICompatRoutesMixin._response_messages_turn_start_index(
        conversation_history, user_message, result)
    parts = [m["reasoning"] for m in messages[start:]
             if isinstance(m, dict) and m.get("role") == "assistant"
             and isinstance(m.get("reasoning"), str) and m["reasoning"].strip()]
    return "\n\n".join(parts)


def _trim_tool_items(items: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """Trim large tool payloads in place so response.completed stays under ~100KB (clients
    already received the full details via the incremental events)."""
    for item in items:
        if item.get("type") == "function_call":
            try:
                raw = item.get("arguments")
                args = json.loads(raw) if isinstance(raw, str) else item.get("arguments", {})
                if isinstance(args, dict):
                    for k in ("content", "query", "pattern", "old_string", "new_string"):
                        if isinstance(args.get(k), str) and len(args[k]) > 500:
                            args[k] = f"[{len(args[k])} chars — truncated for response.completed]"
                    item["arguments"] = json.dumps(args)
            except Exception:
                pass
        elif item.get("type") == "function_call_output":
            output = item.get("output", [])
            if isinstance(output, list) and output:
                first = output[0]
                if isinstance(first, dict) and first.get("type") == "input_text":
                    text = first.get("text", "")
                    if len(text) > 1000:
                        first["text"] = _cap_text(text, 500)
                        item["output"] = [first]
    return items


class _ResponsesStream:
    """Per-request state and event emitters for the POST /v1/responses SSE writer.

    Every event carries a monotonic ``sequence_number`` (canonical Responses SSE schema). Text
    deltas are batched (50ms) against Open WebUI re-render storms; tool events flush first.
    """

    def __init__(self, adapter, response, *, response_id: str, model: str, created_at: int,
                 conversation_history: list[dict[str, str]], user_message: str,
                 instructions: Optional[str], conversation: Optional[str], store: bool, session_id: str,
                 durable_key=None, terminal_replay=False):
        from gateway.platforms import api_server as api
        self._api = api
        self.adapter, self.response, self.response_id = adapter, response, response_id
        self.model, self.created_at, self.conversation_history = model, created_at, conversation_history
        self.user_message, self.instructions = user_message, instructions
        self.conversation, self.store, self.session_id = conversation, store, session_id
        self.durable_key = durable_key
        self.terminal_replay = terminal_replay
        # Resolved in the request's profile scope: a snapshot written after it (disconnect) must not follow another.
        self.response_store = adapter._current_response_store()
        self.final_text_parts: list[str] = []
        self.pending_tool_calls: list[dict[str, Any]] = []  # open function_call items, in order
        self.emitted_items: list[dict[str, Any]] = []  # output items so far (terminal payload)
        self.output_index = 0
        self.call_counter = 0  # call_id fallback when the agent supplies no tool_call_id
        self.sequence_number = 0
        self._identity_counts = {}
        self._published_output_ids = []
        self._suppress_incremental = False
        self.message_item_id = self._item_id('message')
        self.message_output_index: Optional[int] = None
        self.message_opened = False
        self.reasoning_item: Optional[dict[str, Any]] = None  # open ``reasoning`` output item
        self.final_response_text = ""
        self.transformed_final = ""  # non-append transform_llm_output rewrite; replaces the deltas
        self.agent_error: Optional[str] = None
        self.usage: dict[str, int] = {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0}
        self.terminal_snapshot_persisted = False
        self.recorded_events: list[tuple] = []  # (event_type, data) as written, for idempotent replay
        self.result: Any = None
        self._batch_buf: list[str] = []
        self._batch_timer: Optional[asyncio.Task] = None
        self._batch_lock = asyncio.Lock()

    async def write_event(self, event_type: str, data: dict[str, Any]) -> None:
        if self.durable_key and 'output_index' in data:
            if self._suppress_incremental:
                return
            index = data['output_index']
            if event_type == 'response.output_item.added':
                valid = index == len(self._published_output_ids)
                if valid:
                    self._published_output_ids.append(data['item']['id'])
            else:
                identity = data.get('item_id') or data.get('item', {}).get('id')
                valid = 0 <= index < len(self._published_output_ids) and (
                    identity is None or self._published_output_ids[index] == identity)
            if not valid:
                # A late/backpressured observer may miss a globally allocated prefix. Keep the
                # prefix it already published; complete from the canonical terminal snapshot.
                self._suppress_incremental = True
                return
        if "sequence_number" not in data:
            data["sequence_number"] = self.sequence_number
        self.sequence_number += 1
        if self.durable_key:  # the only reader is the keyed replay record (emit_completed)
            # Snapshot: terminal trimming later edits the nested tool output these frames share.
            self.recorded_events.append((event_type, copy.deepcopy(data)))
        await self.response.write(self._api._sse_frame(data, event=event_type))

    def _item_id(self, kind, *, call_id=None, commentary=False):
        from gateway.platforms.api_server_response_identity import next_item_id
        return next_item_id(self.response_id, self._identity_counts, kind,
                            call_id=call_id, commentary=commentary)

    def _output_index(self, item_id):
        index = (self.response_store.output_index(self.durable_key[0], item_id)
                 if self.durable_key else self.output_index)
        self.output_index = max(self.output_index, index + 1)
        return index

    def envelope(self, status: str) -> dict[str, Any]:
        return {"id": self.response_id, "object": "response", "status": status,
                "created_at": self.created_at, "model": self.model}

    def terminal_envelope(self, status: str, output: list[dict[str, Any]], *, error=None) -> dict:
        """``envelope`` + ``output`` (+ ``error`` when given) + ``usage``, in wire key order."""
        env = self.envelope(status)
        env["output"] = output
        if error is not None:
            env["error"] = ({"code": "agent_error", "message": error} if self.durable_key
                            else {"message": error, "type": "server_error"})
        env["usage"] = self._api._responses_usage_payload(self.usage)
        return env

    def _history_with_user(self) -> list[dict[str, Any]]:
        return list(self.conversation_history) + [{"role": "user", "content": self.user_message}]

    def persist_snapshot(self, response_env: dict[str, Any], *, history=None, session_id=None):
        if not self.store:
            return
        if self.durable_key and _settled(self.response_store.get(self.durable_key[0])):
            return  # A concurrent observer already committed the exact terminal envelope.
        advance_conversation = self.conversation and (not self.terminal_replay
            or self.response_store.get_conversation(self.conversation) in (None, self.response_id))
        self.response_store.put(self.response_id, {
            "response": response_env,
            "conversation_history": self._history_with_user() if history is None else history,
            "instructions": self.instructions,
            "session_id": session_id or self.session_id})
        if advance_conversation:
            self.response_store.set_conversation(self.conversation, self.response_id)

    def persist_incomplete_if_needed(self) -> None:
        """Persist an ``incomplete`` snapshot when no terminal one was written (disconnect /
        cancel paths), so GET /v1/responses/{id} and ``previous_response_id`` chaining survive."""
        if not self.store or self.terminal_snapshot_persisted:
            return
        text = "".join(self.final_text_parts) or self.final_response_text
        items = list(self.emitted_items)
        history = self._history_with_user()
        if text:
            items.append(_message_item(text))
            history.append({"role": "assistant", "content": text})
        self.persist_snapshot(self.terminal_envelope("incomplete", items), history=history)

    async def emit_created(self) -> None:
        env = self.envelope("in_progress")
        env["output"] = []
        await self.write_event("response.created", {"type": "response.created", "response": env})
        if not self.terminal_replay:
            self.persist_snapshot(env)

    async def _open_message_item(self) -> None:
        """Emit output_item.added for the assistant message on the first text delta."""
        if self.message_opened:
            return
        self.message_opened = True
        self.message_output_index = self._output_index(self.message_item_id)
        await self.write_event("response.output_item.added", {
            "type": "response.output_item.added", "output_index": self.message_output_index,
            "item": {"id": self.message_item_id, "type": "message", "status": "in_progress",
                     "role": "assistant", "content": []}})

    async def emit_text_delta(self, delta_text: str) -> None:
        if self.terminal_replay:
            self.final_text_parts.append(delta_text)
            return
        await self.close_reasoning_item()
        await self._open_message_item()
        self.final_text_parts.append(delta_text)
        await self.write_event("response.output_text.delta", {
            "type": "response.output_text.delta", "item_id": self.message_item_id,
            "output_index": self.message_output_index, "content_index": 0, "delta": delta_text,
            "logprobs": []})

    async def emit_reasoning_delta(self, delta_text: str) -> None:
        """Responses reasoning-summary family (#99552): one ``reasoning`` output item per
        thinking burst, closed before the next message/tool item opens."""
        if self.reasoning_item is None:
            item = {"id": self._item_id('reasoning'), "type": "reasoning", "status": "in_progress",
                    "summary": []}
            self.reasoning_item = {"item": item, "output_index": self._output_index(item['id']), "parts": []}
            await self.write_event("response.output_item.added", {
                "type": "response.output_item.added",
                "output_index": self.reasoning_item["output_index"], "item": item})
            await self.write_event("response.reasoning_summary_part.added", {
                "type": "response.reasoning_summary_part.added", "item_id": item["id"],
                "output_index": self.reasoning_item["output_index"], "summary_index": 0,
                "part": {"type": "summary_text", "text": ""}})
        rs = self.reasoning_item
        rs["parts"].append(delta_text)
        await self.write_event("response.reasoning_summary_text.delta", {
            "type": "response.reasoning_summary_text.delta", "item_id": rs["item"]["id"],
            "output_index": rs["output_index"], "summary_index": 0, "delta": delta_text})

    async def close_reasoning_item(self) -> None:
        rs, self.reasoning_item = self.reasoning_item, None
        if rs is None:
            return
        text = "".join(rs["parts"])
        item = dict(rs["item"], status="completed", summary=[{"type": "summary_text", "text": text}])
        base = {"item_id": item["id"], "output_index": rs["output_index"], "summary_index": 0}
        await self.write_event("response.reasoning_summary_text.done", {
            "type": "response.reasoning_summary_text.done", **base, "text": text})
        await self.write_event("response.reasoning_summary_part.done", {
            "type": "response.reasoning_summary_part.done", **base,
            "part": {"type": "summary_text", "text": text}})
        self.emitted_items.append(item)
        await self.write_event("response.output_item.done", {
            "type": "response.output_item.done", "output_index": rs["output_index"], "item": item})

    async def emit_commentary(self, text: str) -> None:
        """Mid-turn assistant commentary as its own completed ``message`` item carrying
        ``"phase": "commentary"`` — never appended to ``final_text_parts``, so the final answer
        item stays clean (#67580). Closes any open reasoning item first so a reasoning item
        never straddles a message item."""
        await self.close_reasoning_item()
        item = {"id": self._item_id('message', commentary=True), "status": "completed", "phase": "commentary",
                **_message_item(text)}
        idx = self._output_index(item['id'])
        self.emitted_items.append({"phase": "commentary", **_message_item(text)})
        for event in ("response.output_item.added", "response.output_item.done"):
            await self.write_event(event, {"type": event, "output_index": idx, "item": item})

    async def emit_tool_started(self, payload: dict[str, Any]) -> None:
        """function_call ``output_item.added``; the agent's tool_call_id beats a generated call id."""
        await self.close_reasoning_item()
        self.call_counter += 1
        call_id = payload.get("tool_call_id") or f"call_{self.response_id[5:]}_{self.call_counter}"
        args = payload.get("arguments", {})
        arguments_str = json.dumps(args) if isinstance(args, dict) else str(args)
        name = payload.get("name", "")
        item = {"id": self._item_id('function_call', call_id=call_id), "type": "function_call",
                "status": "in_progress", "name": name, "call_id": call_id, "arguments": arguments_str}
        idx = self._output_index(item['id'])
        self.pending_tool_calls.append({
            "call_id": call_id, "name": name, "arguments": arguments_str, "item_id": item["id"],
            "output_index": idx})
        self.emitted_items.append(
            {"type": "function_call", "name": name, "arguments": arguments_str, "call_id": call_id})
        await self.write_event("response.output_item.added", {
            "type": "response.output_item.added", "output_index": idx, "item": item})

    async def emit_tool_completed(self, payload: dict[str, Any]) -> None:
        """function_call ``output_item.done`` + function_call_output added/done; orphans skipped."""
        call_id = payload.get("tool_call_id")
        pending = next((p for p in self.pending_tool_calls if p["call_id"] == call_id), None)
        if not call_id or pending is None:
            return
        self.pending_tool_calls.remove(pending)
        done_item = {"id": pending["item_id"], "type": "function_call", "status": "completed",
                     "name": pending["name"], "call_id": pending["call_id"],
                     "arguments": pending["arguments"]}
        await self.write_event("response.output_item.done", {
            "type": "response.output_item.done", "output_index": pending["output_index"],
            "item": done_item})
        result = payload.get("result", "")
        result_str = result if isinstance(result, str) else json.dumps(result)
        output_parts = [{"type": "input_text", "text": result_str}]
        output_item = {"id": self._item_id('function_call_output', call_id=call_id), "type": "function_call_output",
                       "call_id": pending["call_id"], "output": output_parts, "status": "completed"}
        idx = self._output_index(output_item['id'])
        self.emitted_items.append(
            {"type": "function_call_output", "call_id": pending["call_id"], "output": output_parts})
        for event in ("response.output_item.added", "response.output_item.done"):
            await self.write_event(event, {"type": event, "output_index": idx, "item": output_item})

    async def emit_status(self, payload: dict[str, Any]) -> None:
        """Lifecycle/warning status (provider wait, auto-recovery countdown, fallback switch) as a
        ``hermes.status`` custom event; not a Responses output item."""
        await self.response.write(self._api._sse_frame(payload, event="hermes.status"))

    # queue tag -> (method name, payload adapter)
    _TAG_HANDLERS = {
        "__tool_started__": ("emit_tool_started", lambda p: p),
        "__tool_completed__": ("emit_tool_completed", lambda p: p),
        "__commentary__": ("emit_commentary", lambda p: p["text"]),
        "__reasoning__": ("emit_reasoning_delta", lambda p: p),
        "__status__": ("emit_status", lambda p: p),
    }

    async def dispatch(self, item: Any) -> None:
        """Route one queue item: tagged tuples emit immediately, strings are batched, others dropped."""
        if self.terminal_replay:
            return
        if isinstance(item, tuple) and len(item) == 2 and isinstance(item[0], str):
            tag, payload = item
            await self.flush_batch()
            handler = self._TAG_HANDLERS.get(tag)
            if handler is not None:
                method, adapt = handler
                await getattr(self, method)(adapt(payload))
        elif isinstance(item, str):
            self._batch_buf.append(item)
            if self._batch_timer is None:
                self._batch_timer = asyncio.create_task(self._batch_flush_after(0.05))

    async def _batch_flush_after(self, delay: float) -> None:
        try:
            await asyncio.sleep(delay)
        except asyncio.CancelledError:
            return
        # Clear the timer BEFORE flushing so new deltas can start a fresh timer while we emit.
        self._batch_timer = None
        await self.flush_batch()

    def cancel_batch_timer(self) -> None:
        if self._batch_timer and not self._batch_timer.done():
            self._batch_timer.cancel()
            self._batch_timer = None

    async def flush_batch(self) -> None:
        """Emit a single delta for all buffered text."""
        if not self._batch_buf:
            return
        async with self._batch_lock:
            if self._batch_buf:
                combined = "".join(self._batch_buf)
                self._batch_buf = []
                await self.emit_text_delta(combined)

    async def collect_result(self, agent_task) -> None:
        """Await the agent; when it produced a final_response but streamed no deltas
        (some providers only emit the full text at the end), emit one fallback delta."""
        try:
            result, agent_usage = await agent_task
            self.result = result
            self.usage = agent_usage or self.usage
            agent_final = result.get("final_response", "") if isinstance(result, dict) else ""
            if self.terminal_replay:
                # The receipt already contains the transformed final; it is not a raw provider
                # stream onto which a verifier/footer transform should be applied a second time.
                self.final_response_text = agent_final or self._api._redact_api_error_text(
                    result.get('error', '(No response generated)'))
                self.final_text_parts = [self.final_response_text]
                return
            tail, appended = _post_stream_transform(result)
            if tail and self.final_text_parts:
                if appended:
                    await self.emit_text_delta(tail)
                else:
                    self.transformed_final = agent_final
            if agent_final and not self.final_text_parts:
                await self.emit_text_delta(agent_final)
            if agent_final and not self.final_response_text:
                self.final_response_text = agent_final
            if isinstance(result, dict) and result.get("error") and not self.final_response_text:
                self.agent_error = self._api._redact_api_error_text(result["error"])
        except Exception as e:
            logger.error("Error running agent for streaming responses: %s", e, exc_info=True)
            self.agent_error = self._api._redact_api_error_text(e)

    async def close_message_item(self) -> None:
        await self.close_reasoning_item()
        self.final_response_text = (
            self.transformed_final or "".join(self.final_text_parts) or self.final_response_text)
        if not self.message_opened:
            return
        await self.write_event("response.output_text.done", {
            "type": "response.output_text.done", "item_id": self.message_item_id,
            "output_index": self.message_output_index, "content_index": 0,
            "text": self.final_response_text, "logprobs": []})
        await self.write_event("response.output_item.done", {
            "type": "response.output_item.done", "output_index": self.message_output_index,
            "item": {"id": self.message_item_id, "type": "message", "status": "completed",
                     "role": "assistant",
                     "content": [{"type": "output_text", "text": self.final_response_text}]}})

    def _final_items(self) -> list[dict[str, Any]]:
        """Emitted items (trimmed) plus a final message item, so clients that only parse
        the terminal payload still see the assistant text (mirrors _extract_output_items)."""
        items = _trim_tool_items(list(self.emitted_items))
        redact = self._api._redact_api_error_text
        text = self.final_response_text or (redact(self.agent_error) if self.agent_error else "")
        items.append(_message_item(text))
        return items

    async def emit_failed(self) -> None:
        redact = self._api._redact_api_error_text
        env = self.terminal_envelope("failed", self._final_items(), error=redact(self.agent_error))
        history = self._history_with_user()
        history.append(
            {"role": "assistant", "content": self.final_response_text or redact(self.agent_error)})
        self.persist_snapshot(env, history=history)
        self.terminal_snapshot_persisted = True
        await self.write_event("response.failed", {"type": "response.failed", "response": env})

    async def emit_completed(self) -> dict[str, Any]:
        result = self.result
        status = _response_status(result)
        output = self._final_items()
        if self.durable_key:
            start = self.adapter._response_messages_turn_start_index(self.conversation_history, self.user_message, result)
            output = self.adapter._extract_output_items(result, start_index=start, response_id=self.response_id)
            output = self.response_store.order_output(self.durable_key[0], output)
        env = self.terminal_envelope(status, output,
            error=self._api._redact_api_error_text(result.get('error') or 'The admitted turn failed.')
            if status == 'failed' else None)
        full_history = self.adapter._build_response_conversation_history(
            self.conversation_history, self.user_message, result, self.final_response_text,
            tool_output_max_chars=self.adapter._history_tool_output_max_chars)
        # Transcript substitution for result["_compressed"] happens in the history builder; only
        # a compression-rotated session_id is propagated so chaining resumes the child session.
        sid = result.get("session_id") if isinstance(result, dict) else None
        self.persist_snapshot(
            env, history=full_history, session_id=sid if isinstance(sid, str) and sid else None)
        self.terminal_snapshot_persisted = True
        if self.durable_key:
            headers = {k: v for k, v in self.response.headers.items() if k.startswith('X-Hermes-')}
            terminal_event = {'type': 'response.' + status, 'response': env, 'sequence_number': self.sequence_number}
            if not self.terminal_replay:
                # Only the publisher recorded the deltas; a replay observer winning the
                # INSERT OR IGNORE would leave every later replay without them.
                self.response_store.put(self.durable_key[0], {
                    'fingerprint': self.durable_key[1], 'response': env, 'headers': headers,
                    'events': [*self.recorded_events, ('response.' + status, terminal_event)]})
            stored = self.response_store.get(self.durable_key[0])
            if _settled(stored):
                env = stored['response']
            status = env['status']
        await self.write_event(
            'response.' + status, {'type': 'response.' + status, 'response': env})
        return env

    async def emit_crash(self, exc: BaseException) -> None:
        error = self._api._redact_api_error_text(exc, limit=500)
        env = self.terminal_envelope("failed", list(self.emitted_items), error=error)
        await self.write_event("response.failed", {"type": "response.failed", "response": env})


def _parse_chat_messages(messages: list[Any]) -> tuple:
    """Chat ``messages`` -> ``(system_prompt, conversation_messages, error_response_or_None)``.

    System messages -> ephemeral system prompt layered ON TOP of core, flattened to text
    (Anthropic rejects images there, OpenAI text models ignore them)."""
    from gateway.platforms.api_server import (
        _multimodal_validation_error, _normalize_chat_content, _normalize_multimodal_content)
    system_prompt = None
    conversation_messages: list[dict[str, str]] = []
    for idx, msg in enumerate(messages):
        role = msg.get("role", "")
        raw_content = msg.get("content", "")
        if role == "system":
            content = _normalize_chat_content(raw_content)
            system_prompt = content if system_prompt is None else system_prompt + "\n" + content
        elif role in {"user", "assistant"}:
            try:
                content = _normalize_multimodal_content(raw_content)
            except ValueError as exc:
                return None, None, _multimodal_validation_error(exc, param=f"messages[{idx}].content")
            conversation_messages.append({"role": role, "content": content})
    return system_prompt, conversation_messages, None


def _parse_responses_input(raw_input: Any) -> tuple:
    """Responses ``input`` (string or item array) -> ``(input_messages, error_response_or_None)``."""
    from gateway.platforms.api_server import (
        _error_response, _multimodal_validation_error, _normalize_multimodal_content)
    input_messages: list[dict[str, Any]] = []
    if isinstance(raw_input, str):
        input_messages = [{"role": "user", "content": raw_input}]
    elif isinstance(raw_input, list):
        for idx, item in enumerate(raw_input):
            if isinstance(item, str):
                input_messages.append({"role": "user", "content": item})
            elif _is_reasoning_input_item(item):
                continue
            elif isinstance(item, dict):
                try:
                    content = _normalize_multimodal_content(item.get("content", ""))
                except ValueError as exc:
                    return None, _multimodal_validation_error(exc, param=f"input[{idx}].content")
                input_messages.append({"role": item.get("role", "user"), "content": content})
    else:
        return None, _error_response("'input' must be a string or array", 400)
    return input_messages, None


def _parse_conversation_history(raw_history: Any) -> tuple:
    """Explicit Responses ``conversation_history`` -> ``(history, error_response_or_None)``."""
    from gateway.platforms.api_server import (
        _error_response, _multimodal_validation_error, _normalize_multimodal_content)
    conversation_history: list[dict[str, Any]] = []
    if not isinstance(raw_history, list):
        return None, _error_response("'conversation_history' must be an array of message objects", 400)
    for i, entry in enumerate(raw_history):
        if _is_reasoning_input_item(entry):
            continue
        if not isinstance(entry, dict) or "role" not in entry or "content" not in entry:
            return None, _error_response(f"conversation_history[{i}] must have 'role' and 'content' fields", 400)
        try:
            entry_content = _normalize_multimodal_content(entry["content"])
        except ValueError as exc:
            return None, _multimodal_validation_error(exc, param=f"conversation_history[{i}].content")
        conversation_history.append({"role": str(entry["role"]), "content": entry_content})
    return conversation_history, None


class OpenAICompatRoutesMixin:
    """/v1/chat/completions and /v1/responses handlers + SSE writers."""

    def _select_request_route(
        self, body: dict[str, Any], *, session_id, gateway_session_key, model_alias) -> tuple:
        """Resolve the model_routes alias + per-request overrides ->
        ``(route, agent_overrides, error_response_or_None)``."""
        from gateway.platforms.api_server import _error_response, _request_agent_overrides
        route = self._resolve_route(model_alias)
        overrides = _request_agent_overrides(
            body, virtual_model=self._model_name, allow_bare_model=self._direct_model_requests)
        err = self._request_route_conflict_error(
            session_id=session_id, gateway_session_key=gateway_session_key,
            requested_model=overrides.get("requested_model"),
            requested_provider=overrides.get("requested_provider"), route=route)
        return route, overrides, (_error_response(err, 400) if err else None)

    def _register_stream_approval(self, request, completion_id, stream_q, session_id) -> tuple:
        """Expose a streaming completion as a run for ``POST /v1/runs/{completion_id}/approval``
        (#51871) -> ``(notify, on_done)``. Owner + status are stamped like the session stream, so
        ``_load_owned_run`` finds it; keyed by the completion id (never the shared session key) so
        concurrent turns can't cross-resolve. ``on_done`` retires it when the agent task ends."""
        from gateway.platforms.api_server import _approval_request_event
        self._run_owners[completion_id] = self._run_idempotency_scope(request)
        self._set_run_status(completion_id, "running", session_id=session_id or "")
        self._run_approval_sessions[completion_id] = completion_id

        def _approval_notify(approval_data):
            event = _approval_request_event(completion_id, approval_data, session_id=session_id or "")
            self._set_run_status(completion_id, "waiting_for_approval", last_event="approval.request",
                                 approval=event)
            stream_q.put_threadsafe(("__approval__", event))

        def _on_done(fut):
            from gateway.platforms.api_server_runs import terminal_run_status
            self._run_approval_sessions.pop(completion_id, None)
            if fut.cancelled():
                status = "cancelled"
            elif fut.exception() is not None:
                status = "failed"
            else:
                result = (fut.result() or (None,))[0]
                status = terminal_run_status(result)[0] if isinstance(result, dict) else "completed"
            self._set_run_status(completion_id, status)
            self._release_run_owner_if_forgotten(completion_id)
        return _approval_notify, _on_done

    def _spawn_stream_agent(self, stream_q, *, on_done=None, run_agent=None, **run_kwargs) -> tuple:
        """Start ``_run_agent`` for an SSE writer -> ``(agent_task, agent_ref)``. ``agent_ref[0]``
        lets the writer interrupt on disconnect; the EOS sentinel is enqueued from the task's done
        callback so drain loops never race a polled ``agent_task.done()``. ``on_done(task)`` runs
        in that same callback, before EOS."""
        def _on_delta(delta):
            # None from the agent is a CLI box-close signal, not EOS — forwarding it would end
            # the stream early. Called from the run_conversation worker thread: put_threadsafe.
            if delta is not None:
                stream_q.put_threadsafe(delta)
        def _on_reasoning(text):
            # Structured reasoning deltas (#99552): the agent's reasoning_callback, not the
            # lossy 500-char ``reasoning.available`` progress preview. Tagged so the writers
            # keep them distinct from answer text.
            if text:
                stream_q.put_threadsafe(("__reasoning__", text))
        def _on_status(kind, message=None):
            # Lifecycle/warning status (provider wait, auto-recovery countdown, fallback switch) as a
            # ``hermes.status`` event, so a client sees why the stream is silent instead of a dead socket.
            from gateway.platforms.api_server import _redact_api_error_text
            text = _redact_api_error_text(message if message is not None else kind or "").strip()
            if text:
                stream_q.put_threadsafe(("__status__", {"kind": str(kind), "text": text}))
        agent_ref = [None]
        agent_task = asyncio.ensure_future((run_agent or self._run_agent)(
            stream_delta_callback=_on_delta, reasoning_callback=_on_reasoning, status_callback=_on_status,
            agent_ref=agent_ref, **run_kwargs))
        def _done(fut):
            try:
                if on_done is not None:
                    on_done(fut)
            except Exception:
                logger.debug("[api_server] stream on_done hook failed", exc_info=True)
            finally:
                stream_q.put_nowait(None)
        agent_task.add_done_callback(_done)
        return agent_task, agent_ref

    async def _handle_chat_completions(self, request: web.Request) -> web.Response:
        """POST /v1/chat/completions — OpenAI Chat Completions format."""
        from gateway.platforms.api_server import (
            ThreadSafeAsyncQueue, _api_request_profile, _chat_usage_payload, _coerce_request_bool,
            _content_has_visible_payload, _derive_chat_session_id, _error_response, _invalid_request,
            _openai_error, _redact_api_error_text, _resolve_media_to_data_urls)
        # Bound total in-flight agent runs (configurable; #7483).
        limited = self._concurrency_limited_response()
        if limited is not None:
            return limited
        try:
            body = await request.json()
        except Exception:
            return _error_response("Invalid JSON in request body", 400)
        from gateway.platforms.api_server import _request_relay_metadata
        relay_metadata = _request_relay_metadata(body)
        messages = body.get("messages")
        if not messages or not isinstance(messages, list):
            return _invalid_request("Missing or invalid 'messages' field")
        stream = _coerce_request_bool(body.get("stream"), default=False)

        system_prompt, conversation_messages, messages_err = _parse_chat_messages(messages)
        if messages_err is not None:
            return messages_err
        user_message: Any = (conversation_messages[-1].get("content", "") if conversation_messages else "")
        history = conversation_messages[:-1]
        if not _content_has_visible_payload(user_message):
            return _invalid_request("No user message found in messages")

        # X-Hermes-Session-Key scopes long-term memory per channel; independent of
        # X-Hermes-Session-Id (the key persists across transcripts, the id rotates on /new).
        gateway_session_key, key_err = self._parse_session_key_header(request)
        if key_err is not None:
            return key_err
        # X-Hermes-Session-Id continues an existing session (history from state.db, not the body);
        # requires a configured API key or any client could read history by guessing ids.
        provided_session_id = request.headers.get("X-Hermes-Session-Id", "").strip()
        if provided_session_id:
            if not self._api_key:
                logger.warning(
                    "Session continuation via X-Hermes-Session-Id rejected: "
                    "no API key configured.  Set API_SERVER_KEY to enable "
                    "session continuity.")
                return _error_response("Session continuation requires API key authentication. "
                        "Configure API_SERVER_KEY to enable this feature.", 403)
            # Same guard as the native gateway: ids are interpolated into on-disk filenames.
            from gateway.session import _is_path_unsafe
            if re.search(r'[\r\n\x00]', provided_session_id) or _is_path_unsafe(provided_session_id):
                return _invalid_request("Invalid session ID")
            if len(provided_session_id) > self._MAX_SESSION_HEADER_LEN:
                return _invalid_request("Session ID too long")
            session_id = provided_session_id
            try:
                db = await self._ensure_session_db_async()
                if db is not None:
                    # #98619/#13437: a client-addressed id from before a compression rotation
                    # must adopt the live continuation tip — history loads from it, the turn and
                    # the wake target bind it, and a detached delegation delivery row persisted
                    # on the tip is what this continuation consumes. Same canonical resolution
                    # the delivery writer (gateway/wake.py) and /v1/runs use; fails open.
                    from gateway.platforms.api_server_runs import _resolve_live_session_id
                    session_id = await _resolve_live_session_id(self, provided_session_id)
                    history = await asyncio.to_thread(db.get_messages_as_conversation, session_id)
            except Exception as e:
                logger.warning("Failed to load session history for %s: %s", session_id, e)
                history = []
        else:
            # Stable id from the conversation fingerprint so Open WebUI-style clients map onto
            # one Hermes session; namespaced by the routed profile (#123989).
            first_user = next(
                (cm.get("content", "") for cm in conversation_messages if cm.get("role") == "user"), "")
            session_id = _derive_chat_session_id(system_prompt, first_user, _api_request_profile.get())
            session_id, lookup_error = await self._implicit_chat_session(request, session_id)
            if lookup_error is not None:
                return lookup_error
        completion_id = f"chatcmpl-{uuid.uuid4().hex[:29]}"
        model_name = body.get("model", self._model_name)
        created = int(time.time())
        route, agent_overrides, selection_error = self._select_request_route(
            body, session_id=session_id, gateway_session_key=gateway_session_key,
            model_alias=model_name)
        if selection_error is not None:
            return selection_error
        run_kwargs = dict(
            user_message=user_message, conversation_history=history,
            ephemeral_system_prompt=system_prompt, session_id=session_id,
            gateway_session_key=gateway_session_key, **agent_overrides, route=route,
            relay_metadata=relay_metadata,
            # #98619: only an explicitly provided X-Hermes-Session-Id is wake-capable (the
            # header is 403-gated on API_SERVER_KEY, so the wake self-post can authenticate
            # and the client can resume the session by sending it again). A fingerprint-derived
            # id from a header-less client is NOT: delegate_task keeps its forced-sync fallback
            # there — the wake would hard-fail or land in history that client never reloads.
            session_history_delivery=("1" if provided_session_id else ""))
        run_kwargs.update(self._authority_chat_kwargs(request, provided_session_id))
        # This is presentation only. The ordinary API-key/session authorization
        # above still applies; it grants no internal ingress or control authority.
        if provided_session_id and body.get("hermes_notification_category") == "diagnostic":
            run_kwargs["notification_category"] = "diagnostic"
        if stream:
            _stream_q = ThreadSafeAsyncQueue()
            # tool_call_ids with an emitted "running": a "completed" without one (internal/
            # filtered tools) is dropped rather than orphaned on the wire.
            _started_tool_call_ids: set[str] = set()

            def _on_tool_start(tool_call_id, function_name, function_args):
                """``hermes.tool.progress`` status=running; ``_``-prefixed tools stay off the wire."""
                if not tool_call_id or function_name.startswith("_"):
                    return
                _started_tool_call_ids.add(tool_call_id)
                from agent.display import build_tool_preview, get_tool_emoji
                label = build_tool_preview(function_name, function_args) or function_name
                _stream_q.put_threadsafe(("__tool_progress__", {
                    "tool": function_name, "emoji": get_tool_emoji(function_name), "label": label,
                    "toolCallId": tool_call_id, "status": "running"}))

            def _on_tool_complete(tool_call_id, function_name, function_args, function_result):
                if not tool_call_id or tool_call_id not in _started_tool_call_ids:
                    return
                _started_tool_call_ids.discard(tool_call_id)
                _stream_q.put_threadsafe(("__tool_progress__", {
                    "tool": function_name, "toolCallId": tool_call_id, "status": "completed"}))

            # tool_progress_callback deliberately NOT wired: it would duplicate the structured
            # start/complete callbacks (which carry the tool_call id).
            # The completion id doubles as the run id; on_done drops the approval mapping once the
            # turn ends (POST /v1/runs/{id}/approval then answers 409) and sets the terminal status.
            approval_notify, end_stream_run = self._register_stream_approval(
                request, completion_id, _stream_q, session_id)
            agent_task, agent_ref = self._spawn_stream_agent(
                _stream_q, on_done=end_stream_run, tool_start_callback=_on_tool_start,
                tool_complete_callback=_on_tool_complete, approval_notify_callback=approval_notify,
                approval_session_key=completion_id, **run_kwargs)
            # #13437 identity contract: an explicit-header client keeps addressing the id it
            # sent; the response echoes that stable id while reads/writes adopt the live tip,
            # so a rotation mid-turn (after these headers are prepared) never changes what the
            # client should send next — it re-sends the same id and the tip resolution above
            # finds whatever session is live by then.
            return await self._write_sse_chat_completion(
                request, completion_id, model_name, created, _stream_q,
                agent_task, agent_ref, session_id=(provided_session_id or session_id),
                gateway_session_key=gateway_session_key)

        async def _compute_completion():
            return await self._run_agent(**run_kwargs)
        outcome, err = await self._run_idempotent(
            request, body, _compute_completion, log_label="chat completions",
            fingerprint_keys=["model", "provider", "model_options", "messages", "tools", "tool_choice", "stream",
                              "hermes_notification_category"],
            route="chat_completions",
        )
        if err is not None:
            return err
        result, usage = outcome
        presentation_muted = result.get("_notification_presentation_suppressed") is True
        final_response = _resolve_media_to_data_urls(result.get("final_response") or "")
        completed, is_partial, is_failed, err_msg = _result_flags(result)
        if err_msg:
            err_msg = _redact_api_error_text(err_msg)
        budget_exhausted = _budget_exhausted(result)
        finish_reason = _finish_reason(completed, is_partial, is_failed, err_msg, budget_exhausted=budget_exhausted)
        # Same #13437 identity contract as the SSE path: an explicit-header client is echoed
        # the stable id it sent; a fingerprint-derived (header-less) turn keeps reporting the
        # id the agent actually resolved, so headerless clients still learn where the turn went.
        response_headers = {"X-Hermes-Session-Id": (provided_session_id or result.get("session_id", session_id))}
        if gateway_session_key:
            response_headers["X-Hermes-Session-Key"] = gateway_session_key
        # Hard fail (no usable text AND a real failure) -> 502 OpenAI error envelope so SDK
        # clients raise instead of rendering the failure string as message.content.
        if not final_response and (is_failed or is_partial):
            err_body = _openai_error(
                "" if presentation_muted else (err_msg or "Agent run did not produce a response."), err_type="server_error",
                code="agent_incomplete")
            err_body["error"]["hermes"] = {
                "completed": completed, "partial": is_partial, "failed": is_failed}
            response_headers["X-Hermes-Completed"] = "false"
            response_headers["X-Hermes-Partial"] = "true" if is_partial else "false"
            return web.json_response(err_body, status=502, headers=response_headers)
        # Soft partial (some text, run incomplete): 200 + finish_reason="length"/Hermes extras.
        response_data = {
            "id": completion_id, "object": "chat.completion", "created": created,
            "model": model_name,
            "choices": [{"index": 0, "message": {"role": "assistant", "content": "" if presentation_muted else final_response},
                         "finish_reason": finish_reason}],
            "usage": _chat_usage_payload(usage)}
        # Non-streaming twin of ``delta.reasoning_content`` (#99552).
        reasoning_text = _turn_reasoning_text(history, user_message, result)
        if reasoning_text and not presentation_muted:
            response_data["choices"][0]["message"]["reasoning_content"] = reasoning_text
        if is_partial or is_failed or not completed:
            response_data["hermes"] = _hermes_extras(
                completed, is_partial, is_failed, "" if presentation_muted else err_msg, finish_reason,
                budget_exhausted=budget_exhausted)
            response_headers["X-Hermes-Completed"] = "false"
            response_headers["X-Hermes-Partial"] = "true" if is_partial else "false"
            if err_msg and not presentation_muted:
                response_headers["X-Hermes-Error"] = _redact_api_error_text(err_msg, limit=200)
        return web.json_response(response_data, headers=response_headers)

    def _chat_request_id(self, request: web.Request) -> Optional[str]:
        """Durable admission identity of a Chat Completions Idempotency-Key. It carries the
        authenticated namespace: a rotated API key (or another profile) reusing a client key is a
        new principal, not a retry."""
        key = request.headers.get('Idempotency-Key')
        return f'chat:{self._run_idempotency_scope(request)}:{key}' if key else None

    def _authority_chat_kwargs(self, request: web.Request, provided_session_id) -> dict:
        """Admission identity of a Chat Completions turn under session authority; none before it."""
        if getattr(self.gateway_runner, 'session_authority', None) is None:
            return {}
        return dict(request_id=self._chat_request_id(request), history_from_session=bool(provided_session_id),
                    run_owner_scope=self._run_idempotency_scope(request))

    async def _implicit_chat_session(self, request: web.Request, session_id: str) -> tuple:
        """A derived (header-less) id whose session was deleted names a new conversation with the
        same opening: the owning profile's store picks its generation (``implicit_api_session``).
        -> ``(session_id, None)`` or ``(session_id, error response)``, before either response starts:
        a refusal keeps its reason and a busy or unreadable store is a retryable 503
        ``storage_unavailable`` (``store_error_response``); anything else is the OpenAI 500."""
        if getattr(self.gateway_runner, 'session_authority', None) is None:
            return session_id, None
        from gateway.session_api import implicit_api_session
        from gateway.session_authorities import active_authority
        authority = active_authority(self.gateway_runner)
        if authority is None:
            return session_id, None  # admission refuses the unserved profile with its own verdict
        import sqlite3
        from hermes_state_runtime import RuntimeStoreError
        try:
            return await asyncio.to_thread(implicit_api_session, authority.db, session_id,
                                           self._chat_request_id(request)), None
        except (RuntimeStoreError, sqlite3.Error, OSError) as exc:
            from gateway.platforms.api_server import _openai_error
            from gateway.platforms.api_server_authority_runs import store_error_response
            return session_id, store_error_response(_openai_error, exc)
        except Exception as e:
            from gateway.platforms.api_server import _error_response
            logger.error("Error resolving the chat completions session: %s", e, exc_info=True)
            return session_id, _error_response(f"Internal server error: {e}", 500, err_type="server_error")

    async def _run_idempotent(
        self, request: web.Request, body: dict[str, Any], compute, *,
        log_label: str, fingerprint_keys: list[str], route: str) -> tuple:
        """Run ``compute()`` once per (principal scope, logical route, Idempotency-Key) + body fingerprint
        -> ``((result, usage), None)`` or ``(None, 500 response)``.

        ``_idem_cache`` is process-global: under ``gateway.multiplex_profiles`` every profile's
        ``/p/<profile>/v1/...`` mirror shares it, so the key carries ``_run_idempotency_scope`` (the same
        ``sha256(profile, expected API key)`` namespace the durable ``/v1/runs`` API uses) — a client key
        colliding across profiles, or a rotated API_SERVER_KEY, never replays another principal's response.
        ``route`` is the logical endpoint (``/v1/...`` and its ``/p/<profile>/v1/...`` alias are the same
        route), folded into the key because the store keeps the fingerprint only as the slot's value.
        """
        from gateway.platforms.api_server import _error_response, _idem_cache, _make_request_fingerprint
        from hermes_state_runtime import RuntimeStoreError
        idempotency_key = request.headers.get("Idempotency-Key")
        try:
            # Durable /v1/runs admissions own idempotency under session authority; the in-memory
            # cache only serves the pre-authority path.
            if idempotency_key and getattr(self.gateway_runner, 'session_authority', None) is None:
                principal_scope = self._run_idempotency_scope(request)
                scoped_key = f"{principal_scope}\0{route}\0{idempotency_key}"
                fp = _make_request_fingerprint(body, keys=fingerprint_keys)
                result, usage = await _idem_cache.get_or_set(scoped_key, fp, compute)
            else:
                result, usage = await compute()
            return (result, usage), None
        except RuntimeStoreError as exc:
            from gateway.platforms.api_server import _openai_error
            from gateway.platforms.api_server_authority_runs import store_error_response
            return None, store_error_response(_openai_error, exc)
        except Exception as e:
            logger.error("Error running agent for %s: %s", log_label, e, exc_info=True)
            message = "" if getattr(e, "_notification_presentation_suppressed", False) is True else f"Internal server error: {e}"
            return None, _error_response(message, 500, err_type="server_error")

    async def _prepare_sse_response(
        self, request: web.Request, session_id: Optional[str], gateway_session_key: Optional[str],
    ) -> web.StreamResponse:
        """Open a prepared SSE StreamResponse with CORS + session headers (the CORS middleware
        can't inject headers after ``prepare()`` flushes them, so they are resolved here)."""
        sse_headers = self._sse_headers(request)
        if session_id:
            sse_headers["X-Hermes-Session-Id"] = session_id
        if gateway_session_key:
            sse_headers["X-Hermes-Session-Key"] = gateway_session_key
        response = web.StreamResponse(status=200, headers=sse_headers)
        await response.prepare(request)
        return response

    async def _write_sse_chat_completion(
        self, request: web.Request, completion_id: str, model: str,
        created: int, stream_q, agent_task, agent_ref=None, session_id: str | None = None,
        gateway_session_key: str | None = None) -> web.StreamResponse:
        """Stream ``chat.completion.chunk`` frames from the agent's delta queue. On client
        disconnect the agent is interrupted (stops LLM calls), then its task wrapper cancelled."""
        from gateway.platforms.api_server import (
            _abandon_agent_task, _chat_usage_payload, _resolve_media_to_data_urls, _sse_frame)
        response = await self._prepare_sse_response(request, session_id, gateway_session_key)

        def _chunk(delta: dict[str, Any], finish_reason=None, **extra) -> dict[str, Any]:
            return {"id": completion_id, "object": "chat.completion.chunk", "created": created,
                    "model": model,
                    "choices": [{"index": 0, "delta": delta, "finish_reason": finish_reason}], **extra}
        content_sent = False
        try:
            await response.write(_sse_frame(_chunk({"role": "assistant"})))
            async for delta in _iter_stream_items(stream_q, agent_task, response):
                if delta is None:
                    break
                if isinstance(delta, tuple) and len(delta) == 2 and delta[0] == "__tool_progress__":
                    # Custom event: tool lifecycle for frontends without markers in history.
                    if not self._tool_progress_events:
                        continue  # opted out for strict OpenAI clients (#12020)
                    await response.write(_sse_frame(delta[1], event="hermes.tool.progress"))
                elif isinstance(delta, tuple) and len(delta) == 2 and delta[0] == "__reasoning__":
                    # DeepSeek-style ``delta.reasoning_content`` (#99552), the field Open WebUI,
                    # opencode and the Vercel AI SDK render as a thinking block.
                    await response.write(_sse_frame(_chunk({"reasoning_content": delta[1]})))
                elif isinstance(delta, tuple) and len(delta) == 2 and delta[0] == "__status__":
                    await response.write(_sse_frame(delta[1], event="hermes.status"))
                elif isinstance(delta, tuple) and len(delta) == 2 and delta[0] == "__approval__":
                    await response.write(_sse_frame(delta[1], event="approval.request"))
                else:
                    if delta:
                        content_sent = True
                    await response.write(_sse_frame(_chunk({"content": delta})))
            # The agent can fail after the queue drains (task raises / result flagged failed or
            # partial): surface a non-"stop" finish_reason like the non-streaming path.
            usage = {"input_tokens": 0, "output_tokens": 0, "total_tokens": 0}
            result = agent_error = None
            try:
                result, agent_usage = await agent_task
                usage = agent_usage or usage
            except Exception as exc:
                agent_error = exc
                logger.error("Agent task %s failed during SSE streaming: %s", completion_id, exc)
            completed, is_partial, is_failed, err_msg = _result_flags(result)
            if agent_error is not None:
                is_failed = True
                err_msg = err_msg or str(agent_error)
            budget_exhausted = _budget_exhausted(result)
            finish_reason = _finish_reason(completed, is_partial, is_failed, err_msg, agent_error,
                                           budget_exhausted=budget_exhausted)
            finish_chunk = _chunk({}, finish_reason, usage=_chat_usage_payload(usage))
            presentation_muted = (
                (isinstance(result, dict) and result.get("_notification_presentation_suppressed") is True)
                or getattr(agent_error, "_notification_presentation_suppressed", False) is True
            )
            # Recovery paths (guardrail halt, partial_stream_recovery, fallback prior-turn
            # content) can return a final_response without firing any content delta; emit it
            # once so the client does not see an empty stream (#31449). Mirrors
            # _ResponsesStream.collect_result for /v1/responses.
            if not content_sent and not presentation_muted and isinstance(result, dict):
                fallback_text = _resolve_media_to_data_urls(result.get("final_response") or "")
                if fallback_text:
                    await response.write(_sse_frame(_chunk({"content": fallback_text})))
            elif not presentation_muted:
                # Chat chunks can only append: a non-append rewrite follows the streamed text (as in the CLI).
                tail, appended = _post_stream_transform(result)
                if tail:
                    await response.write(_sse_frame(_chunk({"content": tail if appended else _transformed_notice() + tail})))
            if finish_reason != "stop":
                if err_msg and not presentation_muted:
                    finish_chunk["error"] = {
                        "message": err_msg,
                        "type": type(agent_error).__name__ if agent_error else "agent_error"}
                finish_chunk["hermes"] = _hermes_extras(
                    completed, is_partial, is_failed, "" if presentation_muted else err_msg, finish_reason,
                    budget_exhausted=budget_exhausted)
            await response.write(_sse_frame(finish_chunk))
            await response.write(b"data: [DONE]\n\n")
        except (ConnectionResetError, ConnectionAbortedError, BrokenPipeError, OSError):
            await _abandon_agent_task(agent_ref, agent_task, "SSE client disconnected")
            logger.info("SSE client disconnected; interrupted agent task %s", completion_id)
        except Exception:
            # Agent crashed mid-stream: an error chunk beats a TransferEncodingError.
            import traceback as _tb
            logger.error("Agent crashed mid-stream for %s: %s", completion_id, _tb.format_exc()[:300])
            with suppress(Exception):
                await response.write(_sse_frame(_chunk({}, "error")))
                await response.write(b"data: [DONE]\n\n")
        return response

    async def _write_sse_responses(
        self, request: web.Request, response_id: str, model: str, created_at: int, stream_q,
        agent_task, agent_ref, conversation_history: list[dict[str, str]], user_message: str,
        instructions: Optional[str], conversation: Optional[str], store: bool, session_id: str,
        gateway_session_key: Optional[str] = None, durable_key=None, terminal_replay=False) -> web.StreamResponse:
        """Write the SSE stream for POST /v1/responses.

        Events: ``response.created`` -> ``output_text.delta/done`` + ``output_item.added/done``
        (reasoning / function_call / function_call_output) + ``reasoning_summary_part/text.*``
        -> ``response.completed`` (non-streaming envelope)
        or ``response.failed``. On disconnect the agent is interrupted and, with ``store=True``,
        an ``incomplete`` snapshot replaces ``in_progress`` so GET / chaining still work.
        ``durable_key`` (canonical Idempotency-Key record) is written only once the admitted turn
        settled, so an exact retry replays this stream instead of admitting different history.
        """
        from gateway.platforms.api_server import _abandon_agent_task, _redact_api_error_text
        response = await self._prepare_sse_response(request, session_id, gateway_session_key)
        st = _ResponsesStream(
            self, response, response_id=response_id, model=model, created_at=created_at,
            conversation_history=conversation_history, user_message=user_message,
            instructions=instructions, conversation=conversation, store=store, session_id=session_id,
            durable_key=durable_key, terminal_replay=terminal_replay)
        try:
            await st.emit_created()
            async for item in _iter_stream_items(stream_q, agent_task, response):
                if item is None:  # EOS sentinel
                    st.cancel_batch_timer()
                    await st.flush_batch()
                    break
                await st.dispatch(item)
            await st.flush_batch()
            await st.collect_result(agent_task)
            await st.close_message_item()
            if st.agent_error and not (durable_key and st.result):
                await st.emit_failed()
            else:
                await st.emit_completed()
        except (ConnectionResetError, ConnectionAbortedError, BrokenPipeError, OSError):
            st.persist_incomplete_if_needed()
            await _abandon_agent_task(agent_ref, agent_task, "SSE client disconnected")
            logger.info("SSE client disconnected; interrupted agent task %s", response_id)
        except asyncio.CancelledError:
            # Server-side cancellation (shutdown, timeout): persist incomplete, then re-raise.
            st.persist_incomplete_if_needed()
            await _abandon_agent_task(
                agent_ref, agent_task, "SSE task cancelled",
                reap_source="api_server_sse_cancelled", await_cancel=False)
            logger.info("SSE task cancelled; persisted incomplete snapshot for %s", response_id)
            raise
        except Exception as exc:
            # Unhandled agent error (BadRequestError, AuthenticationError, ...): emit
            # response.failed and end the stream cleanly (no TransferEncodingError).
            import traceback as _tb
            st.persist_incomplete_if_needed()
            st.agent_error = _redact_api_error_text(_tb.format_exc())
            with suppress(Exception):
                await st.emit_crash(exc)
            logger.error("Agent crashed mid-stream for %s: %s", response_id, str(st.agent_error)[:300])
        return response

    async def _replay_sse_responses(self, request: web.Request, replay: dict[str, Any]) -> web.StreamResponse:
        """Re-emit a settled idempotent record as SSE. A record made by the streaming path replays
        its exact frames; one made by the nonstreaming path gets ``created`` + its terminal event."""
        from gateway.platforms.api_server import _sse_frame
        headers = replay.get('headers') or {}
        response = await self._prepare_sse_response(
            request, headers.get('X-Hermes-Session-Id'), headers.get('X-Hermes-Session-Key'))
        events = replay.get('events')
        if not events:
            env = replay['response']
            created = {k: env[k] for k in ('id', 'object', 'created_at', 'model')}
            created.update(status='in_progress', output=[])
            events = [['response.created', {'type': 'response.created', 'response': created, 'sequence_number': 0}],
                      ['response.' + env['status'], {'type': 'response.' + env['status'], 'response': env,
                                                     'sequence_number': 1}]]
        for event_type, data in events:
            await response.write(_sse_frame(data, event=event_type))
        return response

    async def _responses_durable_replay(self, request: web.Request, body: dict[str, Any], stream: bool) -> tuple:
        """Canonical Idempotency-Key record for POST /v1/responses ->
        ``(durable_key, idempotency_scope, idempotency_key, replayed_response_or_None)``."""
        durable_key = idempotency_scope = None
        idempotency_key = request.headers.get('Idempotency-Key')
        canonical = getattr(self.gateway_runner, 'session_authority', None) is not None
        if canonical and idempotency_key:
            # An exact retry replays the committed response BEFORE the conversation name is
            # expanded: the first success already advanced the conversation, so re-expanding
            # would build a different admission payload and refuse the retry as a conflict.
            # Streaming and nonstreaming share the record: transport mode is not identity.
            from gateway.platforms.api_server import _make_request_fingerprint
            idempotency_scope = self._run_idempotency_scope(request)
            durable_key = (f'idem:{idempotency_scope}:{idempotency_key}',
                           _make_request_fingerprint(body, keys=_responses_fingerprint_keys(body)))
            from gateway.platforms.api_server import _openai_error
            from gateway.platforms.api_server_response_admissions import request_binding
            response_store = self._current_response_store()
            binding = request_binding(self, request, durable_key)
            if not response_store.request_key_matches(durable_key[0], binding):
                return durable_key, idempotency_scope, idempotency_key, web.json_response(
                    _openai_error('admission_conflict', code='admission_conflict'), status=409)
            replay = response_store.get(durable_key[0])
            if replay is not None:
                if replay.get('fingerprint') != durable_key[1]:
                    return durable_key, idempotency_scope, idempotency_key, web.json_response(
                        _openai_error('admission_conflict', code='admission_conflict'), status=409)
                if not _settled(replay):
                    # Never re-run a request whose earlier attempt may have executed.
                    return durable_key, idempotency_scope, idempotency_key, web.json_response(
                        _openai_error('unknown_execution', code='unknown_execution'), status=409)
                if stream:
                    return (durable_key, idempotency_scope, idempotency_key,
                            await self._replay_sse_responses(request, replay))
                return durable_key, idempotency_scope, idempotency_key, web.json_response(
                    replay['response'], headers=replay.get('headers') or {})
        return durable_key, idempotency_scope, idempotency_key, None

    async def _handle_responses(self, request: web.Request) -> web.Response:
        """POST /v1/responses — OpenAI Responses API format."""
        from gateway.platforms.api_server import (
            ThreadSafeAsyncQueue, _coerce_request_bool,
            _error_response, _invalid_request, _redact_api_error_text,
            _resolve_media_to_data_urls, _responses_usage_payload)
        # Bound total in-flight agent runs (configurable; #7483).
        limited = self._concurrency_limited_response()
        if limited is not None:
            return limited
        gateway_session_key, key_err = self._parse_session_key_header(request)
        if key_err is not None:
            return key_err
        try:
            body = await request.json()
        except Exception:
            return _invalid_request("Invalid JSON in request body")
        raw_input = body.get("input")
        if raw_input is None:
            return _error_response("Missing 'input' field", 400)
        instructions = body.get("instructions")
        previous_response_id = body.get("previous_response_id")
        conversation = body.get("conversation")
        store = _coerce_request_bool(body.get("store"), default=True)
        if conversation and previous_response_id:
            return _error_response("Cannot use both 'conversation' and 'previous_response_id'", 400)
        stream = _coerce_request_bool(body.get("stream"), default=False)
        durable_key, idempotency_scope, idempotency_key, replayed = await self._responses_durable_replay(
            request, body, stream)
        if replayed is not None:
            return replayed
        from gateway.platforms.api_server_response_admissions import prepare_context
        from hermes_state_runtime import RuntimeStoreError
        try:
            context, error = await prepare_context(
                self, request, body, gateway_session_key, durable_key, idempotency_scope, idempotency_key)
        except RuntimeStoreError as exc:
            from gateway.platforms.api_server import _openai_error
            from gateway.platforms.api_server_authority_runs import store_error_response
            return store_error_response(_openai_error, exc)
        if error is not None:
            return error
        session_id, user_message = context['session_id'], context['user_message']
        conversation_history, instructions = context['conversation_history'], context['instructions']
        run_kwargs, run_agent = context['run_kwargs'], context['run_agent']
        response_id, created_at = _response_identity(self._current_response_store(), durable_key)
        response_model = body.get('model', self._model_name)
        if durable_key:
            response_model = self._current_response_store().request_model(durable_key[0], response_model)
        if stream:
            _stream_q = ThreadSafeAsyncQueue()

            def _on_tool_progress(event_type, name, preview, args, **kwargs):
                return  # structured start/complete callbacks carry the call id; progress ignored

            def _on_tool_start(tool_call_id, function_name, function_args):
                _stream_q.put_threadsafe(("__tool_started__", {
                    "tool_call_id": tool_call_id, "name": function_name,
                    "arguments": function_args or {}}))

            def _on_tool_complete(tool_call_id, function_name, function_args, function_result):
                _stream_q.put_threadsafe(("__tool_completed__", {
                    "tool_call_id": tool_call_id, "name": function_name,
                    "arguments": function_args or {}, "result": function_result}))

            def _on_commentary(text, *, already_streamed: bool = False):
                # Already-streamed text went out as output_text.delta of the final item; a second
                # copy as a commentary item would duplicate it.
                if not already_streamed and isinstance(text, str) and text.strip():
                    _stream_q.put_threadsafe(("__commentary__", {"text": text}))
            agent_task, agent_ref = self._spawn_stream_agent(
                _stream_q, run_agent=run_agent, tool_progress_callback=_on_tool_progress,
                tool_start_callback=_on_tool_start, tool_complete_callback=_on_tool_complete,
                interim_assistant_callback=_on_commentary, **run_kwargs)
            return await self._write_sse_responses(
                request=request, response_id=response_id,
                model=response_model, created_at=created_at,
                stream_q=_stream_q, agent_task=agent_task, agent_ref=agent_ref,
                conversation_history=conversation_history, user_message=user_message,
                instructions=instructions, conversation=conversation, store=store,
                session_id=session_id, gateway_session_key=gateway_session_key,
                durable_key=durable_key, terminal_replay=context.get('terminal_replay', False))

        async def _compute_response():
            return await run_agent(**run_kwargs)
        outcome, err = await self._run_idempotent(
            request, body, _compute_response, log_label="responses",
            fingerprint_keys=list(_responses_fingerprint_keys(body)), route="responses",
        )
        if err is not None:
            return err
        if durable_key is not None:
            replay = self._current_response_store().get(durable_key[0])
            if _settled(replay):
                return web.json_response(replay['response'], headers=replay.get('headers') or {})
        result, usage = outcome
        final_response = _resolve_media_to_data_urls(result.get("final_response", ""))
        if not final_response:
            final_response = _redact_api_error_text(result.get("error", "(No response generated)"))
        full_history = self._build_response_conversation_history(
            conversation_history, user_message, result, final_response,
            tool_output_max_chars=self._history_tool_output_max_chars)
        # _run_agent's effective session id carries compression rotations; storing it keeps
        # previous_response_id chaining off the pre-rotation session (else compression re-fires).
        _result_sid = result.get("session_id") if isinstance(result, dict) else None
        _effective_session_id = (
            _result_sid if isinstance(_result_sid, str) and _result_sid else session_id)
        # Output items = current turn only (AIAgent returns a full transcript; mocked paths
        # only the current-turn suffix).
        output_start_index = self._response_messages_turn_start_index(
            conversation_history, user_message, result)
        response_data = {
            "id": response_id, "object": "response", "status": _response_status(result),
            "created_at": created_at, "model": response_model,
            "output": self._extract_output_items(result, start_index=output_start_index,
                **({'response_id': response_id} if durable_key else {})),
            "usage": _responses_usage_payload(usage)}
        if durable_key:
            response_data['output'] = self._current_response_store().order_output(durable_key[0], response_data['output'])
        if response_data['status'] == 'failed':
            response_data['error'] = {'code': 'agent_error', 'message': _redact_api_error_text(
                result.get('error') or 'The admitted turn failed.')}
        if store:
            response_store = self._current_response_store()
            advance_conversation = conversation and (not context.get('terminal_replay')
                or response_store.get_conversation(conversation) in (None, response_id))
            response_store.put(response_id, {
                "response": response_data, "conversation_history": full_history,
                "instructions": instructions, "session_id": _effective_session_id})
            if advance_conversation:
                response_store.set_conversation(conversation, response_id)
        response_headers = {"X-Hermes-Session-Id": _effective_session_id}
        if gateway_session_key:
            response_headers["X-Hermes-Session-Key"] = gateway_session_key
        if durable_key is not None:
            self._current_response_store().put(durable_key[0], {
                'fingerprint': durable_key[1], 'response': response_data, 'headers': response_headers})
        return web.json_response(response_data, headers=response_headers)

    async def _handle_get_response(self, request: web.Request) -> web.Response:
        """GET /v1/responses/{response_id} — retrieve a stored response."""
        from gateway.platforms.api_server import _error_response
        auth_err = self._check_auth(request)
        if auth_err:
            return auth_err
        response_id = request.match_info["response_id"]
        stored = self._current_response_store().get(response_id)
        if stored is None:
            return _error_response(f"Response not found: {response_id}", 404)
        return web.json_response(stored["response"])

    async def _handle_delete_response(self, request: web.Request) -> web.Response:
        """DELETE /v1/responses/{response_id} — delete a stored response."""
        from gateway.platforms.api_server import _error_response
        auth_err = self._check_auth(request)
        if auth_err:
            return auth_err
        response_id = request.match_info["response_id"]
        if not self._current_response_store().delete(response_id):
            return _error_response(f"Response not found: {response_id}", 404)
        return web.json_response({"id": response_id, "object": "response", "deleted": True})

    @staticmethod
    def _build_response_conversation_history(
        conversation_history: list[dict[str, Any]], user_message: Any, result: dict[str, Any],
        final_response: Any, *, tool_output_max_chars: int = 0) -> list[dict[str, Any]]:
        """Build the stored Responses transcript without duplicating history.

        A compressed transcript (``result["_compressed"]``) shares no input-history prefix, so
        turn-start detection fails; prepending the uncompressed history would bloat the stored
        context and re-trigger compression every request — it is stored as-is instead.

        ``tool_output_max_chars`` > 0 caps tool outputs / tool-call argument blobs in the
        stored copy (gateway.api_server.history_tool_output_max_chars; 0 = store verbatim).
        """
        from gateway.platforms.api_server import APIServerAdapter
        prior = list(conversation_history)
        current_user = {"role": "user", "content": user_message}
        if isinstance(result, dict) and isinstance(result.get("_turn_user_row"), dict):
            current_user = dict(result["_turn_user_row"])  # settled canonical turn: carries api_content
        agent_messages = result.get("messages") if isinstance(result, dict) else None
        if isinstance(agent_messages, list) and agent_messages:
            turn_start = APIServerAdapter._response_messages_turn_start_index(
                conversation_history, user_message, result)
            # turn_start == 0: compression rewrote the transcript or agent_messages is turn-only.
            if turn_start or result.get("_compressed"):
                history = list(agent_messages)
            else:
                history = prior + [current_user] + agent_messages
        else:
            history = prior + [current_user, {"role": "assistant", "content": final_response}]
        return _cap_history_tool_outputs(history, tool_output_max_chars)

    @staticmethod
    def _response_messages_turn_start_index(
        conversation_history: list[dict[str, Any]], user_message: Any, result: dict[str, Any],
    ) -> int:
        """Index where this turn starts in a transcript-shaped result["messages"] (0 = all)."""
        from gateway.platforms.api_server_turn_boundary import response_turn_start_index
        return response_turn_start_index(conversation_history, user_message, result)

    @classmethod
    def _turn_transcript_messages(
        cls, conversation_history: list[dict[str, Any]], user_message: Any, result: dict[str, Any],
    ) -> list[dict[str, Any]]:
        """This turn's assistant/tool messages in client-safe shape: clients accumulating
        ``assistant.delta`` into one buffer cannot reconstruct assistant segments that preceded
        tool calls, so ``run.completed`` carries the authoritative per-turn transcript.

        Emitting the authoritative per-turn transcript on ``run.completed`` lets any SSE consumer reconcile
        its live view against ground truth without a separate ``GET /messages`` round-trip. Purely additive:
        clients that ignore the field are unaffected. Refs #34703.
        """
        agent_messages = result.get("messages") if isinstance(result, dict) else None
        if not isinstance(agent_messages, list) or not agent_messages:
            return []
        start = cls._response_messages_turn_start_index(conversation_history, user_message, result)
        out: list[dict[str, Any]] = []
        for msg in agent_messages[start:]:
            if not isinstance(msg, dict) or msg.get("role") not in {"assistant", "tool"}:
                continue
            # _message_response projects compaction scaffolding; pure handoffs are "hidden".
            projected = cls._message_response(msg)
            if projected.get("display_kind") != "hidden":
                out.append(projected)
        return out

    @staticmethod
    def _extract_output_items(result: dict[str, Any], start_index: int = 0, *, response_id=None) -> list[dict[str, Any]]:
        """Output items from ``result["messages"][start_index:]``: ``function_call`` per assistant
        tool_call, ``function_call_output`` per tool message, then the final ``message``."""
        from gateway.platforms.api_server import _redact_api_error_text
        items: list[dict[str, Any]] = []
        messages = result.get("messages", [])
        if start_index > 0:
            messages = messages[start_index:]
        for msg in messages:
            role = msg.get("role")
            reasoning = msg.get("reasoning") if role == "assistant" else None
            if isinstance(reasoning, str) and reasoning.strip():
                # Precedes this message's function_call items, like the SSE writer closes a
                # thinking burst before the next tool item opens (#99552).
                items.append(_reasoning_item(reasoning))
            if role == "assistant" and msg.get("tool_calls"):
                for tc in msg["tool_calls"]:
                    func = tc.get("function", {})
                    # Already executed server-side; replayed for structured tool UI only, so
                    # marked completed (matching the SSE path) — never pending client calls.
                    items.append({
                        "id": f"fc_{uuid.uuid4().hex[:24]}", "type": "function_call",
                        "status": "completed", "name": func.get("name", ""),
                        "arguments": func.get("arguments", ""),
                        "call_id": tc.get("id", "")})
            elif role == "tool":
                items.append({
                    "id": f"fco_{uuid.uuid4().hex[:24]}", "type": "function_call_output",
                    "status": "completed", "call_id": msg.get("tool_call_id", ""),
                    "output": msg.get("content", "")})
        final = result.get("final_response", "") or _redact_api_error_text(
            result.get("error", "(No response generated)"))
        items.append(_message_item(final))
        if response_id is not None:
            from gateway.platforms.api_server_response_identity import identify_output
            return identify_output(items, response_id)
        return items