Files
odysseus/src/llm_core.py
T
c4369305f0 refactor(model-routing): centralize explicit foreground fallback policy (#6020)
* refactor(model-routing): centralize explicit foreground fallback policy

Make foreground fallback an explicit per-user, availability-only policy shared by streaming Chat, non-stream Chat, and Agent runs.

Preserve strict defaults, owner/model and credential boundaries, pinned Agent routes, and truthful per-round provenance/accounting. Carry provider-reported model identifiers through native streaming adapters, non-stream responses, and caches, and keep legacy default_model_fallbacks as tombstoned raw storage that generic settings APIs and agent tools cannot expose or mutate.

* fix(agent-loop): restore rebase-dropped qwen routing, workspace prompt, and temperature clamp

* fix(model-routing): thread selected endpoint identity, fix cost classification and fallback eligibility

* fix(chat): restore stream helpers and harden run stop lifecycle

* fix(model-routing): let numeric provider codes win over symbolic rate-limit statuses

* fix(agent-loop): apply qwen temperature and notes-tool clamps per fallback candidate

* fix(chat): honor queued stop across resend and reload canonical terminal on EOF

* fix(chat): track stop queue and cleanup ownership by per-send generation

* fix(agent-loop): preserve requested temperature for non-qwen fallback candidates

* fix(chat): reserve send ownership before any await and scope stop to the current send

* fix(chat): clear the previous run identity at send reservation

---------

Co-authored-by: RaresKeY <158580472+RaresKeY@users.noreply.github.com>
Co-authored-by: StressTestor <212606152+StressTestor@users.noreply.github.com>
2026-08-14 08:10:30 +01:00

3731 lines
168 KiB
Python

# src/llm_core.py
import httpx
import asyncio
import copy
import time
import json
import logging
import hashlib
import threading
import re
import os
import math
from contextlib import asynccontextmanager
from fastapi import HTTPException
from typing import Optional, Dict, List, Tuple
from src.model_context import get_context_length, DEFAULT_CONTEXT, is_local_endpoint
from urllib.parse import urlparse
logger = logging.getLogger(__name__)
_LOCAL_MODEL_LOCK = asyncio.Lock()
_LOCAL_MODEL_WAITING_FOREGROUND = 0
_LOCAL_MODEL_CURRENT: Dict[str, object] = {}
def _normalize_usage_counts(input_value=0, output_value=0):
"""Return safe integer token counts, or ``None`` for malformed usage."""
def _count(value):
if isinstance(value, bool) or not isinstance(value, (int, float)):
return None
if isinstance(value, int):
count = value
else:
if not math.isfinite(value) or not value.is_integer():
return None
count = int(value)
if count < 0 or count > (2**63 - 1):
return None
return count
input_tokens = _count(input_value)
output_tokens = _count(output_value)
if input_tokens is None or output_tokens is None:
return None
return {
"input_tokens": input_tokens,
"output_tokens": output_tokens,
}
def _normalize_http_status(value) -> Optional[int]:
"""Accept only genuine three-digit integral HTTP status values."""
if isinstance(value, bool) or value is None:
return None
if isinstance(value, int):
status = value
elif isinstance(value, float):
if not math.isfinite(value) or not value.is_integer():
return None
status = int(value)
elif isinstance(value, str):
text = value.strip()
if not re.fullmatch(r"\d{3}", text):
return None
status = int(text)
else:
return None
return status if 100 <= status <= 599 else None
def _local_model_gate_enabled() -> bool:
return os.getenv("ODYSSEUS_LOCAL_MODEL_GATE", "true").lower() not in {"0", "false", "no", "off"}
def _gate_workload(workload: Optional[str]) -> str:
return "background" if str(workload or "").lower() == "background" else "foreground"
@asynccontextmanager
async def _local_model_slot(target_url: str, model: str, workload: Optional[str] = None):
"""Serialize local model traffic, with foreground chat taking priority.
Most local servers expose one GPU/CPU generation pipe even when their HTTP
API accepts multiple requests. Letting scheduled email/tasks and foreground
chat hit that pipe together creates the user-visible "streams crossed" and
"prompt waited behind a task" failure mode. Cloud providers are left alone.
"""
if not _local_model_gate_enabled() or not is_local_endpoint(target_url):
yield
return
global _LOCAL_MODEL_WAITING_FOREGROUND
kind = _gate_workload(workload)
current_task = asyncio.current_task()
if kind == "foreground":
_LOCAL_MODEL_WAITING_FOREGROUND += 1
current = dict(_LOCAL_MODEL_CURRENT)
if current.get("workload") == "background":
task = current.get("task")
if isinstance(task, asyncio.Task) and not task.done():
logger.info(
"[model-gate] cancelling background local model call for foreground request model=%s",
model,
)
task.cancel()
else:
# Background work should not jump in while the browser/chat is active
# or while a foreground request is waiting to acquire the local model.
try:
from src.interactive_gate import has_foreground_activity
except Exception:
has_foreground_activity = lambda: False # type: ignore
while _LOCAL_MODEL_WAITING_FOREGROUND > 0 or has_foreground_activity():
await asyncio.sleep(0.25)
acquired = False
try:
await _LOCAL_MODEL_LOCK.acquire()
acquired = True
if kind == "foreground":
_LOCAL_MODEL_WAITING_FOREGROUND = max(0, _LOCAL_MODEL_WAITING_FOREGROUND - 1)
_LOCAL_MODEL_CURRENT.clear()
_LOCAL_MODEL_CURRENT.update({
"task": current_task,
"workload": kind,
"url": target_url,
"model": model,
"started": time.time(),
})
yield
finally:
if kind == "foreground":
_LOCAL_MODEL_WAITING_FOREGROUND = max(0, _LOCAL_MODEL_WAITING_FOREGROUND - 1)
if acquired and _LOCAL_MODEL_LOCK.locked():
owner = _LOCAL_MODEL_CURRENT.get("task")
if owner is current_task:
_LOCAL_MODEL_CURRENT.clear()
_LOCAL_MODEL_LOCK.release()
class LLMConfig:
"""Configuration constants for LLM operations."""
DEFAULT_TIMEOUT = 30
DEFAULT_TEMPERATURE = 1.0
DEFAULT_MAX_TOKENS = 0
MAX_RETRIES = 3
RETRY_DELAY = 0.5
STREAM_TIMEOUT = 300
# TCP+TLS connect budget for a SINGLE attempt. The old hard-coded 3.0s
# assumed LAN/Tailscale peers ('SYN in <100ms'); it is too tight for public
# cloud endpoints (offshore APIs take ~0.5-1.5s cold, with jitter), so a
# brief blip on the first connect of an idle chat surfaced as a 503 on the
# streaming path (which, unlike llm_call, does not retry the connect). A
# genuinely dead upstream stays bounded by the dead-host cooldown. Override
# with env LLM_CONNECT_TIMEOUT (seconds).
CONNECT_TIMEOUT = float(os.getenv('LLM_CONNECT_TIMEOUT', '10') or '10')
class _FallbackIneligibleHTTPException(HTTPException):
"""HTTP-shaped provider failure that must never advance a route chain."""
fallback_eligible = False
def _call_timeout(read_timeout) -> httpx.Timeout:
"""Per-request timeout for non-streaming LLM calls (connect from config)."""
return httpx.Timeout(connect=LLMConfig.CONNECT_TIMEOUT, read=float(read_timeout), write=10.0, pool=5.0)
def _stream_timeout(read_timeout) -> httpx.Timeout:
"""Per-request timeout for streaming LLM calls (connect from config)."""
return httpx.Timeout(connect=LLMConfig.CONNECT_TIMEOUT, read=float(read_timeout), write=30.0, pool=5.0)
# Cache for LLM responses
def _cache_header_identity(headers) -> str:
"""Return a non-secret identity for credential-distinct request routes."""
if isinstance(headers, str):
try:
headers = json.loads(headers)
except (TypeError, ValueError, json.JSONDecodeError):
headers = {"_raw": headers}
if not isinstance(headers, dict):
headers = {}
canonical = [
(str(key).strip().lower(), str(value))
for key, value in headers.items()
]
canonical.sort()
encoded = json.dumps(canonical, ensure_ascii=False, separators=(",", ":"))
return hashlib.sha256(encoded.encode()).hexdigest()
def _get_cache_key(url: str, model: str, messages: List[Dict],
temperature: float, max_tokens: int, headers=None) -> str:
"""Generate a cache key partitioned by endpoint and credential identity."""
hashable_messages = []
for msg in messages:
sorted_items = tuple(sorted(msg.items()))
hashable_messages.append(sorted_items)
content = json.dumps({
'url': url,
'model': model,
'messages': hashable_messages,
'temp': temperature,
'max_tokens': max_tokens,
# Never put credentials in a cache key or loggable cache payload. The
# digest only prevents responses from one configured account/route
# being returned under another route with the same URL and model.
'header_identity': _cache_header_identity(headers),
}, sort_keys=True)
return hashlib.sha256(content.encode()).hexdigest()
_response_cache = {}
_response_model_cache = {}
# Dead-host cooldown: maps host (scheme://host:port) -> unix ts when cooldown expires.
# When a connect to a host fails, we mark it dead for DEAD_HOST_COOLDOWN seconds so
# subsequent calls fail instantly instead of waiting on the connect timeout. Keeps
# one unreachable upstream from jamming chat across the rest of the app.
#
# But a SINGLE transient blip (local model briefly busy, a momentary
# Tailscale hiccup) used to trip a full 60s lockout — the user saw a
# 503 and thought the model died when it was fine a second later. So:
# - require FAIL_THRESHOLD consecutive failures before cooling
# - shorter cooldown so recovery is quick
# - any success resets the failure counter immediately
DEAD_HOST_COOLDOWN = 20.0
_HOST_FAIL_THRESHOLD = 2
_dead_hosts: Dict[str, float] = {}
_host_fails: Dict[str, int] = {}
# Guards the two maps above. The synchronous llm_call() runs inside FastAPI's
# threadpool (sync routes such as /sessions/auto-sort) while llm_call_async()
# runs on the event loop, so these maps are mutated from multiple OS threads.
# Without the lock the get()+1+set on _host_fails is a read-modify-write that
# loses failure counts under concurrent connect errors (issue #659).
_host_health_lock = threading.Lock()
_model_activity: Dict[str, float] = {}
_HARMONY_MARKER_RE = re.compile(
r"<\|channel\|>(analysis|commentary|final)"
r"|<\|start\|>(?:assistant|system|user|tool)?"
r"|<\|message\|>"
r"|<\|end\|>"
r"|<\|return\|>"
r"|<\|call\|>"
)
_HARMONY_MARKERS = (
"<|channel|>analysis",
"<|channel|>commentary",
"<|channel|>final",
"<|start|>assistant",
"<|start|>system",
"<|start|>user",
"<|start|>tool",
"<|start|>",
"<|message|>",
"<|end|>",
"<|return|>",
"<|call|>",
)
_HARMONY_MAX_MARKER_LEN = max(len(marker) for marker in _HARMONY_MARKERS)
_VISIBLE_CHAT_TEMPLATE_ARTIFACT_RE = re.compile(
r"(?:\|end\|)+\|?assistan(?:t)?\|?"
r"|\|assistan(?:t)?\|"
r"|<\|im_start\|>\s*assistant"
r"|<\|im_end\|>",
re.IGNORECASE,
)
def _strip_visible_chat_template_artifacts(text: str) -> str:
return _VISIBLE_CHAT_TEMPLATE_ARTIFACT_RE.sub("", text or "")
def _harmony_suffix_hold_len(text: str) -> int:
"""Return how many trailing chars could be the start of a harmony marker."""
limit = min(len(text), _HARMONY_MAX_MARKER_LEN - 1)
for n in range(limit, 0, -1):
suffix = text[-n:]
if any(marker.startswith(suffix) for marker in _HARMONY_MARKERS):
return n
return 0
class _HarmonyStreamRouter:
"""Route OpenAI harmony analysis/final channels without leaking markers."""
def __init__(self) -> None:
self._buf = ""
self._seen_harmony = False
self._channel: Optional[str] = None
self._in_message = False
def feed(self, text: str) -> List[Tuple[str, bool]]:
if not text:
return []
self._buf += text
return self._drain(final=False)
def flush(self) -> List[Tuple[str, bool]]:
return self._drain(final=True)
def _append_text(self, out: List[Tuple[str, bool]], text: str) -> None:
if not text:
return
if not self._seen_harmony:
out.append((text, False))
return
if self._in_message:
# analysis + commentary (tool-call preambles / function-arg bodies)
# are internal, not user-facing — route them to thinking so they
# don't leak into the visible answer; only `final` is visible.
out.append((text, self._channel in ("analysis", "commentary")))
def _handle_marker(self, match: re.Match[str]) -> None:
marker = match.group(0)
self._seen_harmony = True
if marker.startswith("<|channel|>"):
self._channel = match.group(1)
self._in_message = False
elif marker == "<|message|>":
self._in_message = True
else:
self._in_message = False
if marker in {"<|end|>", "<|return|>", "<|call|>"}:
self._channel = None
def _drain(self, *, final: bool) -> List[Tuple[str, bool]]:
out: List[Tuple[str, bool]] = []
while True:
match = _HARMONY_MARKER_RE.search(self._buf)
if not match:
break
self._append_text(out, self._buf[:match.start()])
self._handle_marker(match)
self._buf = self._buf[match.end():]
hold = 0 if final else _harmony_suffix_hold_len(self._buf)
emit = self._buf if hold == 0 else self._buf[:-hold]
self._buf = "" if hold == 0 else self._buf[-hold:]
self._append_text(out, emit)
return out
def _stream_delta_event(text: str, *, thinking: bool = False) -> str:
payload = {"delta": text}
if thinking:
payload["thinking"] = True
return f"data: {json.dumps(payload)}\n\n"
_DEGENERATE_WORD_RE = re.compile(r"[A-Za-z0-9_\u0370-\u03ff\u0400-\u04ff]+")
class _DegenerateStreamGuard:
"""Detect local-model token collapse before it floods the UI.
Some self-hosted models fail by repeating one token forever ("Var Var Var",
"Summer Summer ..."). This is not a useful response and can burn context,
browser memory, and GPU time. Keep the guard conservative: only fire on long
same-token runs or a very dominant repeated token in the recent window.
"""
def __init__(self, model: str):
self.model = model or "model"
self.last_token = ""
self.same_run = 0
self.recent_tokens: List[str] = []
self.total_chars = 0
def check(self, text: str) -> Optional[str]:
if not text:
return None
self.total_chars += len(text)
tokens = [t.lower() for t in _DEGENERATE_WORD_RE.findall(text) if len(t) >= 2]
if not tokens:
return None
for token in tokens:
if token == self.last_token:
self.same_run += 1
else:
self.last_token = token
self.same_run = 1
self.recent_tokens.append(token)
if len(self.recent_tokens) > 96:
self.recent_tokens = self.recent_tokens[-96:]
reason = None
if self.same_run >= 28 and self.total_chars >= 100:
reason = f"repeated '{self.last_token}' {self.same_run} times"
elif len(self.recent_tokens) >= 72:
top = max(set(self.recent_tokens), key=self.recent_tokens.count)
count = self.recent_tokens.count(top)
if count >= 60 and count / max(len(self.recent_tokens), 1) >= 0.78:
reason = f"repeated '{top}' {count}/{len(self.recent_tokens)} recent tokens"
if not reason and len(self.recent_tokens) >= 80:
# Phrase loops are common on some local quantized MLX/MoE models:
# "Also be a software developer mode?" repeated forever will not
# trip the single-token guard above, but it is still a wedged
# generation. Require many repeats of the same 4-gram so normal
# prose/list formatting is not interrupted.
grams = [tuple(self.recent_tokens[i:i + 4]) for i in range(0, len(self.recent_tokens) - 3)]
if grams:
top_gram = max(set(grams), key=grams.count)
gram_count = grams.count(top_gram)
if gram_count >= 10:
reason = f"repeated phrase '{' '.join(top_gram)}' {gram_count} times"
if not reason:
return None
logger.warning("[degenerate-stream] aborting model=%s reason=%s", self.model, reason)
message = (
f"Stopped generation: {self.model} started repeating tokens "
f"({reason}). Try a different model or lower temperature."
)
return f'event: error\ndata: {json.dumps({"status": 502, "text": message, "error": message, "fallback_eligible": False})}\n\n'
def _model_activity_key(url: str, model: str) -> str:
return f"{(url or '').strip()}|{(model or '').strip()}"
def _same_model_identity(left: str, right: str) -> bool:
return (left or "").strip().lower() == (right or "").strip().lower()
def _reported_model_name(value) -> str:
"""Return a provider model identifier only when it is usable metadata."""
return value.strip() if isinstance(value, str) and value.strip() else ""
def _model_actual_event(requested_model: str, reported_model) -> Optional[str]:
"""Build a provenance event when a provider resolves a different model."""
actual_model = _reported_model_name(reported_model)
if not actual_model or _same_model_identity(actual_model, requested_model):
return None
return f'data: {json.dumps({"type": "model_actual", "requested_model": requested_model, "model": actual_model})}\n\n'
def _annotate_usage_model(usage: dict, requested_model: str, actual_model: str) -> dict:
"""Attach provider model provenance to a normalized usage payload."""
actual_model = _reported_model_name(actual_model)
if actual_model:
usage["model"] = actual_model
if not _same_model_identity(actual_model, requested_model):
usage["requested_model"] = requested_model
return usage
def note_model_activity(url: str, model: str):
"""Record that a real upstream request used this endpoint/model."""
if not url or not model:
return
_model_activity[_model_activity_key(url, model)] = time.time()
def seconds_since_model_activity(url: str, model: str) -> Optional[float]:
"""Seconds since the endpoint/model was last used in this process."""
ts = _model_activity.get(_model_activity_key(url, model))
if not ts:
return None
return max(0.0, time.time() - ts)
def _host_key(url: str) -> str:
from urllib.parse import urlsplit
s = urlsplit(url)
return f"{s.scheme}://{s.netloc}" if s.scheme and s.netloc else url
def _is_host_dead(url: str) -> bool:
key = _host_key(url)
with _host_health_lock:
exp = _dead_hosts.get(key)
if exp is None:
return False
if time.time() >= exp:
_dead_hosts.pop(key, None)
return False
return True
def _mark_host_dead(url: str) -> bool:
"""Record a connect failure. Only actually cools the host after
_HOST_FAIL_THRESHOLD consecutive failures. Returns True if the host
is now cooled (so callers can log accurately), False if it's still
within its allowed-failure grace."""
key = _host_key(url)
with _host_health_lock:
n = _host_fails.get(key, 0) + 1
_host_fails[key] = n
if n >= _HOST_FAIL_THRESHOLD:
_dead_hosts[key] = time.time() + DEAD_HOST_COOLDOWN
return True
return False
def _clear_host_dead(url: str) -> None:
key = _host_key(url)
with _host_health_lock:
_dead_hosts.pop(key, None)
_host_fails.pop(key, None)
# Shared async HTTP client. Reusing one client keeps connections warm:
# repeat calls to api.anthropic.com / api.openai.com / openrouter skip the
# 100-500ms TCP+TLS handshake. Lazy init so we bind to the running event loop.
_http_client: Optional[httpx.AsyncClient] = None
_http_limits = httpx.Limits(max_connections=100, max_keepalive_connections=30, keepalive_expiry=30.0)
def _get_http_client() -> httpx.AsyncClient:
"""Return process-wide AsyncClient. Per-request timeout is passed at call time."""
global _http_client
if _http_client is None or _http_client.is_closed:
from src.tls_overrides import llm_verify
_http_client = httpx.AsyncClient(
limits=_http_limits, http2=False, verify=llm_verify(),
)
return _http_client
def _get_cached_response(cache_key: str) -> Optional[str]:
"""Get cached response if it exists."""
return _response_cache.get(cache_key)
def _get_cached_response_model(cache_key: str) -> Optional[str]:
"""Return provider-reported model metadata paired with a cached reply."""
model = _response_model_cache.get(cache_key)
return model if isinstance(model, str) and model.strip() else None
def _set_cached_response(
cache_key: str,
response: str,
*,
actual_model: Optional[str] = None,
) -> None:
"""Store response in cache."""
if len(_response_cache) > 128:
keys_to_remove = list(_response_cache.keys())[:64]
for key in keys_to_remove:
# pop(), not del: another thread (sync llm_call runs in FastAPI's
# threadpool) may have already evicted the same snapshotted key,
# and del would raise KeyError mid-eviction (issue #659).
_response_cache.pop(key, None)
_response_model_cache.pop(key, None)
_response_cache[cache_key] = response
if isinstance(actual_model, str) and actual_model.strip():
_response_model_cache[cache_key] = actual_model.strip()
else:
_response_model_cache.pop(cache_key, None)
# ── Anthropic native API adapter ──
ANTHROPIC_MODELS = [
"claude-opus-4-20250514", "claude-opus-4",
"claude-sonnet-4-20250514", "claude-sonnet-4", "claude-sonnet-4-5-20250929", "claude-sonnet-4-5",
"claude-haiku-4-20250514", "claude-haiku-4", "claude-haiku-3-5-20241022", "claude-haiku-3-5",
]
def _is_ollama_native_url(url: str) -> bool:
"""Return True for native Ollama API URLs, including Ollama Cloud."""
try:
parsed = urlparse(url or "")
except Exception as e:
logger.warning("Failed to parse URL for Ollama detection", exc_info=e)
return False
host = parsed.hostname or ""
path = (parsed.path or "").rstrip("/")
if _host_match(url, "ollama.com"):
return True
if path.startswith("/v1"):
return False
local_ollama_host = host in {"localhost", "127.0.0.1", "0.0.0.0", "::1"} or parsed.port == 11434
return local_ollama_host and (path == "" or path == "/api" or path.startswith("/api/"))
def _is_ollama_openai_compat_url(url: str) -> bool:
"""Return True for local Ollama's OpenAI-compatible /v1 surface.
Mirrors the host detection used by ``_is_ollama_native_url`` so that the
two helpers stay in lockstep: a localhost Ollama on a non-default port
(custom ``OLLAMA_HOST``, reverse proxy, container port remap) is treated
the same way here as it is on the native ``/api`` path.
"""
try:
parsed = urlparse(url or "")
except Exception:
return False
host = parsed.hostname or ""
path = (parsed.path or "").rstrip("/")
local_ollama_host = host in {"localhost", "127.0.0.1", "0.0.0.0", "::1"} or parsed.port == 11434
return local_ollama_host and (path == "/v1" or path.startswith("/v1/"))
def _ollama_api_root(url: str) -> str:
"""Return a native Ollama API root such as https://ollama.com/api."""
url = (url or "").strip().rstrip("/")
parsed = urlparse(url)
path = (parsed.path or "").rstrip("/")
if path.endswith("/api/chat"):
return url[: -len("/chat")]
if path.endswith("/api/tags"):
return url[: -len("/tags")]
if path.endswith("/api/generate"):
return url[: -len("/generate")]
if path.endswith("/api"):
return url
if path == "":
return url + "/api"
if _host_match(url, "ollama.com"):
root = f"{parsed.scheme}://{parsed.netloc}" if parsed.scheme and parsed.netloc else "https://ollama.com"
return root.rstrip("/") + "/api"
return url
def _normalize_ollama_url(url: str) -> str:
"""Ensure a native Ollama URL points at /api/chat."""
base = _ollama_api_root(url)
return base.rstrip("/") + "/chat"
def _normalize_openai_chat_url(url: str) -> str:
"""Ensure an OpenAI-compatible base URL points at /chat/completions."""
base = (url or "").strip().rstrip("/")
if not base:
return base
if base.endswith("/chat/completions") or base.endswith("/completions"):
return base
if base.endswith("/models"):
base = base[: -len("/models")].rstrip("/")
return base + "/chat/completions"
def _ollama_normalize_messages(messages: List[Dict]) -> List[Dict]:
"""Adapt Odysseus' canonical OpenAI-style messages to native Ollama /api/chat.
Two shape mismatches silently break requests:
1. Tool calls: Odysseus carries `function.arguments` as a JSON *string*.
Native Ollama expects a JSON *object* and rejects the string form with
HTTP 400 ("Value looks like object, but can't find closing '}' symbol"),
aborting every follow-up (tool-result) round. Parse the arguments back
into an object here, on a shallow copy, leaving non-tool messages
untouched. The opaque Gemini `extra_content` (thought_signature) is
dropped — it is meaningless to Ollama and only matters when the
conversation is replayed to Gemini.
2. Images (issue #4723): Odysseus carries multimodal user content as an
OpenAI-style list ``[{type: "text", ...}, {type: "image_url",
image_url: {url: "data:image/...;base64,XXX"}}, ...]``. Native Ollama
does not accept a list for ``content`` — it wants ``content`` as a
string plus a separate ``images`` array of raw base64 strings (no
``data:`` prefix). Without this conversion the image blocks pass
through untouched, the vision-capable model never sees the picture,
and the user gets "I can't see any image" even though the request
succeeded.
"""
out: List[Dict] = []
for m in messages or []:
if not isinstance(m, dict):
out.append(m)
continue
nm = dict(m)
# 1. Tool-call argument strings -> objects.
tcs = nm.get("tool_calls")
if tcs:
new_calls = []
for tc in tcs:
fn = tc.get("function") or {}
args = fn.get("arguments")
if isinstance(args, str):
try:
args = json.loads(args) if args.strip() else {}
except (json.JSONDecodeError, TypeError):
args = {}
call: Dict = {"function": {"name": fn.get("name", ""), "arguments": args or {}}}
if tc.get("id"):
call["id"] = tc["id"]
new_calls.append(call)
nm["tool_calls"] = new_calls
# 2. Multimodal content list -> native content string + images array.
content = nm.get("content")
if isinstance(content, list):
text_parts: List[str] = []
images: List[str] = list(nm.get("images") or [])
for block in content:
if not isinstance(block, dict):
continue
btype = block.get("type")
if btype == "text":
t = block.get("text")
if t:
text_parts.append(str(t))
elif btype == "image_url":
url = (block.get("image_url") or {}).get("url", "")
if not url:
continue
if url.startswith("data:"):
# Strip the ``data:[...];base64,`` prefix — native
# Ollama wants only the base64 bytes.
_, _, b64 = url.partition(",")
if b64:
images.append(b64)
else:
# Native Ollama images[] is base64-only; it does
# not fetch HTTP URLs. Skip unsupported schemes
# rather than sending a non-base64 string that the
# model silently ignores.
logger.warning(
"Skipping non-data image_url (Ollama images[] "
"requires base64): %s",
url[:80],
)
nm["content"] = "\n".join(text_parts).strip()
if images:
nm["images"] = images
out.append(nm)
return out
# Backward-compatible alias for callers/tests that imported the older name
# (it only handled tool messages originally — issue #4723 broadened scope).
_ollama_normalize_tool_messages = _ollama_normalize_messages
def _build_ollama_payload(
model: str,
messages: List[Dict],
temperature: float,
max_tokens: int,
stream: bool = False,
tools: Optional[List[Dict]] = None,
num_ctx: Optional[int] = None,
) -> Dict:
"""Build the JSON payload for Ollama's /api/chat endpoint.
``num_ctx`` sets the input context window. Ollama defaults to 2048
when the option is omitted, so a model with a larger advertised
window is silently truncated there, and a model with a smaller one
gets an oversized window it can't service. Pass the discovered
context length through ``num_ctx``; this builder only emits it when
the value is trusted (not the ``DEFAULT_CONTEXT`` fallback), so we
don't guess for unknown models but do tell Ollama the real window
when we know it — even if it's smaller than 2048.
"""
payload: Dict = {
"model": model,
"messages": _ollama_normalize_messages(messages),
"stream": stream,
}
options: Dict = {}
if temperature is not None:
options["temperature"] = temperature
if max_tokens and max_tokens > 0:
options["num_predict"] = max_tokens
if num_ctx is not None and num_ctx > 0 and num_ctx != DEFAULT_CONTEXT:
options["num_ctx"] = num_ctx
if options:
payload["options"] = options
if tools:
payload["tools"] = _alias_harmony_tools(tools, model)
return payload
def _parse_ollama_response(data: dict) -> str:
message = data.get("message") or {}
return message.get("content") or data.get("response") or ""
def _host_match(url: str, *domains: str) -> bool:
"""Return True if url's hostname equals any of `domains` or is a subdomain of one.
Used by helpers that want "is this Anthropic?" / "is this OpenRouter?"
style checks. Prefer this over substring matching on the URL: the
substring form gives wrong answers for unrelated paths or query strings
that happen to contain the domain text.
"""
if not url:
return False
try:
# rstrip(".") so a fully-qualified host with a trailing dot
# ("api.anthropic.com.") still matches "anthropic.com".
host = (urlparse(url).hostname or "").lower().rstrip(".")
except Exception:
return False
if not host:
return False
return any(host == d or host.endswith("." + d) for d in domains)
# Kimi Code subscription keys (api.kimi.com/coding/v1) require a whitelisted
# coding-agent User-Agent; otherwise the API returns 403 access_terminated_error.
# Tried in order; first success is cached per base URL for later requests.
KIMI_CODE_USER_AGENTS: tuple[str, ...] = (
"claude-code/0.1.0",
"claude-code/1.0.0",
"KimiCLI/1.0",
"Kilo-Code/1.0",
"Roo-Code/1.0",
"Cursor/1.0",
)
KIMI_CODE_USER_AGENT = KIMI_CODE_USER_AGENTS[0]
_kimi_code_ua_cache: dict[str, str] = {}
def _is_kimi_code_url(url: str) -> bool:
if not url or not _host_match(url, "kimi.com"):
return False
try:
return "/coding" in (urlparse(url).path or "")
except Exception:
return False
def _kimi_code_base_key(url: str) -> str:
"""Normalize a Kimi Code chat/models URL to its OpenAI base (.../coding/v1)."""
parsed = urlparse(url)
path = (parsed.path or "").rstrip("/")
for suffix in ("/chat/completions", "/models", "/completions"):
if path.endswith(suffix):
path = path[: -len(suffix)]
path = path.rstrip("/") or "/coding/v1"
return f"{parsed.scheme}://{parsed.netloc}{path}"
def _is_kimi_code_access_denied(status: int, body: bytes | str) -> bool:
if status != 403:
return False
text = body.decode("utf-8", errors="replace") if isinstance(body, bytes) else (body or "")
lower = text.lower()
return (
"access_terminated_error" in lower
or "coding agents" in lower
or "only available for coding" in lower
)
def _kimi_code_ua_candidates(url: str) -> list[str]:
if not _is_kimi_code_url(url):
return []
base_key = _kimi_code_base_key(url)
cached = _kimi_code_ua_cache.get(base_key)
if cached:
return [cached] + [ua for ua in KIMI_CODE_USER_AGENTS if ua != cached]
return list(KIMI_CODE_USER_AGENTS)
def _remember_kimi_code_user_agent(url: str, user_agent: str) -> None:
_kimi_code_ua_cache[_kimi_code_base_key(url)] = user_agent
def apply_kimi_code_headers(headers: Optional[Dict], url: str) -> Dict[str, str]:
"""Pick a Kimi Code User-Agent (cached probe when possible)."""
h = dict(headers or {})
if not _is_kimi_code_url(url):
return h
base_key = _kimi_code_base_key(url)
cached = _kimi_code_ua_cache.get(base_key)
if cached:
h["User-Agent"] = cached
return h
models_url = base_key.rstrip("/") + "/models"
from src.tls_overrides import llm_verify
for ua in KIMI_CODE_USER_AGENTS:
trial = dict(h)
trial["User-Agent"] = ua
try:
r = httpx.get(models_url, headers=trial, timeout=8, verify=llm_verify())
except Exception:
continue
if _is_kimi_code_access_denied(r.status_code, r.content):
logger.debug("Kimi Code rejected User-Agent %s (403), trying next", ua)
continue
if r.status_code < 400:
_remember_kimi_code_user_agent(url, ua)
h["User-Agent"] = ua
return h
break
h.setdefault("User-Agent", KIMI_CODE_USER_AGENT)
return h
async def apply_kimi_code_headers_async(client, headers: Optional[Dict], url: str) -> Dict[str, str]:
"""Pick a Kimi Code User-Agent without blocking the event loop."""
h = dict(headers or {})
if not _is_kimi_code_url(url):
return h
base_key = _kimi_code_base_key(url)
cached = _kimi_code_ua_cache.get(base_key)
if cached:
h["User-Agent"] = cached
return h
models_url = base_key.rstrip("/") + "/models"
for ua in KIMI_CODE_USER_AGENTS:
trial = dict(h)
trial["User-Agent"] = ua
try:
r = await client.get(models_url, headers=trial, timeout=8)
except Exception:
continue
if _is_kimi_code_access_denied(r.status_code, r.content):
logger.debug("Kimi Code rejected User-Agent %s (403), trying next", ua)
continue
if r.status_code < 400:
_remember_kimi_code_user_agent(url, ua)
h["User-Agent"] = ua
return h
break
h.setdefault("User-Agent", KIMI_CODE_USER_AGENT)
return h
def httpx_get_kimi_aware(url: str, headers: Optional[Dict], **kwargs):
h = apply_kimi_code_headers(headers, url)
if not _is_kimi_code_url(url):
return httpx.get(url, headers=h, **kwargs)
last = None
for ua in _kimi_code_ua_candidates(url):
trial = dict(h)
trial["User-Agent"] = ua
last = httpx.get(url, headers=trial, **kwargs)
if not _is_kimi_code_access_denied(last.status_code, last.content):
if last.status_code < 400:
_remember_kimi_code_user_agent(url, ua)
return last
return last
def httpx_post_kimi_aware(url: str, headers: Optional[Dict], **kwargs):
h = apply_kimi_code_headers(headers, url)
if not _is_kimi_code_url(url):
return httpx.post(url, headers=h, **kwargs)
last = None
for ua in _kimi_code_ua_candidates(url):
trial = dict(h)
trial["User-Agent"] = ua
last = httpx.post(url, headers=trial, **kwargs)
if not _is_kimi_code_access_denied(last.status_code, last.content):
if last.status_code < 400:
_remember_kimi_code_user_agent(url, ua)
return last
return last
async def httpx_post_kimi_aware_async(client, url: str, headers: Optional[Dict], **kwargs):
h = await apply_kimi_code_headers_async(client, headers, url)
if not _is_kimi_code_url(url):
return await client.post(url, headers=h, **kwargs)
last = None
for ua in _kimi_code_ua_candidates(url):
trial = dict(h)
trial["User-Agent"] = ua
last = await client.post(url, headers=trial, **kwargs)
if not _is_kimi_code_access_denied(last.status_code, last.content):
if last.status_code < 400:
_remember_kimi_code_user_agent(url, ua)
return last
return last
def _detect_provider(url: str) -> str:
"""Detect the API provider from a configured endpoint URL.
Matches on hostname (exact or subdomain) rather than substring, so a URL
that merely contains a provider's domain in its path or query — or a
look-alike host such as ``anthropic.com.example`` — is not misclassified.
Unknown hosts fall back to the OpenAI-compatible default, which the
majority of providers implement.
"""
if _is_ollama_native_url(url):
return "ollama"
if _host_match(url, "anthropic.com"):
return "anthropic"
if _host_match(url, "opencode.ai/zen/go"):
return "opencode-go"
if _host_match(url, "opencode.ai/zen"):
return "opencode-zen"
if _host_match(url, "openrouter.ai"):
return "openrouter"
if _host_match(url, "groq.com"):
return "groq"
if _host_match(url, "nvidia.com"):
return "nvidia"
if _host_match(url, "moonshot.ai") or _host_match(url, "moonshot.cn"):
return "moonshot"
from src.chatgpt_subscription import is_chatgpt_subscription_base
if is_chatgpt_subscription_base(url):
return "chatgpt-subscription"
from src.copilot import is_copilot_base
if is_copilot_base(url):
return "copilot"
if _host_match(url, "cerebras.ai"):
return "cerebras"
if _host_match(url, "mistral.ai"):
return "mistral"
return "openai"
def _is_self_hosted_openai_compatible(url: str) -> bool:
"""True for custom/local OpenAI-compatible servers (llama.cpp, LM Studio,
vLLM, text-generation-webui, etc.) as opposed to cloud APIs.
Used to gate llama.cpp-server-specific payload extras (``session_id``,
``cache_prompt``) used for KV-cache slot affinity (issue #2927). Strict
cloud providers reject unrecognized top-level fields (api.openai.com
returns 400, Mistral returns 422 "extra_forbidden", issue #3793), and any
unknown OpenAI-compatible host used to be treated as self-hosted, so those
fields leaked to every strict provider added as a custom endpoint.
A server only counts as self-hosted when it also resolves as local:
loopback/private/tailscale host, or the endpoint explicitly configured
with kind "local". A self-hosted server exposed via a public hostname
loses the affinity hint unless its endpoint kind is set to "local" -
a lost perf hint, versus a hard 4xx on every request the other way.
"""
if _detect_provider(url) != "openai" or _host_match(url, "openai.com"):
return False
from src.model_context import is_local_endpoint
return is_local_endpoint(url)
def _apply_local_cache_affinity(payload: Dict, url: str, session_id: Optional[str]) -> None:
"""Add llama.cpp-server slot-affinity hints to an outgoing payload, in place.
As diagnosed in issue #2927, llama.cpp assigns requests to processing
slots via LRU when no stable identifier is present ("session_id=<empty>
server-selected (LCP/LRU)"), which means consecutive turns of the same
chat can land on different slots and lose their cached prefix entirely.
Sending a stable ``session_id`` (derived from the Odysseus session) lets
the server keep routing the same conversation to the same slot, and
``cache_prompt: true`` asks it to retain/reuse the prefix it already has.
Both fields are llama.cpp / LM Studio extensions to the OpenAI schema; we
only set them for self-hosted OpenAI-compatible endpoints (never
api.openai.com or other cloud providers, which reject unrecognized
top-level request fields).
"""
if not session_id:
return
if not _is_self_hosted_openai_compatible(url):
return
payload.setdefault("session_id", str(session_id))
payload.setdefault("cache_prompt", True)
def _is_local_minimax_mlx_request(url: str, model: str) -> bool:
"""Local MLX MiniMax-family endpoints need conservative sampling defaults.
The OpenAI-compatible MLX server accepts repetition/frequency penalties.
Some large quantized MiniMax/MoE ports otherwise fall into visible reasoning
loops ("Also be...", "No.", etc.) even for trivial prompts.
"""
if not model:
return False
m = model.lower()
if "minimax" not in m and "mini-max" not in m:
return False
try:
from src.model_context import is_local_endpoint
return is_local_endpoint(url)
except Exception:
return False
def _apply_local_generation_stability(payload: Dict, url: str, model: str) -> None:
if not _is_local_minimax_mlx_request(url, model):
return
if "temperature" in payload:
try:
# MiniMax MLX quantized ports are very sensitive to chat/agent
# harness size. Character presets can ask for a warmer voice, but
# local MiniMax needs a final compatibility clamp or trivial
# prompts can fall into visible reasoning/repetition loops.
payload["temperature"] = min(float(payload.get("temperature") or 0.2), 0.2)
except (TypeError, ValueError):
payload["temperature"] = 0.2
payload.setdefault("top_p", 0.9)
payload.setdefault("top_k", 20)
payload.setdefault("repetition_penalty", 1.12)
payload.setdefault("repetition_context_size", 256)
payload.setdefault("frequency_penalty", 0.08)
payload.setdefault("frequency_context_size", 256)
payload.setdefault("presence_penalty", 0.02)
payload.setdefault("presence_context_size", 256)
payload.setdefault("stop", ["<|im_end|>", "<|endoftext|>", "</s>"])
# A max_tokens of 0 means "server default/unbounded" for many local
# endpoints. Keep simple chats from running forever when the model loops.
if not payload.get("max_tokens") and not payload.get("max_completion_tokens"):
payload["max_tokens"] = 2048
def _provider_headers(provider: str, headers: Optional[Dict] = None) -> Dict[str, str]:
h = {"Content-Type": "application/json"}
if isinstance(headers, dict):
h.update(headers)
if provider == "openrouter":
h.setdefault("HTTP-Referer", "https://github.com/odysseus-dev/odysseus")
h.setdefault("X-OpenRouter-Title", "Odysseus")
if provider == "copilot":
# Ensure the Copilot-required headers are present even when the caller
# didn't pass pre-built headers (e.g. model listing). build_headers()
# already injects these for the live chat path; setdefault keeps any
# request-specific values (x-initiator/vision) the caller set.
from src.copilot import copilot_headers
for k, v in copilot_headers(None).items():
h.setdefault(k, v)
return h
def _provider_label(url: str) -> str:
"""Human-friendly provider name for error messages."""
if not url:
return "provider"
if _host_match(url, "anthropic.com"): return "Anthropic"
if _host_match(url, "ollama.com"): return "Ollama Cloud"
if _host_match(url, "x.ai"): return "xAI"
if _host_match(url, "openai.com"): return "OpenAI"
if _host_match(url, "openrouter.ai"): return "OpenRouter"
if _host_match(url, "opencode.ai/zen/go"): return "OpenCode Go"
if _host_match(url, "opencode.ai/zen"): return "OpenCode Zen"
if _host_match(url, "groq.com"): return "Groq"
from src.chatgpt_subscription import is_chatgpt_subscription_base
if is_chatgpt_subscription_base(url): return "ChatGPT Subscription"
from src.copilot import is_copilot_base
if is_copilot_base(url): return "GitHub Copilot"
if _host_match(url, "cerebras.ai"):
return "cerebras"
if _host_match(url, "mistral.ai"): return "Mistral"
if _host_match(url, "deepseek.com"): return "DeepSeek"
if _host_match(url, "nvidia.com"): return "NVIDIA"
if _host_match(url, "googleapis.com"): return "Google"
if _host_match(url, "together.xyz", "together.ai"): return "Together"
if _host_match(url, "fireworks.ai"): return "Fireworks"
if _host_match(url, "kimi.com"):
try:
if "/coding" in (urlparse(url).path or ""):
return "Kimi Code"
except Exception:
pass
if _is_ollama_native_url(url): return "Ollama"
try:
_parsed_local = urlparse(url)
host = (_parsed_local.hostname or "").lower()
port = _parsed_local.port
except Exception:
return "provider"
if host in {"localhost", "127.0.0.1", "::1", "0.0.0.0"}:
# A port alone is not authoritative: vLLM, SGLang, llama.cpp and plain
# OpenAI-compatible servers all routinely share 8000/8080, so naming the
# serving tool from the port here would mislabel real setups. The tool is
# identified by probing llama-server's native /props endpoint during
# discovery (see ModelDiscovery._fingerprint_provider); this stays neutral.
return "local endpoint"
return host or "provider"
def _is_openai_hosted_chat_url(url: str) -> bool:
try:
parsed = urlparse(url or "")
except Exception:
return False
path = (parsed.path or "").rstrip("/")
return _host_match(url, "openai.com") and path.endswith("/chat/completions")
def _model_disallows_reasoning_effort_with_chat_tools(model: str) -> bool:
"""OpenAI GPT 5.x variants reject reasoning_effort + tools on chat completions."""
m = (model or "").strip().lower()
return bool(re.match(r"^(?:openai/)?gpt-5(?:[.\-]\d+)?(?:[-_:].*)?$", m))
# gpt-oss (harmony) ships BUILT-IN tools named `python` and `browser`, invoked
# with the raw body as the argument (`to=python` + bare source), while custom
# functions use `to=functions.NAME` + JSON. A tool we expose under a built-in's
# name therefore gets called with the built-in convention: the model emits raw
# code, the server tries to parse it as JSON, and the whole request dies
# ("error parsing tool call: raw='import sys, ...'"). In streaming mode Ollama
# does not even report it — it truncates the stream, so the turn looks like an
# empty response. `bash` collides the same way in practice.
#
# Measured on gpt-oss:20b via Ollama /v1 with a fixed agentic prompt:
# tools named python+bash ............ 2/6 succeeded (4 parse failures)
# python renamed ..................... 5/6
# python and bash renamed ............ 6/6
#
# So rename the colliding tools on the way out and map the names back on the
# way in. Confined to the transport layer: callers keep using the real names.
_HARMONY_TOOL_ALIASES = {
"python": "run_python_code",
"bash": "run_shell_command",
"browser": "web_browser_tool",
}
_HARMONY_TOOL_ALIASES_REVERSE = {v: k for k, v in _HARMONY_TOOL_ALIASES.items()}
def _is_harmony_model(model: str) -> bool:
"""True for gpt-oss / harmony-format models, which have built-in tool names."""
return "gpt-oss" in (model or "").lower()
def _alias_harmony_tools(tools: Optional[List[Dict]], model: str) -> Optional[List[Dict]]:
"""Rename tools that collide with harmony built-ins. Returns a copy."""
if not tools or not _is_harmony_model(model):
return tools
out = []
for t in tools:
fn = t.get("function") or {}
alias = _HARMONY_TOOL_ALIASES.get(fn.get("name"))
if alias:
t = copy.deepcopy(t)
t["function"]["name"] = alias
out.append(t)
return out
def _unalias_harmony_tool_name(name: str, model: str) -> str:
"""Map an aliased tool name in a model response back to the real name."""
if not _is_harmony_model(model):
return name
return _HARMONY_TOOL_ALIASES_REVERSE.get(name, name)
def _scrub_openai_chat_tool_reasoning(payload: Dict, target_url: str, model: str) -> None:
if not payload.get("tools"):
return
if not _is_openai_hosted_chat_url(target_url):
return
if not _model_disallows_reasoning_effort_with_chat_tools(model):
return
payload["reasoning_effort"] = "none"
def _normalize_chatgpt_subscription_url(url: str) -> str:
base = (url or "").strip().rstrip("/")
if base.endswith("/responses"):
return base
return base + "/responses"
def _message_content_as_text(content) -> str:
if isinstance(content, str):
return content
if isinstance(content, list):
parts: list[str] = []
for part in content:
if not isinstance(part, dict):
if part:
parts.append(str(part))
continue
if isinstance(part.get("text"), str):
parts.append(part["text"])
continue
if isinstance(part.get("content"), str):
parts.append(part["content"])
return "\n".join(parts)
return "" if content is None else str(content)
def _chatgpt_subscription_instructions(messages: List[Dict]) -> str:
instructions = [
_message_content_as_text(msg.get("content")).strip()
for msg in messages or []
if (msg.get("role") or "") == "system"
]
instructions = [part for part in instructions if part]
if instructions:
return "\n\n".join(instructions)
return "You are a helpful AI assistant."
def _build_chatgpt_responses_payload(
model: str,
messages: List[Dict],
temperature: float,
max_tokens: int,
*,
stream: bool = False,
) -> Dict:
from src.chatgpt_subscription import build_responses_input
conversation = [msg for msg in (messages or []) if (msg.get("role") or "") != "system"]
payload: Dict = {
"model": model,
"instructions": _chatgpt_subscription_instructions(messages),
"input": build_responses_input(conversation),
"stream": stream,
"store": False,
}
if not _restricts_temperature(model):
payload["temperature"] = temperature
# ChatGPT Subscription Codex API does not support max_output_tokens —
# passing it returns HTTP 400 "Unsupported parameter: max_output_tokens".
# Do not include it in the payload.
return payload
def _format_chatgpt_subscription_error(status_code: int, text: str) -> str:
if status_code in (401, 403):
return "ChatGPT Subscription credentials expired or were rejected. Reconnect the provider."
if status_code == 429:
return "ChatGPT Subscription quota or rate limit was reached. Retry after the upstream limit resets."
return _format_upstream_error(status_code, text, "https://chatgpt.com/backend-api/codex")
def _format_upstream_error(status: int, body: bytes | str, url: str) -> str:
"""Turn an upstream HTTP error into a user-readable sentence.
Auth failures (401/403) become 'xAI rejected the API key' etc., so the UI
stops showing raw JSON like '{"error":{"message":"User not found."}}'.
"""
if isinstance(body, bytes):
try:
body = body.decode("utf-8", errors="replace")
except Exception:
body = str(body)
provider = _provider_label(url)
# Try to pull a message out of the body
detail = ""
try:
j = json.loads(body) if body else {}
if isinstance(j, dict):
err = j.get("error") or j
if isinstance(err, dict):
detail = (err.get("message") or err.get("detail") or "").strip()
elif isinstance(err, str):
detail = err.strip()
except Exception:
detail = (body or "").strip()[:240]
if status in (401, 403):
msg = f"{provider} rejected the API key"
if status == 403:
msg = f"{provider} denied access (403)"
if detail:
msg += f" — {detail}"
msg += ". Check Model Endpoints → {} and re-paste the key.".format(provider)
return msg
if status == 404:
return f"{provider} returned 404 — check the base URL and model name." + (f" ({detail})" if detail else "")
if status == 429:
return f"{provider} rate-limited the request (429)." + (f" {detail}" if detail else "")
if status >= 500:
return f"{provider} is having an outage (HTTP {status})." + (f" {detail}" if detail else "")
return f"{provider} returned HTTP {status}" + (f": {detail}" if detail else "")
# Models that require max_completion_tokens instead of max_tokens
_MAX_COMPLETION_TOKENS_MODELS = {"o1", "o3", "o4", "gpt-4.5", "gpt-5"}
def _uses_max_completion_tokens(model: str) -> bool:
"""Check if a model requires max_completion_tokens instead of max_tokens."""
if not model:
return False
m = model.lower()
return any(m.startswith(p) or f"/{p}" in m for p in _MAX_COMPLETION_TOKENS_MODELS)
# OpenAI reasoning models (o1, o3, o4, gpt-5 families) only accept the default
# temperature. Sending any explicit value — even 0.0 — returns HTTP 400
# ("Only the default (1) value is supported"). That otherwise breaks chat when a
# preset sets a non-default temperature, and makes endpoint probing report a
# perfectly good model as failing. For these models we omit the field and let
# the API use its required default. (gpt-4.5 is intentionally excluded — it is
# not a reasoning model and accepts temperature normally.)
_FIXED_TEMPERATURE_MODELS = ("o1", "o3", "o4", "gpt-5", "kimi-for-coding")
def _restricts_temperature(model: str) -> bool:
"""Check if a model rejects any non-default temperature."""
if not model:
return False
m = model.lower()
return any(m.startswith(p) or f"/{p}" in m for p in _FIXED_TEMPERATURE_MODELS)
# The official Moonshot API fixes temperature at 1.0 in thinking mode and 0.6
# when thinking is explicitly disabled for Kimi K2.5/K2.6. Any other explicit
# value returns HTTP 400. Odysseus does not currently send the `thinking` mode
# control, so omit temperature and let Moonshot use its default thinking mode.
# Keep the gate provider-specific: self-hosted Kimi deployments may accept
# custom sampling values, and older Moonshot models have different defaults.
def _moonshot_rejects_custom_temperature(provider: str, model: str) -> bool:
"""Check if the official Moonshot API fixes temperature for this model."""
if provider != "moonshot" or not isinstance(model, str):
return False
model_id = model.lower().rsplit("/", 1)[-1]
return bool(re.match(r"^kimi-k2\.(?:5|6)(?:$|[-_:])", model_id))
def _omit_temperature(provider: str, model: str) -> bool:
"""Check if a request should use the provider's default temperature."""
return _restricts_temperature(model) or _moonshot_rejects_custom_temperature(
provider, model
)
# Anthropic removed the sampling parameters (temperature, top_p, top_k) starting
# with Claude Opus 4.7. On Opus 4.7 and later, sending `temperature` at all —
# even 0.0 — returns HTTP 400. Earlier Claude models (Opus 4.6 and below, every
# Sonnet/Haiku) still accept temperature in [0.0, 1.0], so the omission must be
# version-gated rather than applied to all `claude-*` models.
def _anthropic_rejects_temperature(model: str) -> bool:
"""Check if a native-Anthropic model rejects the temperature field (Opus 4.7+)."""
if not isinstance(model, str) or not model:
return False
# `(?<![a-z])` anchors "opus" to a word boundary so a substring match like
# `oct-opus`/`octopus-4-8` can't be read as Opus (it would otherwise strip
# temperature). Both version components are capped at 1-2 digits and forbid a
# trailing digit, so an 8-digit date can never be read as a version number:
# `claude-opus-4-20250514` (Opus 4.0) parses as major-only rather than reading
# `20250514` as a giant minor, and `claude-3-opus-20240229` (legacy Claude 3
# Opus, date directly after "opus-") fails to match at all rather than reading
# the date as a giant major. Dated 4.7+ snapshots (`claude-opus-4-7-20260201`)
# keep their explicit minor and are still matched.
#
# The minor is optional and a missing minor reads as `.0`, so major-only ids
# like `claude-opus-5` are correctly treated as >= 4.7 (issue #5753). Without
# this, every Opus 5 call kept `temperature` and failed with HTTP 400 — visible
# only on paths that pass a temperature, e.g. scheduled tasks inheriting
# `stream_agent_loop`'s 0.3 default, which returned empty responses.
match = re.search(
r"(?<![a-z])opus[-_]?(\d{1,2})(?!\d)(?:[-_.](\d{1,2})(?!\d))?", model.lower()
)
if not match:
return False
major = int(match.group(1))
minor = int(match.group(2)) if match.group(2) else 0
return (major, minor) >= (4, 7)
# Reasoning effort level sent to Mistral thinking-capable models. Mistral's
# API accepts "high", "medium", "low", "none" — see
# https://docs.mistral.ai/capabilities/reasoning/. Override via env var
# ODYSSEUS_MISTRAL_REASONING_EFFORT (e.g. set to "medium" for cheaper chat).
_MISTRAL_REASONING_EFFORT = os.getenv("ODYSSEUS_MISTRAL_REASONING_EFFORT", "high")
# Models that support structured thinking — may output </think> without opening tag
_THINKING_MODEL_PATTERNS = (
"qwen3", "qwq", "deepseek-r1", "deepseek-reasoner", "deepseek-v4",
"minimax", "m2-reap", "gemma", "stepfun", "step-3", "step3",
"magistral", "mistral-small", "mistral-medium",
)
def _supports_thinking(model: str) -> bool:
"""Check if model supports structured thinking output."""
if not model:
return False
m = model.lower()
return any(p in m for p in _THINKING_MODEL_PATTERNS)
def _normalize_mistral_content(content):
"""Mistral returns content as a structured array when reasoning is on:
[{"type": "thinking", "thinking": [{"type": "text", "text": "..."}], "closed": true},
{"type": "text", "text": "...final answer..."}]
Convert to (text, thinking) tuple of plain strings. Pass through strings
unchanged so non-Mistral OpenAI-compat endpoints are unaffected.
"""
if isinstance(content, str):
return content, ""
if not isinstance(content, list):
return "", ""
text_parts = []
thinking_parts = []
for block in content:
if not isinstance(block, dict):
continue
btype = block.get("type")
if btype == "text":
t = block.get("text", "")
if t:
text_parts.append(t)
elif btype == "thinking":
inner = block.get("thinking", [])
if isinstance(inner, list):
for tb in inner:
if isinstance(tb, dict) and tb.get("text"):
thinking_parts.append(tb["text"])
elif isinstance(inner, str):
thinking_parts.append(inner)
return "".join(text_parts), "".join(thinking_parts)
def _convert_openai_content_to_anthropic(content):
"""Convert OpenAI multimodal content blocks to Anthropic format.
Converts image_url blocks (data URI) → Anthropic image blocks.
Passes text blocks through unchanged.
"""
if not isinstance(content, list):
return content
converted = []
for block in content:
if not isinstance(block, dict):
converted.append(block)
continue
if block.get("type") == "image_url":
url = (block.get("image_url") or {}).get("url", "")
# Parse data URI: data:image/<fmt>;base64,<data>
if url.startswith("data:"):
try:
header, b64_data = url.split(",", 1)
media_type = header.split(";")[0].replace("data:", "")
except (ValueError, IndexError):
continue
converted.append({
"type": "image",
"source": {
"type": "base64",
"media_type": media_type,
"data": b64_data,
},
})
else:
# External URL — use Anthropic's URL source
converted.append({
"type": "image",
"source": {"type": "url", "url": url},
})
elif block.get("type") == "text":
converted.append(block)
else:
converted.append(block)
return converted
def _build_anthropic_payload(model, messages, temperature, max_tokens, stream=False, tools=None):
"""Convert OpenAI-style messages to Anthropic format."""
system_parts = []
chat_messages = []
for m in messages:
if m.get("role") == "system":
system_parts.append(m.get("content") or "")
elif m.get("role") == "tool":
# Convert OpenAI tool result to Anthropic format
chat_messages.append({
"role": "user",
"content": [{
"type": "tool_result",
"tool_use_id": m.get("tool_call_id", ""),
"content": m.get("content", ""),
}],
})
elif m.get("role") == "assistant" and isinstance(m.get("tool_calls"), list):
# Convert OpenAI assistant tool_calls to Anthropic format
content = []
if m.get("content"):
content.append({"type": "text", "text": m["content"]})
for tc in m["tool_calls"]:
fn = tc.get("function") or {}
args_str = fn.get("arguments") or "{}"
try:
args = json.loads(args_str) if isinstance(args_str, str) else args_str
except (json.JSONDecodeError, TypeError):
args = {}
content.append({
"type": "tool_use",
"id": tc.get("id", ""),
"name": fn.get("name", ""),
"input": args,
})
chat_messages.append({"role": "assistant", "content": content})
else:
# Convert multimodal content (image_url → image) for Anthropic
content = _convert_openai_content_to_anthropic(m["content"])
chat_messages.append({"role": m["role"], "content": content})
# Anthropic only accepts temperature in [0.0, 1.0] and 400s on anything above
# 1.0. Clamp here (in the Anthropic builder only) so presets/sliders that use
# the wider OpenAI 0.0-2.0 range — e.g. the shipped "Nietzsche" preset at 1.2
# — don't hard-break every Claude request. OpenAI's own path is left untouched.
if temperature is not None:
temperature = max(0.0, min(temperature, 1.0))
payload = {
"model": model,
"messages": chat_messages,
"max_tokens": max_tokens if max_tokens and max_tokens > 0 else 4096,
}
# Opus 4.7+ removed the sampling parameters — sending `temperature` (even 0.0)
# returns HTTP 400. Omit it for those models; older Claude models still take it.
if not _anthropic_rejects_temperature(model):
payload["temperature"] = temperature
if system_parts:
system_text = "\n\n".join(system_parts)
# Send `system` as a structured text block so we can attach a prompt-cache
# breakpoint. The agent loop re-sends this same large prefix every round;
# caching it makes Anthropic re-read it from cache (~90% cheaper, lower TTFB)
# instead of re-billing it. Skip caching tiny one-off prompts, where the
# cache-WRITE premium wouldn't pay back (no reuse). Presence of `tools`
# means an agentic/multi-round call, where the prefix is always reused.
system_block = {"type": "text", "text": system_text}
if tools or len(system_text) > 4000:
system_block["cache_control"] = {"type": "ephemeral"}
payload["system"] = [system_block]
if stream:
payload["stream"] = True
# Convert OpenAI-format tools to Anthropic format
if tools:
anthropic_tools = []
for t in tools:
if t.get("type") == "function":
fn = t["function"]
anthropic_tools.append({
"name": fn["name"],
"description": fn.get("description", ""),
"input_schema": fn.get("parameters", {"type": "object", "properties": {}}),
})
if anthropic_tools:
# Cache the tool schemas too — they're stable for the whole agent run.
# The breakpoint caches all tool defs preceding it in the request.
anthropic_tools[-1]["cache_control"] = {"type": "ephemeral"}
payload["tools"] = anthropic_tools
return payload
def _build_anthropic_headers(headers):
"""Convert Bearer auth to x-api-key for Anthropic."""
h = {"Content-Type": "application/json", "anthropic-version": "2023-06-01"}
if headers:
for k, v in headers.items():
if k.lower() == "authorization" and isinstance(v, str) and v.startswith("Bearer "):
h["x-api-key"] = v[7:]
else:
h[k] = v
return h
def _parse_anthropic_response(data: dict) -> str:
"""Extract text from an Anthropic response.
The Messages API `content` is an array that can hold more than one text
block (e.g. text split around a tool_use block, or citation-segmented
text). Concatenate them all instead of returning only the first, which
silently dropped the rest of the reply.
"""
return "".join(
block.get("text", "")
for block in data.get("content", [])
if isinstance(block, dict) and block.get("type") == "text"
)
def _as_content_blocks(content) -> List[Dict]:
"""Coerce a message `content` into a list of content blocks.
A list (multimodal: text + image parts) passes through; a non-empty string
becomes a single text block; None/empty yields no blocks. Used when merging
consecutive user messages so multimodal content isn't str()-ed away.
"""
if isinstance(content, list):
return content
if content:
return [{"type": "text", "text": str(content)}]
return []
def _is_untrusted_context_content(content) -> bool:
if isinstance(content, str):
return (
content.startswith("UNTRUSTED SOURCE DATA\n")
or "<<<UNTRUSTED_SOURCE_DATA>>>" in content
)
if isinstance(content, list):
return any(
isinstance(block, dict)
and block.get("type") == "text"
and _is_untrusted_context_content(block.get("text") or "")
for block in content
)
return False
_REFERENCE_CONTEXT_BOUNDARY = "Reference context received."
def _sanitize_llm_messages(messages: List[Dict]) -> List[Dict]:
"""Strip Odysseus-only metadata before sending messages to providers.
Per the OpenAI chat format: user/system messages must have content; a tool
message needs content + tool_call_id; an assistant message may carry content,
tool_calls, or both. The old guard required content on every message, which
dropped a valid assistant message that has only tool_calls — e.g. the
follow-up message _append_tool_results builds for a no-prose native tool call
(content=None, since Gemini/Ollama reject tool_calls alongside ""). Dropping
it leaves the tool result dangling and breaks the next round.
"""
allowed = {"role", "content", "name", "tool_call_id", "tool_calls", "function_call", "reasoning_content"}
cleaned = []
for msg in messages or []:
if not isinstance(msg, dict):
continue
item = {k: v for k, v in msg.items() if k in allowed and v is not None}
role = item.get("role")
if not role:
continue
if role == "assistant":
# Re-add an explicit content=None when the message is tool-calls-only
# (the None was stripped above) so the provider gets the spec-correct
# `content: null`, not an omitted key.
if "content" not in item and item.get("tool_calls"):
item["content"] = None
if "content" in item or item.get("tool_calls"):
cleaned.append(item)
elif role == "tool":
if "content" in item and "tool_call_id" in item:
cleaned.append(item)
elif "content" in item:
cleaned.append(item)
# Repair tool-call adjacency before sending to any OpenAI-compatible
# provider. Trimming/compaction/retries can leave `role:"tool"` messages
# without their immediately-preceding assistant `tool_calls` parent, which
# DeepSeek rejects with:
# "Messages with role 'tool' must be a response to a preceding message with
# 'tool_calls'". Also strip unanswered assistant tool_calls; some providers
# reject those as incomplete conversations.
repaired: List[Dict] = []
i = 0
while i < len(cleaned):
msg = cleaned[i]
role = msg.get("role")
if role == "tool":
# Orphan tool result. There is no valid assistant tool_calls parent
# immediately before this batch, so it cannot be sent.
logger.debug("Dropping orphan tool message before provider request")
i += 1
continue
tool_calls = msg.get("tool_calls") if role == "assistant" else None
if not tool_calls:
repaired.append(msg)
i += 1
continue
call_ids = [
str(tc.get("id"))
for tc in tool_calls
if isinstance(tc, dict) and tc.get("id")
]
expected = set(call_ids)
answered_ids = []
tool_batch = []
j = i + 1
while j < len(cleaned) and cleaned[j].get("role") == "tool":
tid = str(cleaned[j].get("tool_call_id") or "")
if tid in expected and tid not in answered_ids:
answered_ids.append(tid)
tool_batch.append(cleaned[j])
else:
logger.debug("Dropping unmatched/duplicate tool message before provider request")
j += 1
if not tool_batch:
plain = {k: v for k, v in msg.items() if k != "tool_calls"}
if (plain.get("content") or "").strip():
repaired.append(plain)
else:
logger.debug("Dropping unanswered assistant tool_calls before provider request")
i = j
continue
answered = set(answered_ids)
pruned_calls = [
tc for tc in tool_calls
if isinstance(tc, dict) and str(tc.get("id")) in answered
]
fixed = dict(msg)
fixed["tool_calls"] = pruned_calls
if "content" not in fixed:
fixed["content"] = None
repaired.append(fixed)
repaired.extend(tool_batch)
if len(pruned_calls) != len(tool_calls):
logger.debug("Pruned unanswered assistant tool_calls before provider request")
i = j
# Merge consecutive user messages to satisfy strict role alternation
# requirements after invalid tool-call fragments have been removed.
merged: List[Dict] = []
for item in repaired:
if not merged:
merged.append(item)
continue
last = merged[-1]
if last.get("role") == "user" and item.get("role") == "user":
if _is_untrusted_context_content(last.get("content")):
merged.append({"role": "assistant", "content": _REFERENCE_CONTEXT_BOUNDARY})
merged.append(item)
continue
last_copy = dict(last)
lc = last_copy.get("content")
ic = item.get("content")
if isinstance(lc, list) or isinstance(ic, list):
# Preserve multimodal content blocks (e.g. an image part) by
# concatenating the block lists. str()-ing a list turned an
# image message into its Python repr and dropped the image.
merged_blocks = _as_content_blocks(lc) + _as_content_blocks(ic)
if merged_blocks:
last_copy["content"] = merged_blocks
else:
last_copy.pop("content", None)
else:
last_str = str(lc) if lc is not None else ""
item_str = str(ic) if ic is not None else ""
new_content = "\n\n".join(part for part in (last_str, item_str) if part)
if new_content:
last_copy["content"] = new_content
else:
last_copy.pop("content", None)
merged[-1] = last_copy
else:
merged.append(item)
return merged
def _normalize_anthropic_url(url: str) -> str:
"""Ensure Anthropic URL points to /v1/messages."""
url = url.rstrip("/")
if url.endswith("/v1/messages"):
return url
if url.endswith("/v1"):
return url + "/messages"
return url + "/v1/messages"
def _model_list_base(url: str) -> str:
"""Normalize model/chat URLs to the configured endpoint base."""
base = (url or "").strip().rstrip("/")
for suffix in ("/models", "/chat/completions", "/completions", "/v1/messages", "/responses"):
if base.endswith(suffix):
base = base[: -len(suffix)].rstrip("/")
for suffix in ("/chat", "/tags", "/generate"):
if base.endswith("/api" + suffix):
base = base[: -len(suffix)].rstrip("/")
return base
def _parse_model_cache(raw) -> List[str]:
if not raw:
return []
try:
models = json.loads(raw) if isinstance(raw, str) else raw
except Exception:
return []
if not isinstance(models, list):
return []
out = []
seen = set()
for item in models:
mid = str(item or "").strip()
if not mid or mid in seen:
continue
out.append(mid)
seen.add(mid)
return out
def _configured_cached_model_ids(
endpoint_url: str,
*,
owner: Optional[str] = None,
endpoint_id: Optional[str] = None,
) -> List[str]:
"""Return cached models for a configured endpoint matching endpoint_url."""
target = _model_list_base(endpoint_url)
if not target:
return []
try:
from src.database import SessionLocal, ModelEndpoint
except Exception:
return []
db = SessionLocal()
try:
q = db.query(ModelEndpoint).filter(ModelEndpoint.is_enabled == True)
if endpoint_id:
q = q.filter(ModelEndpoint.id == endpoint_id)
if owner:
from src.auth_helpers import owner_filter
q = owner_filter(q, ModelEndpoint, owner)
rows = q.all()
for ep in rows:
if _model_list_base(getattr(ep, "base_url", "")) != target:
continue
models = _parse_model_cache(getattr(ep, "cached_models", None) or getattr(ep, "models", None))
if not models:
continue
hidden = set(_parse_model_cache(getattr(ep, "hidden_models", None)))
return [m for m in models if m not in hidden]
except Exception:
return []
finally:
try:
db.close()
except Exception:
pass
return []
def list_model_ids(
base_chat_url: str,
timeout: int = LLMConfig.DEFAULT_TIMEOUT,
headers: Optional[Dict] = None,
*,
owner: Optional[str] = None,
endpoint_id: Optional[str] = None,
) -> List[str]:
"""List available model IDs from an endpoint."""
cached = _configured_cached_model_ids(base_chat_url, owner=owner, endpoint_id=endpoint_id)
if cached:
return cached
provider = _detect_provider(base_chat_url)
if provider == "anthropic":
return list(ANTHROPIC_MODELS)
try:
h = {}
if headers:
h.update(headers)
if provider == "ollama":
models_url = _ollama_api_root(base_chat_url) + "/tags"
else:
from src.endpoint_resolver import build_models_url
models_url = build_models_url(base_chat_url)
r = httpx_get_kimi_aware(models_url, h, timeout=timeout)
r.raise_for_status()
data = r.json()
# Some OpenAI-compatible APIs (e.g. Together) return a bare list here.
items = data if isinstance(data, list) else (data.get("data") or [])
model_ids = [m.get("id") for m in items if isinstance(m, dict) and m.get("id")]
if not model_ids and isinstance(data, dict):
model_ids = [
m.get("name") or m.get("model")
for m in (data.get("models") or [])
if m.get("name") or m.get("model")
]
return model_ids
except Exception:
try:
if ":11434" in base_chat_url or "ollama" in base_chat_url.lower():
root = base_chat_url.replace("/v1/chat/completions", "").replace("/chat/completions", "").rstrip("/")
r = httpx.get(root + "/api/tags", timeout=timeout)
r.raise_for_status()
return [m.get("name") or m.get("model") for m in (r.json().get("models") or []) if m.get("name") or m.get("model")]
except Exception as e:
logger.warning("Failed to fetch model list from configured endpoint", exc_info=e)
return []
def normalize_model_id(
endpoint_url: str,
requested: str,
timeout: int = LLMConfig.DEFAULT_TIMEOUT,
*,
owner: Optional[str] = None,
endpoint_id: Optional[str] = None,
) -> Optional[str]:
"""Normalize a model ID to match available models."""
avail = list_model_ids(endpoint_url, timeout, owner=owner, endpoint_id=endpoint_id)
if not avail:
return None
if requested in avail:
return requested
import os as _os
req_base = _os.path.basename(requested.rstrip("/"))
for a in avail:
if _os.path.basename(a.rstrip("/")) == req_base:
return a
return None
def llm_call(url: str, model: str, messages: List[Dict], temperature: float = LLMConfig.DEFAULT_TEMPERATURE,
max_tokens: int = LLMConfig.DEFAULT_MAX_TOKENS, headers: Optional[Dict] = None,
timeout: int = LLMConfig.DEFAULT_TIMEOUT, prompt_type: Optional[str] = None) -> str:
"""Synchronous LLM call with optional prompt type enhancement."""
h = _provider_headers(_detect_provider(url))
# Tolerate headers that arrive as a JSON string (some sessions stored them
# double-encoded) — otherwise h.update() throws "dictionary update sequence
# element #0 has length 1; 2 is required".
if isinstance(headers, str):
try:
headers = json.loads(headers)
except Exception:
headers = None
if isinstance(headers, dict):
h.update(headers)
messages_copy = _sanitize_llm_messages(messages)
# Consolidate multiple system messages into one at the start.
sys_parts = []
non_sys = []
for m in messages_copy:
if m.get("role") == "system":
sys_parts.append(m.get('content') or '')
else:
non_sys.append(m)
if sys_parts:
messages_copy = [{"role": "system", "content": "\n\n".join(sys_parts)}] + non_sys
else:
messages_copy = non_sys
provider = _detect_provider(url)
cache_key = _get_cache_key(
url, model, messages_copy, temperature, max_tokens, headers=headers,
)
cached_response = _get_cached_response(cache_key)
if cached_response:
logger.debug(f"Returning cached response for key: {cache_key}")
return cached_response
if provider == "anthropic":
target_url = _normalize_anthropic_url(url)
h = _build_anthropic_headers(headers)
payload = _build_anthropic_payload(model, messages_copy, temperature, max_tokens)
elif provider == "ollama":
target_url = _normalize_ollama_url(url)
payload = _build_ollama_payload(
model, messages_copy, temperature, max_tokens,
stream=False, num_ctx=get_context_length(url, model),
)
else:
target_url = _normalize_openai_chat_url(url)
if provider == "copilot":
from src.copilot import apply_request_headers
apply_request_headers(h, messages_copy)
payload = {
"model": model,
"messages": messages_copy,
"temperature": temperature,
}
if _omit_temperature(provider, model):
payload.pop("temperature", None)
if max_tokens and max_tokens > 0:
tok_key = "max_completion_tokens" if _uses_max_completion_tokens(model) else "max_tokens"
payload[tok_key] = max_tokens
_apply_local_generation_stability(payload, target_url, model)
if provider == "mistral" and _supports_thinking(model):
payload["reasoning_effort"] = _MISTRAL_REASONING_EFFORT
try:
note_model_activity(target_url, model)
r = httpx_post_kimi_aware(target_url, h, json=payload, timeout=timeout)
except Exception as e:
raise HTTPException(502, f"POST {target_url} failed: {e}")
if not r.is_success:
raise HTTPException(502, f"Upstream {target_url} -> {r.status_code}: {r.text}")
data = r.json()
try:
if provider == "anthropic":
response = _parse_anthropic_response(data)
elif provider == "ollama":
response = _parse_ollama_response(data)
else:
msg = data["choices"][0]["message"]
content = msg.get("content")
if isinstance(content, list):
# Mistral structured content — extract thinking + text
text_part, thinking_part = _normalize_mistral_content(content)
if thinking_part:
response = thinking_part + "\n\n" + (text_part or "")
else:
response = text_part or msg.get("reasoning_content") or ""
else:
response = content or msg.get("reasoning_content") or ""
_set_cached_response(cache_key, response)
return response
except Exception:
raise HTTPException(502, f"Unexpected schema from {target_url}: {str(data)[:400]}")
def _candidate_is_configured(candidate) -> bool:
return bool(
isinstance(candidate, (tuple, list))
and len(candidate) == 3
and isinstance(candidate[0], str)
and candidate[0].strip()
and isinstance(candidate[1], str)
and candidate[1].strip()
)
def _safe_route_descriptor(value) -> dict:
value = value if isinstance(value, dict) else {}
endpoint_id = value.get("endpoint_id")
endpoint_label = value.get("endpoint_label")
endpoint_cost_tracked = value.get("endpoint_cost_tracked")
return {
"endpoint_id": endpoint_id if isinstance(endpoint_id, str) and endpoint_id else None,
"endpoint_label": (
endpoint_label
if isinstance(endpoint_label, str) and endpoint_label.strip()
else "Selected route"
),
"endpoint_cost_tracked": (
endpoint_cost_tracked
if isinstance(endpoint_cost_tracked, bool)
else None
),
}
def _dedupe_model_candidates_with_descriptors(candidates, descriptors=None):
"""Dedupe routes and their parallel non-secret descriptors together."""
seen = []
out = []
out_descriptors = []
descriptors = list(descriptors or [])
for index, candidate in enumerate(candidates or []):
if not _candidate_is_configured(candidate):
continue
route = (candidate[0], candidate[1], candidate[2] or {})
if any(route == prior for prior in seen):
continue
seen.append(route)
out.append(candidate)
raw_descriptor = descriptors[index] if index < len(descriptors) else {}
out_descriptors.append(_safe_route_descriptor(raw_descriptor))
return out, out_descriptors
def dedupe_model_candidates(candidates):
"""Filter malformed entries and drop a later repeat of an already-seen
``(url, model, headers)`` route, preserving order (first occurrence wins).
The chain is the primary target followed by any caller-authorized
fallbacks. A fallback that repeats the session's current model would
otherwise make the chain re-attempt the very route that just failed: a
wasted round-trip plus a spurious ``fallback`` notice for a switch that did
not happen. Credentials are part of route identity: two configured
endpoints may intentionally use the same provider URL/model with different
keys, and rate limiting on one must not discard the other candidate.
"""
out, _descriptors = _dedupe_model_candidates_with_descriptors(candidates)
return out
def llm_call_with_fallback(candidates, messages, **kwargs) -> str:
"""Sync `llm_call` with an ordered fallback chain.
`candidates` is a list of (url, model, headers). The first one that returns
without an exception wins. Connection / 5xx-style failures fall through to
the next candidate. The dead-host cooldown inside `llm_call` makes repeat
attempts at an offline primary effectively free.
"""
cands = dedupe_model_candidates(candidates)
if not cands:
raise HTTPException(503, "No model endpoint configured")
last_err = None
for i, (url, model, headers) in enumerate(cands):
try:
return llm_call(url, model, messages, headers=headers, **kwargs)
except Exception as e:
last_err = e
tag = "primary" if i == 0 else "candidate"
logger.warning(f"[fallback] {tag} {model} failed ({type(e).__name__}); trying next")
continue
raise last_err if last_err else HTTPException(503, "All fallback candidates failed")
async def llm_call_async_with_fallback(candidates, messages, **kwargs) -> str:
"""Async variant of `llm_call_with_fallback` — same semantics."""
cands = dedupe_model_candidates(candidates)
if not cands:
raise HTTPException(503, "No model endpoint configured")
last_err = None
for i, (url, model, headers) in enumerate(cands):
try:
return await llm_call_async(url, model, messages, headers=headers, **kwargs)
except Exception as e:
last_err = e
tag = "primary" if i == 0 else "candidate"
logger.warning(f"[fallback] {tag} {model} failed ({type(e).__name__}); trying next")
continue
raise last_err if last_err else HTTPException(503, "All fallback candidates failed")
def _nonstream_error_status(error: Exception) -> Optional[int]:
"""Normalize a non-stream provider failure for explicit fallback policy."""
status = getattr(error, "status_code", None)
if not isinstance(status, bool) and status is not None:
return _normalize_http_status(status)
if isinstance(error, (httpx.ConnectError, httpx.ConnectTimeout)):
return 503
if isinstance(error, httpx.ReadTimeout):
return 504
return None
async def llm_call_async_with_route_fallback(
candidates,
messages,
*,
fallback_statuses,
**kwargs,
):
"""Call an ordered non-stream route chain and return route provenance.
Unlike the legacy utility helper, this advances only for an explicitly
eligible status. A successful empty response still commits the current
candidate; empty output is not availability evidence. The third return
value is the provider-reported model when available, otherwise the exact
configured candidate model.
"""
raw_candidates = list(candidates or [])
if not raw_candidates or not _candidate_is_configured(raw_candidates[0]):
raise _FallbackIneligibleHTTPException(400, "Selected model endpoint is not configured")
candidate_request_factory = kwargs.pop("candidate_request_factory", None)
cands = dedupe_model_candidates(raw_candidates)
if not cands:
raise HTTPException(503, "No model endpoint configured")
eligible_statuses = frozenset(fallback_statuses or ())
for index, candidate in enumerate(cands):
url, model, headers = candidate
try:
candidate_messages = messages
candidate_kwargs = kwargs
if candidate_request_factory is not None:
request = candidate_request_factory(index, url, model, headers) or {}
if hasattr(request, "__await__"):
request = await request
candidate_messages = request.get("messages", messages)
candidate_kwargs = {**kwargs, **(request.get("kwargs") or {})}
candidate_kwargs = {
**candidate_kwargs,
"availability_only_transport": True,
}
response = await llm_call_async(
url,
model,
candidate_messages,
headers=headers,
return_model_metadata=True,
**candidate_kwargs,
)
actual_model = model
if (
isinstance(response, tuple)
and len(response) == 2
and isinstance(response[0], str)
):
response, reported_model = response
if isinstance(reported_model, str) and reported_model.strip():
actual_model = reported_model.strip()
return response, candidate, actual_model
except Exception as error:
if getattr(error, "fallback_eligible", None) is False:
raise
status = _nonstream_error_status(error)
if index >= len(cands) - 1 or status not in eligible_statuses:
raise
tag = "primary" if index == 0 else "candidate"
logger.warning(
"[fallback] %s %s failed with eligible status %s; trying next",
tag,
model,
status,
)
raise HTTPException(503, "All fallback candidates failed")
async def llm_call_async(
url: str,
model: str,
messages: List[Dict],
temperature: float = LLMConfig.DEFAULT_TEMPERATURE,
max_tokens: int = LLMConfig.DEFAULT_MAX_TOKENS,
headers: Optional[Dict] = None,
timeout: int = LLMConfig.STREAM_TIMEOUT,
max_retries: int = LLMConfig.MAX_RETRIES,
prompt_type: Optional[str] = None,
session_id: Optional[str] = None,
workload: str = "foreground",
availability_only_transport: bool = False,
return_model_metadata: bool = False,
) -> str | tuple[str, str]:
"""Asynchronous LLM call using httpx with connection pooling, timeout, retry logic, and performance logging."""
provider = _detect_provider(url)
messages_copy = _sanitize_llm_messages(messages)
# Consolidate multiple system messages into one at the start.
sys_parts = []
non_sys = []
for m in messages_copy:
if m.get("role") == "system":
sys_parts.append(m.get('content') or '')
else:
non_sys.append(m)
if sys_parts:
messages_copy = [{"role": "system", "content": "\n\n".join(sys_parts)}] + non_sys
else:
messages_copy = non_sys
cache_key = _get_cache_key(
url, model, messages_copy, temperature, max_tokens, headers=headers,
)
cached_response = _get_cached_response(cache_key)
if cached_response:
logger.debug(f"Returning cached response for key: {cache_key}")
if return_model_metadata:
return cached_response, (_get_cached_response_model(cache_key) or model)
return cached_response
if provider == "chatgpt-subscription":
# ChatGPT/Codex requires streamed Responses requests even for callers
# that want a plain string (auto-title, memory extraction, etc.).
# Reuse stream_llm's validated Codex SSE path and collect deltas.
parts: List[str] = []
actual_model = model
async for chunk in stream_llm(
url,
model,
messages_copy,
temperature=temperature,
max_tokens=max_tokens,
headers=headers,
timeout=timeout,
workload=workload,
):
event_is_error = False
for line in str(chunk).splitlines():
if line.startswith("event:"):
event_is_error = line[6:].strip() == "error"
continue
if not line.startswith("data:"):
continue
raw = line[5:].strip()
if not raw:
continue
if raw == "[DONE]":
response = "".join(parts)
_set_cached_response(
cache_key,
response,
actual_model=actual_model,
)
return (
(response, actual_model)
if return_model_metadata
else response
)
try:
data = json.loads(raw)
except json.JSONDecodeError:
continue
if event_is_error or data.get("error") or (data.get("status") and data.get("text")):
status = int(data.get("status") or 502)
text = data.get("text") or data.get("error") or "ChatGPT Subscription request failed"
error_type = (
_FallbackIneligibleHTTPException
if data.get("fallback_eligible") is False
else HTTPException
)
raise error_type(status, text)
if data.get("type") == "model_actual":
reported_model = data.get("model")
if isinstance(reported_model, str) and reported_model.strip():
actual_model = reported_model.strip()
delta = data.get("delta")
if isinstance(delta, str):
parts.append(delta)
response = "".join(parts)
_set_cached_response(cache_key, response, actual_model=actual_model)
return (response, actual_model) if return_model_metadata else response
if provider == "anthropic":
target_url = _normalize_anthropic_url(url)
h = _build_anthropic_headers(headers)
payload = _build_anthropic_payload(model, messages_copy, temperature, max_tokens)
elif provider == "ollama":
target_url = _normalize_ollama_url(url)
h = {"Content-Type": "application/json"}
if headers:
h.update(headers)
payload = _build_ollama_payload(
model, messages_copy, temperature, max_tokens,
stream=False, num_ctx=get_context_length(url, model),
)
else:
target_url = _normalize_openai_chat_url(url)
h = _provider_headers(provider, headers)
if provider == "copilot":
from src.copilot import apply_request_headers
apply_request_headers(h, messages_copy)
payload = {
"model": model,
"messages": messages_copy,
"temperature": temperature,
}
if _omit_temperature(provider, model):
payload.pop("temperature", None)
if max_tokens and max_tokens > 0:
tok_key = "max_completion_tokens" if _uses_max_completion_tokens(model) else "max_tokens"
payload[tok_key] = max_tokens
# Suppress thinking for qwen3/gemma4 on Ollama /v1 — same as stream_llm.
if _is_ollama_openai_compat_url(url) and _supports_thinking(model):
payload["think"] = False
if provider == "mistral" and _supports_thinking(model):
payload["reasoning_effort"] = _MISTRAL_REASONING_EFFORT
_apply_local_cache_affinity(payload, url, session_id)
_apply_local_generation_stability(payload, target_url, model)
if _is_host_dead(target_url):
raise HTTPException(503, f"Upstream {_host_key(target_url)} marked unreachable (cooldown active)")
call_timeout = _call_timeout(timeout)
attempt = 0
while attempt < max_retries:
attempt += 1
start = time.time()
try:
async with _local_model_slot(target_url, model, workload):
note_model_activity(target_url, model)
client = _get_http_client()
r = await httpx_post_kimi_aware_async(client, target_url, h, json=payload, timeout=call_timeout)
duration = time.time() - start
if not r.is_success:
friendly = _format_upstream_error(r.status_code, r.text, target_url)
logger.warning(
f"LLM async call to {target_url} failed in {duration:.2f}s "
f"(attempt {attempt}): HTTP {r.status_code} {friendly}"
)
if r.status_code in (429, 502, 503, 504) and attempt < max_retries:
await asyncio.sleep(LLMConfig.RETRY_DELAY)
continue
raise HTTPException(r.status_code, friendly)
logger.info(f"LLM async call to {target_url} succeeded in {duration:.2f}s (attempt {attempt})")
_clear_host_dead(target_url)
data = r.json()
if isinstance(data, dict) and data.get("error"):
provider_error = data["error"]
status = _provider_stream_error_status(provider_error, default=400)
if isinstance(provider_error, dict):
detail = provider_error.get("message") or provider_error.get("type") or str(provider_error)
else:
detail = str(provider_error)
raise HTTPException(status, detail or "Upstream request failed")
try:
reported_model = data.get("model") if isinstance(data, dict) else None
actual_model = (
reported_model.strip()
if isinstance(reported_model, str) and reported_model.strip()
else model
)
if provider == "anthropic":
response = _parse_anthropic_response(data)
elif provider == "ollama":
response = _parse_ollama_response(data)
else:
msg = data["choices"][0]["message"]
content = msg.get("content")
if isinstance(content, list):
# Mistral structured content — extract thinking + text
# (same contract as llm_call / stream_llm; see #5435).
text_part, thinking_part = _normalize_mistral_content(content)
if thinking_part:
response = thinking_part + "\n\n" + (text_part or "")
else:
response = text_part or msg.get("reasoning_content") or ""
else:
response = content or msg.get("reasoning_content") or ""
_set_cached_response(
cache_key,
response,
actual_model=actual_model,
)
return (
(response, actual_model)
if return_model_metadata
else response
)
except HTTPException:
raise
except Exception:
raise _FallbackIneligibleHTTPException(
502,
f"Unexpected schema from {target_url}: {str(data)[:400]}",
)
except (httpx.ConnectError, httpx.ConnectTimeout) as e:
_cooled = _mark_host_dead(target_url)
duration = time.time() - start
_tail = f" — host cooled for {DEAD_HOST_COOLDOWN:.0f}s" if _cooled else " — transient, will retry"
logger.warning(f"LLM async connect to {target_url} failed after {duration:.2f}s: {e}{_tail}")
if _cooled or attempt >= max_retries:
raise HTTPException(503, f"Cannot reach {_host_key(target_url)}: {e}")
await asyncio.sleep(LLMConfig.RETRY_DELAY)
except httpx.ReadTimeout as e:
duration = time.time() - start
logger.warning(f"LLM async read timed out after {duration:.2f}s: {e}")
if attempt >= max_retries:
raise HTTPException(504, f"POST {target_url} timed out after {max_retries} attempts")
await asyncio.sleep(LLMConfig.RETRY_DELAY)
except httpx.PoolTimeout as e:
duration = time.time() - start
logger.warning(f"LLM async connection pool timed out after {duration:.2f}s: {e}")
if availability_only_transport:
raise HTTPException(
504,
f"POST {target_url} could not acquire an upstream connection",
)
if attempt >= max_retries:
raise HTTPException(504, f"POST {target_url} timed out after {max_retries} attempts")
await asyncio.sleep(LLMConfig.RETRY_DELAY)
except httpx.WriteTimeout as e:
duration = time.time() - start
logger.warning(f"LLM async upstream timeout after {duration:.2f}s: {e}")
if availability_only_transport:
raise _FallbackIneligibleHTTPException(
504,
f"POST {target_url} failed during request delivery",
)
if attempt >= max_retries:
raise HTTPException(504, f"POST {target_url} timed out after {max_retries} attempts")
await asyncio.sleep(LLMConfig.RETRY_DELAY)
except httpx.ProtocolError as e:
duration = time.time() - start
logger.warning(f"LLM async protocol failure after {duration:.2f}s: {e}")
if availability_only_transport:
raise _FallbackIneligibleHTTPException(
502,
f"POST {target_url} failed with a protocol error",
)
if attempt >= max_retries:
raise HTTPException(502, f"POST {target_url} failed after {max_retries} attempts: {e}")
await asyncio.sleep(LLMConfig.RETRY_DELAY)
except httpx.NetworkError as e:
duration = time.time() - start
logger.warning(f"LLM async network failure after {duration:.2f}s: {e}")
if availability_only_transport:
raise _FallbackIneligibleHTTPException(
502,
f"POST {target_url} failed with a network error",
)
if attempt >= max_retries:
raise HTTPException(502, f"POST {target_url} failed after {max_retries} attempts: {e}")
await asyncio.sleep(LLMConfig.RETRY_DELAY)
except httpx.HTTPStatusError as e:
status = e.response.status_code if e.response is not None else 502
raise HTTPException(status, str(e))
except httpx.RequestError as e:
duration = time.time() - start
logger.warning(f"LLM async request configuration failed after {duration:.2f}s: {e}")
raise _FallbackIneligibleHTTPException(
502,
f"POST {target_url} could not be configured: {e}",
)
def _stream_target_url(url: str) -> str:
provider = _detect_provider(url)
if provider == "anthropic":
return _normalize_anthropic_url(url)
if provider == "ollama":
return _normalize_ollama_url(url)
if provider == "chatgpt-subscription":
return _normalize_chatgpt_subscription_url(url)
return _normalize_openai_chat_url(url)
async def stream_llm(url: str, model: str, messages: List[Dict], temperature: float = LLMConfig.DEFAULT_TEMPERATURE,
max_tokens: int = LLMConfig.DEFAULT_MAX_TOKENS, headers: Optional[Dict] = None,
timeout: int = LLMConfig.STREAM_TIMEOUT, prompt_type: Optional[str] = None,
tools: Optional[List[Dict]] = None, session_id: Optional[str] = None,
tool_choice_none: bool = False, workload: str = "foreground"):
target_url = _stream_target_url(url)
async with _local_model_slot(target_url, model, workload):
async for chunk in _stream_llm_inner(
url,
model,
messages,
temperature=temperature,
max_tokens=max_tokens,
headers=headers,
timeout=timeout,
prompt_type=prompt_type,
tools=tools,
session_id=session_id,
tool_choice_none=tool_choice_none,
):
yield chunk
async def _stream_llm_inner(url: str, model: str, messages: List[Dict], temperature: float = LLMConfig.DEFAULT_TEMPERATURE,
max_tokens: int = LLMConfig.DEFAULT_MAX_TOKENS, headers: Optional[Dict] = None,
timeout: int = LLMConfig.STREAM_TIMEOUT, prompt_type: Optional[str] = None,
tools: Optional[List[Dict]] = None, session_id: Optional[str] = None,
tool_choice_none: bool = False):
"""Stream LLM responses with improved error handling.
Yields SSE chunks:
- data: {"delta": "text"} — text content
- data: {"type": "tool_calls", ...} — accumulated native tool calls (before DONE)
- event: error — errors
- data: [DONE] — end of stream
"""
provider = _detect_provider(url)
messages_copy = _sanitize_llm_messages(messages)
# Consolidate multiple system messages into one at the start.
# Some models (e.g. Qwen3.5) reject system messages that aren't first.
sys_parts = []
non_sys = []
for m in messages_copy:
if m.get("role") == "system":
sys_parts.append(m.get('content') or '')
else:
non_sys.append(m)
if sys_parts:
messages_copy = [{"role": "system", "content": "\n\n".join(sys_parts)}] + non_sys
else:
messages_copy = non_sys
if provider == "anthropic":
target_url = _normalize_anthropic_url(url)
h = _build_anthropic_headers(headers)
payload = _build_anthropic_payload(model, messages_copy, temperature, max_tokens, stream=True, tools=tools)
elif provider == "ollama":
target_url = _normalize_ollama_url(url)
h = {"Content-Type": "application/json"}
if headers:
h.update(headers)
payload = _build_ollama_payload(
model, messages_copy, temperature, max_tokens,
stream=True, tools=tools, num_ctx=get_context_length(url, model),
)
elif provider == "chatgpt-subscription":
target_url = _normalize_chatgpt_subscription_url(url)
h = _provider_headers(provider, headers)
payload = _build_chatgpt_responses_payload(model, messages_copy, temperature, max_tokens, stream=True)
else:
target_url = _normalize_openai_chat_url(url)
payload = {
"model": model,
"messages": messages_copy,
"temperature": temperature,
"stream": True,
}
if _omit_temperature(provider, model):
payload.pop("temperature", None)
if provider not in {"openrouter", "groq"}:
payload["stream_options"] = {"include_usage": True}
if max_tokens and max_tokens > 0:
tok_key = "max_completion_tokens" if _uses_max_completion_tokens(model) else "max_tokens"
payload[tok_key] = max_tokens
if tools:
payload["tools"] = _alias_harmony_tools(tools, model)
elif tool_choice_none:
payload["tool_choice"] = "none"
# Mistral thinking-capable models — send reasoning_effort so Mistral
# activates thinking mode and returns structured reasoning_content.
# Effort level is configurable via ODYSSEUS_MISTRAL_REASONING_EFFORT
# (high / medium / low / none); default "high".
if provider == "mistral" and _supports_thinking(model):
payload["reasoning_effort"] = _MISTRAL_REASONING_EFFORT
# For Ollama's OpenAI-compat /v1 endpoint with thinking models (qwen3,
# gemma4, etc.), suppress thinking so tool calls aren't swallowed inside
# <think> blocks. Ollama /v1 accepts "think": false as a top-level param.
if _is_ollama_openai_compat_url(url) and _supports_thinking(model):
payload["think"] = False
_apply_local_cache_affinity(payload, url, session_id)
_apply_local_generation_stability(payload, target_url, model)
_scrub_openai_chat_tool_reasoning(payload, target_url, model)
h = _provider_headers(provider, headers)
if provider == "copilot":
from src.copilot import apply_request_headers
apply_request_headers(h, messages_copy)
# Connect budget from LLMConfig.CONNECT_TIMEOUT (env LLM_CONNECT_TIMEOUT).
# The dead-host cooldown still bounds a genuinely unreachable upstream, so a
# wider connect budget only affects first contact and stops a brief cold
# connect blip (offshore/public endpoints) surfacing as a 503 on this stream
# path, which -- unlike llm_call -- does not retry the connect.
stream_timeout = _stream_timeout(timeout)
if _is_host_dead(target_url):
yield f'event: error\ndata: {json.dumps({"error": f"Upstream {_host_key(target_url)} unreachable (cooldown active)", "status": 503})}\n\n'
return
note_model_activity(target_url, model)
degenerate_guard = _DegenerateStreamGuard(model)
# ── ChatGPT Subscription / Codex Responses streaming ──
if provider == "chatgpt-subscription":
event_name = ""
input_tokens = 0
output_tokens = 0
_responses_actual_model = ""
_responses_model_announced = False
try:
client = _get_http_client()
async with client.stream('POST', target_url, json=payload, headers=h, timeout=stream_timeout) as r:
_clear_host_dead(target_url)
if r.status_code != 200:
raw = (await r.aread()).decode(errors="replace")
friendly = _format_chatgpt_subscription_error(r.status_code, raw)
yield f'event: error\ndata: {json.dumps({"status": r.status_code, "text": friendly, "raw": raw[:500]})}\n\n'
return
async for line in r.aiter_lines():
if not line:
continue
if line.startswith("event:"):
event_name = line[6:].strip()
continue
if not line.startswith("data:"):
continue
raw = line[5:].strip()
if not raw:
continue
try:
data = json.loads(raw)
except json.JSONDecodeError:
continue
evt = data.get("type") or event_name
response_data = data.get("response") or {}
reported_model = (
response_data.get("model")
if isinstance(response_data, dict)
else None
)
reported_model = _reported_model_name(
reported_model or data.get("model")
)
if reported_model:
_responses_actual_model = reported_model
if not _responses_model_announced:
model_event = _model_actual_event(model, reported_model)
if model_event:
_responses_model_announced = True
yield model_event
if evt == "response.output_text.delta":
delta = data.get("delta") or ""
if delta:
_degenerate = degenerate_guard.check(delta)
if _degenerate:
yield _degenerate
return
yield f'data: {json.dumps({"delta": delta})}\n\n'
elif evt == "response.completed":
usage = (data.get("response") or {}).get("usage") or data.get("usage") or {}
if isinstance(usage, dict):
raw_input = (
usage.get("input_tokens")
if "input_tokens" in usage
else usage.get("prompt_tokens", input_tokens)
)
raw_output = (
usage.get("output_tokens")
if "output_tokens" in usage
else usage.get("completion_tokens", output_tokens)
)
normalized_usage = _normalize_usage_counts(
raw_input,
raw_output,
)
if normalized_usage and (
"input_tokens" in usage
or "prompt_tokens" in usage
or "output_tokens" in usage
or "completion_tokens" in usage
):
_annotate_usage_model(
normalized_usage,
model,
_responses_actual_model,
)
yield f'data: {json.dumps({"type": "usage", "data": normalized_usage})}\n\n'
yield "data: [DONE]\n\n"
return
elif evt in ("response.failed", "error"):
err = data.get("error") or (data.get("response") or {}).get("error") or {}
if evt == "error" and not err:
# Responses API ``error`` events carry code/message
# at the top level, unlike ``response.failed``.
err = {
key: data[key]
for key in ("type", "code", "message", "status", "status_code", "http_status")
if key in data
}
text = err.get("message") if isinstance(err, dict) else str(err or "ChatGPT Subscription request failed")
status = _provider_stream_error_status(err, default=400)
yield f'event: error\ndata: {json.dumps({"status": status, "text": text})}\n\n'
return
yield "data: [DONE]\n\n"
except (httpx.ConnectError, httpx.ConnectTimeout) as e:
_cooled = _mark_host_dead(target_url)
_tail = f" — host cooled for {DEAD_HOST_COOLDOWN:.0f}s" if _cooled else " — transient, will retry"
logger.warning(f"ChatGPT Subscription stream connect to {target_url} failed: {e}{_tail}")
yield f'event: error\ndata: {json.dumps({"error": f"Cannot reach {_host_key(target_url)}", "status": 503})}\n\n'
except httpx.ReadTimeout:
yield f'event: error\ndata: {json.dumps({"error": "Read timeout", "status": 504})}\n\n'
except httpx.PoolTimeout:
yield f'event: error\ndata: {json.dumps({"error": "Connection pool timeout", "status": 504})}\n\n'
except httpx.WriteTimeout:
yield f'event: error\ndata: {json.dumps({"error": "Upstream timeout", "status": 504, "fallback_eligible": False})}\n\n'
except httpx.ProtocolError:
yield f'event: error\ndata: {json.dumps({"error": "Upstream protocol error", "status": 502, "fallback_eligible": False})}\n\n'
except httpx.NetworkError:
yield f'event: error\ndata: {json.dumps({"error": "Network error", "status": 502, "fallback_eligible": False})}\n\n'
except Exception as e:
logger.error(f"ChatGPT Subscription stream error: {e}")
yield f'event: error\ndata: {json.dumps({"error": str(e), "status": 502, "fallback_eligible": False})}\n\n'
return
# ── Native Ollama streaming ──
if provider == "ollama":
_ollama_tool_calls: List[Dict] = []
_harmony_router = _HarmonyStreamRouter()
_ollama_actual_model = ""
_ollama_model_announced = False
try:
client = _get_http_client()
async with client.stream('POST', target_url, json=payload, headers=h, timeout=stream_timeout) as r:
_clear_host_dead(target_url)
if r.status_code != 200:
raw = (await r.aread()).decode(errors="replace")
friendly = _format_upstream_error(r.status_code, raw, target_url)
yield f'event: error\ndata: {json.dumps({"status": r.status_code, "text": friendly, "raw": raw[:500]})}\n\n'
return
async for line in r.aiter_lines():
if not line:
continue
try:
j = json.loads(line)
except json.JSONDecodeError:
continue
if j.get("error"):
err = j.get("error")
status = _provider_stream_error_status(err, default=400)
text = err.get("message") if isinstance(err, dict) else str(err)
yield f'event: error\ndata: {json.dumps({"error": text or "Ollama request failed", "status": status})}\n\n'
return
reported_model = _reported_model_name(j.get("model"))
if reported_model:
_ollama_actual_model = reported_model
if not _ollama_model_announced:
model_event = _model_actual_event(model, reported_model)
if model_event:
_ollama_model_announced = True
yield model_event
message = j.get("message") or {}
thinking = message.get("thinking") or ""
if thinking:
yield _stream_delta_event(thinking, thinking=True)
content = message.get("content") or ""
if content:
for part, is_thinking in _harmony_router.feed(content):
yield _stream_delta_event(part, thinking=is_thinking)
for tc in message.get("tool_calls") or []:
fn = tc.get("function") or {}
if fn.get("name"):
_ollama_tool_calls.append({
"id": tc.get("id") or f"call_{len(_ollama_tool_calls)}",
"name": _unalias_harmony_tool_name(fn.get("name") or "", model),
"arguments": json.dumps(fn.get("arguments") or {}),
})
if j.get("done"):
for part, is_thinking in _harmony_router.flush():
yield _stream_delta_event(part, thinking=is_thinking)
if _ollama_tool_calls:
yield f'data: {json.dumps({"type": "tool_calls", "calls": _ollama_tool_calls})}\n\n'
if j.get("prompt_eval_count") is not None or j.get("eval_count") is not None:
normalized_usage = _normalize_usage_counts(
j.get("prompt_eval_count", 0),
j.get("eval_count", 0),
)
if normalized_usage:
_annotate_usage_model(
normalized_usage,
model,
_ollama_actual_model,
)
yield f'data: {json.dumps({"type": "usage", "data": normalized_usage})}\n\n'
yield "data: [DONE]\n\n"
return
for part, is_thinking in _harmony_router.flush():
yield _stream_delta_event(part, thinking=is_thinking)
yield "data: [DONE]\n\n"
except (httpx.ConnectError, httpx.ConnectTimeout) as e:
_cooled = _mark_host_dead(target_url)
_tail = f" — host cooled for {DEAD_HOST_COOLDOWN:.0f}s" if _cooled else " — transient, will retry"
logger.warning(f"Ollama stream connect to {target_url} failed: {e}{_tail}")
yield f'event: error\ndata: {json.dumps({"error": f"Cannot reach {_host_key(target_url)}", "status": 503})}\n\n'
except httpx.ReadTimeout:
yield f'event: error\ndata: {json.dumps({"error": "Read timeout", "status": 504})}\n\n'
except httpx.PoolTimeout:
yield f'event: error\ndata: {json.dumps({"error": "Connection pool timeout", "status": 504})}\n\n'
except httpx.WriteTimeout:
yield f'event: error\ndata: {json.dumps({"error": "Upstream timeout", "status": 504, "fallback_eligible": False})}\n\n'
except httpx.ProtocolError:
yield f'event: error\ndata: {json.dumps({"error": "Upstream protocol error", "status": 502, "fallback_eligible": False})}\n\n'
except httpx.NetworkError:
yield f'event: error\ndata: {json.dumps({"error": "Network error", "status": 502, "fallback_eligible": False})}\n\n'
except Exception as e:
logger.error(f"Ollama stream error: {e}")
yield f'event: error\ndata: {json.dumps({"error": str(e), "status": 502, "fallback_eligible": False})}\n\n'
return
# ── Anthropic streaming ──
if provider == "anthropic":
_anth_input_tokens = 0
_anth_output_tokens = 0
_anth_usage_seen = False
_anth_actual_model = ""
_anth_model_announced = False
# Track tool_use blocks: {index: {id, name, arguments_json}}
_anth_tool_blocks: Dict[int, Dict] = {}
_anth_block_idx = -1
_anth_block_type = ""
try:
client = _get_http_client()
async with client.stream('POST', target_url, json=payload, headers=h, timeout=stream_timeout) as r:
_clear_host_dead(target_url)
if r.status_code != 200:
raw = (await r.aread()).decode(errors="replace")
friendly = _format_upstream_error(r.status_code, raw, target_url)
yield f'event: error\ndata: {json.dumps({"status": r.status_code, "text": friendly, "raw": raw[:500]})}\n\n'
return
async for line in r.aiter_lines():
# SSE allows "data:value" with no space after the colon
# (the space is optional per the spec). Some gateways and
# local servers omit it; gating on "data: " dropped their
# entire stream.
if not line or not line.startswith("data:"):
continue
data = line[5:].strip()
if not data or not data.startswith("{"):
continue
try:
j = json.loads(data)
evt = j.get("type", "")
if evt == "content_block_start":
_anth_block_idx = j.get("index", _anth_block_idx + 1)
cb = j.get("content_block") or {}
_anth_block_type = cb.get("type", "text")
if _anth_block_type == "tool_use":
_anth_tool_blocks[_anth_block_idx] = {
"id": cb.get("id") or f"call_{_anth_block_idx}",
"name": cb.get("name") or "",
"arguments": "",
}
elif evt == "content_block_delta":
delta = j.get("delta") or {}
delta_type = delta.get("type", "")
if delta_type == "text_delta":
text = delta.get("text") or ""
if text:
yield f'data: {json.dumps({"delta": text})}\n\n'
elif delta_type == "input_json_delta":
# Accumulate tool arguments JSON
idx = j.get("index", _anth_block_idx)
if idx in _anth_tool_blocks:
partial = delta.get("partial_json") or ""
_anth_tool_blocks[idx]["arguments"] += partial
# Stream tool arg deltas for doc tools
if partial and _anth_tool_blocks[idx].get("name") in ("create_document", "update_document", "edit_document"):
yield f'data: {json.dumps({"type": "tool_call_delta", "index": idx, "name": _anth_tool_blocks[idx]["name"], "arg_delta": partial})}\n\n'
elif evt == "message_start":
message_data = j.get("message") or {}
reported_model = _reported_model_name(
message_data.get("model")
if isinstance(message_data, dict)
else None
)
if reported_model:
_anth_actual_model = reported_model
if not _anth_model_announced:
model_event = _model_actual_event(model, reported_model)
if model_event:
_anth_model_announced = True
yield model_event
_u = (
message_data.get("usage")
if isinstance(message_data, dict)
else {}
) or {}
if not isinstance(_u, dict):
_u = {}
if "input_tokens" in _u:
_anth_usage_seen = True
_anth_input_tokens = _u.get("input_tokens", 0)
# Surface prompt-cache effectiveness: cache_read > 0 means the
# stable system+tools prefix was served from cache this round.
_c_read = _u.get("cache_read_input_tokens", 0)
_c_write = _u.get("cache_creation_input_tokens", 0)
if _c_read or _c_write:
logger.info(
"[anthropic-cache] read=%s write=%s fresh_input=%s",
_c_read, _c_write, _anth_input_tokens,
)
elif evt == "message_delta":
_u = j.get("usage") or {}
if not isinstance(_u, dict):
_u = {}
if "output_tokens" in _u:
_anth_usage_seen = True
_anth_output_tokens = _u.get("output_tokens", 0)
elif evt == "message_stop":
# Emit accumulated tool calls in OpenAI-compatible format
if _anth_tool_blocks:
calls = []
for idx in sorted(_anth_tool_blocks):
tb = _anth_tool_blocks[idx]
calls.append({
"id": tb["id"],
"name": tb["name"],
"arguments": tb["arguments"],
})
yield f'data: {json.dumps({"type": "tool_calls", "calls": calls})}\n\n'
normalized_usage = _normalize_usage_counts(
_anth_input_tokens,
_anth_output_tokens,
)
if normalized_usage and _anth_usage_seen:
_annotate_usage_model(
normalized_usage,
model,
_anth_actual_model,
)
yield f'data: {json.dumps({"type": "usage", "data": normalized_usage})}\n\n'
yield "data: [DONE]\n\n"
return
elif evt == "error":
err = j.get("error") or {}
err_msg = err.get("message", "Unknown error") if isinstance(err, dict) else str(err)
status = _provider_stream_error_status(err, default=400)
yield f'event: error\ndata: {json.dumps({"error": err_msg, "status": status})}\n\n'
return
except json.JSONDecodeError:
continue
yield "data: [DONE]\n\n"
except (httpx.ConnectError, httpx.ConnectTimeout) as e:
_cooled = _mark_host_dead(target_url)
_tail = f" — host cooled for {DEAD_HOST_COOLDOWN:.0f}s" if _cooled else " — transient, will retry"
logger.warning(f"Anthropic stream connect to {target_url} failed: {e}{_tail}")
yield f'event: error\ndata: {json.dumps({"error": f"Cannot reach {_host_key(target_url)}", "status": 503})}\n\n'
except httpx.ReadTimeout:
yield f'event: error\ndata: {json.dumps({"error": "Read timeout", "status": 504})}\n\n'
except httpx.PoolTimeout:
yield f'event: error\ndata: {json.dumps({"error": "Connection pool timeout", "status": 504})}\n\n'
except httpx.WriteTimeout:
yield f'event: error\ndata: {json.dumps({"error": "Upstream timeout", "status": 504, "fallback_eligible": False})}\n\n'
except httpx.ProtocolError:
yield f'event: error\ndata: {json.dumps({"error": "Upstream protocol error", "status": 502, "fallback_eligible": False})}\n\n'
except httpx.NetworkError:
yield f'event: error\ndata: {json.dumps({"error": "Network error", "status": 502, "fallback_eligible": False})}\n\n'
except Exception as e:
logger.error(f"Anthropic stream error: {e}")
yield f'event: error\ndata: {json.dumps({"error": str(e), "status": 502, "fallback_eligible": False})}\n\n'
return
# ── OpenAI-compatible streaming ──
# Accumulate native tool_calls across streaming chunks
_tc_acc: Dict[int, Dict] = {} # index -> {id, name, arguments}
_tc_last_idx = [-1] # most-recently-touched slot, for providers that omit `index`
# For thinking models: prepend <think> to first content delta so frontend
# can detect thinking-in-progress (some models output </think> but no <think>)
_thinking_model = _supports_thinking(model)
_first_content_sent = False
_in_think_tag = False # True while consuming <think>…</think> content
_think_open_stripped = False # opening <think> tag already removed
_harmony_router = _HarmonyStreamRouter()
_harmony_active = False # sticky: gpt-oss harmony <|channel|> stream detected
_actual_model = ""
_actual_model_announced = False
def _emit_tool_calls():
"""Build the tool_calls event string if any were accumulated."""
if not _tc_acc:
return None
calls = [_tc_acc[i] for i in sorted(_tc_acc)]
return f'data: {json.dumps({"type": "tool_calls", "calls": calls})}\n\n'
def _format_routed_content(parts: List[Tuple[str, bool]]) -> List[str]:
nonlocal _first_content_sent
events = []
for part, is_thinking in parts:
if is_thinking:
events.append(_stream_delta_event(part, thinking=True))
continue
# Some thinking backends start normal content with a stray closing
# tag. Repair only that shape; do not wrap every first token for
# model families like MiniMax, which often stream ordinary answers.
if _thinking_model and not _first_content_sent and part.lstrip().lower().startswith("</think"):
part = "<think>" + part
_first_content_sent = True
events.append(_stream_delta_event(part))
return events
try:
client = _get_http_client()
h = await apply_kimi_code_headers_async(client, h, target_url)
async with client.stream('POST', target_url, json=payload, headers=h, timeout=stream_timeout) as r:
_clear_host_dead(target_url)
if r.status_code != 200:
raw = (await r.aread()).decode(errors="replace")
friendly = _format_upstream_error(r.status_code, raw, target_url)
yield f'event: error\ndata: {json.dumps({"status": r.status_code, "text": friendly, "raw": raw[:500]})}\n\n'
return
async for line in r.aiter_lines():
if not line:
continue
# SSE allows "data:value" with no space after the colon; gating
# on "data: " silently dropped content + usage from providers
# that omit it.
if line.startswith("data:"):
data = line[5:].strip()
if data == "[DONE]":
for event in _format_routed_content(_harmony_router.flush()):
yield event
tc_event = _emit_tool_calls()
if tc_event:
yield tc_event
yield "data: [DONE]\n\n"
return
try:
if data.strip():
if data.startswith("{"):
j = json.loads(data)
if j.get("error"):
err = j.get("error")
status = _provider_stream_error_status(err, default=400)
text = err.get("message") if isinstance(err, dict) else str(err)
yield f'event: error\ndata: {json.dumps({"error": text or "Upstream request failed", "status": status})}\n\n'
return
chunk_model = j.get("model")
if isinstance(chunk_model, str) and chunk_model.strip():
_actual_model = chunk_model.strip()
if (
not _actual_model_announced
and not _same_model_identity(_actual_model, model)
):
_actual_model_announced = True
yield f'data: {json.dumps({"type": "model_actual", "requested_model": model, "model": _actual_model})}\n\n'
# Usage chunk (from stream_options)
_choices = j.get("choices") or []
_delta0 = _choices[0].get("delta") if (_choices and _choices[0] is not None) else None
# Capture usage whenever the chunk carries it and
# the delta has no actual output. Some gateways /
# local servers attach usage to the FINAL delta,
# which also carries role/finish_reason (so it is
# not exactly None/{}/{"content": None}); gating on
# those exact shapes discarded their token counts.
_delta_has_output = isinstance(_delta0, dict) and (
_delta0.get("content")
or _delta0.get("reasoning_content")
or _delta0.get("reasoning")
or _delta0.get("thinking")
or _delta0.get("tool_calls")
)
u = j.get("usage")
_has_genuine_usage = (
isinstance(u, dict)
and (
"prompt_tokens" in u
or "completion_tokens" in u
)
)
if _has_genuine_usage and not _delta_has_output:
_usage_data = _normalize_usage_counts(
u.get("prompt_tokens", 0),
u.get("completion_tokens", 0),
)
if _usage_data is None:
continue
# llama.cpp puts a `timings` block alongside `usage` with the
# TRUE generation speed (predicted_per_second) — pure decode,
# excluding prefill/network. Pass it through so the UI shows the
# real gen t/s instead of recomputing tokens/wall-clock (which
# includes prefill and reads ~20-40% low). Prefill speed too.
_tm = j.get("timings")
if isinstance(_tm, dict):
if _tm.get("predicted_per_second"):
_usage_data["gen_tps"] = round(_tm["predicted_per_second"], 2)
if _tm.get("prompt_per_second"):
_usage_data["prefill_tps"] = round(_tm["prompt_per_second"], 2)
if _actual_model:
_usage_data["model"] = _actual_model
if not _same_model_identity(_actual_model, model):
_usage_data["requested_model"] = model
yield f'data: {json.dumps({"type": "usage", "data": _usage_data})}\n\n'
elif "choices" in j:
_c0 = (j["choices"] or [None])[0]
if _c0 is None:
continue
delta = _c0.get("delta") or {}
if isinstance(delta, dict):
# Text content
# Reasoning tokens (VLLM --reasoning-parser, e.g. Qwen3/DeepSeek-R1, Nemotron). vLLM 0.20.2 / NIM emit the field as `reasoning`; older builds use `reasoning_content`. Some OpenAI-compatible Ollama builds use `thinking`.
reasoning = delta.get("reasoning_content") or delta.get("reasoning") or delta.get("thinking") or ""
content = delta.get("content") or ""
# Mistral structured content: content is a list of typed blocks
# ({"type": "thinking", ...}, {"type": "text", ...}). Split into
# reasoning + text so thinking streams into the thinking panel.
if isinstance(content, list):
text_part, thinking_part = _normalize_mistral_content(content)
if thinking_part:
reasoning = (reasoning + thinking_part) if reasoning else thinking_part
content = text_part
if reasoning:
_degenerate = degenerate_guard.check(reasoning)
if _degenerate:
yield _degenerate
return
yield _stream_delta_event(reasoning, thinking=True)
if content:
content = _strip_visible_chat_template_artifacts(content)
if not content:
continue
_degenerate = degenerate_guard.check(content)
if _degenerate:
yield _degenerate
return
content = re.sub(r"<mm:think(\s+[^>]*)?>", r"<think\1>", content, flags=re.IGNORECASE)
content = re.sub(r"</mm:think>", "</think>", content, flags=re.IGNORECASE)
stripped = content.lstrip()
# gpt-oss harmony format (<|channel|>analysis/final): route via the harmony
# stream router. Sticky once the first marker appears — distinct from the
# <think> path below (handled in the else, preserving #2588 behaviour).
if _harmony_active or "<|" in content:
_harmony_active = True
for event in _format_routed_content(_harmony_router.feed(content)):
yield event
else:
# Auto-detect <think>…</think> in content stream.
# Covers Qwen3-derived models (Qwopus, QwQ forks) whose
# names don't match _THINKING_MODEL_PATTERNS but still
# emit literal <think> markup via llama.cpp --jinja.
if not _first_content_sent and not _thinking_model and not _in_think_tag and stripped.lower().startswith("<think"):
_thinking_model = True
_in_think_tag = True
if _in_think_tag:
close_idx = content.lower().find("</think>")
if close_idx != -1:
# Split: up-to-</think> → thinking, remainder → content
think_part = content[:close_idx]
if not _think_open_stripped:
# Strip the opening <think[...] > from the first chunk.
# Use a dedicated flag — _first_content_sent stays False
# throughout the think block, so it must not be reused.
tag_end = think_part.lower().find(">")
if tag_end != -1:
think_part = think_part[tag_end + 1:]
_think_open_stripped = True
regular_part = content[close_idx + len("</think>"):]
_in_think_tag = False
if think_part:
yield f'data: {json.dumps({"delta": think_part, "thinking": True})}\n\n'
if regular_part:
_first_content_sent = True
yield f'data: {json.dumps({"delta": regular_part})}\n\n'
else:
# Still inside <think>: route to thinking channel
if not _think_open_stripped:
# Strip the opening <think[...] > tag (first chunk only)
tag_end = stripped.lower().find(">")
if tag_end != -1:
content = stripped[tag_end + 1:]
_think_open_stripped = True
if content:
yield f'data: {json.dumps({"delta": content, "thinking": True})}\n\n'
else:
# Some thinking backends start normal content with a
# stray closing tag. Repair only that shape; do not
# wrap every first token for model families like
# MiniMax, which often stream ordinary answers.
if _thinking_model and not _first_content_sent and stripped.lower().startswith("</think"):
content = "<think>" + content
_first_content_sent = True
yield f'data: {json.dumps({"delta": content})}\n\n'
# Native tool calls — accumulate across chunks
for tc in delta.get("tool_calls") or []:
if tc is None:
continue
func = tc.get("function") or {}
raw_idx = tc.get("index")
if raw_idx is None:
# Gemini's OpenAI-compat layer omits `index` on
# parallel tool calls (every delta arrives as
# index=None) and sends each call complete in one
# delta. Without this, all parallel calls collide
# into slot 0 — later calls overwrite the first's
# name and CORRUPT its arguments by concatenation,
# so only one malformed call survives and the
# follow-up round 400s. A function name marks the
# start of a new call → allocate a fresh slot;
# an arg-only continuation attaches to the last.
if func.get("name") or _tc_last_idx[0] < 0:
# Next free slot ABOVE any existing key (not
# len()), so a provider mixing integer indices
# with index=None can never collide.
idx = max(_tc_acc, default=-1) + 1
else:
idx = _tc_last_idx[0]
else:
idx = raw_idx
_tc_last_idx[0] = idx
if idx not in _tc_acc:
_tc_acc[idx] = {"id": "", "name": "", "arguments": ""}
if tc.get("id"):
_tc_acc[idx]["id"] = tc["id"]
# Gemini 3 returns an opaque thought_signature in
# extra_content on the function-call delta. It MUST be
# echoed back on the assistant tool_call next round or the
# follow-up request 400s ("Function call is missing a
# thought_signature"). Preserve it verbatim; other
# providers never send it, so this is a no-op for them.
if tc.get("extra_content"):
_tc_acc[idx]["extra_content"] = tc["extra_content"]
if func.get("name"):
# Map harmony aliases back to real
# tool names before anything
# downstream sees them.
_tc_acc[idx]["name"] = _unalias_harmony_tool_name(func["name"], model)
if "arguments" in func:
# Guard against a null arguments delta: `func` can be
# {"arguments": None} (JSON null), and a raw `+= None`
# raises TypeError that the broad except swallows,
# silently dropping the rest of the chunk. Matches the
# Anthropic accumulator (`partial = ... or ""`) above.
_tc_acc[idx]["arguments"] += func["arguments"] or ""
# Stream tool arg deltas for doc tools
if func["arguments"] and _tc_acc[idx].get("name") in ("create_document", "update_document", "edit_document"):
yield f'data: {json.dumps({"type": "tool_call_delta", "index": idx, "name": _tc_acc[idx]["name"], "arg_delta": func["arguments"]})}\n\n'
elif "text" in j:
if j["text"]:
for event in _format_routed_content(_harmony_router.feed(j["text"])):
yield event
else:
if data.strip():
for event in _format_routed_content(_harmony_router.feed(data)):
yield event
except Exception as e:
logger.error(f"Error parsing stream data: {e}")
continue
# End of stream (no explicit [DONE] received)
for event in _format_routed_content(_harmony_router.flush()):
yield event
tc_event = _emit_tool_calls()
if tc_event:
yield tc_event
yield "data: [DONE]\n\n"
except (httpx.ConnectError, httpx.ConnectTimeout) as e:
_cooled = _mark_host_dead(target_url)
_tail = f" — host cooled for {DEAD_HOST_COOLDOWN:.0f}s" if _cooled else " — transient, will retry"
logger.warning(f"Stream connect to {target_url} failed: {e}{_tail}")
yield f'event: error\ndata: {json.dumps({"error": f"Cannot reach {_host_key(target_url)}", "status": 503})}\n\n'
except httpx.ReadTimeout:
yield f'event: error\ndata: {json.dumps({"error": "Read timeout", "status": 504})}\n\n'
except httpx.PoolTimeout:
yield f'event: error\ndata: {json.dumps({"error": "Connection pool timeout", "status": 504})}\n\n'
except httpx.WriteTimeout:
yield f'event: error\ndata: {json.dumps({"error": "Upstream timeout", "status": 504, "fallback_eligible": False})}\n\n'
except httpx.ProtocolError:
yield f'event: error\ndata: {json.dumps({"error": "Upstream protocol error", "status": 502, "fallback_eligible": False})}\n\n'
except httpx.NetworkError:
yield f'event: error\ndata: {json.dumps({"error": "Network error", "status": 502, "fallback_eligible": False})}\n\n'
except Exception as e:
logger.error(f"Stream error: {e}")
yield f'event: error\ndata: {json.dumps({"error": str(e), "status": 502, "fallback_eligible": False})}\n\n'
def _summarize_stream_error(err_chunk: Optional[str]) -> str:
"""Pull a short human reason out of an `event: error` SSE chunk for the
fallback notice. Returns a generic message if it can't be parsed."""
if not err_chunk:
return "primary model failed"
try:
for line in err_chunk.split("\n"):
if line.startswith("data: "):
j = json.loads(line[6:])
txt = j.get("text") or j.get("error") or ""
status = j.get("status")
msg = (f"HTTP {status}: " if status else "") + str(txt)
return msg[:200].strip() or "primary model failed"
except Exception:
pass
return "primary model failed"
def _stream_error_status(err_chunk: Optional[str]) -> Optional[int]:
"""Return the integer status from an SSE error chunk when present."""
if not err_chunk:
return None
try:
for line in err_chunk.split("\n"):
if not line.startswith("data: "):
continue
status = json.loads(line[6:]).get("status")
return _normalize_http_status(status)
except (TypeError, ValueError, json.JSONDecodeError):
return None
return None
def _stream_error_fallback_override(err_chunk: Optional[str]) -> Optional[bool]:
"""Return an adapter's explicit eligibility decision when present."""
if not err_chunk:
return None
try:
for line in err_chunk.split("\n"):
if not line.startswith("data: "):
continue
value = json.loads(line[6:]).get("fallback_eligible")
return value if isinstance(value, bool) else None
except (TypeError, ValueError, json.JSONDecodeError):
return None
return None
def _request_factory_error_chunk(error: Exception, status: Optional[int]) -> str:
"""Convert route-request preparation failures into a safe SSE error."""
wire_status = status if status is not None else 500
payload = {
"error": f"Model request preparation failed (HTTP {wire_status})",
"status": wire_status,
}
override = getattr(error, "fallback_eligible", None)
if isinstance(override, bool):
payload["fallback_eligible"] = override
elif status is None:
# An unclassified internal/configuration failure must never become an
# availability fallback merely because its safe wire status is 500.
payload["fallback_eligible"] = False
return f'event: error\ndata: {json.dumps(payload)}\n\n'
# Symbolic-only rate-limit statuses providers emit without a numeric code.
# RESOURCE_EXHAUSTED is the gRPC/Google symbol for 429; the other two appear
# in OpenAI-compatible proxies. Any other symbolic status still fails closed.
_SYMBOLIC_RATE_LIMIT_STATUSES = frozenset({
"RATE_LIMITED",
"RATE_LIMIT_EXCEEDED",
"RESOURCE_EXHAUSTED",
})
def _provider_stream_error_status(error, *, default: int = 400) -> int:
"""Classify structured provider stream errors without making them eligible by default.
Some streaming APIs report an HTTP 200 handshake and put the real failure
in a later event. Unknown application errors are request failures, not
availability evidence; only explicit transient/server markers become 5xx
or rate-limit statuses.
"""
if isinstance(error, dict):
# A structured numeric status is authoritative. Text heuristics are
# only a fallback for providers that omit it.
saw_explicit_status = False
symbolic_rate_limited = False
for key in ("status", "status_code", "http_status", "code"):
if key not in error:
continue
value = error.get(key)
if value is None:
continue
# Google-style errors use a symbolic ``status`` together with a
# numeric HTTP ``code``. A symbolic ``code`` remains part of the
# marker heuristics below; it is not itself an explicit status.
if key != "code":
saw_explicit_status = True
if (
key == "status"
and isinstance(value, str)
and value.strip().upper() in _SYMBOLIC_RATE_LIMIT_STATUSES
):
# Only availability evidence when no numeric status follows:
# a payload pairing a symbolic status with e.g. code=401 must
# surface the numeric truth, not advance fallback.
symbolic_rate_limited = True
continue
status = _normalize_http_status(value)
if status is not None:
return status
if symbolic_rate_limited:
return 429
if saw_explicit_status:
return default
marker = " ".join(str(error.get(key) or "") for key in ("type", "code", "message")).lower()
else:
marker = str(error or "").lower()
if "insufficient_quota" in marker or "billing" in marker:
return 402
if any(token in marker for token in ("authentication", "unauthorized", "invalid api key", "invalid_api_key")):
return 401
if any(token in marker for token in ("permission", "forbidden")):
return 403
if any(token in marker for token in ("not_found", "not found", "unknown model")):
return 404
if any(token in marker for token in ("invalid_request", "invalid request", "unsupported", "malformed", "bad request")):
return 400
if any(token in marker for token in ("rate_limit", "rate limit", "too many requests")):
return 429
if any(token in marker for token in ("overloaded", "over capacity")):
return 529
if any(token in marker for token in ("timeout", "timed out")):
return 504
if any(token in marker for token in ("api_error", "server_error", "server error", "internal error", "temporarily unavailable")):
return 500
return default
async def stream_llm_with_fallback(candidates, messages, **kwargs):
"""Wrap stream_llm with an ordered fallback chain.
`candidates` is a list of (url, model, headers). Each is tried in order,
but only retried on an eligible *pre-content* failure. Callers can restrict
errors with ``fallback_statuses`` and disable empty-completion switching
with ``fallback_on_empty=False``. Omitting both preserves the generic
fallback behavior for non-foreground call sites.
Metadata is held until substantive output commits the candidate.
Once a candidate has emitted real output we never switch (that would
duplicate streamed tokens); a later error from that candidate passes
through unchanged. The dead-host cooldown in stream_llm makes repeat
attempts at an offline primary effectively instant.
Yields the same SSE chunk protocol as stream_llm.
"""
fallback_statuses = kwargs.pop("fallback_statuses", None)
fallback_on_empty = bool(kwargs.pop("fallback_on_empty", True))
candidate_request_factory = kwargs.pop("candidate_request_factory", None)
candidate_route_descriptors = kwargs.pop("candidate_route_descriptors", None)
eligible_statuses = None if fallback_statuses is None else frozenset(fallback_statuses)
raw_candidates = list(candidates or [])
if not raw_candidates or not _candidate_is_configured(raw_candidates[0]):
yield f'event: error\ndata: {json.dumps({"error": "Selected model endpoint is not configured", "status": 400, "fallback_eligible": False})}\n\n'
return
cands, route_descriptors = _dedupe_model_candidates_with_descriptors(
raw_candidates,
candidate_route_descriptors,
)
primary_model = cands[0][1]
primary_route = route_descriptors[0]
last_error = None
failures = []
for i, (url, model, headers) in enumerate(cands):
is_last = (i == len(cands) - 1)
emitted = False
retried = False
pending_metadata = []
candidate_messages = messages
candidate_kwargs = kwargs
if candidate_request_factory is not None:
try:
request = candidate_request_factory(i, url, model, headers) or {}
if hasattr(request, "__await__"):
request = await request
candidate_messages = request.get("messages", messages)
candidate_kwargs = {**kwargs, **(request.get("kwargs") or {})}
except Exception as error:
status = _nonstream_error_status(error)
eligibility_override = getattr(error, "fallback_eligible", None)
eligible = (
True
if eligible_statuses is None
else (
eligibility_override
if isinstance(eligibility_override, bool)
else status in eligible_statuses
)
)
error_chunk = _request_factory_error_chunk(error, status)
if not is_last and eligible:
last_error = error_chunk
failures.append({
"candidate_index": i,
"model": model,
"status": status,
"reason": _summarize_stream_error(error_chunk),
})
tag = "primary" if i == 0 else "candidate"
logger.warning(
"[fallback] %s %s request preparation failed with "
"eligible status %s; trying next",
tag,
model,
status,
)
continue
yield error_chunk
return
candidate_stream = stream_llm(
url,
model,
candidate_messages,
headers=headers,
**candidate_kwargs,
)
try:
async for chunk in candidate_stream:
if chunk.startswith("event: error"):
status = _stream_error_status(chunk)
eligibility_override = _stream_error_fallback_override(chunk)
eligible = (
True
if eligible_statuses is None
else (
eligibility_override
if eligibility_override is not None
else status in eligible_statuses
)
)
if not emitted and not is_last and eligible:
# Pre-content failure with fallbacks left — swallow and
# move to the next candidate.
last_error = chunk
failures.append({
"candidate_index": i,
"model": model,
"status": status,
"reason": _summarize_stream_error(chunk),
})
retried = True
if i == 0:
logger.warning(f"[fallback] primary {model} failed before output; trying fallback")
else:
logger.warning(f"[fallback] candidate {model} failed; trying next")
break
if not emitted:
# A last-candidate error is already the clearest terminal
# result; do not append an empty-completion error as well.
yield chunk
return
yield chunk
continue
event_data = {}
is_done = chunk.startswith("data: [DONE]")
if chunk.startswith("data: ") and not is_done:
try:
event_data = json.loads(chunk[6:])
except Exception:
pass
delta = event_data.get("delta")
event_type = event_data.get("type")
substantive = (
isinstance(delta, str) and bool(delta.strip())
) or (
event_type == "tool_calls"
and bool(event_data.get("calls"))
)
if substantive and not emitted:
# First real output from a NON-primary candidate: tell the client
# the selected model failed and another answered. Without this the
# fallback is invisible — a misconfigured provider looks like it
# works because the reply is shown under the originally selected
# model's name (e.g. a Bedrock/Claude endpoint that 400s every
# request but appears fine because another model silently answered).
if i > 0:
primary_reason = (
failures[0]["reason"]
if failures
else _summarize_stream_error(last_error)
)
yield ('data: ' + json.dumps({
"type": "fallback",
"selected_model": primary_model,
"answered_by": model,
"selected_endpoint_id": primary_route.get("endpoint_id"),
"selected_endpoint_label": primary_route.get("endpoint_label"),
"selected_endpoint_cost_tracked": primary_route.get("endpoint_cost_tracked"),
"answered_by_endpoint_id": route_descriptors[i].get("endpoint_id"),
"answered_by_endpoint_label": route_descriptors[i].get("endpoint_label"),
"answered_by_endpoint_cost_tracked": route_descriptors[i].get("endpoint_cost_tracked"),
"candidate_index": i,
"reason": primary_reason,
"failures": [
{
"candidate_index": failure["candidate_index"],
"model": failure["model"],
"status": failure["status"],
}
for failure in failures
],
}) + '\n\n')
# Metadata must not commit a candidate. Once real output arrives,
# flush it after any fallback notice and before the output itself.
for metadata_chunk in pending_metadata:
yield metadata_chunk
pending_metadata.clear()
emitted = True
if substantive or emitted:
yield chunk
elif not is_done:
pending_metadata.append(chunk)
finally:
close_candidate = getattr(candidate_stream, "aclose", None)
if callable(close_candidate):
try:
await close_candidate()
except Exception as close_error:
logger.warning(
"[fallback] failed to close candidate %s stream: %s",
model,
type(close_error).__name__,
)
if emitted:
return
if retried:
continue
if not is_last and fallback_on_empty:
last_error = f'event: error\ndata: {json.dumps({"error": f"Model {model} returned no substantive output", "status": 502})}\n\n'
failures.append({
"candidate_index": i,
"model": model,
"status": 502,
"reason": _summarize_stream_error(last_error),
})
tag = "primary" if i == 0 else "candidate"
logger.warning(f"[fallback] {tag} {model} returned no substantive output; trying next")
continue
if not is_last:
yield f'event: error\ndata: {json.dumps({"error": f"Model {model} returned no substantive output", "status": 502})}\n\n'
return
yield f'event: error\ndata: {json.dumps({"error": "All model candidates returned no substantive output", "status": 502})}\n\n'
return