diff --git a/gateway/platforms/api_server_openai_routes.py b/gateway/platforms/api_server_openai_routes.py index 045af0b306..9dbaa819d6 100644 --- a/gateway/platforms/api_server_openai_routes.py +++ b/gateway/platforms/api_server_openai_routes.py @@ -1,10 +1,9 @@ """OpenAI-compatible routes for the API server adapter. -``OpenAICompatRoutesMixin`` carries ``/v1/chat/completions``, ``/v1/responses`` -(+ GET/DELETE), their two SSE writers, and the Responses-transcript helpers. -``APIServerAdapter`` inherits it. api_server-internal helpers are imported lazily -inside each method (the origin imports this module, so a top-level import would be -a cycle), which also keeps ``patch("gateway.platforms.api_server.X")`` effective. +``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 @@ -25,11 +24,10 @@ except ImportError: # pragma: no cover - mirrors api_server's optional import logger = logging.getLogger("gateway.platforms.api_server") async def _iter_stream_items(stream_q, agent_task, response): - """Yield agent stream items until end-of-stream, writing SSE keepalives while idle. + """Yield agent stream items until EOS, writing SSE keepalives while idle. - Woken directly by ``put_threadsafe`` (no executor hop / poll latency). 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. + 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 @@ -89,10 +87,8 @@ def _message_item(text: Any) -> Dict[str, Any]: 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. - """ + """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: @@ -120,9 +116,8 @@ def _trim_tool_items(items: List[Dict[str, Any]]) -> List[Dict[str, Any]]: class _ResponsesStream: """Per-request state and event emitters for the POST /v1/responses SSE writer. - Every event carries a monotonically increasing ``sequence_number`` (canonical - Responses SSE schema). Text deltas are batched (50ms) to reduce Open WebUI - re-render storms; tool events flush the buffer first. + 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, @@ -130,17 +125,10 @@ class _ResponsesStream: instructions: Optional[str], conversation: Optional[str], store: bool, session_id: str): from gateway.platforms import api_server as api self._api = api - self.adapter = adapter - self.response = response - self.response_id = response_id - self.model = model - self.created_at = created_at - self.conversation_history = conversation_history - self.user_message = user_message - self.instructions = instructions - self.conversation = conversation - self.store = store - self.session_id = session_id + 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.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) @@ -169,6 +157,15 @@ class _ResponsesStream: 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"] = {"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}] @@ -184,11 +181,8 @@ class _ResponsesStream: self.adapter._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. - - Used by the disconnect and cancellation paths so GET /v1/responses/{id} and - ``previous_response_id`` chaining keep working after abrupt termination. - """ + """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 @@ -197,10 +191,7 @@ class _ResponsesStream: if text: items.append(_message_item(text)) history.append({"role": "assistant", "content": text}) - env = self.envelope("incomplete") - env["output"] = items - env["usage"] = self._api._responses_usage_payload(self.usage) - self.persist_snapshot(env, history=history) + self.persist_snapshot(self.terminal_envelope("incomplete", items), history=history) async def emit_created(self) -> None: env = self.envelope("in_progress") @@ -250,14 +241,10 @@ class _ResponsesStream: 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 = None - if call_id: - for i, p in enumerate(self.pending_tool_calls): - if p["call_id"] == call_id: - pending = self.pending_tool_calls.pop(i) - break - if pending is None: + 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"]} @@ -357,10 +344,7 @@ class _ResponsesStream: async def emit_failed(self) -> None: redact = self._api._redact_api_error_text - env = self.envelope("failed") - env["output"] = self._final_items() - env["error"] = {"message": redact(self.agent_error), "type": "server_error"} - env["usage"] = self._api._responses_usage_payload(self.usage) + 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)}) @@ -369,15 +353,12 @@ class _ResponsesStream: await self.write_event("response.failed", {"type": "response.failed", "response": env}) async def emit_completed(self) -> None: - env = self.envelope("completed") - env["output"] = self._final_items() - env["usage"] = self._api._responses_usage_payload(self.usage) + env = self.terminal_envelope("completed", self._final_items()) result = self.result full_history = self.adapter._build_response_conversation_history( self.conversation_history, self.user_message, result, self.final_response_text) - # Compression-aware transcript substitution happens in _build_response_conversation_history - # (result["_compressed"]); only a compression-rotated session_id is propagated here so - # previous_response_id chaining resumes the child session. + # 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) @@ -386,11 +367,8 @@ class _ResponsesStream: "response.completed", {"type": "response.completed", "response": env}) async def emit_crash(self, exc: BaseException) -> None: - env = self.envelope("failed") - env["output"] = list(self.emitted_items) - env["error"] = {"message": self._api._redact_api_error_text(exc, limit=500), - "type": "server_error"} - env["usage"] = self._api._responses_usage_payload(self.usage) + 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}) @@ -399,12 +377,8 @@ class OpenAICompatRoutesMixin: 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. - - Returns ``(route, agent_overrides, error_response_or_None)``. A matching alias - makes this request's agent use that route's model/provider instead of the - global default. - """ + """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( @@ -416,12 +390,9 @@ class OpenAICompatRoutesMixin: return route, overrides, (_error_response(err, 400) if err else None) def _spawn_stream_agent(self, stream_q, **run_kwargs) -> tuple: - """Start ``_run_agent`` for an SSE writer; returns ``(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()``. - """ + """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()``.""" 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. @@ -452,8 +423,8 @@ class OpenAICompatRoutesMixin: return _invalid_request("Missing or invalid 'messages' field") stream = _coerce_request_bool(body.get("stream"), default=False) - # System messages become the ephemeral system prompt layered ON TOP of core; they are - # flattened to text (Anthropic rejects images there, OpenAI text models ignore them). + # System messages -> ephemeral system prompt layered ON TOP of core, flattened to text + # (Anthropic rejects images there, OpenAI text models ignore them). system_prompt = None conversation_messages: List[Dict[str, str]] = [] for idx, msg in enumerate(messages): @@ -468,11 +439,8 @@ class OpenAICompatRoutesMixin: except ValueError as exc: return _multimodal_validation_error(exc, param=f"messages[{idx}].content") conversation_messages.append({"role": role, "content": content}) - user_message: Any = "" - history = [] - if conversation_messages: - user_message = conversation_messages[-1].get("content", "") - history = conversation_messages[:-1] + 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") @@ -481,9 +449,8 @@ class OpenAICompatRoutesMixin: 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, otherwise any client could read arbitrary - # history by guessing session ids. + # 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: @@ -493,8 +460,7 @@ class OpenAICompatRoutesMixin: "session continuity.") return _error_response("Session continuation requires API key authentication. " "Configure API_SERVER_KEY to enable this feature.", 403) - # Same entry-boundary guard as the native gateway (gateway.session._is_path_unsafe): - # ids are interpolated into on-disk artifact filenames. + # 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") @@ -509,8 +475,8 @@ class OpenAICompatRoutesMixin: logger.warning("Failed to load session history for %s: %s", session_id, e) history = [] else: - # Stable id from the conversation fingerprint (system prompt + first user message are - # constant across turns) so Open WebUI-style clients map onto one Hermes session. + # Stable id from the conversation fingerprint so Open WebUI-style clients map onto + # one Hermes session. 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) @@ -528,13 +494,12 @@ class OpenAICompatRoutesMixin: gateway_session_key=gateway_session_key, **agent_overrides, route=route) if stream: _stream_q = ThreadSafeAsyncQueue() - # tool_call_ids with an emitted "running" event: a "completed" without a matching - # "running" (internal/filtered tools) is dropped rather than orphaned on the wire. + # 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`` (legacy tool/emoji/label payload + - toolCallId/status correlation). Names starting with ``_`` stay off the wire.""" + """``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) @@ -551,8 +516,8 @@ class OpenAICompatRoutesMixin: _stream_q.put_threadsafe(("__tool_progress__", { "tool": function_name, "toolCallId": tool_call_id, "status": "completed"})) - # tool_progress_callback is deliberately NOT wired: it fires alongside the structured - # start/complete callbacks (which carry the tool_call id) and would duplicate emits. + # tool_progress_callback deliberately NOT wired: it would duplicate the structured + # start/complete callbacks (which carry the tool_call id). agent_task, agent_ref = self._spawn_stream_agent( _stream_q, tool_start_callback=_on_tool_start, tool_complete_callback=_on_tool_complete, **run_kwargs) @@ -578,8 +543,8 @@ class OpenAICompatRoutesMixin: response_headers = {"X-Hermes-Session-Id": 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) → 5xx with an OpenAI error envelope so - # SDK clients raise instead of rendering the failure string as message.content. + # 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( err_msg or "Agent run did not produce a response.", err_type="server_error", @@ -591,9 +556,7 @@ class OpenAICompatRoutesMixin: 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, + "id": completion_id, "object": "chat.completion", "created": created, "model": model_name, "choices": [{"index": 0, "message": {"role": "assistant", "content": final_response}, "finish_reason": finish_reason}], @@ -610,10 +573,8 @@ class OpenAICompatRoutesMixin: async def _run_idempotent( self, request: "web.Request", body: Dict[str, Any], compute, *, log_label: str, fingerprint_keys: List[str]) -> tuple: - """Run ``compute()`` once per Idempotency-Key + body fingerprint. - - Returns ``((result, usage), None)`` or ``(None, 500 response)``. - """ + """Run ``compute()`` once per Idempotency-Key + body fingerprint -> + ``((result, usage), None)`` or ``(None, 500 response)``.""" from gateway.platforms.api_server import ( _error_response, _idem_cache, _make_request_fingerprint) idempotency_key = request.headers.get("Idempotency-Key") @@ -631,19 +592,13 @@ class OpenAICompatRoutesMixin: 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. - - CORS middleware can't inject headers after ``prepare()`` flushes them, - so they are resolved up front here. - """ + """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 = { - "Content-Type": "text/event-stream", - "Cache-Control": "no-cache", - "X-Accel-Buffering": "no"} + "Content-Type": "text/event-stream", "Cache-Control": "no-cache", "X-Accel-Buffering": "no"} origin = request.headers.get("Origin", "") - cors = self._cors_headers_for_origin(origin) if origin else None - if cors: - sse_headers.update(cors) + if origin: + sse_headers.update(self._cors_headers_for_origin(origin) or {}) if session_id: sse_headers["X-Hermes-Session-Id"] = session_id if gateway_session_key: @@ -656,11 +611,8 @@ class OpenAICompatRoutesMixin: self, request: "web.Request", completion_id: str, model: str, created: int, stream_q, agent_task, agent_ref=None, session_id: str = None, gateway_session_key: str = None) -> "web.StreamResponse": - """Stream ``chat.completion.chunk`` frames from the agent's delta queue. - - On client disconnect the agent is interrupted (``agent.interrupt()``) so it - stops making LLM calls, then the asyncio task wrapper is cancelled. - """ + """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, _sse_frame) response = await self._prepare_sse_response(request, session_id, gateway_session_key) @@ -675,13 +627,12 @@ class OpenAICompatRoutesMixin: if delta is None: break if isinstance(delta, tuple) and len(delta) == 2 and delta[0] == "__tool_progress__": - # Custom event so frontends can render tool lifecycle without the - # markers landing in conversation history. + # Custom event: tool lifecycle for frontends without markers in history. await response.write(_sse_frame(delta[1], event="hermes.tool.progress")) else: await response.write(_sse_frame(_chunk({"content": delta}))) - # The agent can fail after the queue drains cleanly (task raises, or result flagged - # failed/partial); surface it as a non-"stop" finish_reason like the non-streaming path. + # 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: @@ -709,8 +660,7 @@ class OpenAICompatRoutesMixin: 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: emit an error chunk so the client gets a proper - # response instead of a TransferEncodingError from incomplete chunked encoding. + # 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): @@ -723,14 +673,12 @@ class OpenAICompatRoutesMixin: 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) -> "web.StreamResponse": - """Write the SSE stream for POST /v1/responses (OpenAI Responses API). + """Write the SSE stream for POST /v1/responses. - Events: ``response.created`` → ``response.output_text.delta/done`` and - ``response.output_item.added/done`` (function_call / function_call_output) - → ``response.completed`` (same envelope as non-streaming) or ``response.failed``. - On disconnect the agent is interrupted and, when ``store=True``, an - ``incomplete`` snapshot replaces the ``in_progress`` one so GET / - ``previous_response_id`` chaining still work. + Events: ``response.created`` -> ``output_text.delta/done`` + ``output_item.added/done`` + (function_call / function_call_output) -> ``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. """ 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) @@ -758,8 +706,7 @@ class OpenAICompatRoutesMixin: 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, request timeout): persist the incomplete - # snapshot, then re-raise so the runtime's cancellation semantics are respected. + # 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", @@ -767,9 +714,8 @@ class OpenAICompatRoutesMixin: logger.info("SSE task cancelled; persisted incomplete snapshot for %s", response_id) raise except Exception as exc: - # Unhandled agent error (e.g. BadRequestError, AuthenticationError): emit - # response.failed and terminate the stream cleanly so the client doesn't get a - # TransferEncodingError from incomplete chunked encoding. + # 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()) @@ -824,8 +770,7 @@ class OpenAICompatRoutesMixin: else: return _error_response("'input' must be a string or array", 400) - # Explicit conversation_history (stateless clients) takes precedence over - # previous_response_id chaining. + # Explicit conversation_history (stateless clients) beats previous_response_id chaining. conversation_history: List[Dict[str, Any]] = [] raw_history = body.get("conversation_history") if raw_history: @@ -859,8 +804,8 @@ class OpenAICompatRoutesMixin: conversation_history = _auto_truncate_response_history(conversation_history) # Session precedence: previous_response_id chain > declared X-Hermes-Session-Key > fresh - # id. Binding the declared key is gated on that same precedence — a chain-selected - # session must not have its routing key rewritten to this header. + # id. Binding the declared key follows the same precedence: a chain-selected session must + # not have its routing key rewritten to this header. _declared_selected = not stored_session_id and bool(gateway_session_key) session_id = ( stored_session_id @@ -881,8 +826,7 @@ class OpenAICompatRoutesMixin: _stream_q = ThreadSafeAsyncQueue() def _on_tool_progress(event_type, name, preview, args, **kwargs): - # Structured start/complete callbacks carry the call id; progress events ignored. - return + 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__", { @@ -921,31 +865,24 @@ class OpenAICompatRoutesMixin: created_at = int(time.time()) full_history = self._build_response_conversation_history( conversation_history, user_message, result, final_response) - # The effective session id from _run_agent carries compression-triggered rotations; - # storing it keeps previous_response_id chaining off the pre-rotation session (which - # would re-trigger compression on every request). - _effective_session_id = session_id + # _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 - if isinstance(_result_sid, str) and _result_sid: - _effective_session_id = _result_sid - # Output items come from the current turn only: AIAgent returns a full transcript in - # result["messages"], older/mocked paths only the current-turn suffix. + _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": "completed", - "created_at": created_at, - "model": body.get("model", self._model_name), + "id": response_id, "object": "response", "status": "completed", + "created_at": created_at, "model": body.get("model", self._model_name), "output": self._extract_output_items(result, start_index=output_start_index), "usage": _responses_usage_payload(usage)} if store: self._response_store.put(response_id, { - "response": response_data, - "conversation_history": full_history, - "instructions": instructions, - "session_id": _effective_session_id}) + "response": response_data, "conversation_history": full_history, + "instructions": instructions, "session_id": _effective_session_id}) if conversation: self._response_store.set_conversation(conversation, response_id) response_headers = {"X-Hermes-Session-Id": _effective_session_id} @@ -982,10 +919,9 @@ class OpenAICompatRoutesMixin: final_response: Any) -> List[Dict[str, Any]]: """Build the stored Responses transcript without duplicating history. - A compressed transcript (``result["_compressed"]``) does not share the input - history prefix, so turn-start detection fails; concatenating the uncompressed - history on front would bloat the stored context and re-trigger compression - every request — the compressed transcript is stored as-is instead. + 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. """ from gateway.platforms.api_server import APIServerAdapter prior = list(conversation_history) @@ -994,8 +930,7 @@ class OpenAICompatRoutesMixin: 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 (use as-is) or - # agent_messages is only the current turn. + # turn_start == 0: compression rewrote the transcript or agent_messages is turn-only. if turn_start or result.get("_compressed"): return list(agent_messages) return prior + [current_user] + agent_messages @@ -1021,13 +956,9 @@ class OpenAICompatRoutesMixin: def _turn_transcript_messages( cls, conversation_history: List[Dict[str, Any]], user_message: Any, result: Dict[str, Any], ) -> List[Dict[str, Any]]: - """Return this turn's assistant/tool messages in client-safe shape. - - Clients accumulating ``assistant.delta`` events into one buffer cannot - reconstruct intermediate assistant segments that preceded tool calls; the - authoritative per-turn transcript on ``run.completed`` lets them reconcile - without a separate ``GET /messages`` round-trip. Purely additive. - """ + """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.""" agent_messages = result.get("messages") if isinstance(result, dict) else None if not isinstance(agent_messages, list) or not agent_messages: return [] @@ -1036,8 +967,7 @@ class OpenAICompatRoutesMixin: 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 itself and marks pure - # handoffs display_kind == "hidden" (avoids re-running the content classifier). + # _message_response projects compaction scaffolding; pure handoffs are "hidden". projected = cls._message_response(msg) if projected.get("display_kind") != "hidden": out.append(projected) @@ -1057,24 +987,19 @@ class OpenAICompatRoutesMixin: 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", - # Already executed server-side; replayed for structured tool UI only, so - # marked completed (matching the SSE path) — never pending client calls. - "status": "completed", - "name": func.get("name", ""), + "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", ""), + "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", "") - if not final: - final = _redact_api_error_text(result.get("error", "(No response generated)")) + final = result.get("final_response", "") or _redact_api_error_text( + result.get("error", "(No response generated)")) items.append(_message_item(final)) return items