ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 10 Packages Projects Releases Wiki Activity

refactor(gateway/platforms): api_server_openai_routes — shared terminal_envelope, tuple assigns, compact docstrings/comments

Browse Source
This commit is contained in:
Teknium
2026-09-02 22:15:40 -07:00
parent 73a91014df
commit b676f0e9de
1 changed files with 109 additions and 184 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway/platforms/api_server_openai_routes.py
+109 -184
View File
@@ -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