"""
AbstractCore Server - Universal LLM Gateway with Media Processing

A focused FastAPI server that provides OpenAI-compatible endpoints with support for
multiple agent formats, tool calling, and comprehensive media processing capabilities.

Key Features:
- Universal tool call syntax conversion (OpenAI, Codex, Qwen3, LLaMA3, custom)
- Auto-detection of target agent format
- Media processing for images, documents, and data files
- OpenAI Vision API compatible format support
- Streaming responses with media attachments
- Clean delegation to AbstractCore
- Proper ReAct loop support
- Comprehensive model listing from AbstractCore providers

Media Support:
- Images: PNG, JPEG, GIF, WEBP, BMP, TIFF
- Documents: PDF, DOCX, XLSX, PPTX
- Data: CSV, TSV, JSON, XML, TXT, MD
- Size limits: 10MB per file, 32MB total per request
- Both base64 data URLs and HTTP URLs supported
"""

import os
import json
import time
import uuid
import hashlib
import base64
import tempfile
import urllib.request
import urllib.parse
import argparse
import sys
import logging
import threading
import warnings
import socket
import queue
import httpx
import ipaddress
import fnmatch
import posixpath
import hmac
from pathlib import Path
from typing import List, Dict, Any, Optional, Literal, Union, Iterator, Tuple, Annotated
from enum import Enum
from concurrent.futures import ThreadPoolExecutor
from fastapi import FastAPI, HTTPException, Request, Query, Body, Path as FastAPIPath
from fastapi.openapi.docs import get_redoc_html, get_swagger_ui_oauth2_redirect_html
from fastapi.openapi.utils import get_openapi
from fastapi.responses import StreamingResponse, JSONResponse, HTMLResponse, Response
from fastapi.middleware.cors import CORSMiddleware
from fastapi.exceptions import RequestValidationError
from pydantic import BaseModel, Field, ValidationError, AliasChoices, model_validator
from starlette.exceptions import HTTPException as StarletteHTTPException

from ..core.bloc_kv import (
    BlocKVArtifactInUseError,
    delete_bloc,
    delete_bloc_kv_artifact,
    ensure_bloc_kv_artifact,
    find_bloc_kv_live_bindings,
    list_bloc_kv_artifacts,
    load_bloc_kv_artifact,
    prune_bloc_kv_artifacts,
    read_bloc_kv_manifest,
)
from ..capabilities.registry import CapabilityRegistry
from ..capabilities.errors import CapabilityUnavailableError
from ..core.factory import create_llm
from ..core.file_blocs import FileBlocStore
from ..embeddings.models import (
    list_available_providers as _list_embedding_providers,
    list_direct_embedding_providers as _list_direct_embedding_providers,
)
from ..exceptions import AuthenticationError, InvalidRequestError, ModelNotFoundError, ProviderAPIError, RateLimitError
from ..providers.base import PromptCacheError
from ..utils.structured_logging import get_logger, configure_logging
from ..utils.version import __version__
from ..utils.message_preprocessor import MessagePreprocessor
# Removed simple_model_discovery import - now using provider methods directly
from ..tools.syntax_rewriter import (
    ToolCallSyntaxRewriter,
    SyntaxFormat,
    auto_detect_format,
    create_openai_rewriter,
    create_codex_rewriter,
    create_passthrough_rewriter
)

# ============================================================================
# Dev sys.path helpers
# ============================================================================

def _maybe_prefer_sibling_dev_packages() -> None:
    """Prefer sibling checkouts (../abstractvision, ../abstractmusic, ../abstractvoice) when present.

    AbstractCore is often developed alongside sibling repos inside a monorepo-like
    folder. The server loads capability plugins via installed entry points, but
    the entry points import Python modules by name. By prepending sibling repo
    paths to `sys.path`, we ensure those imports resolve to the local working
    copy instead of an older site-packages install.
    """
    raw = str(os.getenv("ABSTRACTCORE_DEV_PREFER_SIBLINGS", "1") or "").strip().lower()
    if raw in {"0", "false", "no", "off"}:
        return

    try:
        repo_root = Path(__file__).resolve().parents[2]  # .../abstractcore
        mono_root = repo_root.parent  # .../abstractframework
        candidates = [
            mono_root / "abstractvision" / "src",
            mono_root / "abstractmusic" / "src",
            mono_root / "abstractvoice",
        ]
        for path in reversed([p for p in candidates if p.exists()]):
            text = str(path)
            if text not in sys.path:
                sys.path.insert(0, text)
    except Exception:
        # Never fail server import due to dev convenience.
        return


_maybe_prefer_sibling_dev_packages()

# ============================================================================
# Configuration
# ============================================================================

# Initialize with default logging configuration (can be overridden later).
#
# IMPORTANT: default console verbosity is controlled by AbstractCore's centralized logging defaults
# (and env overrides like ABSTRACTCORE_CONSOLE_LOG_LEVEL). The server must not force INFO-level
# console logs on startup.
debug_mode = os.getenv("ABSTRACTCORE_DEBUG", "false").lower() == "true"

if debug_mode:
    configure_logging(
        console_level=logging.DEBUG,
        file_level=logging.DEBUG,
        log_dir="logs",
        verbatim_enabled=True,
        console_json=False,
        file_json=True,
    )

def _configure_warning_logging() -> None:
    """Route Python warnings through structured logging."""
    def _showwarning(message, category, filename, lineno, file=None, line=None):
        log = get_logger("server")
        log.warning(
            "Python warning",
            category=getattr(category, "__name__", str(category)),
            warning_message=str(message),
            location=f"{filename}:{lineno}",
        )

    warnings.showwarning = _showwarning


# Get initial logger
logger = get_logger("server")
_configure_warning_logging()


def _apply_centralized_config_env() -> None:
    """Load centralized config so persisted keys/server settings reach os.environ.

    The server's auth middleware runs before provider creation, so the server
    auth token and auth hardening knobs must be available at module import time,
    not only when `create_llm(...)` eventually imports the provider registry.
    """
    raw_disable = str(os.getenv("ABSTRACTCORE_SERVER_DISABLE_CENTRALIZED_CONFIG") or "").strip().lower()
    if raw_disable in {"1", "true", "yes", "on"}:
        return

    try:
        from ..config import get_config_manager

        get_config_manager()
    except Exception as exc:
        logger.debug(
            "Centralized config env injection skipped",
            error_type=type(exc).__name__,
            error=str(exc),
        )


_apply_centralized_config_env()

# Log initial startup with debug mode status (may be suppressed by console level).
logger.info("🚀 AbstractCore Server Initializing", version=__version__, debug_mode=debug_mode)


_REDACTED = "[REDACTED]"
_SENSITIVE_FIELD_NAMES = {
    "api_key",
    "apikey",
    "api-key",
    "access_token",
    "auth_token",
    "authorization",
    "bearer",
    "cookie",
    "id_token",
    "key",
    "password",
    "refresh_token",
    "secret",
    "set-cookie",
    "token",
    "x-api-key",
}
_SENSITIVE_FIELD_FRAGMENTS = ("api_key", "api-key", "apikey", "authorization", "password", "secret", "token")
_PLACEHOLDER_API_KEYS = {"not-needed", "not_needed", "notneeded", "unused", "dummy", "empty", "none"}
_SERVER_AUTH_TOKEN_ENV_VAR = "ABSTRACTCORE_AUTH_TOKEN"
_PROVIDER_API_KEY_HEADERS = (
    "x-abstractcore-provider-api-key",
    "x-provider-api-key",
)


def _is_sensitive_name(name: Any) -> bool:
    key = str(name or "").strip().lower()
    if not key:
        return False
    return key in _SENSITIVE_FIELD_NAMES or any(fragment in key for fragment in _SENSITIVE_FIELD_FRAGMENTS)


def _redact_sensitive_data(value: Any) -> Any:
    """Recursively redact known secret-bearing fields for logs and validation responses."""
    if isinstance(value, dict):
        return {
            k: (_REDACTED if _is_sensitive_name(k) else _redact_sensitive_data(v))
            for k, v in value.items()
        }
    if isinstance(value, list):
        return [_redact_sensitive_data(v) for v in value]
    if isinstance(value, tuple):
        return tuple(_redact_sensitive_data(v) for v in value)
    return value


def _redact_url(url: Any) -> str:
    """Redact sensitive query params and userinfo from a URL before logging."""
    raw = str(url or "")
    try:
        parsed = urllib.parse.urlsplit(raw)
    except Exception:
        return raw

    query = urllib.parse.parse_qsl(parsed.query, keep_blank_values=True)
    redacted_query = [
        (k, _REDACTED if _is_sensitive_name(k) else v)
        for k, v in query
    ]

    netloc = parsed.netloc
    if parsed.hostname:
        host = parsed.hostname
        if ":" in host and not host.startswith("["):
            host = f"[{host}]"
        try:
            parsed_port = parsed.port
        except Exception:
            parsed_port = None
        port = f":{parsed_port}" if parsed_port is not None else ""
        netloc = f"{host}{port}"

    return urllib.parse.urlunsplit(
        (
            parsed.scheme,
            netloc,
            parsed.path,
            urllib.parse.urlencode(redacted_query, doseq=True),
            parsed.fragment,
        )
    )


def _redact_headers(headers: Any) -> Dict[str, Any]:
    """Redact sensitive request/response headers before logging."""
    try:
        items = headers.items()
    except Exception:
        return {}
    return {
        str(k): (_REDACTED if _is_sensitive_name(k) else v)
        for k, v in items
    }


def _redact_text(text: Any) -> str:
    """Best-effort redaction for strings that may contain URLs with secret query params."""
    raw = str(text or "")
    try:
        return _redact_url(raw)
    except Exception:
        return raw


def _server_auth_token() -> str:
    """Return the configured inbound server auth token, if any."""
    value = os.getenv(_SERVER_AUTH_TOKEN_ENV_VAR)
    if isinstance(value, str) and value.strip():
        return value.strip()
    return ""


def _server_auth_enabled() -> bool:
    return bool(_server_auth_token())


def _server_allows_unauthenticated() -> bool:
    raw = str(os.getenv("ABSTRACTCORE_SERVER_ALLOW_UNAUTHENTICATED") or "").strip().lower()
    return raw in {"1", "true", "yes", "on"}


def _server_protect_docs() -> bool:
    raw = str(os.getenv("ABSTRACTCORE_SERVER_PROTECT_DOCS") or "").strip().lower()
    return raw in {"1", "true", "yes", "on"}


def _extract_bearer_token(auth_header: Any) -> str:
    auth = str(auth_header or "").strip()
    if not auth.lower().startswith("bearer "):
        return ""
    return auth[7:].strip()


def _extract_secret_header(header_value: Any) -> str:
    value = str(header_value or "").strip()
    if value.lower().startswith("bearer "):
        return value[7:].strip()
    return value


def _is_placeholder_api_key(token: Any) -> bool:
    t = str(token or "").strip()
    return not t or t.lower() in _PLACEHOLDER_API_KEYS


def _request_is_auth_exempt(request: Request) -> bool:
    path = str(getattr(request.url, "path", "") or "")
    if request.method.upper() == "OPTIONS":
        return True
    if path in {"/", "/favicon.ico", "/health"}:
        return True
    if not _server_protect_docs() and path in {"/docs", "/docs-lite", "/docs/oauth2-redirect", "/openapi.json", "/redoc"}:
        return True
    return False


def _unauthorized_response(message: str) -> JSONResponse:
    return JSONResponse(
        status_code=401,
        content={"error": {"message": message, "type": "authentication_error"}},
        headers={"WWW-Authenticate": "Bearer"},
    )


async def _enforce_server_auth(request: Request, call_next):
    """Enforce the server auth-token boundary before endpoint/provider creation."""
    if _request_is_auth_exempt(request):
        return await call_next(request)

    expected = _server_auth_token()
    if not expected:
        # No server auth token is configured. In this mode the server is either:
        # - explicitly configured to allow unauthenticated access (local/dev), or
        # - used with an explicit upstream provider key override header.
        if (
            _server_allows_unauthenticated()
            or _provider_api_key_from_request(request)
        ):
            return await call_next(request)
        return JSONResponse(
            status_code=503,
            content={
                "error": {
                    "message": (
                        "Server authentication is required but ABSTRACTCORE_AUTH_TOKEN is not configured. "
                        "Set ABSTRACTCORE_AUTH_TOKEN, set "
                        "ABSTRACTCORE_SERVER_ALLOW_UNAUTHENTICATED=1 only for "
                        "intentional local/dev use, or pass an explicit upstream provider key via "
                        "X-AbstractCore-Provider-API-Key or Authorization: Bearer <provider-key>."
                    ),
                    "type": "server_auth_not_configured",
                }
            },
        )

    provided = _extract_bearer_token(request.headers.get("authorization"))
    if not provided:
        return _unauthorized_response("Missing server Authorization bearer token")
    if not hmac.compare_digest(provided, expected):
        return _unauthorized_response("Invalid server Authorization bearer token")

    request.state.abstractcore_server_authenticated = True
    return await call_next(request)


def _request_has_server_auth(http_request: Optional[Request]) -> bool:
    if http_request is None:
        return False
    return bool(getattr(http_request.state, "abstractcore_server_authenticated", False))


def _provider_api_key_from_request(http_request: Request) -> Optional[str]:
    """Resolve the explicit upstream provider key for this request.

    Contract:
    - `Authorization` is the AbstractCore server auth token whenever server auth is configured.
    - `X-AbstractCore-Provider-API-Key` is the per-request upstream provider override.
    - For backwards compatibility and Swagger UI convenience, when server auth is
      NOT configured, `Authorization: Bearer <token>` is treated as an upstream
      provider key.
    """
    for header_name in _PROVIDER_API_KEY_HEADERS:
        token = _extract_secret_header(http_request.headers.get(header_name))
        if not _is_placeholder_api_key(token):
            return token

    # When server auth is enabled, `Authorization` is reserved for the server
    # auth token and must never be forwarded upstream.
    if _server_auth_enabled():
        return None

    # When server auth is disabled, accept `Authorization` as the explicit
    # upstream provider key (legacy behavior, and what Swagger UI "Authorize"
    # naturally supports).
    token = _extract_bearer_token(http_request.headers.get("authorization"))
    if _is_placeholder_api_key(token):
        return None
    return token

def reconfigure_for_debug():
    """Reconfigure logging for debug mode when --debug flag is used."""
    global debug_mode, logger

    debug_mode = True

    # Reconfigure with debug levels
    configure_logging(
        console_level=logging.DEBUG,
        file_level=logging.DEBUG,
        log_dir="logs",
        verbatim_enabled=True,
        console_json=False,
        file_json=True
    )

    # Update logger instance
    logger = get_logger("server")
    _configure_warning_logging()

    return logger

# Create FastAPI app (will be initialized after argument parsing)
app = FastAPI(
    title="AbstractCore Server",
    description=(
        "Universal LLM Gateway with Multi-Agent Tool Call Syntax Support and Media Processing.\n\n"
        "When server authentication is enabled, click Authorize in Swagger UI and enter the "
        "`ABSTRACTCORE_AUTH_TOKEN` value. "
        "The UI sends it as `Authorization: Bearer <token>` "
        "for protected API calls."
    ),
    version=__version__,
    docs_url=None,
    redoc_url=None,
    swagger_ui_parameters={
        "persistAuthorization": True,
        "displayRequestDuration": True,
        "tryItOutEnabled": True,
    },
)


def _offline_openapi_docs_html(*, title: str, openapi_url: str = "/openapi.json") -> str:
    """Return a dependency-free OpenAPI viewer for offline installs."""
    safe_title = str(title or "API docs")
    safe_openapi_url = str(openapi_url or "/openapi.json")
    return (
        """<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <title>__TITLE__</title>
  <style>
    :root { color-scheme: dark; --bg:#0b1020; --panel:#121a30; --text:#e8edff; --muted:#9ca8cc; --line:#2b3860; --accent:#35d0ff; --green:#39d98a; --blue:#6ea8fe; --orange:#ffb454; --red:#ff6b6b; }
    * { box-sizing: border-box; }
    body { margin:0; min-height:100vh; font:15px/1.55 ui-sans-serif,-apple-system,BlinkMacSystemFont,"Segoe UI",sans-serif; color:var(--text); background:radial-gradient(circle at 20% 10%,rgba(53,208,255,.14),transparent 30rem),radial-gradient(circle at 80% 0%,rgba(255,107,107,.12),transparent 28rem),var(--bg); }
    main { max-width:1180px; margin:0 auto; padding:40px 24px 64px; }
    header { display:flex; align-items:flex-start; justify-content:space-between; gap:24px; margin-bottom:26px; }
    h1 { margin:0 0 8px; font-size:clamp(2rem,4vw,3.4rem); letter-spacing:-.04em; }
    p { margin:0; color:var(--muted); }
    a { color:var(--accent); text-decoration:none; }
    a:hover { text-decoration:underline; }
    .actions { display:flex; gap:10px; flex-wrap:wrap; justify-content:flex-end; }
    .button { display:inline-flex; align-items:center; border:1px solid var(--line); border-radius:999px; padding:10px 14px; background:rgba(255,255,255,.05); color:var(--text); font-weight:700; }
    .notice { border:1px solid rgba(53,208,255,.35); border-radius:18px; padding:16px 18px; background:rgba(53,208,255,.08); margin-bottom:24px; }
    .toolbar { display:grid; grid-template-columns:minmax(0,1fr) auto; gap:12px; margin-bottom:18px; }
    input { width:100%; border:1px solid var(--line); border-radius:14px; padding:13px 14px; background:rgba(0,0,0,.18); color:var(--text); font:inherit; }
    .status { color:var(--muted); align-self:center; white-space:nowrap; }
    .endpoint { border:1px solid var(--line); border-radius:18px; background:rgba(18,26,48,.88); margin:12px 0; overflow:hidden; }
    .endpoint summary { cursor:pointer; padding:15px 18px; display:grid; grid-template-columns:82px minmax(0,1fr); gap:14px; align-items:center; }
    .method { display:inline-flex; justify-content:center; border-radius:999px; padding:5px 10px; font:800 12px/1 ui-monospace,SFMono-Regular,Menlo,monospace; letter-spacing:.08em; color:#06101f; background:var(--blue); }
    .method.POST { background:var(--green); } .method.PUT,.method.PATCH { background:var(--orange); } .method.DELETE { background:var(--red); }
    .path { font:700 15px/1.35 ui-monospace,SFMono-Regular,Menlo,monospace; overflow-wrap:anywhere; }
    .details { padding:0 18px 18px 114px; color:var(--muted); }
    .details code { display:inline-block; margin:4px 6px 4px 0; padding:4px 8px; border-radius:999px; background:rgba(255,255,255,.07); color:var(--text); }
    .error { border:1px solid rgba(255,107,107,.55); background:rgba(255,107,107,.1); border-radius:18px; padding:18px; color:#ffd0d0; }
    @media (max-width:720px) { header,.toolbar { display:block; } .actions { justify-content:flex-start; margin-top:18px; } .status { margin-top:10px; display:block; } .endpoint summary { grid-template-columns:1fr; } .details { padding-left:18px; } }
  </style>
</head>
<body>
  <main>
    <header>
      <div>
        <h1>__TITLE__</h1>
        <p id="description">Offline OpenAPI documentation. No CDN, no internet required.</p>
      </div>
      <nav class="actions">
        <a class="button" href="__OPENAPI_URL__">Open JSON</a>
        <a class="button" href="/health">Health</a>
      </nav>
    </header>
    <section class="notice">This lightweight documentation view lists endpoints from <a href="__OPENAPI_URL__">__OPENAPI_URL__</a>. Protected routes still require configured server authentication when called by clients.</section>
    <section class="toolbar"><input id="filter" autocomplete="off" placeholder="Filter endpoints, methods, summaries..."><span class="status" id="status">Loading OpenAPI...</span></section>
    <section id="routes"></section>
  </main>
  <script>
    const openapiUrl = "__OPENAPI_URL__";
    const methods = new Set(["get","post","put","patch","delete","head","options"]);
    const state = { entries: [] };
    const text = (value) => value == null ? "" : String(value);
    function render() {
      const q = document.getElementById("filter").value.trim().toLowerCase();
      const routes = document.getElementById("routes");
      routes.textContent = "";
      const filtered = state.entries.filter((entry) => !q || [entry.method, entry.path, entry.summary, entry.description, entry.tags.join(" ")].join(" ").toLowerCase().includes(q));
      document.getElementById("status").textContent = `${filtered.length} / ${state.entries.length} endpoints`;
      for (const entry of filtered) {
        const details = document.createElement("details");
        details.className = "endpoint";
        const summary = document.createElement("summary");
        const method = document.createElement("span");
        method.className = `method ${entry.method}`;
        method.textContent = entry.method;
        const path = document.createElement("span");
        path.className = "path";
        path.textContent = entry.path;
        summary.append(method, path);
        const body = document.createElement("div");
        body.className = "details";
        const line = document.createElement("p");
        line.textContent = entry.summary || entry.description || "No summary.";
        body.append(line);
        if (entry.tags.length) {
          const tags = document.createElement("p");
          tags.append("Tags: ");
          for (const tag of entry.tags) {
            const code = document.createElement("code");
            code.textContent = tag;
            tags.append(code);
          }
          body.append(tags);
        }
        if (entry.parameters.length) {
          const params = document.createElement("p");
          params.append("Parameters: ");
          for (const param of entry.parameters) {
            const code = document.createElement("code");
            code.textContent = `${param.name || "?"}:${param.in || "?"}`;
            params.append(code);
          }
          body.append(params);
        }
        if (entry.requestBody) {
          const req = document.createElement("p");
          req.textContent = "Request body: yes";
          body.append(req);
        }
        details.append(summary, body);
        routes.append(details);
      }
    }
    async function load() {
      try {
        const response = await fetch(openapiUrl, { headers: { "Accept": "application/json" } });
        if (!response.ok) throw new Error(`HTTP ${response.status}`);
        const schema = await response.json();
        document.title = `${schema.info?.title || "__TITLE__"} - API docs`;
        document.getElementById("description").textContent = schema.info?.description || "Offline OpenAPI documentation.";
        const entries = [];
        for (const [path, operations] of Object.entries(schema.paths || {})) {
          for (const [method, op] of Object.entries(operations || {})) {
            if (!methods.has(method)) continue;
            entries.push({ method: method.toUpperCase(), path, summary: text(op.summary), description: text(op.description), tags: Array.isArray(op.tags) ? op.tags.map(text) : [], parameters: Array.isArray(op.parameters) ? op.parameters : [], requestBody: Boolean(op.requestBody) });
          }
        }
        entries.sort((a, b) => a.path.localeCompare(b.path) || a.method.localeCompare(b.method));
        state.entries = entries;
        render();
      } catch (error) {
        const routes = document.getElementById("routes");
        routes.textContent = "";
        const box = document.createElement("div");
        box.className = "error";
        box.textContent = `Could not load ${openapiUrl}: ${error}`;
        routes.append(box);
        document.getElementById("status").textContent = "OpenAPI load failed";
      }
    }
    document.getElementById("filter").addEventListener("input", render);
    load();
  </script>
</body>
</html>"""
        .replace("__TITLE__", safe_title)
        .replace("__OPENAPI_URL__", safe_openapi_url)
    )


@app.get("/docs-lite", include_in_schema=False)
async def offline_openapi_docs_html() -> HTMLResponse:
    """Serve dependency-free OpenAPI docs for offline installs."""
    return HTMLResponse(_offline_openapi_docs_html(title=f"{app.title} API Docs"))


@app.get("/docs", include_in_schema=False)
async def custom_swagger_ui_html() -> HTMLResponse:
    """Serve Swagger UI with a small media-preview fix for binary audio POSTs."""

    html = """
<!DOCTYPE html>
<html>
<head>
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<link type="text/css" rel="stylesheet" href="https://cdn.jsdelivr.net/npm/swagger-ui-dist@5.32.6/swagger-ui.css">
<link rel="shortcut icon" href="https://fastapi.tiangolo.com/img/favicon.png">
<title>AbstractCore Server - Swagger UI</title>
<style>
body { margin: 0; }
</style>
</head>
<body>
<div id="swagger-ui"></div>
<script src="https://cdn.jsdelivr.net/npm/swagger-ui-dist@5.32.6/swagger-ui-bundle.js"></script>
<script>
window.__abstractcoreLatestAudioPreviewUrl = null;
const abstractcoreServerAuthEnabled = __SERVER_AUTH_ENABLED__;

function abstractcoreHeader(response, name) {
  const headers = response && response.headers ? response.headers : {};
  return headers[name] || headers[name.toLowerCase()] || headers[name.toUpperCase()] || "";
}

function abstractcoreBytesFromString(value) {
  const bytes = new Uint8Array(value.length);
  for (let index = 0; index < value.length; index += 1) {
    bytes[index] = value.charCodeAt(index) & 255;
  }
  return bytes;
}

function abstractcoreBlobFromResponse(response, contentType) {
  const data = response ? response.data : null;
  if (data instanceof Blob) {
    return data;
  }
  if (data instanceof ArrayBuffer) {
    return new Blob([data], { type: contentType });
  }
  if (data && data.buffer instanceof ArrayBuffer) {
    return new Blob([data], { type: contentType });
  }
  if (typeof data === "string" && data.length > 0) {
    return new Blob([abstractcoreBytesFromString(data)], { type: contentType });
  }
  return null;
}

function abstractcorePatchAudioPlayers() {
  const previewUrl = window.__abstractcoreLatestAudioPreviewUrl;
  if (!previewUrl) {
    return;
  }
  document.querySelectorAll("audio").forEach((audio) => {
    const current = audio.getAttribute("src") || "";
    if (current.includes("/v1/audio/speech") || current === "" || current === window.location.href) {
      if (audio.src !== previewUrl) {
        audio.src = previewUrl;
        audio.load();
      }
    }
  });
}

async function abstractcoreValidateServerToken(candidate) {
  if (!abstractcoreServerAuthEnabled) {
    return { ok: true };
  }

  const token = String(candidate || "").trim();
  if (!token) {
    return { ok: false, message: "Enter the AbstractCore server token." };
  }

  try {
    const response = await fetch("/acore/auth/validate", {
      method: "GET",
      headers: {
        "Authorization": `Bearer ${token}`,
        "Accept": "application/json"
      }
    });
    const payload = await response.json().catch(() => null);
    if (!response.ok) {
      const message = payload && payload.error && payload.error.message
        ? String(payload.error.message)
        : `Invalid server token (HTTP ${response.status})`;
      return { ok: false, message };
    }
    if (!payload || payload.authenticated !== true) {
      return { ok: false, message: "Server token was not accepted." };
    }
    return { ok: true };
  } catch (error) {
    return { ok: false, message: `Token validation failed: ${error}` };
  }
}

function abstractcoreAuthErrorsClear(system, authId) {
  if (!(system && system.errActions && typeof system.errActions.clearBy === "function")) {
    return;
  }
  system.errActions.clearBy((error) => {
    if (!(error && typeof error.get === "function")) {
      return true;
    }
    return !(error.get("source") === "auth" && error.get("authId") === authId);
  });
}

function abstractcoreAuthError(system, authId, message) {
  if (!(system && system.errActions && typeof system.errActions.newAuthErr === "function")) {
    return;
  }
  system.errActions.newAuthErr({
    authId,
    source: "auth",
    level: "error",
    message: String(message || "Invalid server token"),
  });
}

function abstractcoreAuthValue(payload, authId) {
  if (!(payload && Object.prototype.hasOwnProperty.call(payload, authId))) {
    return "";
  }
  const auth = payload[authId] || {};
  const raw = auth.value;
  if (typeof raw === "string") {
    return raw.trim();
  }
  if (raw && typeof raw.value === "string") {
    return raw.value.trim();
  }
  if (raw == null) {
    return "";
  }
  return String(raw).trim();
}

const AbstractCoreSwaggerAuthPlugin = function(system) {
  return {
    statePlugins: {
      auth: {
        wrapActions: {
          authorize: (oriAction, wrappedSystem) => (payload) => {
            const authId = "AbstractCoreBearerAuth";
            if (!abstractcoreServerAuthEnabled || !(payload && Object.prototype.hasOwnProperty.call(payload, authId))) {
              return oriAction(payload);
            }

            abstractcoreAuthErrorsClear(wrappedSystem, authId);
            const token = abstractcoreAuthValue(payload, authId);
            if (!token) {
              abstractcoreAuthError(wrappedSystem, authId, "Enter the AbstractCore server token.");
              return undefined;
            }

            return abstractcoreValidateServerToken(token).then((result) => {
              if (!result || result.ok !== true) {
                abstractcoreAuthError(wrappedSystem, authId, result && result.message ? result.message : "Invalid server token.");
                return undefined;
              }
              abstractcoreAuthErrorsClear(wrappedSystem, authId);
              return oriAction(payload);
            });
          },
        },
      },
    },
  };
};

const ui = SwaggerUIBundle({
  url: "/openapi.json",
  dom_id: "#swagger-ui",
  layout: "BaseLayout",
  deepLinking: true,
  showExtensions: true,
  showCommonExtensions: true,
  persistAuthorization: false,
  displayRequestDuration: true,
  tryItOutEnabled: true,
  oauth2RedirectUrl: window.location.origin + "/docs/oauth2-redirect",
  responseInterceptor: (response) => {
    const contentType = String(abstractcoreHeader(response, "content-type")).split(";")[0].trim().toLowerCase();
    if (contentType.startsWith("audio/")) {
      const blob = abstractcoreBlobFromResponse(response, contentType);
      if (blob && blob.size > 0) {
        if (window.__abstractcoreLatestAudioPreviewUrl) {
          URL.revokeObjectURL(window.__abstractcoreLatestAudioPreviewUrl);
        }
        const previewUrl = URL.createObjectURL(blob);
        window.__abstractcoreLatestAudioPreviewUrl = previewUrl;
        response.url = previewUrl;
        response.text = previewUrl;
        response.obj = previewUrl;
        window.setTimeout(abstractcorePatchAudioPlayers, 0);
        window.setTimeout(abstractcorePatchAudioPlayers, 100);
        window.setTimeout(abstractcorePatchAudioPlayers, 500);
      }
    }
    return response;
  },
  presets: [
    SwaggerUIBundle.presets.apis,
    SwaggerUIBundle.SwaggerUIStandalonePreset
  ],
  plugins: [
    AbstractCoreSwaggerAuthPlugin
  ],
});

const abstractcoreSwaggerRoot = document.getElementById("swagger-ui");
if (abstractcoreSwaggerRoot) {
  const abstractcoreObserver = new MutationObserver(abstractcorePatchAudioPlayers);
  abstractcoreObserver.observe(abstractcoreSwaggerRoot, {
    childList: true,
    subtree: true,
    attributes: true,
    attributeFilter: ["src"]
  });
}
</script>
</body>
</html>
    """
    return HTMLResponse(html.replace("__SERVER_AUTH_ENABLED__", "true" if _server_auth_enabled() else "false"))


@app.get("/docs/oauth2-redirect", include_in_schema=False)
async def swagger_ui_redirect() -> HTMLResponse:
    return get_swagger_ui_oauth2_redirect_html()


@app.get("/redoc", include_in_schema=False)
async def redoc_html() -> HTMLResponse:
    return get_redoc_html(openapi_url="/openapi.json", title=f"{app.title} - ReDoc")


@app.get("/acore/auth/validate", include_in_schema=False)
async def validate_server_auth_token(request: Request) -> JSONResponse:
    """Validate the inbound server token for the docs UI."""
    return JSONResponse(
        {
            "ok": True,
            "server_auth_enabled": _server_auth_enabled(),
            "authenticated": _request_has_server_auth(request),
        }
    )
def _custom_openapi() -> Dict[str, Any]:
    if app.openapi_schema:
        return app.openapi_schema

    schema = get_openapi(
        title=app.title,
        version=app.version,
        description=app.description,
        routes=app.routes,
    )
    components = schema.setdefault("components", {})
    if _server_auth_enabled():
        security_schemes = components.setdefault("securitySchemes", {})
        security_schemes["AbstractCoreBearerAuth"] = {
            "type": "http",
            "scheme": "bearer",
            "bearerFormat": "server auth token",
            "description": (
                "Inbound AbstractCore server token. Use the value configured in "
                "ABSTRACTCORE_AUTH_TOKEN. Provider-key overrides still use the "
                "X-AbstractCore-Provider-API-Key header."
            ),
        }
    schemas = components.setdefault("schemas", {})

    def _register_schema_model(model_cls: type[BaseModel]) -> None:
        try:
            model_schema = model_cls.model_json_schema(ref_template="#/components/schemas/{model}")
        except Exception:
            return
        defs = model_schema.pop("$defs", {})
        if isinstance(defs, dict):
            for name, definition in defs.items():
                if isinstance(name, str) and isinstance(definition, dict):
                    schemas.setdefault(name, definition)
        schemas.setdefault(model_cls.__name__, model_schema)

    _register_schema_model(OpenAIResponsesRequest)
    _register_schema_model(ChatCompletionRequest)

    schemas.setdefault(
        "AbstractCoreError",
        {
            "type": "object",
            "required": ["error"],
            "properties": {
                "error": {
                    "type": "object",
                    "required": ["message", "type"],
                    "properties": {
                        "message": {"type": "string", "description": "Human-readable error message."},
                        "type": {"type": "string", "description": "Stable error category such as `authentication_error`, `invalid_request_error`, or `http_error`."},
                        "code": {"type": "string", "nullable": True, "description": "Optional provider/server error code."},
                    },
                }
            },
            "examples": [
                {
                    "error": {
                        "message": "Invalid server Authorization bearer token",
                        "type": "authentication_error",
                    }
                }
            ],
        },
    )

    def _component_examples(name: str, example: Dict[str, Any]) -> None:
        comp = schemas.get(name)
        if isinstance(comp, dict):
            comp["examples"] = [example]

    def _request_example(path: str, method: str, content_type: str, name: str, summary: str, value: Dict[str, Any]) -> None:
        operation = schema.get("paths", {}).get(path, {}).get(method.lower())
        if not isinstance(operation, dict):
            return
        content = operation.get("requestBody", {}).get("content", {})
        media = content.get(content_type)
        if isinstance(media, dict):
            examples = media.get("examples")
            if not isinstance(examples, dict):
                examples = {}
                media["examples"] = examples
            examples[name] = {"summary": summary, "value": value}

    _request_example(
        "/v1/responses",
        "post",
        "application/json",
        "responses_openai_format",
        "Responses API format",
        {
            "model": "openai/gpt-4o",
            "input": [
                {
                    "role": "user",
                    "content": [
                        {"type": "input_text", "text": "Analyze this document"},
                        {"type": "input_file", "file_url": "https://example.com/doc.pdf"},
                    ],
                }
            ],
            "thinking": "off",
            "prompt_cache_key": "tenantA:doc-review",
            "stream": False,
        },
    )
    _request_example(
        "/v1/responses",
        "post",
        "application/json",
        "responses_legacy_format",
        "Legacy chat format",
        {
            "model": "openai/gpt-4",
            "messages": [{"role": "user", "content": "Tell me a story"}],
            "thinking": "high",
            "stream": False,
        },
    )

    _component_examples(
        "AudioSpeechRequest",
        {
            "model": "openai/gpt-4o-mini-tts",
            "input": "Hello from AbstractCore.",
            "text": None,
            "voice": "coral",
            "profile": None,
            "response_format": "wav",
            "format": None,
            "speed": 1.0,
            "quality_preset": "standard",
            "quality": None,
            "instructions": "Speak clearly and calmly.",
            "provider": {},
            "base_url": None,
        },
    )
    _component_examples(
        "AudioMusicRequest",
        {
            "model": "acemusic/ace-step-api",
            "provider": "acemusic",
            "prompt": "A short calm piano loop.",
            "input": None,
            "text": None,
            "lyrics": None,
            "duration_s": 8,
            "response_format": "wav",
            "format": None,
        },
    )
    image_generation_example = {
        "model": "openai-compatible/gpt-image-1",
        "provider": "openai-compatible",
        "prompt": "A precise product photo of a red ceramic mug on a white table.",
        "n": 1,
        "base_url": None,
        "size": "1024x1024",
        "width": 1024,
        "height": 1024,
        "response_format": "b64_json",
        "negative_prompt": None,
        "seed": None,
        "steps": None,
        "guidance_scale": None,
        "quality": "low",
        "style": None,
        "user": "swagger-user",
        "background": "auto",
        "output_format": "png",
        "output_compression": None,
        "moderation": "auto",
        "extra": {},
    }
    _component_examples("ImageGenerationBody", image_generation_example)
    _component_examples(
        "PromptCacheSetProxyRequest",
        {"base_url": "http://localhost:8001/v1", "api_key": None, "key": "project-default", "make_default": True, "ttl_s": 3600},
    )
    _component_examples(
        "PromptCacheUpdateProxyRequest",
        {
            "base_url": "http://localhost:8001/v1",
            "api_key": None,
            "key": "project-default",
            "prompt": "You are a helpful assistant.",
            "messages": None,
            "system_prompt": None,
            "tools": None,
            "thinking": "off",
            "add_generation_prompt": False,
            "ttl_s": 3600,
        },
    )
    _component_examples(
        "PromptCacheForkProxyRequest",
        {"base_url": "http://localhost:8001/v1", "api_key": None, "from_key": "project-default", "to_key": "project-branch", "make_default": False, "ttl_s": 3600},
    )
    _component_examples(
        "PromptCacheClearProxyRequest",
        {"base_url": "http://localhost:8001/v1", "api_key": None, "key": "project-default"},
    )
    _component_examples(
        "PromptCachePrepareModulesProxyRequest",
        {
            "base_url": "http://localhost:8001/v1",
            "api_key": None,
            "namespace": "abstractcore.tools",
            "modules": [{"name": "calculator", "version": "1.0"}],
            "make_default": False,
            "ttl_s": 3600,
            "version": 1,
        },
    )
    _request_example(
        "/v1/audio/speech",
        "post",
        "application/json",
        "remote_openai",
        "Remote OpenAI speech",
        {
            "model": "openai/gpt-4o-mini-tts",
            "input": "Hello from AbstractCore.",
            "text": None,
            "voice": "coral",
            "profile": None,
            "response_format": "wav",
            "format": None,
            "speed": 1.0,
            "quality_preset": "standard",
            "quality": None,
            "instructions": "Speak clearly and calmly.",
            "provider": {},
            "base_url": None,
        },
    )
    _request_example(
        "/v1/audio/music",
        "post",
        "application/json",
        "remote_ace_music",
        "Remote ACE Music",
        {
            "model": "acemusic/ace-step-api",
            "provider": "acemusic",
            "prompt": "A short calm piano loop.",
            "input": None,
            "text": None,
            "lyrics": None,
            "duration_s": 8,
            "response_format": "wav",
            "format": None,
        },
    )
    _request_example(
        "/v1/audio/transcriptions",
        "post",
        "multipart/form-data",
        "remote_openai",
        "Remote OpenAI transcription",
        {
            "file": "<upload audio.mp3>",
            "model": "openai/gpt-4o-mini-transcribe",
            "provider": None,
            "language": "en",
            "prompt": "Technical discussion about AbstractCore endpoints.",
            "response_format": "json",
            "temperature": 0.0,
            "format": None,
            "base_url": None,
        },
    )
    _request_example(
        "/{provider}/v1/audio/transcriptions",
        "post",
        "multipart/form-data",
        "provider_scoped_stt",
        "Provider-scoped transcription",
        {
            "file": "<upload audio.mp3>",
            "model": "gpt-4o-mini-transcribe",
            "provider": None,
            "language": "en",
            "prompt": "Technical discussion about AbstractCore endpoints.",
            "response_format": "json",
            "temperature": 0.0,
            "format": None,
            "base_url": None,
        },
    )
    _request_example("/v1/images/generations", "post", "application/json", "remote_openai_compatible", "Remote OpenAI-compatible image generation", image_generation_example)
    _request_example("/{provider}/v1/images/generations", "post", "application/json", "provider_scoped_image_generation", "Provider-scoped image generation", {**image_generation_example, "model": "gpt-image-1", "provider": None})
    _request_example("/v1/vision/jobs/images/generations", "post", "application/json", "async_image_generation", "Async image generation job", image_generation_example)
    video_generation_example = {
        "prompt": "A slow camera move through a luminous data center.",
        "provider": "mlx-gen",
        "model": "mlx-gen/AbstractFramework/wan2.2-t2v-a14b-diffusers-8bit",
        "base_url": None,
        "width": 432,
        "height": 240,
        "fps": 24,
        "num_frames": 41,
        "steps": 20,
        "guidance_scale": 4.0,
        "response_format": "b64_json",
        "extra": {"max_sequence_length": 256},
    }
    _request_example("/v1/videos/generations", "post", "application/json", "text_to_video", "Text-to-video generation", video_generation_example)
    _request_example("/{provider}/v1/videos/generations", "post", "application/json", "provider_scoped_video_generation", "Provider-scoped text-to-video generation", {**video_generation_example, "model": "AbstractFramework/wan2.2-t2v-a14b-diffusers-8bit", "provider": None})
    _request_example("/v1/vision/jobs/videos/generations", "post", "application/json", "async_video_generation", "Async text-to-video generation job", video_generation_example)
    _request_example(
        "/v1/audio/translations",
        "post",
        "multipart/form-data",
        "not_implemented",
        "Compatibility placeholder",
        {"file": "<upload audio.mp3>", "model": "openai/whisper-1"},
    )
    image_edit_example = {
        "prompt": "Make the mug blue and keep the white background.",
        "image": "<upload source.png>",
        "mask": None,
        "reference_images": ["<optional style-reference.png>"],
        "provider": "openai-compatible",
        "model": "openai-compatible/gpt-image-1",
        "base_url": None,
        "size": "1024x1024",
        "response_format": "b64_json",
        "negative_prompt": None,
        "seed": None,
        "steps": None,
        "guidance_scale": None,
        "extra_json": '{"quality":"low","background":"auto","output_format":"png"}',
    }
    _request_example("/v1/images/edits", "post", "multipart/form-data", "remote_openai_compatible", "Remote OpenAI-compatible image edit", image_edit_example)
    _request_example("/{provider}/v1/images/edits", "post", "multipart/form-data", "provider_scoped_image_edit", "Provider-scoped image edit", {**image_edit_example, "model": "gpt-image-1"})
    _request_example("/v1/vision/jobs/images/edits", "post", "multipart/form-data", "async_image_edit", "Async image edit job", image_edit_example)
    image_upscale_example = {
        "image": "<upload source.png>",
        "provider": "mlx-gen",
        "model": "mlx-gen/AbstractFramework/seedvr2-3b-8bit",
        "base_url": None,
        "response_format": "b64_json",
        "scale": "2x",
        "resolution": "2x",
        "softness": "0.0",
        "seed": "1234",
        "quantize": None,
        "vae_tiling": "false",
    }
    _request_example("/v1/images/upscale", "post", "multipart/form-data", "local_seedvr2", "Local SeedVR2 image upscale", image_upscale_example)
    _request_example(
        "/{provider}/v1/images/upscale",
        "post",
        "multipart/form-data",
        "provider_scoped_seedvr2",
        "Provider-scoped SeedVR2 image upscale",
        {**image_upscale_example, "model": "AbstractFramework/seedvr2-3b-8bit", "provider": None},
    )
    _request_example("/v1/vision/jobs/images/upscale", "post", "multipart/form-data", "async_seedvr2", "Async SeedVR2 image upscale job", image_upscale_example)
    video_edit_example = {
        "prompt": "Slow camera push-in.",
        "image": "<upload first-frame.png>",
        "provider": "mlx-gen",
        "model": "mlx-gen/AbstractFramework/wan2.2-i2v-a14b-diffusers-8bit",
        "base_url": None,
        "width": 432,
        "height": 240,
        "fps": 24,
        "num_frames": 41,
        "steps": 20,
        "guidance_scale": 4.0,
        "response_format": "b64_json",
        "extra_json": '{"max_sequence_length":256}',
    }
    _request_example("/v1/videos/edits", "post", "multipart/form-data", "image_to_video", "Image-to-video generation", video_edit_example)
    _request_example("/{provider}/v1/videos/edits", "post", "multipart/form-data", "provider_scoped_image_to_video", "Provider-scoped image-to-video generation", {**video_edit_example, "model": "AbstractFramework/wan2.2-i2v-a14b-diffusers-8bit"})
    _request_example("/v1/vision/jobs/videos/edits", "post", "multipart/form-data", "async_image_to_video", "Async image-to-video generation job", video_edit_example)
    _request_example(
        "/v1/voice/clone",
        "post",
        "multipart/form-data",
        "compatible_clone_server",
        "AbstractVoice-compatible clone server",
        {
            "file": "<upload reference.wav>",
            "model": "openai-compatible/default",
            "provider": None,
            "name": "my_voice",
            "reference_text": "Hello from AbstractCore voice cloning.",
            "validate": True,
            "base_url": "http://127.0.0.1:5000/v1",
            "clone_path": "/voice/clone",
            "file_field": "file",
            "consent": None,
        },
    )
    _request_example(
        "/{provider}/v1/voice/clone",
        "post",
        "multipart/form-data",
        "provider_scoped_clone_server",
        "Provider-scoped voice clone",
        {
            "file": "<upload reference.wav>",
            "model": "default",
            "provider": None,
            "name": "my_voice",
            "reference_text": "Hello from AbstractCore voice cloning.",
            "validate": True,
            "base_url": "http://127.0.0.1:5000/v1",
            "clone_path": "/voice/clone",
            "file_field": "file",
            "consent": None,
        },
    )
    _request_example("/acore/prompt_cache/set", "post", "application/json", "set_cache_key", "Set prompt cache key", {"base_url": "http://localhost:8001/v1", "api_key": None, "key": "project-default", "make_default": True, "ttl_s": 3600})
    _request_example(
        "/acore/prompt_cache/update",
        "post",
        "application/json",
        "update_cache_key",
        "Update prompt cache",
        {
            "base_url": "http://localhost:8001/v1",
            "api_key": None,
            "key": "project-default",
            "prompt": "You are a helpful assistant.",
            "messages": None,
            "system_prompt": None,
            "tools": None,
            "add_generation_prompt": False,
            "ttl_s": 3600,
        },
    )
    _request_example("/acore/prompt_cache/fork", "post", "application/json", "fork_cache_key", "Fork prompt cache", {"base_url": "http://localhost:8001/v1", "api_key": None, "from_key": "project-default", "to_key": "project-branch", "make_default": False, "ttl_s": 3600})
    _request_example("/acore/prompt_cache/clear", "post", "application/json", "clear_cache_key", "Clear prompt cache", {"base_url": "http://localhost:8001/v1", "api_key": None, "key": "project-default"})
    _request_example(
        "/acore/prompt_cache/prepare_modules",
        "post",
        "application/json",
        "prepare_modules",
        "Prepare module cache",
        {
            "base_url": "http://localhost:8001/v1",
            "api_key": None,
            "namespace": "abstractcore.tools",
            "modules": [{"name": "calculator", "version": "1.0"}],
            "make_default": False,
            "ttl_s": 3600,
            "version": 1,
        },
    )

    error_response = {
        "description": "AbstractCore error response.",
        "content": {
            "application/json": {
                "schema": {"$ref": "#/components/schemas/AbstractCoreError"},
                "examples": {
                    "error": {
                        "summary": "Error",
                        "value": {
                            "error": {
                                "message": "Invalid server Authorization bearer token",
                                "type": "authentication_error",
                            }
                        },
                    }
                },
            }
        },
    }

    for path, path_item in schema.get("paths", {}).items():
        if not isinstance(path_item, dict):
            continue
        for method, operation in path_item.items():
            if method.lower() not in {"get", "put", "post", "delete", "patch", "options", "head"}:
                continue
            if not isinstance(operation, dict):
                continue
            if _server_auth_enabled():
                operation["security"] = [] if path == "/health" else [{"AbstractCoreBearerAuth": []}]
            else:
                operation.pop("security", None)
            if path != "/health":
                responses = operation.setdefault("responses", {})
                for code, description in (
                    ("400", "Bad request."),
                    ("401", "Unauthorized."),
                    ("413", "Payload too large."),
                    ("500", "Internal server error."),
                    ("501", "Capability not implemented or not configured."),
                    ("502", "Upstream/provider error."),
                ):
                    if code not in responses:
                        item = dict(error_response)
                        item["description"] = description
                        responses[code] = item

    app.openapi_schema = schema
    return app.openapi_schema


app.openapi = _custom_openapi  # type: ignore[method-assign]

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# Optional: OpenAI-compatible vision generation endpoints (/v1/images/*).
# These are safe-by-default and require explicit configuration; see `vision_endpoints.py`.
try:
    from .vision_endpoints import provider_router as _vision_provider_router
    from .vision_endpoints import router as _vision_router

    app.include_router(_vision_router, prefix="/v1")
    app.include_router(_vision_provider_router)
    logger.info("🖼️ Vision endpoints enabled at /v1/images/*")
except Exception as e:
    logger.debug(f"Vision endpoints not loaded: {e}")

# Optional: OpenAI-compatible audio endpoints (/v1/audio/*).
# These delegate to capability plugins (e.g. AbstractVoice) and degrade to 501 when unavailable.
try:
    from .audio_endpoints import provider_router as _provider_audio_router
    from .audio_endpoints import router as _audio_router

    app.include_router(_audio_router, prefix="/v1")
    app.include_router(_provider_audio_router)
    logger.info("🔊 Audio endpoints enabled at /v1/audio/*")
except Exception as e:
    logger.debug(f"Audio endpoints not loaded: {e}")

# ============================================================================
# Enhanced Error Handling and Logging Middleware
# ============================================================================

@app.middleware("http")
async def debug_logging_middleware(request: Request, call_next):
    """Enhanced logging middleware for debug mode."""
    start_time = time.time()

    # Log request details in debug mode
    if debug_mode:
        logger.debug(
            "📥 HTTP Request",
            method=request.method,
            url=_redact_url(str(request.url)),
            headers=_redact_headers(request.headers),
            client=request.client.host if request.client else "unknown"
        )

    response = await _enforce_server_auth(request, call_next)

    process_time = time.time() - start_time

    # Log response details
    log_data = {
        "method": request.method,
        "url": _redact_url(str(request.url)),
        "status_code": response.status_code,
        "process_time_ms": round(process_time * 1000, 2)
    }

    if response.status_code >= 400:
        logger.error("❌ HTTP Error Response", **log_data)
    elif debug_mode:
        logger.debug("📤 HTTP Response", **log_data)
    else:
        logger.info("✅ HTTP Request", **log_data)

    return response

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    """Enhanced handler for 422 validation errors with detailed logging."""
    error_details = []
    for error in exc.errors():
        error_details.append({
            "field": " -> ".join(str(loc) for loc in error["loc"]),
            "message": error["msg"],
            "type": error["type"],
            "input": _redact_sensitive_data(error.get("input")),
        })

    # Log detailed validation error information
    logger.error(
        "🔴 Request Validation Error (422)",
        method=request.method,
        url=_redact_url(str(request.url)),
        error_count=len(error_details),
        errors=error_details,
        client=request.client.host if request.client else "unknown"
    )

    # In debug mode, also try to log the request body if possible
    if debug_mode:
        try:
            # Try to get the request body for debugging
            body = await request.body()
            if body:
                try:
                    import json
                    body_json = json.loads(body)
                    logger.debug(
                        "📋 Request Body (Validation Error)",
                        body=_redact_sensitive_data(body_json),
                    )
                except json.JSONDecodeError:
                    raw = body.decode("utf-8", errors="replace")
                    body_text = raw
                    if len(body_text) > 1000:
                        #[WARNING:TRUNCATION] bounded request-body preview for debug logs
                        body_text = body_text[:980].rstrip() + "\n… (truncated)"
                    logger.debug(
                        "📋 Request Body (Validation Error)",
                        body_text=body_text,
                    )
        except Exception as e:
            logger.debug(f"Could not read request body for debugging: {e}")

    # Return detailed error response
    return JSONResponse(
        status_code=422,
        content={
            "error": {
                "message": "Request validation failed",
                "type": "validation_error",
                "details": error_details
            }
        }
    )

@app.exception_handler(StarletteHTTPException)
async def http_exception_handler(request: Request, exc: StarletteHTTPException):
    """Enhanced handler for HTTP exceptions with detailed logging."""
    safe_detail = _redact_sensitive_data(exc.detail)
    logger.error(
        "🔴 HTTP Exception",
        method=request.method,
        url=_redact_url(str(request.url)),
        status_code=exc.status_code,
        detail=safe_detail,
        client=request.client.host if request.client else "unknown"
    )

    return JSONResponse(
        status_code=exc.status_code,
        content={
            "error": {
                "message": str(safe_detail),
                "type": "http_error"
            }
        },
        headers=getattr(exc, "headers", None),
    )

@app.exception_handler(Exception)
async def general_exception_handler(request: Request, exc: Exception):
    """Handler for unexpected exceptions with detailed logging."""
    logger.error(
        "💥 Unexpected Server Error",
        method=request.method,
        url=_redact_url(str(request.url)),
        exception_type=type(exc).__name__,
        exception_message=_redact_text(str(exc)),
        client=request.client.host if request.client else "unknown",
        exc_info=True  # This will include the full stack trace
    )

    return JSONResponse(
        status_code=500,
        content={
            "error": {
                "message": "Internal server error",
                "type": "server_error"
            }
        }
    )

# ============================================================================
# Model Type Detection
# ============================================================================

# Import the core capability enums directly
from ..providers.model_capabilities import (
    ModelInputCapability,
    ModelOutputCapability,
    normalize_capability_route_filter,
)


# ============================================================================
# Provider Model Discovery (Using Centralized Registry)
# ============================================================================

def get_models_from_provider(
    provider_name: str, 
    input_capabilities=None, 
    output_capabilities=None,
    capability_routes=None,
    **kwargs,
) -> List[str]:
    """
    Get available models from a specific provider using the centralized provider registry.

    Args:
        provider_name: Name of the provider
        input_capabilities: Optional list of ModelInputCapability enums
        output_capabilities: Optional list of ModelOutputCapability enums
        capability_routes: Optional list of route keys such as input.image/output.text

    Returns:
        List of model names from the provider, optionally filtered
    """
    try:
        from ..providers.registry import get_available_models_for_provider
        return get_available_models_for_provider(
            provider_name, 
            input_capabilities=input_capabilities,
            output_capabilities=output_capabilities,
            capability_routes=capability_routes,
            **kwargs,
        )
    except Exception as e:
        if bool(kwargs.get("raise_on_error", False)):
            raise
        logger.debug(f"Failed to get models from provider {provider_name}: {_redact_text(str(e))}")
        return []



# ============================================================================
# OpenAI Responses API Models (100% Compatible)
# ============================================================================

class OpenAIInputContent(BaseModel):
    """OpenAI Responses API content item"""
    type: Literal["input_text", "input_file", "input_image"] = Field(
        description="Content type - 'input_text' for text, 'input_file' for files, or 'input_image' for images"
    )
    text: Optional[str] = Field(
        default=None,
        description="Text content (required when type='input_text')"
    )
    file_url: Optional[str] = Field(
        default=None,
        description="Direct file URL (required when type='input_file')"
    )
    image_url: Optional[Union[str, Dict[str, Any]]] = Field(
        default=None,
        description="Image URL (required when type='input_image'). Can be a string URL/data URL, or an OpenAI-style object.",
    )

class OpenAIResponsesInput(BaseModel):
    """OpenAI Responses API message-style input item.

    OpenAI clients commonly send either:
    - "easy" message items (`role` + `content`, with optional `type`)
    - explicit `{"type":"message", ...}` items
    """
    type: Literal["message"] = Field(
        default="message",
        description="Input item type (default: 'message').",
        example="message",
    )
    role: Literal["system", "developer", "user", "assistant", "tool"] = Field(description="Message role")
    content: Union[str, List[OpenAIInputContent]] = Field(
        description="Message content as a string, or an array of input content items."
    )
    tool_call_id: Optional[str] = Field(
        default=None,
        description="Tool call id this tool message is responding to (legacy/tool-loop compatibility).",
        example="call_abc123",
    )
    tool_calls: Optional[List[Dict[str, Any]]] = Field(
        default=None,
        description="Assistant tool calls (legacy/tool-loop compatibility).",
        example=None,
    )
    name: Optional[str] = Field(
        default=None,
        description="Optional participant name.",
        example="User1",
    )


class OpenAIResponsesFunctionCallOutput(BaseModel):
    """OpenAI Responses API tool result input item."""
    type: Literal["function_call_output"] = Field(
        description="Input item type for tool results.",
        example="function_call_output",
    )
    call_id: str = Field(description="Tool call id this output is responding to.", example="call_abc123")
    output: str = Field(description="Tool output as a string.", example="{\"results\":[]}")
    id: Optional[str] = Field(default=None, description="Optional item id.", example="fco_123")
    status: Optional[Literal["in_progress", "completed", "incomplete"]] = Field(
        default=None,
        description="Optional item status.",
        example="completed",
    )


class OpenAIResponsesFunctionCall(BaseModel):
    """OpenAI Responses API function/tool call input item (conversation replay compatibility)."""
    type: Literal["function_call"] = Field(description="Input item type for function calls.", example="function_call")
    call_id: str = Field(description="Function call id.", example="call_abc123")
    name: str = Field(description="Tool/function name.", example="web_search")
    arguments: str = Field(description="JSON-encoded arguments.", example="{\"query\":\"example\"}")
    id: Optional[str] = Field(default=None, description="Optional item id.", example="fc_123")
    status: Optional[Literal["in_progress", "completed", "incomplete"]] = Field(default=None)

class OpenAIResponsesRequest(BaseModel):
    """OpenAI Responses API request format (100% compatible)"""
    model: str = Field(
        description="Model identifier",
        example="gpt-4o"
    )
    input: Union[
        str,
        List[
            Annotated[
                Union[
                    OpenAIResponsesInput,
                    OpenAIResponsesFunctionCallOutput,
                    OpenAIResponsesFunctionCall,
                ],
                Field(discriminator="type"),
            ]
        ],
    ] = Field(description="Text input (string) or array of Responses input items.")
    max_tokens: Optional[int] = Field(
        default=None,
        validation_alias=AliasChoices("max_tokens", "max_output_tokens"),
        description="Maximum tokens to generate (OpenAI: max_output_tokens)"
    )
    temperature: Optional[float] = Field(
        default=None,
        description="Sampling temperature"
    )
    top_p: Optional[float] = Field(
        default=None,
        description="Top-p sampling"
    )
    tools: Optional[List[Dict[str, Any]]] = Field(
        default=None,
        description=(
            "Tools available to the model. Supports OpenAI Responses-style tools (built-ins like "
            "'web_search' plus function tools) and will be normalized for provider prompting."
        ),
    )
    tool_choice: Optional[Union[str, Dict[str, Any]]] = Field(
        default="auto",
        description="Controls which (if any) tool is called by the model.",
        example="auto",
    )
    instructions: Optional[str] = Field(
        default=None,
        description="Optional system-level instructions (OpenAI Responses API compatible).",
        example="You are a helpful assistant.",
    )
    stream: Optional[bool] = Field(
        default=False,
        description="Enable streaming (false by default, set to true for real-time responses)"
    )
    stop: Optional[List[str]] = Field(
        default=None,
        description="Up to 4 sequences where the API will stop generating further tokens. "
                    "The returned text will not contain the stop sequence.",
        example=None,
    )
    seed: Optional[int] = Field(
        default=None,
        description="If specified, the system will make a best effort to sample deterministically, "
                    "such that repeated requests with the same seed and parameters should return the same result. "
                    "Determinism is not guaranteed.",
        example=None,
    )
    frequency_penalty: Optional[float] = Field(
        default=0.0,
        ge=-2.0,
        le=2.0,
        description="Number between -2.0 and 2.0. Positive values penalize new tokens based on their existing frequency in the text so far, "
                    "decreasing the model's likelihood to repeat the same line verbatim.",
        example=0.0,
    )
    presence_penalty: Optional[float] = Field(
        default=0.0,
        ge=-2.0,
        le=2.0,
        description="Number between -2.0 and 2.0. Positive values penalize new tokens based on whether they appear in the text so far, "
                    "increasing the model's likelihood to talk about new topics.",
        example=0.0,
    )
    thinking: Optional[Union[bool, str]] = Field(
        default=None,
        description="Unified thinking/reasoning control (best-effort across providers/models). "
                    "Accepted values: null/'auto'/'on'/'off'/'none' or 'low'/'medium'/'high'/'xhigh' when supported. "
                    "Note: 'none' is treated as an alias for 'off'.",
        example="off",
    )
    prompt_cache_key: Optional[str] = Field(
        default=None,
        description="Provider-specific prompt cache key for prefix caching (best-effort).",
        example="tenantA:session123"
    )
    prompt_cache_binding: Optional[Union[Dict[str, Any], str]] = Field(
        default=None,
        description="Optional exact durable bloc binding returned by /acore/blocs/kv/load.",
        example={"key": "tenantA:doc-review", "binding_id": "sha256..."},
    )
    prompt_cache_retention: Optional[str] = Field(
        default=None,
        description="Prompt cache retention policy (provider-specific; OpenAI: 'in_memory' or '24h').",
        example="24h",
    )
    agent_format: Optional[str] = Field(
        default=None,
        description="Target agent format for tool call syntax conversion (AbstractCore-specific feature). "
                    "Options: 'auto' (auto-detect), 'openai', 'codex', 'qwen3', 'llama3', 'passthrough'. "
                    "Use 'auto' for automatic format detection based on model and user-agent.",
        example="auto",
    )
    api_key: Optional[str] = Field(
        default=None,
        description="Deprecated/disabled for HTTP requests. Provider API keys must be supplied via "
                    "X-AbstractCore-Provider-API-Key, or configured on the server (e.g., OPENAI_API_KEY, ANTHROPIC_API_KEY, "
                    "OPENROUTER_API_KEY, PORTKEY_API_KEY).",
        example=None,
    )
    base_url: Optional[str] = Field(
        default=None,
        description="Base URL for the provider API endpoint (AbstractCore-specific feature). "
                    "Useful for OpenAI-compatible providers (lmstudio, vllm, openrouter, portkey, openai-compatible) and custom/proxied endpoints. "
                    "Example: 'http://localhost:1234/v1' for LMStudio, 'http://localhost:8080/v1' for llama.cpp. "
                    "If not specified, uses provider's default or environment variable.",
        example="http://localhost:1234/v1",
    )
    timeout_s: Optional[float] = Field(
        default=None,
        description="Per-request provider HTTP timeout in seconds (AbstractCore-specific feature). "
                    "Intended for orchestrators (e.g. AbstractRuntime) to enforce execution policy. "
                    "If omitted, the server uses its own defaults. "
                    "Values <= 0 are treated as unlimited.",
        example=7200.0,
    )
    unload_after: bool = Field(
        default=False,
        description="If true, call `llm.unload_model(model)` after the request completes (AbstractCore-specific feature). "
                    "This is useful for explicit memory hygiene in single-tenant or batch scenarios. "
                    "WARNING: for providers that unload shared server state (e.g. Ollama), this can disrupt other "
                    "clients and is disabled by default unless explicitly enabled by the server operator.",
        example=False,
    )

    @model_validator(mode="before")
    @classmethod
    def _normalize_input_items(cls, data: Any) -> Any:
        """Best-effort: coerce easy message items into explicit type='message' items for union discrimination."""
        if not isinstance(data, dict):
            return data
        raw = data.get("input")
        if not isinstance(raw, list):
            return data
        normalized: List[Any] = []
        for item in raw:
            if isinstance(item, dict) and "type" not in item:
                # Heuristic: treat dicts with (role, content) as message input items.
                if "role" in item and "content" in item:
                    coerced = dict(item)
                    coerced["type"] = "message"
                    normalized.append(coerced)
                    continue
            normalized.append(item)
        out = dict(data)
        out["input"] = normalized
        return out

# ============================================================================
# Models
# ============================================================================

class ContentItem(BaseModel):
    """Individual content item within a message (OpenAI Vision API format with file support)"""
    type: Literal["text", "image_url", "file"] = Field(
        description="Content type - 'text' for text content, 'image_url' for images, or 'file' for file attachments"
    )
    text: Optional[str] = Field(
        default=None,
        description="Text content (required when type='text')"
    )
    image_url: Optional[Dict[str, Any]] = Field(
        default=None,
        description="Image URL object (required when type='image_url'). Should contain 'url' field with base64 data URL or HTTP(S) URL"
    )
    file_url: Optional[Dict[str, Any]] = Field(
        default=None,
        description="File URL object (required when type='file'). Should contain 'url' field with HTTP(S) URL, local path, or base64 data URL"
    )

class ChatMessage(BaseModel):
    """Enhanced OpenAI-compatible message format with media support"""
    role: Literal["system", "developer", "user", "assistant", "tool"] = Field(
        description=(
            "The role of the message author. One of 'system', 'developer', 'user', 'assistant', or 'tool'. "
            "Developer messages are accepted for OpenAI compatibility and normalized for providers that do not "
            "support that role natively."
        ),
        example="user"
    )
    content: Optional[Union[str, List[ContentItem]]] = Field(
        default=None,
        description="Message content - can be a string or array of content objects for multimodal messages. "
                   "For multimodal messages, use array format with text, image_url, and file objects.",
        example="What is the capital of France?"
    )
    tool_call_id: Optional[str] = Field(
        default=None,
        description="Tool call that this message is responding to (required for role='tool').",
        example="call_abc123"
    )
    tool_calls: Optional[List[Dict[str, Any]]] = Field(
        default=None,
        description="The tool calls generated by the model (only for assistant messages)."
    )
    name: Optional[str] = Field(
        default=None,
        description="An optional name for the participant. Provides the model information to differentiate between participants of the same role.",
        example="User1"
    )

class ChatCompletionRequest(BaseModel):
    """OpenAI-compatible chat completion request"""
    model: str = Field(
        description="ID of the model to use. Use provider/model format (e.g., 'openai/gpt-4', 'ollama/llama3:latest', "
                    "'anthropic/claude-3-opus-20240229'). You can use the List models API to see all available models, "
                    "or filter by input_type=text&output_type=text.",
        example="openai/gpt-4"
    )
    messages: List[ChatMessage] = Field(
        description="A list of messages comprising the conversation so far. Each message has a role "
                    "(system/developer/user/assistant/tool) and content. System and developer messages set "
                    "assistant behavior, user messages are from the end user, assistant messages are from the AI, "
                    "and tool messages contain function call results."
    )

    # Core parameters
    temperature: Optional[float] = Field(
        default=0.7,
        ge=0.0,
        le=2.0,
        description="What sampling temperature to use, between 0 and 2. Higher values like 0.8 will make the output more random, "
                    "while lower values like 0.2 will make it more focused and deterministic. "
                    "We generally recommend altering this or top_p but not both.",
        example=0.7
    )
    max_tokens: Optional[int] = Field(
        default=None,
        ge=1,
        description="The maximum number of tokens that can be generated in the chat completion. "
                    "The total length of input tokens and generated tokens is limited by the model's context length. "
                    "If not specified, uses the model's default maximum output tokens.",
        example=2048
    )
    top_p: Optional[float] = Field(
        default=1.0,
        ge=0.0,
        le=1.0,
        description="An alternative to sampling with temperature, called nucleus sampling. "
                    "The model considers the results of the tokens with top_p probability mass. "
                    "For example, 0.1 means only the tokens comprising the top 10% probability mass are considered. "
                    "We generally recommend altering this or temperature but not both.",
        example=1.0
    )
    stream: Optional[bool] = Field(
        default=False,
        description="If set, partial message deltas will be sent as server-sent events. "
                    "Tokens will be sent as data-only server-sent events as they become available, "
                    "with the stream terminated by a 'data: [DONE]' message.",
        example=False
    )

    # Unified thinking/reasoning control (AbstractCore-specific feature)
    thinking: Optional[Union[bool, str]] = Field(
        default=None,
        description="Unified thinking/reasoning control (best-effort across providers/models). "
                    "Accepted values: null/'auto'/'on'/'off'/'none' or 'low'/'medium'/'high'/'xhigh' when supported. "
                    "Note: 'none' is treated as an alias for 'off'.",
        example="off",
    )

    # Tool calling
    tools: Optional[List[Dict[str, Any]]] = Field(
        default=None,
        description="A list of tools the model may call. Currently, only functions are supported as a tool. "
                    "Use this to provide a list of functions the model may generate JSON inputs for. "
                    "Maximum of 128 functions supported."
    )
    tool_choice: Optional[Union[str, Dict[str, Any]]] = Field(
        default="auto",
        description="Controls which (if any) tool is called by the model. "
                    "'none' means the model will not call any tool and instead generates a message. "
                    "'auto' means the model can pick between generating a message or calling one or more tools. "
                    "'required' means the model must call one or more tools. "
                    "Specifying a particular tool via {\"type\": \"function\", \"function\": {\"name\": \"my_function\"}} forces the model to call that tool.",
        example="auto"
    )

    # Other OpenAI parameters
    stop: Optional[List[str]] = Field(
        default=None,
        description="Up to 4 sequences where the API will stop generating further tokens. "
                    "The returned text will not contain the stop sequence.",
        example=None
    )
    seed: Optional[int] = Field(
        default=None,
        description="If specified, the system will make a best effort to sample deterministically, "
                    "such that repeated requests with the same seed and parameters should return the same result. "
                    "Determinism is not guaranteed.",
        example=None
    )
    frequency_penalty: Optional[float] = Field(
        default=0.0,
        ge=-2.0,
        le=2.0,
        description="Number between -2.0 and 2.0. Positive values penalize new tokens based on their existing frequency in the text so far, "
                    "decreasing the model's likelihood to repeat the same line verbatim.",
        example=0.0
    )
    presence_penalty: Optional[float] = Field(
        default=0.0,
        ge=-2.0,
        le=2.0,
        description="Number between -2.0 and 2.0. Positive values penalize new tokens based on whether they appear in the text so far, "
                    "increasing the model's likelihood to talk about new topics.",
        example=0.0
    )

    # OpenAI prompt caching (2025+): forwarded best-effort by providers that support it.
    prompt_cache_key: Optional[str] = Field(
        default=None,
        description="Provider-specific prompt cache key for prefix caching (best-effort).",
        example="tenantA:session123"
    )
    prompt_cache_binding: Optional[Union[Dict[str, Any], str]] = Field(
        default=None,
        description="Optional exact durable bloc binding returned by /acore/blocs/kv/load.",
        example={"key": "tenantA:doc-review", "binding_id": "sha256..."},
    )
    prompt_cache_retention: Optional[str] = Field(
        default=None,
        description="Prompt cache retention policy (provider-specific; OpenAI: 'in_memory' or '24h').",
        example="24h",
    )

    # Agent format control (AppV2 feature)
    agent_format: Optional[str] = Field(
        default=None,
        description="Target agent format for tool call syntax conversion (AbstractCore-specific feature). "
                    "Options: 'auto' (auto-detect), 'openai', 'codex', 'qwen3', 'llama3', 'passthrough'. "
                    "Use 'auto' for automatic format detection based on model and user-agent.",
        example="auto"
    )

    # Provider-specific parameters (AbstractCore-specific feature)
    api_key: Optional[str] = Field(
        default=None,
        description="Deprecated/disabled for HTTP requests. Provider API keys must be supplied via "
                    "X-AbstractCore-Provider-API-Key, or configured on the server (e.g., OPENAI_API_KEY, ANTHROPIC_API_KEY, "
                    "OPENROUTER_API_KEY, PORTKEY_API_KEY).",
        example=None
    )
    base_url: Optional[str] = Field(
        default=None,
        description="Base URL for the provider API endpoint (AbstractCore-specific feature). "
                    "Useful for OpenAI-compatible providers (lmstudio, vllm, openrouter, portkey, openai-compatible) and custom/proxied endpoints. "
                    "Example: 'http://localhost:1234/v1' for LMStudio, 'http://localhost:8080/v1' for llama.cpp. "
                    "If not specified, uses provider's default or environment variable.",
        example="http://localhost:1234/v1"
    )

    # Runtime/orchestrator policy (AbstractCore-specific feature)
    timeout_s: Optional[float] = Field(
        default=None,
        description="Per-request provider HTTP timeout in seconds (AbstractCore-specific feature). "
                    "Intended for orchestrators (e.g. AbstractRuntime) to enforce execution policy. "
                    "If omitted, the server uses its own defaults. "
                    "Values <= 0 are treated as unlimited.",
        example=7200.0,
    )
    unload_after: bool = Field(
        default=False,
        description="If true, call `llm.unload_model(model)` after the request completes (AbstractCore-specific feature). "
                    "This is useful for explicit memory hygiene in single-tenant or batch scenarios. "
                    "WARNING: for providers that unload shared server state (e.g. Ollama), this can disrupt other "
                    "clients and is disabled by default unless explicitly enabled by the server operator.",
        example=False,
    )

    class Config:
        json_schema_extra = {
            "examples": list({
                "basic_text": {
                    "summary": "Basic Text Chat",
                    "description": "Simple text-based conversation",
                    "value": {
                        "model": "openai/gpt-4",
                        "messages": [
                            {
                                "role": "system",
                                "content": "You are a helpful assistant."
                            },
                            {
                                "role": "user",
                                "content": "What is the capital of France?"
                            }
                        ],
                        "temperature": 0.7,
                        "max_tokens": 150,
                        "stream": False
                    }
                },
                "vision_image": {
                    "summary": "Image Analysis",
                    "description": "Analyze images using vision-capable models with OpenAI Vision API format",
                    "value": {
                        "model": "ollama/qwen2.5vl:7b",
                        "messages": [
                            {
                                "role": "user",
                                "content": [
                                    {
                                        "type": "text",
                                        "text": "What's in this image?"
                                    },
                                    {
                                        "type": "image_url",
                                        "image_url": {
                                            "url": "data:image/jpeg;base64,/9j/4AAQSkZJRgABAQAAAQABAAD/2wBDAAYEBQYFBAYGBQYHBwYIChAKCgkJChQODwwQFxQYGBcUFhYaHSUfGhsjHBYWICwgIyYnKSopGR8tMC0oMCUoKSj/2wBDAQcHBwoIChMKChMoGhYaKCgoKCgoKCgoKCgoKCgoKCgoKCgoKCgoKCgoKCgoKCgoKCgoKCgoKCgoKCgoKCgoKCj/wAARCAABAAEDASIAAhEBAxEB/8QAFQABAQAAAAAAAAAAAAAAAAAAAAv/xAAUEAEAAAAAAAAAAAAAAAAAAAAA/8QAFQEBAQAAAAAAAAAAAAAAAAAAAAX/xAAUEQEAAAAAAAAAAAAAAAAAAAAA/9oADAMBAAIRAxEAPwCdABmX/9k="
                                        }
                                    }
                                ]
                            }
                        ],
                        "temperature": 0.7,
                        "max_tokens": 200
                    }
                },
                "document_analysis": {
                    "summary": "Document Analysis",
                    "description": "Process documents (PDF, CSV, Excel, etc.) with file attachments",
                    "value": {
                        "model": "lmstudio/qwen/qwen3-next-80b",
                        "messages": [
                            {
                                "role": "user",
                                "content": [
                                    {
                                        "type": "text",
                                        "text": "Analyze this CSV file and calculate the total sales"
                                    },
                                    {
                                        "type": "image_url",
                                        "image_url": {
                                            "url": "data:text/csv;base64,RGF0ZSxQcm9kdWN0LFNhbGVzCjIwMjQtMDEtMDEsUHJvZHVjdCBBLDEwMDAwCjIwMjQtMDEtMDIsUHJvZHVjdCBCLDE1MDAwCjIwMjQtMDEtMDMsUHJvZHVjdCBDLDI1MDAw"
                                        }
                                    }
                                ]
                            }
                        ],
                        "temperature": 0.3,
                        "max_tokens": 300
                    }
                },
                "mixed_media": {
                    "summary": "Mixed Media Analysis",
                    "description": "Process multiple file types in a single request",
                    "value": {
                        "model": "ollama/qwen2.5vl:7b",
                        "messages": [
                            {
                                "role": "user",
                                "content": [
                                    {
                                        "type": "text",
                                        "text": "Compare this chart image with the data in this PDF report"
                                    },
                                    {
                                        "type": "image_url",
                                        "image_url": {
                                            "url": "data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNkYPhfDwAChwGA60e6kgAAAABJRU5ErkJggg=="
                                        }
                                    },
                                    {
                                        "type": "image_url",
                                        "image_url": {
                                            "url": "data:application/pdf;base64,JVBERi0xLjQKJdPr6eEKMSAwIG9iago8PAovVHlwZSAvQ2F0YWxvZwovUGFnZXMgMiAwIFIKPj4KZW5kb2JqCjIgMCBvYmoKPDwKL1R5cGUgL1BhZ2VzCi9LaWRzIFszIDAgUl0KL0NvdW50IDEKPJ4KZW5kb2JqCjMgMCBvYmoKPDwKL1R5cGUgL1BhZ2UKL1BhcmVudCAyIDAgUgovTWVkaWFCb3ggWzAgMCA2MTIgNzkyXQo+PgplbmRvYmoKeHJlZgowIDQKMDAwMDAwMDAwMCA2NTUzNSBmIAowMDAwMDAwMDA5IDAwMDAwIG4gCjAwMDAwMDAwNTggMDAwMDAgbiAKMDAwMDAwMDExNSAwMDAwMCBuIAp0cmFpbGVyCjw8Ci9TaXplIDQKL1Jvb3QgMSAwIFIKPj4Kc3RhcnR4cmVmCjE5NQolJUVPRgo="
                                        }
                                    }
                                ]
                            }
                        ],
                        "temperature": 0.5,
                        "max_tokens": 500,
                        "stream": False
                    }
                },
                "tools_with_media": {
                    "summary": "Tools + Media",
                    "description": "Combine tool usage with file attachments for complex workflows",
                    "value": {
                        "model": "openai/gpt-4",
                        "messages": [
                            {
                                "role": "user",
                                "content": [
                                    {
                                        "type": "text",
                                        "text": "Analyze this financial data and create a summary chart"
                                    },
                                    {
                                        "type": "image_url",
                                        "image_url": {
                                            "url": "data:text/csv;base64,Q29tcGFueSxRMSxRMixRMyxRNApBY21lIEluYywyMDAsMjUwLDMwMCwzNTAKVGVjaCBDb3JwLDE1MCwyMDAsMjUwLDMwMApCaXogTHRkLDEwMCwxMjAsMTQwLDE2MA=="
                                        }
                                    }
                                ]
                            }
                        ],
                        "temperature": 0.7,
                        "max_tokens": 2048,
                        "stream": False,
                        "tools": [
                            {
                                "type": "function",
                                "function": {
                                    "name": "create_chart",
                                    "description": "Create a chart from data",
                                    "parameters": {
                                        "type": "object",
                                        "properties": {
                                            "chart_type": {"type": "string"},
                                            "data": {"type": "array"}
                                        }
                                    }
                                }
                            }
                        ],
                        "tool_choice": "auto"
                    }
                },
                "complete_request": {
                    "summary": "Complete Request with Media",
                    "description": "Full example showing all possible fields with file attachment",
                    "value": {
                        "model": "openai/gpt-4",
                        "messages": [
                            {
                                "role": "user",
                                "content": [
                                    {
                                        "type": "text",
                                        "text": "Analyze this CSV file and provide insights"
                                    },
                                    {
                                        "type": "image_url",
                                        "image_url": {
                                            "url": "data:text/csv;base64,RGF0ZSxQcm9kdWN0LFNhbGVzCjIwMjQtMDEtMDEsUHJvZHVjdCBBLDEwMDAwCjIwMjQtMDEtMDIsUHJvZHVjdCBCLDE1MDAwCjIwMjQtMDEtMDMsUHJvZHVjdCBDLDI1MDAw"
                                        }
                                    }
                                ],
                                "tool_call_id": None,
                                "tool_calls": None,
                                "name": "DataAnalyst"
                            }
                        ],
                        "temperature": 0.7,
                        "max_tokens": 2048,
                        "top_p": 1,
                        "stream": False,
                        "tools": [
                            {
                                "type": "function",
                                "function": {
                                    "name": "analyze_data",
                                    "description": "Analyze data and generate insights",
                                    "parameters": {
                                        "type": "object",
                                        "properties": {
                                            "analysis_type": {"type": "string"},
                                            "metrics": {"type": "array"}
                                        }
                                    }
                                }
                            }
                        ],
                        "tool_choice": "auto",
                        "stop": ["END"],
                        "seed": 12345,
                        "frequency_penalty": 0.0,
                        "presence_penalty": 0.0,
                        "agent_format": "auto",
                        "base_url": None
                    }
                }
            }.values())
        }

class EmbeddingRequest(BaseModel):
    """OpenAI-compatible embedding request"""
    input: Union[str, List[str]] = Field(
        description="Input text to embed, encoded as a string or array of strings. "
                    "To embed multiple inputs in a single request, pass an array of strings. "
                    "The input must not exceed the max input tokens for the model (8192 tokens for most embedding models), "
                    "cannot be an empty string, and any array must be 2048 dimensions or less.",
        example="this is the story of starship lost in space"
    )
    model: str = Field(
        description="ID of the model to use. Use provider/model format (e.g., 'huggingface/sentence-transformers/all-MiniLM-L6-v2', "
                    "'ollama/granite-embedding:278m', 'lmstudio/text-embedding-all-minilm-l6-v2'). "
                    "You can use the List models API to see all available models, or filter by output_type=embeddings.",
        example="huggingface/sentence-transformers/all-MiniLM-L6-v2"
    )
    encoding_format: Optional[str] = Field(
        default="float",
        description="The format to return the embeddings in. Can be either 'float' or 'base64'. Defaults to 'float'.",
        example="float"
    )
    dimensions: Optional[int] = Field(
        default=None,
        description="The number of dimensions the resulting output embeddings should have. "
                    "Only supported in some models (e.g., text-embedding-3 and later models). "
                    "If specified, embeddings will be truncated to this dimension. "
                    "Set to 0 or None to use the model's default dimension.",
        example=0
    )
    user: Optional[str] = Field(
        default=None,
        description="A unique identifier representing your end-user, which can help OpenAI/providers to monitor and detect abuse. "
                    "This is optional but recommended for production applications.",
        example="user-123"
    )
    base_url: Optional[str] = Field(
        default=None,
        description=(
            "AbstractCore extension: request-level OpenAI-compatible base URL override. "
            "Loopback URLs are allowed by default; non-loopback URLs require "
            "ABSTRACTCORE_SERVER_BASE_URL_ALLOWLIST."
        ),
        example="http://127.0.0.1:1234/v1",
    )
    api_key: Optional[str] = Field(
        default=None,
        description="Deprecated/disabled for HTTP request bodies. Use X-AbstractCore-Provider-API-Key for provider overrides.",
        example=None,
    )

    class Config:
        json_schema_extra = {
            "example": {
                "input": "this is the story of starship lost in space",
                "model": "huggingface/sentence-transformers/all-MiniLM-L6-v2",
                "encoding_format": "float",
                "dimensions": 0,
                "user": "user-123"
            }
        }

# ============================================================================
# Union Request Model for /v1/responses endpoint
# ============================================================================

class ResponsesAPIRequest(BaseModel):
    """
    Union request model for /v1/responses endpoint supporting both OpenAI and legacy formats.

    The endpoint automatically detects the format based on the presence of 'input' vs 'messages' field.
    """
    class Config:
        json_schema_extra = {
            "oneOf": [
                {
                    "title": "OpenAI Responses API Format",
                    "description": "OpenAI-compatible responses format with input_file support",
                    "$ref": "#/components/schemas/OpenAIResponsesRequest"
                },
                {
                    "title": "Legacy Format (ChatCompletionRequest)",
                    "description": "Backward-compatible format using messages array",
                    "$ref": "#/components/schemas/ChatCompletionRequest"
                }
            ],
            "examples": list({
                "openai_format": {
                    "summary": "OpenAI Responses API Format",
                    "description": "Use input array with input_text and input_file objects",
                    "value": {
                        "model": "gpt-4o",
                        "input": [
                            {
                                "role": "user",
                                "content": [
                                    {"type": "input_text", "text": "Analyze this document"},
                                    {"type": "input_file", "file_url": "https://example.com/doc.pdf"}
                                ]
                            }
                        ],
                        "stream": False,
                        "thinking": "off",
                        "prompt_cache_key": "tenantA:doc-review",
                    }
                },
                "legacy_format": {
                    "summary": "Legacy Format (Backward Compatible)",
                    "description": "Use messages array like standard chat completions",
                    "value": {
                        "model": "openai/gpt-4",
                        "messages": [
                            {"role": "user", "content": "Tell me a story"}
                        ],
                        "stream": False
                    }
                }
            }.values())
        }

# ============================================================================
# OpenAI Responses API Compatibility
# ============================================================================

_RESPONSES_WEB_SEARCH_TOOL_TYPES = {
    "web_search",
    "web_search_preview",
    "web_search_2025_08_26",
    "web_search_preview_2025_03_11",
}


def _normalize_openai_responses_tools(tools: Optional[List[Dict[str, Any]]]) -> Optional[List[Dict[str, Any]]]:
    """Normalize OpenAI Responses-style tools into ChatCompletions-style tools.

    AbstractCore does not execute tools server-side. Normalization exists only to
    keep tool declarations visible to chat-style providers and local models.
    """
    if tools is None:
        return None
    if not isinstance(tools, list):
        raise ValueError("tools must be an array")

    normalized: List[Dict[str, Any]] = []
    for idx, tool in enumerate(tools):
        if not isinstance(tool, dict):
            raise ValueError(f"tools[{idx}] must be an object")

        tool_type = tool.get("type")
        if not isinstance(tool_type, str) or not tool_type.strip():
            raise ValueError(f"tools[{idx}].type must be a non-empty string")
        tool_type = tool_type.strip()

        if tool_type == "function":
            # Accept both Responses-style (flattened) and ChatCompletions-style (nested) shapes.
            function_obj = tool.get("function")
            if isinstance(function_obj, dict):
                name = function_obj.get("name")
                if not isinstance(name, str) or not name.strip():
                    raise ValueError(f"tools[{idx}].function.name must be a non-empty string")
                normalized.append({"type": "function", "function": dict(function_obj)})
                continue

            name = tool.get("name")
            if not isinstance(name, str) or not name.strip():
                raise ValueError(f"tools[{idx}].name must be a non-empty string for type=function")
            function_def: Dict[str, Any] = {
                "name": name.strip(),
            }
            if isinstance(tool.get("description"), str) and tool["description"].strip():
                function_def["description"] = tool["description"]
            if isinstance(tool.get("parameters"), dict):
                function_def["parameters"] = tool["parameters"]
            if "strict" in tool:
                function_def["strict"] = tool.get("strict")
            normalized.append({"type": "function", "function": function_def})
            continue

        if tool_type in _RESPONSES_WEB_SEARCH_TOOL_TYPES:
            function_def = {
                "name": tool_type,
                "description": "Search the web for current information.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "query": {"type": "string"},
                        "recency": {"type": "integer"},
                    },
                    "required": ["query"],
                },
            }
            normalized.append({"type": "function", "function": function_def})
            continue

        raise ValueError(f"Unsupported Responses tool type: {tool_type}")

    return normalized


def _normalize_openai_responses_tool_choice(
    tool_choice: Optional[Union[str, Dict[str, Any]]]
) -> Optional[Union[str, Dict[str, Any]]]:
    """Normalize OpenAI Responses-style tool_choice into ChatCompletions-style tool_choice."""
    if tool_choice is None:
        return None
    if isinstance(tool_choice, str):
        return tool_choice
    if not isinstance(tool_choice, dict):
        raise ValueError("tool_choice must be a string or object")

    choice_type = tool_choice.get("type")
    if isinstance(choice_type, str):
        choice_type = choice_type.strip()

    if choice_type == "function":
        if isinstance(tool_choice.get("function"), dict):
            return dict(tool_choice)
        name = tool_choice.get("name")
        if isinstance(name, str) and name.strip():
            return {"type": "function", "function": {"name": name.strip()}}
        raise ValueError("tool_choice.name must be provided when tool_choice.type=function")

    if isinstance(choice_type, str) and choice_type in _RESPONSES_WEB_SEARCH_TOOL_TYPES:
        return {"type": "function", "function": {"name": choice_type}}

    # Unknown object shape: pass through unchanged (best-effort).
    return dict(tool_choice)


def _normalize_openai_responses_tools_for_response(
    tools: Optional[List[Dict[str, Any]]],
) -> List[Dict[str, Any]]:
    """Normalize `tools` into OpenAI Responses API tool objects for Response payloads.

    This differs from `_normalize_openai_responses_tools`, which normalizes tools into
    Chat Completions tool objects for provider prompting.
    """
    if not tools:
        return []
    if not isinstance(tools, list):
        raise ValueError("tools must be an array")

    normalized: List[Dict[str, Any]] = []
    for idx, tool in enumerate(tools):
        if not isinstance(tool, dict):
            raise ValueError(f"tools[{idx}] must be an object")

        tool_type = tool.get("type")
        if not isinstance(tool_type, str) or not tool_type.strip():
            raise ValueError(f"tools[{idx}].type must be a non-empty string")
        tool_type = tool_type.strip()

        if tool_type == "function":
            function_obj = tool.get("function")
            if isinstance(function_obj, dict):
                name = function_obj.get("name")
                if not isinstance(name, str) or not name.strip():
                    raise ValueError(f"tools[{idx}].function.name must be a non-empty string")
                flat: Dict[str, Any] = {"type": "function", "name": name.strip()}
                if isinstance(function_obj.get("description"), str):
                    flat["description"] = function_obj.get("description")
                if isinstance(function_obj.get("parameters"), dict):
                    flat["parameters"] = function_obj.get("parameters")
                if "strict" in function_obj:
                    flat["strict"] = function_obj.get("strict")
                normalized.append(flat)
                continue

            name = tool.get("name")
            if not isinstance(name, str) or not name.strip():
                raise ValueError(f"tools[{idx}].name must be a non-empty string for type=function")
            flat = {"type": "function", "name": name.strip()}
            if isinstance(tool.get("description"), str):
                flat["description"] = tool.get("description")
            if isinstance(tool.get("parameters"), dict):
                flat["parameters"] = tool.get("parameters")
            if "strict" in tool:
                flat["strict"] = tool.get("strict")
            normalized.append(flat)
            continue

        if tool_type in _RESPONSES_WEB_SEARCH_TOOL_TYPES:
            normalized.append(dict(tool))
            continue

        raise ValueError(f"Unsupported Responses tool type: {tool_type}")

    return normalized


def _normalize_openai_responses_tool_choice_for_response(
    tool_choice: Optional[Union[str, Dict[str, Any]]],
) -> Union[str, Dict[str, Any]]:
    """Normalize `tool_choice` into OpenAI Responses API tool_choice objects for Response payloads."""
    if tool_choice is None:
        return "auto"
    if isinstance(tool_choice, str):
        if tool_choice in {"none", "auto", "required"}:
            return tool_choice
        return "auto"
    if not isinstance(tool_choice, dict):
        raise ValueError("tool_choice must be a string or object")

    choice_type = tool_choice.get("type")
    if isinstance(choice_type, str):
        choice_type = choice_type.strip()

    if choice_type == "function":
        function_obj = tool_choice.get("function")
        if isinstance(function_obj, dict):
            name = function_obj.get("name")
        else:
            name = tool_choice.get("name")
        if isinstance(name, str) and name.strip():
            return {"type": "function", "name": name.strip()}
        raise ValueError("tool_choice.name must be provided when tool_choice.type=function")

    # Only return builtin tool_choice objects that are known to be accepted by the OpenAI SDK types.
    if choice_type in {
        "file_search",
        "web_search_preview",
        "web_search_preview_2025_03_11",
        "computer_use_preview",
        "image_generation",
        "code_interpreter",
    }:
        return {"type": choice_type}

    # Some clients send {"type":"web_search"} but OpenAI SDKs may not accept it everywhere yet;
    # keep the response valid by falling back to "auto".
    return "auto"


def convert_to_openai_responses_response(
    response,
    provider: str,
    model: str,
    syntax_rewriter: ToolCallSyntaxRewriter,
    request_id: str,
    openai_request: "OpenAIResponsesRequest",
) -> Dict[str, Any]:
    """Convert an AbstractCore response into an OpenAI Responses API `object: response` payload."""
    if response is None:
        logger.warning("Received None response", request_id=request_id)
        content = "Error: No response generated"
        tool_calls = None
        usage_in = None
    else:
        content = response.content if hasattr(response, "content") else str(response)
        tool_calls = getattr(response, "tool_calls", None) if hasattr(response, "tool_calls") else None
        usage_in = getattr(response, "usage", None) if hasattr(response, "usage") else None

    if syntax_rewriter.target_format in [SyntaxFormat.OPENAI, SyntaxFormat.CODEX]:
        if isinstance(content, str) and any(
            pattern in content for pattern in ["<function_call>", "<tool_call>", "<|tool_call|>", "```tool_code"]
        ):
            content = syntax_rewriter.remove_tool_call_patterns(content)
    elif syntax_rewriter.target_format != SyntaxFormat.PASSTHROUGH:
        if tool_calls:
            content = syntax_rewriter.rewrite_content(content, detected_tool_calls=tool_calls)
        else:
            content = syntax_rewriter.rewrite_content(content)

    output: List[Dict[str, Any]] = []

    if isinstance(content, str) and content:
        output.append(
            {
                "id": f"msg_{uuid.uuid4().hex}",
                "type": "message",
                "status": "completed",
                "role": "assistant",
                "content": [
                    {
                        "type": "output_text",
                        "text": content,
                        "annotations": [],
                        "logprobs": [],
                    }
                ],
            }
        )

    if tool_calls:
        openai_tool_calls = syntax_rewriter.convert_to_openai_format(tool_calls)
        for tool_call in openai_tool_calls:
            call_id = tool_call.get("id")
            function_obj = tool_call.get("function") if isinstance(tool_call, dict) else None
            if not isinstance(call_id, str) or not call_id.strip() or not isinstance(function_obj, dict):
                continue
            name = function_obj.get("name")
            arguments = function_obj.get("arguments")
            if not isinstance(name, str) or not name.strip():
                continue
            if not isinstance(arguments, str):
                arguments = json.dumps(arguments, ensure_ascii=False)
            output.append(
                {
                    "id": f"fc_{uuid.uuid4().hex}",
                    "type": "function_call",
                    "status": "completed",
                    "call_id": call_id,
                    "name": name,
                    "arguments": arguments,
                    }
            )

    usage_out: Optional[Dict[str, Any]] = None
    if isinstance(usage_in, dict):
        prompt_tokens = int(usage_in.get("prompt_tokens", 0) or 0)
        completion_tokens = int(usage_in.get("completion_tokens", 0) or 0)
        total_tokens = int(usage_in.get("total_tokens", prompt_tokens + completion_tokens) or 0)
        usage_out = {
            "input_tokens": prompt_tokens,
            "input_tokens_details": {"cached_tokens": 0},
            "output_tokens": completion_tokens,
            "output_tokens_details": {"reasoning_tokens": 0},
            "total_tokens": total_tokens,
        }

    tools_out = _normalize_openai_responses_tools_for_response(openai_request.tools)
    tool_choice_out = _normalize_openai_responses_tool_choice_for_response(openai_request.tool_choice)

    payload: Dict[str, Any] = {
        "id": f"resp_{uuid.uuid4().hex}",
        "object": "response",
        "created_at": float(int(time.time())),
        "model": openai_request.model,
        "output": output,
        "parallel_tool_calls": True,
        "tool_choice": tool_choice_out,
        "tools": tools_out,
        "status": "completed",
        "error": None,
        "incomplete_details": None,
        "usage": usage_out,
        "prompt_cache_key": openai_request.prompt_cache_key,
        "instructions": openai_request.instructions,
        "metadata": {},
    }

    # Omit null optional fields for closer parity with OpenAI payloads.
    for k in ("instructions", "prompt_cache_key", "usage"):
        if payload.get(k) is None:
            payload.pop(k, None)

    return payload


def generate_streaming_responses_response(
    llm,
    gen_kwargs: Dict[str, Any],
    provider: str,
    model: str,
    syntax_rewriter: ToolCallSyntaxRewriter,
    request_id: str,
    openai_request: "OpenAIResponsesRequest",
    temp_files_to_cleanup: List[str] = None,
    *,
    unload_after: bool = False,
    ollama_key: Optional[Tuple[str, str, str]] = None,
    allow_unsafe_unload_after: bool = False,
    loaded_runtime: Optional["_GatewayLoadedRuntime"] = None,
) -> Iterator[str]:
    """Generate OpenAI Responses API streaming events (SSE)."""
    provider_normalized = provider.strip().lower()
    sequence_number = 0
    resp_id = f"resp_{uuid.uuid4().hex}"
    created_at = float(int(time.time()))

    tools_out = _normalize_openai_responses_tools_for_response(openai_request.tools)
    tool_choice_out = _normalize_openai_responses_tool_choice_for_response(openai_request.tool_choice)

    base_response: Dict[str, Any] = {
        "id": resp_id,
        "object": "response",
        "created_at": created_at,
        "model": openai_request.model,
        "output": [],
        "parallel_tool_calls": True,
        "tool_choice": tool_choice_out,
        "tools": tools_out,
        "status": "in_progress",
        "error": None,
        "incomplete_details": None,
        "usage": None,
        "prompt_cache_key": openai_request.prompt_cache_key,
        "instructions": openai_request.instructions,
        "metadata": {},
    }
    for k in ("instructions", "prompt_cache_key"):
        if base_response.get(k) is None:
            base_response.pop(k, None)

    try:
        def _emit(event: Dict[str, Any]) -> Iterator[str]:
            nonlocal sequence_number
            event = dict(event)
            event["sequence_number"] = sequence_number
            sequence_number += 1
            yield f"data: {json.dumps(event)}\n\n"

        # response.created
        yield from _emit({"type": "response.created", "response": base_response})

        output_items: List[Dict[str, Any]] = []

        msg_id = f"msg_{uuid.uuid4().hex}"
        output_index = 0
        current_text = ""
        started_message = False
        tool_items_emitted: Dict[str, int] = {}

        def _iter_chunks() -> Iterator[Any]:
            if loaded_runtime is not None:
                for item in _stream_loaded_gateway_runtime(
                    loaded_runtime,
                    lambda: llm.generate(**gen_kwargs),
                ):
                    yield item
                return
            for item in llm.generate(**gen_kwargs):
                yield item

        for chunk in _iter_chunks():
            if hasattr(chunk, "content") and chunk.content:
                chunk_text = chunk.content
                if syntax_rewriter.target_format in [SyntaxFormat.OPENAI, SyntaxFormat.CODEX]:
                    if any(
                        pattern in chunk_text
                        for pattern in ["<function_call>", "<tool_call>", "<|tool_call|>", "```tool_code"]
                    ):
                        chunk_text = syntax_rewriter.remove_tool_call_patterns(chunk_text)
                elif syntax_rewriter.target_format != SyntaxFormat.PASSTHROUGH:
                    chunk_text = syntax_rewriter.rewrite_content(chunk_text)

                if chunk_text:
                    if not started_message:
                        started_message = True
                        # output_item.added (message)
                        yield from _emit(
                            {
                                "type": "response.output_item.added",
                                "output_index": output_index,
                                "item": {
                                    "id": msg_id,
                                    "status": "in_progress",
                                    "type": "message",
                                    "role": "assistant",
                                    "content": [],
                                },
                            }
                        )
                        # content_part.added (output_text)
                        yield from _emit(
                            {
                                "type": "response.content_part.added",
                                "item_id": msg_id,
                                "output_index": output_index,
                                "content_index": 0,
                                "part": {"type": "output_text", "text": "", "annotations": []},
                            }
                        )

                    current_text += chunk_text
                    # output_text.delta
                    yield from _emit(
                        {
                            "type": "response.output_text.delta",
                            "item_id": msg_id,
                            "output_index": output_index,
                            "content_index": 0,
                            "delta": chunk_text,
                            "logprobs": [],
                        }
                    )

            if hasattr(chunk, "tool_calls") and chunk.tool_calls:
                openai_tool_calls = syntax_rewriter.convert_to_openai_format(chunk.tool_calls)
                for tool_call in openai_tool_calls:
                    call_id = tool_call.get("id")
                    function_obj = tool_call.get("function") if isinstance(tool_call, dict) else None
                    if not isinstance(call_id, str) or not call_id.strip() or not isinstance(function_obj, dict):
                        continue
                    name = function_obj.get("name")
                    arguments = function_obj.get("arguments")
                    if not isinstance(name, str) or not name.strip():
                        continue
                    if not isinstance(arguments, str):
                        arguments = json.dumps(arguments, ensure_ascii=False)

                    if call_id in tool_items_emitted:
                        continue

                    tool_output_index = output_index if not started_message else (output_index + len(tool_items_emitted) + 1)
                    tool_items_emitted[call_id] = tool_output_index

                    fc_id = f"fc_{uuid.uuid4().hex}"
                    item_obj = {
                        "id": fc_id,
                        "type": "function_call",
                        "status": "completed",
                        "call_id": call_id,
                        "name": name,
                        "arguments": arguments,
                    }

                    yield from _emit(
                        {
                            "type": "response.output_item.added",
                            "output_index": tool_output_index,
                            "item": {
                                **item_obj,
                                "status": "in_progress",
                            },
                        }
                    )
                    yield from _emit(
                        {
                            "type": "response.function_call_arguments.done",
                            "item_id": fc_id,
                            "output_index": tool_output_index,
                            "arguments": arguments,
                        }
                    )
                    yield from _emit(
                        {
                            "type": "response.output_item.done",
                            "output_index": tool_output_index,
                            "item": item_obj,
                        }
                    )
                    output_items.append(item_obj)

        if started_message:
            message_item = {
                "id": msg_id,
                "type": "message",
                "status": "completed",
                "role": "assistant",
                "content": [
                    {
                        "type": "output_text",
                        "text": current_text,
                        "annotations": [],
                        "logprobs": [],
                    }
                ],
            }

            # output_text.done
            yield from _emit(
                {
                    "type": "response.output_text.done",
                    "item_id": msg_id,
                    "output_index": output_index,
                    "content_index": 0,
                    "text": current_text,
                    "logprobs": [],
                }
            )
            # content_part.done
            yield from _emit(
                {
                    "type": "response.content_part.done",
                    "item_id": msg_id,
                    "output_index": output_index,
                    "content_index": 0,
                    "part": {"type": "output_text", "text": current_text, "annotations": []},
                }
            )
            # output_item.done (message)
            yield from _emit(
                {
                    "type": "response.output_item.done",
                    "output_index": output_index,
                    "item": message_item,
                }
            )
            output_items.insert(0, message_item)

        final_response = dict(base_response)
        final_response["status"] = "completed"
        final_response["output"] = output_items

        # response.completed
        yield from _emit({"type": "response.completed", "response": final_response})
        yield "data: [DONE]\n\n"

        logger.info(
            "✅ Responses streaming completed",
            request_id=request_id,
            output_items=len(output_items),
        )

        if temp_files_to_cleanup:
            import threading

            def delayed_streaming_cleanup():
                time.sleep(2)
                for temp_file in temp_files_to_cleanup:
                    try:
                        if os.path.exists(temp_file) and (
                            "abstractcore_img_" in temp_file
                            or "abstractcore_file_" in temp_file
                            or "abstractcore_b64_" in temp_file
                        ):
                            os.unlink(temp_file)
                    except Exception as cleanup_error:
                        logger.warning(f"Failed to cleanup temporary file {temp_file}: {cleanup_error}")

            cleanup_thread = threading.Thread(target=delayed_streaming_cleanup, daemon=True)
            cleanup_thread.start()

    except Exception as e:
        logger.error(
            "❌ Responses streaming failed",
            request_id=request_id,
            error=str(e),
        )
        error_chunk = {"error": {"message": str(e), "type": "server_error"}}
        yield f"data: {json.dumps(error_chunk)}\n\n"
        yield "data: [DONE]\n\n"
    finally:
        if loaded_runtime is not None and unload_after:
            _best_effort_unload_loaded_gateway_runtime(loaded_runtime, request_id=request_id)
        elif provider_normalized == "ollama" and ollama_key is not None:
            try:
                should_unload = _ollama_inflight_exit(ollama_key, unload_after_requested=unload_after)
            except Exception as e:
                logger.warning(
                    "⚠️ Failed to update in-flight unload state",
                    request_id=request_id,
                    provider=provider,
                    model=model,
                    error=str(e),
                    error_type=type(e).__name__,
                )
                should_unload = False

            if should_unload and allow_unsafe_unload_after:
                _best_effort_unload(llm, request_id=request_id, provider=provider, model=model)
            elif should_unload:
                logger.warning(
                    "⚠️ Unload requested but disabled by server policy",
                    request_id=request_id,
                    provider=provider,
                    model=model,
                )
        elif unload_after:
            _best_effort_unload(llm, request_id=request_id, provider=provider, model=model)


def convert_openai_responses_to_chat_completion(openai_request: OpenAIResponsesRequest) -> ChatCompletionRequest:
    """
    Convert OpenAI Responses API format to internal ChatCompletionRequest format.

    Transforms:
    - input -> messages
    - input_text -> text
    - input_file -> file with file_url

    Args:
        openai_request: OpenAI responses API request

    Returns:
        ChatCompletionRequest compatible with our internal processing
    """
    messages: List[ChatMessage] = []

    if isinstance(openai_request.instructions, str) and openai_request.instructions.strip():
        messages.append(ChatMessage(role="system", content=openai_request.instructions.strip()))

    if isinstance(openai_request.input, str):
        raw = openai_request.input
        if raw.strip():
            messages.append(ChatMessage(role="user", content=raw))
    else:
        for input_item in openai_request.input:
            if isinstance(input_item, OpenAIResponsesFunctionCallOutput):
                messages.append(
                    ChatMessage(
                        role="tool",
                        tool_call_id=input_item.call_id,
                        content=input_item.output,
                    )
                )
                continue

            if isinstance(input_item, OpenAIResponsesFunctionCall):
                messages.append(
                    ChatMessage(
                        role="assistant",
                        content=None,
                        tool_calls=[
                            {
                                "id": input_item.call_id,
                                "type": "function",
                                "function": {
                                    "name": input_item.name,
                                    "arguments": input_item.arguments,
                                },
                            }
                        ],
                    )
                )
                continue

            # Message item
            role = input_item.role
            tool_call_id = input_item.tool_call_id
            tool_calls = input_item.tool_calls
            name = input_item.name

            if isinstance(input_item.content, str):
                messages.append(
                    ChatMessage(
                        role=role,
                        content=input_item.content,
                        tool_call_id=tool_call_id,
                        tool_calls=tool_calls,
                        name=name,
                    )
                )
                continue

            # Build content array as list of dicts (not ContentItem objects)
            content_items: List[Dict[str, Any]] = []
            for content in input_item.content:
                if content.type == "input_text":
                    content_items.append({"type": "text", "text": content.text})
                elif content.type == "input_file":
                    content_items.append({"type": "file", "file_url": {"url": content.file_url}})
                elif content.type == "input_image":
                    url_value = content.image_url
                    if isinstance(url_value, dict):
                        image_url_obj = dict(url_value)
                    else:
                        image_url_obj = {"url": url_value}
                    content_items.append({"type": "image_url", "image_url": image_url_obj})

            messages.append(
                ChatMessage(
                    role=role,
                    content=content_items,
                    tool_call_id=tool_call_id,
                    tool_calls=tool_calls,
                    name=name,
                )
            )

    # Build ChatCompletionRequest
    return ChatCompletionRequest(
        model=openai_request.model,
        messages=messages,
        max_tokens=openai_request.max_tokens,
        temperature=openai_request.temperature,
        top_p=openai_request.top_p,
        stream=openai_request.stream,
        stop=openai_request.stop,
        seed=openai_request.seed,
        frequency_penalty=openai_request.frequency_penalty,
        presence_penalty=openai_request.presence_penalty,
        thinking=openai_request.thinking,
        tools=_normalize_openai_responses_tools(openai_request.tools),
        tool_choice=_normalize_openai_responses_tool_choice(openai_request.tool_choice),
        prompt_cache_key=openai_request.prompt_cache_key,
        prompt_cache_binding=openai_request.prompt_cache_binding,
        prompt_cache_retention=openai_request.prompt_cache_retention,
        agent_format=openai_request.agent_format,
        api_key=openai_request.api_key,
        base_url=openai_request.base_url,
        timeout_s=openai_request.timeout_s,
        unload_after=bool(openai_request.unload_after),
    )

# ============================================================================
# Helper Functions
# ============================================================================

def _parse_bool_env(var_name: str) -> bool:
    """Parse a boolean environment variable (1/true/yes/on)."""
    val = os.getenv(var_name)
    if val is None:
        return False
    return str(val).strip().lower() in {"1", "true", "yes", "on"}


def _csv_env(var_name: str) -> List[str]:
    """Parse a comma-separated env var into a list of non-empty items."""
    raw = os.getenv(var_name)
    if not isinstance(raw, str) or not raw.strip():
        return []
    items = [p.strip() for p in raw.split(",")]
    return [p for p in items if p]


def _resolve_host_ips(hostname: str) -> List[str]:
    """Best-effort: resolve hostname to IP strings (v4/v6)."""
    host = str(hostname or "").strip()
    if not host:
        return []
    try:
        infos = socket.getaddrinfo(host, None)
    except Exception:
        return []
    out: List[str] = []
    for _fam, _typ, _proto, _canon, sockaddr in infos:
        try:
            ip = sockaddr[0]
        except Exception:
            continue
        if isinstance(ip, str) and ip and ip not in out:
            out.append(ip)
    return out


def _is_loopback_host(hostname: str) -> bool:
    host = str(hostname or "").strip().lower()
    if not host:
        return False
    if host == "localhost":
        return True
    try:
        return bool(ipaddress.ip_address(host).is_loopback)
    except Exception:
        pass
    ips = _resolve_host_ips(host)
    if not ips:
        return False
    try:
        return all(ipaddress.ip_address(ip).is_loopback for ip in ips)
    except Exception:
        return False


def _canonical_allowlist_host(hostname: Any) -> str:
    """Canonicalize a parsed hostname for allowlist comparisons."""
    host = str(hostname or "").strip().rstrip(".").lower()
    if not host:
        return ""
    try:
        return ipaddress.ip_address(host).compressed.lower()
    except Exception:
        pass
    try:
        return host.encode("idna").decode("ascii").lower()
    except Exception:
        return host


def _effective_url_port(scheme: str, port: Optional[int]) -> Optional[int]:
    if port is not None:
        return port
    if scheme == "http":
        return 80
    if scheme == "https":
        return 443
    return None


def _safe_split_http_url(value: str) -> Optional[Tuple[Any, str, str, Optional[int]]]:
    """Parse an http(s) URL into canonical pieces, rejecting malformed ports."""
    try:
        parsed = urllib.parse.urlsplit(str(value or "").strip())
        port = parsed.port
    except Exception:
        return None

    scheme = str(parsed.scheme or "").lower()
    if scheme not in {"http", "https"}:
        return None
    if parsed.username is not None or parsed.password is not None:
        return None

    host = _canonical_allowlist_host(parsed.hostname)
    if not host:
        return None

    return parsed, scheme, host, _effective_url_port(scheme, port)


def _canonical_allowlist_path(path: Any) -> str:
    """Normalize URL paths for path-prefix allowlists.

    Repeated percent-decoding and dot-segment normalization make `/v1/../admin`
    and `/v1/%252e%252e/admin` compare as `/admin`, avoiding prefix bypasses.
    """
    p = str(path or "")
    if not p.startswith("/"):
        p = "/" + p

    for _ in range(3):
        decoded = urllib.parse.unquote(p)
        if decoded == p:
            break
        p = decoded

    p = p.replace("\\", "/")
    normalized = posixpath.normpath(p)
    if not normalized.startswith("/"):
        normalized = "/" + normalized
    if p.endswith("/") and normalized != "/" and not normalized.endswith("/"):
        normalized += "/"
    return normalized


def _allowlist_path_prefix_matches(target_path: Any, allowed_path: Any) -> bool:
    allowed = _canonical_allowlist_path(allowed_path)
    target = _canonical_allowlist_path(target_path)
    if allowed == "/":
        return True
    allowed = allowed.rstrip("/")
    return target == allowed or target.startswith(f"{allowed}/")


def _split_host_allowlist_item(item: str) -> Tuple[str, Optional[int]]:
    """Split a host/glob allowlist entry into host pattern and optional port."""
    raw = str(item or "").strip()
    if not raw:
        return "", None

    if raw.startswith("["):
        end = raw.find("]")
        if end > 0:
            host = raw[1:end]
            rest = raw[end + 1:]
            if rest.startswith(":") and rest[1:].isdigit():
                return host, int(rest[1:])
            if not rest:
                return host, None
            return raw, None

    if raw.count(":") == 1:
        host, port_s = raw.rsplit(":", 1)
        if port_s.isdigit():
            return host, int(port_s)

    return raw, None


def _allowlist_matches_url(url: str, allowlist: List[str]) -> bool:
    """Match a URL/host against an operator-provided allowlist.

    Allowlist items:
    - If item contains '://': match scheme, host, effective port, and path prefix
      with a path-segment boundary.
    - Otherwise: match a hostname glob (optionally `:port`), e.g. '*.example.com'.
    """
    target = _safe_split_http_url(url)
    if target is None or not allowlist:
        return False
    target_parsed, target_scheme, target_host, target_port = target

    for item in allowlist:
        it = str(item or "").strip()
        if not it:
            continue
        if "://" in it:
            allowed = _safe_split_http_url(it)
            if allowed is None:
                continue
            allowed_parsed, allowed_scheme, allowed_host, allowed_port = allowed

            if allowed_parsed.username is not None or allowed_parsed.password is not None:
                continue
            if allowed_parsed.fragment:
                continue

            if target_scheme != allowed_scheme:
                continue
            if target_host != allowed_host:
                continue
            if target_port != allowed_port:
                continue
            if allowed_parsed.query and target_parsed.query != allowed_parsed.query:
                continue
            if _allowlist_path_prefix_matches(target_parsed.path, allowed_parsed.path):
                return True
            continue

        host_pattern, allowed_port = _split_host_allowlist_item(it)
        if allowed_port is not None and target_port != allowed_port:
            continue
        if any(ch in host_pattern for ch in "*?["):
            pattern = host_pattern.strip().rstrip(".").lower()
        else:
            pattern = _canonical_allowlist_host(host_pattern)
        if pattern and fnmatch.fnmatchcase(target_host, pattern):
            return True
    return False


def _require_public_url(url: str, *, allowlist_env: str) -> None:
    """Enforce that a URL does not resolve to private/link-local/loopback addresses.

    Operator escape hatch: set an allowlist env var (comma-separated) to permit specific
    structured URL entries or host globs for controlled environments.
    """
    u = str(url or "").strip()
    if not u:
        raise HTTPException(status_code=400, detail={"error": {"message": "Empty URL", "type": "invalid_request"}})

    try:
        parsed = urllib.parse.urlsplit(u)
        _ = parsed.port
    except Exception:
        raise HTTPException(
            status_code=400,
            detail={"error": {"message": "Invalid URL", "type": "invalid_request"}},
        )
    if parsed.scheme not in {"http", "https"}:
        raise HTTPException(
            status_code=400,
            detail={"error": {"message": "Only http/https URLs are allowed", "type": "invalid_request"}},
        )
    host = parsed.hostname
    if not host:
        raise HTTPException(
            status_code=400,
            detail={"error": {"message": "Invalid URL (missing host)", "type": "invalid_request"}},
        )

    allowlist = _csv_env(allowlist_env)
    if _allowlist_matches_url(u, allowlist):
        return

    ips = _resolve_host_ips(host)
    if not ips:
        raise HTTPException(
            status_code=403,
            detail={
                "error": {
                    "message": f"URL host did not resolve: {host}",
                    "type": "forbidden",
                }
            },
        )
    # Block anything that is not globally routable (covers loopback, RFC1918, link-local, etc).
    for ip in ips:
        try:
            if not ipaddress.ip_address(ip).is_global:
                raise HTTPException(
                    status_code=403,
                    detail={
                        "error": {
                            "message": (
                                "Refusing to fetch non-public URL target (SSRF protection). "
                                f"Host={host} resolved to {ip}. "
                                f"To allow this, set {allowlist_env} with an explicit allowlist."
                            ),
                            "type": "forbidden",
                        }
                    },
                )
        except HTTPException:
            raise
        except Exception:
            # If parsing fails, be conservative.
            raise HTTPException(
                status_code=403,
                detail={
                    "error": {
                        "message": f"Refusing to fetch URL with unparseable IP for host={host!r} (SSRF protection).",
                        "type": "forbidden",
                    }
                },
            )


def _provider_api_key_env_var(provider: str) -> Optional[str]:
    """Best-effort mapping of provider -> env var used for auth."""
    p = str(provider or "").strip().lower()
    if p == "openai":
        return "OPENAI_API_KEY"
    if p == "anthropic":
        return "ANTHROPIC_API_KEY"
    if p == "openrouter":
        return "OPENROUTER_API_KEY"
    if p == "portkey":
        return "PORTKEY_API_KEY"
    if p == "openai-compatible":
        return "OPENAI_API_KEY"
    if p == "vllm":
        return "VLLM_API_KEY"
    return None


def _server_has_provider_api_key(provider: str) -> bool:
    p = str(provider or "").strip().lower()
    env_var = _provider_api_key_env_var(provider)
    if env_var and str(os.getenv(env_var) or "").strip():
        return True

    try:
        from ..config import manager as config_manager_module

        cfg = getattr(config_manager_module, "_config_manager", None)
        if cfg is not None:
            runtime_config = cfg.get_provider_config(p)
            for key_name in ("api_key", "provider_api_key"):
                if str(runtime_config.get(key_name) or "").strip():
                    return True
    except Exception:
        pass

    return False


def _reject_body_api_key(api_key: Any) -> None:
    if api_key is None:
        return
    raise HTTPException(
        status_code=400,
        detail={
            "error": {
                "message": (
                    "Per-request provider api_key fields are disabled for the HTTP server because they leak "
                    "through bodies, logs, and client history. Use X-AbstractCore-Provider-API-Key for a "
                    "provider override, or configure provider keys on the server."
                ),
                "type": "invalid_request",
            }
        },
    )


def _reject_query_api_key(api_key: Any) -> None:
    if api_key is None:
        return
    raise HTTPException(
        status_code=400,
        detail={
            "error": {
                "message": (
                    "Query-string api_key is disabled because URLs are routinely logged. Use Authorization "
                    "for the AbstractCore server auth token when configured, use X-AbstractCore-Provider-API-Key "
                    "for a provider override, or configure provider keys on the server."
                ),
                "type": "invalid_request",
            }
        },
    )


def _guard_unauthenticated_server_provider_key_use(
    provider: str,
    *,
    explicit_provider_key: bool,
    http_request: Optional[Request] = None,
) -> None:
    """Do not let unauthenticated HTTP callers spend or exfiltrate server-held provider keys."""
    if _request_has_server_auth(http_request) or explicit_provider_key:
        return
    if not _server_has_provider_api_key(provider):
        return

    env_var = _provider_api_key_env_var(provider) or "provider API key"
    raise HTTPException(
        status_code=401,
        detail={
            "error": {
                "message": (
                    f"Server-held {env_var} is configured, but inbound server auth is not. "
                    "Set ABSTRACTCORE_AUTH_TOKEN "
                    "and send Authorization: Bearer <server-token>, "
                    "or pass an explicit provider key with X-AbstractCore-Provider-API-Key."
                ),
                "type": "authentication_error",
            }
        },
        headers={"WWW-Authenticate": "Bearer"},
    )


def _server_allows_request_base_url(base_url: str) -> bool:
    """Allow request-level provider base_url overrides only when explicitly safe.

    Default: allow loopback only. Operators can expand via `ABSTRACTCORE_SERVER_BASE_URL_ALLOWLIST`.
    """
    u = str(base_url or "").strip()
    if not u:
        return False
    parsed = _safe_split_http_url(u)
    if parsed is None:
        return False
    _parsed_url, _scheme, host, _port = parsed
    if _is_loopback_host(host):
        return True
    allowlist = _csv_env("ABSTRACTCORE_SERVER_BASE_URL_ALLOWLIST")
    return _allowlist_matches_url(u, allowlist)


def _server_media_root() -> Optional[Path]:
    """Optional safe-root for local file attachments in HTTP requests.

    If set, local paths are only allowed when they resolve under this directory.
    """
    raw = os.getenv("ABSTRACTCORE_SERVER_MEDIA_ROOT")
    if not isinstance(raw, str) or not raw.strip():
        return None
    try:
        return Path(raw).expanduser().resolve()
    except Exception:
        return None


def _server_allows_local_files() -> bool:
    """Dangerous: allow arbitrary local file paths from HTTP requests."""
    return _parse_bool_env("ABSTRACTCORE_SERVER_ALLOW_LOCAL_FILES")


def _resolve_local_media_path(path_value: str) -> str:
    """Resolve and validate a local file path under server policy."""
    raw = str(path_value or "").strip()
    if not raw:
        raise HTTPException(
            status_code=400,
            detail={"error": {"message": "Empty local file path", "type": "invalid_request"}},
        )

    root = _server_media_root()
    allow_all = _server_allows_local_files()
    if root is None and not allow_all:
        raise HTTPException(
            status_code=403,
            detail={
                "error": {
                    "message": (
                        "Local file paths are disabled for HTTP requests. "
                        "Use http(s) URLs or data: base64 attachments, or configure "
                        "ABSTRACTCORE_SERVER_MEDIA_ROOT (safe) / ABSTRACTCORE_SERVER_ALLOW_LOCAL_FILES=1 (unsafe)."
                    ),
                    "type": "forbidden",
                }
            },
        )

    p = Path(raw).expanduser()
    if root is not None and not p.is_absolute():
        p = root / p

    try:
        resolved = p.resolve()
    except Exception as e:
        raise HTTPException(
            status_code=400,
            detail={"error": {"message": f"Invalid local path: {e}", "type": "invalid_request"}},
        ) from e

    if root is not None:
        try:
            if not resolved.is_relative_to(root):
                raise HTTPException(
                    status_code=403,
                    detail={
                        "error": {
                            "message": (
                                "Local file path is outside ABSTRACTCORE_SERVER_MEDIA_ROOT. "
                                f"root={str(root)} path={str(resolved)}"
                            ),
                            "type": "forbidden",
                        }
                    },
                )
        except HTTPException:
            raise
        except Exception:
            raise HTTPException(
                status_code=403,
                detail={
                    "error": {
                        "message": "Local file path rejected by server policy.",
                        "type": "forbidden",
                    }
                },
            )

    if not resolved.exists() or not resolved.is_file():
        raise HTTPException(
            status_code=400,
            detail={"error": {"message": f"File not found: {str(resolved)}", "type": "file_not_found"}},
        )

    return str(resolved)


_OLLAMA_INFLIGHT_LOCK = threading.Lock()
_OLLAMA_INFLIGHT_COUNTS: Dict[Tuple[str, str, str], int] = {}
_OLLAMA_UNLOAD_REQUESTED: Dict[Tuple[str, str, str], bool] = {}


def _ollama_inflight_key(provider: str, base_url: Optional[str], model: str) -> Tuple[str, str, str]:
    """Build a stable key for tracking in-flight Ollama requests."""
    return (provider.strip().lower(), (base_url or "").strip(), model)


def _ollama_inflight_enter(key: Tuple[str, str, str]) -> None:
    """Increment in-flight counter for an Ollama (provider/base_url/model) key."""
    with _OLLAMA_INFLIGHT_LOCK:
        _OLLAMA_INFLIGHT_COUNTS[key] = _OLLAMA_INFLIGHT_COUNTS.get(key, 0) + 1


def _ollama_inflight_exit(key: Tuple[str, str, str], *, unload_after_requested: bool) -> bool:
    """Decrement in-flight counter and return True if an unload should happen now."""
    with _OLLAMA_INFLIGHT_LOCK:
        if unload_after_requested:
            _OLLAMA_UNLOAD_REQUESTED[key] = True

        current = _OLLAMA_INFLIGHT_COUNTS.get(key, 0)
        if current <= 1:
            _OLLAMA_INFLIGHT_COUNTS.pop(key, None)
            return bool(_OLLAMA_UNLOAD_REQUESTED.pop(key, False))

        _OLLAMA_INFLIGHT_COUNTS[key] = current - 1
        return False


def _best_effort_unload(llm: Any, *, request_id: str, provider: str, model: str) -> None:
    """Unload provider resources without failing the request lifecycle."""
    try:
        if not hasattr(llm, "unload_model"):
            raise AttributeError("Provider does not implement unload_model(model_name)")
        llm.unload_model(model)
        logger.info("🧹 Provider Unloaded", request_id=request_id, provider=provider, model=model)
    except Exception as e:
        logger.warning(
            "⚠️ Provider unload failed",
            request_id=request_id,
            provider=provider,
            model=model,
            error=str(e),
            error_type=type(e).__name__,
        )


def parse_model_string(model_string: str) -> tuple[str, str]:
    """Parse model string to extract provider and model."""
    if not model_string:
        return "ollama", "qwen3-coder:30b"  # Default

    # Explicit provider/model format
    if '/' in model_string:
        parts = model_string.split('/', 1)
        provider = parts[0].strip()
        model = parts[1].strip()
        return provider, model

    # Auto-detect provider from model name
    model_lower = model_string.lower()

    if any(pattern in model_lower for pattern in ['gpt-', 'text-davinci', 'text-embedding']):
        return "openai", model_string
    elif any(pattern in model_lower for pattern in ['claude']):
        return "anthropic", model_string
    elif any(pattern in model_lower for pattern in ['llama', 'mistral', 'gemma', 'phi', 'qwen']):
        return "ollama", model_string
    elif any(pattern in model_lower for pattern in ['-4bit', 'mlx-community']):
        return "mlx", model_string
    else:
        return "ollama", model_string  # Default


class _GatewayLoadedRuntime:
    def __init__(
        self,
        *,
        provider: str,
        model: str,
        base_url: Optional[str],
        explicit_provider_key_hash: Optional[str],
        llm: Any,
    ) -> None:
        self.provider = provider
        self.model = model
        self.base_url = base_url
        self.explicit_provider_key_hash = explicit_provider_key_hash
        self.llm = llm
        signature = f"{provider}|{model}|{base_url or ''}|{explicit_provider_key_hash or ''}"
        self.runtime_id = hashlib.sha256(signature.encode("utf-8")).hexdigest()[:16]
        # Streaming responses may be iterated and finalized by different ASGI
        # worker threads when the client disconnects or cancels a request.
        # `threading.RLock` is owner-bound, so finalization from another thread
        # can leave the loaded runtime permanently wedged. Use a plain Lock:
        # runtime generation still stays serialized, but release is not tied to
        # the thread that first acquired it.
        self.lock = threading.Lock()
        now = time.time()
        self.created_at = now
        self.last_used_at = now
        self.request_count = 0
        self.provider_executor = ThreadPoolExecutor(
            max_workers=1,
            thread_name_prefix=f"abstractcore-gateway-{self.runtime_id}",
        )


_GATEWAY_RUNTIME_LOCK = threading.RLock()
_GATEWAY_LOADED_RUNTIMES: Dict[str, _GatewayLoadedRuntime] = {}
_GATEWAY_RUNTIME_IDS: Dict[str, str] = {}
_SERVER_BLOC_STORE = FileBlocStore()


def _normalize_loaded_runtime_base_url(base_url: Optional[str]) -> Optional[str]:
    if not isinstance(base_url, str):
        return None
    value = base_url.strip()
    if not value:
        return None
    return value.rstrip("/")


def _provider_key_fingerprint(provider_api_key: Optional[str]) -> Optional[str]:
    if not isinstance(provider_api_key, str) or not provider_api_key.strip():
        return None
    return hashlib.sha256(provider_api_key.strip().encode("utf-8")).hexdigest()


def _gateway_runtime_registry_key(
    *,
    provider: str,
    model: str,
    base_url: Optional[str],
    explicit_provider_key_hash: Optional[str],
) -> str:
    return "|".join(
        [
            str(provider or "").strip().lower(),
            str(model or "").strip(),
            _normalize_loaded_runtime_base_url(base_url) or "",
            explicit_provider_key_hash or "",
        ]
    )


def _gateway_runtime_to_dict(runtime: _GatewayLoadedRuntime) -> Dict[str, Any]:
    return {
        "runtime_id": runtime.runtime_id,
        "provider": runtime.provider,
        "model": runtime.model,
        "base_url": runtime.base_url,
        "uses_explicit_provider_key": bool(runtime.explicit_provider_key_hash),
        "resolved_model_id": getattr(runtime.llm, "_resolved_model_id", None),
        "created_at": int(runtime.created_at),
        "last_used_at": int(runtime.last_used_at),
        "request_count": int(runtime.request_count),
        "supports_prompt_cache": bool(getattr(runtime.llm, "supports_prompt_cache", lambda: False)()),
    }


def _gateway_residency_bool(value: Any) -> Optional[bool]:
    if isinstance(value, bool):
        return value
    if isinstance(value, str):
        raw = value.strip().lower()
        if raw in {"1", "true", "yes", "on"}:
            return True
        if raw in {"0", "false", "no", "off"}:
            return False
    return None


def _unknown_gateway_provider_residency(*, warning: str) -> Dict[str, Any]:
    return {
        "provider_residency_verified": False,
        "provider_resident": None,
        "provider_residency_source": "abstractcore.provider",
        "provider_state": "provider_residency_unknown",
        "warnings": [warning],
    }


def _gateway_provider_residency_claim(runtime: _GatewayLoadedRuntime) -> Dict[str, Any]:
    method = getattr(runtime.llm, "get_model_residency", None)
    if not callable(method):
        return _unknown_gateway_provider_residency(
            warning="AbstractCore provider does not expose verified model residency."
        )

    try:
        raw_claim = _run_loaded_gateway_runtime(
            runtime,
            lambda: method(task="text_generation", model=runtime.model),
            touch=False,
        )
    except Exception as exc:
        return _unknown_gateway_provider_residency(
            warning=f"AbstractCore provider residency query failed: {exc}"
        )
    if not isinstance(raw_claim, dict):
        return _unknown_gateway_provider_residency(
            warning="AbstractCore provider residency query returned a non-mapping response."
        )

    verified = _gateway_residency_bool(raw_claim.get("provider_residency_verified")) is True
    provider_resident = _gateway_residency_bool(raw_claim.get("provider_resident"))
    if provider_resident is None and verified:
        provider_resident = _gateway_residency_bool(raw_claim.get("loaded"))

    source = raw_claim.get("provider_residency_source") or raw_claim.get("source") or "abstractcore.provider"
    claim: Dict[str, Any] = {
        "provider_residency_verified": verified,
        "provider_resident": provider_resident,
        "provider_residency_source": str(source),
    }
    provider_state = raw_claim.get("provider_state") or raw_claim.get("state")
    if isinstance(provider_state, str) and provider_state.strip():
        claim["provider_state"] = provider_state.strip()

    blocked = {
        "task",
        "provider",
        "model",
        "runtime_id",
        "resident",
        "loaded",
        "state",
        "source",
        "isolation",
        "default",
        "pinned",
        "cache_state",
        "runtime_cached",
        "provider_residency_verified",
        "provider_resident",
        "provider_residency_source",
        "provider_state",
    }
    for key, value in raw_claim.items():
        if key in blocked or value is None:
            continue
        claim[str(key)] = value
    return claim


def _gateway_text_residency_record(runtime: _GatewayLoadedRuntime) -> Dict[str, Any]:
    provider_claim = _gateway_provider_residency_claim(runtime)
    verified = bool(provider_claim.get("provider_residency_verified"))
    provider_resident_raw = provider_claim.get("provider_resident")
    provider_resident = provider_resident_raw if isinstance(provider_resident_raw, bool) else None
    if verified:
        loaded = bool(provider_resident)
        state = "provider_loaded" if loaded else "provider_not_loaded"
    else:
        loaded = False
        state = "provider_residency_unknown"
    return {
        **_gateway_runtime_to_dict(runtime),
        "task": "text_generation",
        "state": state,
        "loaded": loaded,
        "runtime_cached": True,
        "cache_state": "gateway_runtime_cached",
        "pinned": True,
        "isolation": "in_process",
        "health": "ok",
        "error": None,
        **provider_claim,
    }


def _drop_loaded_gateway_runtime(runtime: _GatewayLoadedRuntime) -> None:
    runtime.provider_executor.shutdown(wait=False, cancel_futures=True)
    with _GATEWAY_RUNTIME_LOCK:
        registry_key = _GATEWAY_RUNTIME_IDS.pop(runtime.runtime_id, None)
        if registry_key:
            _GATEWAY_LOADED_RUNTIMES.pop(registry_key, None)


def _gateway_load_provider_residency(runtime: _GatewayLoadedRuntime, options: Optional[Dict[str, Any]] = None) -> Any:
    method = getattr(runtime.llm, "load_model", None)
    if not callable(method):
        return None

    load_options = dict(options or {}) if isinstance(options, dict) else {}
    return _run_loaded_gateway_runtime(
        runtime,
        lambda: method(runtime.model, **load_options),
    )


def _get_loaded_gateway_runtime(
    *,
    provider: Optional[str] = None,
    model: Optional[str] = None,
    base_url: Optional[str] = None,
    explicit_provider_key_hash: Optional[str] = None,
    runtime_id: Optional[str] = None,
) -> Optional[_GatewayLoadedRuntime]:
    with _GATEWAY_RUNTIME_LOCK:
        if isinstance(runtime_id, str) and runtime_id.strip():
            registry_key = _GATEWAY_RUNTIME_IDS.get(runtime_id.strip())
            if registry_key:
                return _GATEWAY_LOADED_RUNTIMES.get(registry_key)
            return None
        if not isinstance(provider, str) or not provider.strip() or not isinstance(model, str) or not model.strip():
            return None
        registry_key = _gateway_runtime_registry_key(
            provider=provider,
            model=model,
            base_url=base_url,
            explicit_provider_key_hash=explicit_provider_key_hash,
        )
        return _GATEWAY_LOADED_RUNTIMES.get(registry_key)


def _touch_loaded_gateway_runtime(runtime: _GatewayLoadedRuntime) -> None:
    runtime.last_used_at = time.time()
    runtime.request_count += 1


def _run_loaded_gateway_runtime(runtime: _GatewayLoadedRuntime, operation, *, touch: bool = True):
    """Run provider work on the runtime's stable worker thread.

    MLX prompt caches and generation state are thread-affine. A loaded gateway
    runtime must keep cache operations, bloc KV loading, and generation on the
    same provider thread or a later request can inherit incompatible local
    state and wedge the loaded runtime.
    """
    with runtime.lock:
        if touch:
            _touch_loaded_gateway_runtime(runtime)
        return runtime.provider_executor.submit(operation).result()


def _stream_loaded_gateway_runtime(runtime: _GatewayLoadedRuntime, operation) -> Iterator[Any]:
    # Do not bound the bridge queue here. A loaded runtime has a single provider
    # worker thread; if the client stops draining a bounded queue, that worker
    # can block forever in `put()` and wedge every later cache-control or chat
    # request for the runtime.
    out: "queue.Queue[Tuple[str, Any]]" = queue.Queue()

    with runtime.lock:
        _touch_loaded_gateway_runtime(runtime)

    def _run() -> None:
        try:
            chunks = operation()
            if not hasattr(chunks, "__iter__"):
                out.put_nowait(("error", RuntimeError("provider did not return an iterator for stream=True")))
                return
            for chunk in chunks:
                out.put_nowait(("chunk", chunk))
        except Exception as e:
            out.put_nowait(("error", e))
        finally:
            out.put_nowait(("done", None))

    runtime.provider_executor.submit(_run)
    while True:
        kind, value = out.get()
        if kind == "done":
            break
        if kind == "error":
            raise value
        yield value


def _ensure_loaded_gateway_runtime(
    *,
    provider: str,
    model: str,
    provider_kwargs: Dict[str, Any],
    explicit_provider_key_hash: Optional[str],
) -> Tuple[_GatewayLoadedRuntime, bool]:
    registry_key = _gateway_runtime_registry_key(
        provider=provider,
        model=model,
        base_url=provider_kwargs.get("base_url"),
        explicit_provider_key_hash=explicit_provider_key_hash,
    )
    with _GATEWAY_RUNTIME_LOCK:
        existing = _GATEWAY_LOADED_RUNTIMES.get(registry_key)
        if existing is not None:
            return existing, False
        llm = create_llm(provider, model=model, **provider_kwargs)
        runtime = _GatewayLoadedRuntime(
            provider=str(provider).strip().lower(),
            model=str(model).strip(),
            base_url=_normalize_loaded_runtime_base_url(provider_kwargs.get("base_url")),
            explicit_provider_key_hash=explicit_provider_key_hash,
            llm=llm,
        )
        _GATEWAY_LOADED_RUNTIMES[registry_key] = runtime
        _GATEWAY_RUNTIME_IDS[runtime.runtime_id] = registry_key
        return runtime, True


def _unload_loaded_gateway_runtime(runtime: _GatewayLoadedRuntime) -> None:
    if not hasattr(runtime.llm, "unload_model"):
        raise AttributeError("Provider does not implement unload_model(model_name)")
    _run_loaded_gateway_runtime(runtime, lambda: runtime.llm.unload_model(runtime.model))
    _drop_loaded_gateway_runtime(runtime)


def _best_effort_unload_loaded_gateway_runtime(runtime: _GatewayLoadedRuntime, *, request_id: str) -> None:
    try:
        _unload_loaded_gateway_runtime(runtime)
        logger.info(
            "🧹 Gateway Runtime Unloaded",
            request_id=request_id,
            provider=runtime.provider,
            model=runtime.model,
            runtime_id=runtime.runtime_id,
        )
    except Exception as e:
        logger.warning(
            "⚠️ Gateway runtime unload failed",
            request_id=request_id,
            provider=runtime.provider,
            model=runtime.model,
            runtime_id=runtime.runtime_id,
            error=str(e),
            error_type=type(e).__name__,
        )


def _resolve_local_runtime_for_control(
    *,
    provider: Optional[str],
    model: Optional[str],
    http_request: Optional[Request],
    base_url: Optional[str] = None,
    runtime_id: Optional[str] = None,
) -> _GatewayLoadedRuntime:
    explicit_provider_key_hash = _provider_key_fingerprint(_provider_api_key_from_request(http_request)) if http_request is not None else None
    runtime = _get_loaded_gateway_runtime(
        provider=provider,
        model=model,
        base_url=base_url,
        explicit_provider_key_hash=explicit_provider_key_hash,
        runtime_id=runtime_id,
    )
    if runtime is None:
        raise HTTPException(
            status_code=404,
            detail={
                "error": {
                    "message": (
                        "Loaded gateway runtime not found. Load it first with `/acore/models/load` "
                        "or proxy to an AbstractEndpoint via `base_url`."
                    ),
                    "type": "not_found",
                }
            },
        )
    return runtime


def _gateway_prompt_cache_capabilities_dict(llm: Any) -> Dict[str, Any]:
    getter = getattr(llm, "get_prompt_cache_capabilities", None)
    if callable(getter):
        try:
            caps = getter()
            to_dict = getattr(caps, "to_dict", None)
            if callable(to_dict):
                return to_dict()
        except Exception:
            pass
    return {
        "supported": bool(getattr(llm, "supports_prompt_cache", lambda: False)()),
    }


def _gateway_prompt_cache_error_payload(llm: Any, error: Exception, *, operation: str) -> Dict[str, Any]:
    caps = _gateway_prompt_cache_capabilities_dict(llm)
    if isinstance(error, PromptCacheError):
        payload = error.to_dict()
        payload.setdefault("capabilities", caps)
        return {
            "supported": False,
            "operation": payload.get("operation") or operation,
            "code": payload.get("code") or "prompt_cache_error",
            "error": payload.get("message") or str(error),
            "capabilities": payload.get("capabilities") or caps,
        }
    return {
        "supported": False,
        "operation": operation,
        "code": "prompt_cache_error",
        "error": str(error),
        "capabilities": caps,
    }


def _prompt_cache_binding_status(error: Exception) -> int:
    if not isinstance(error, PromptCacheError):
        return 400
    if error.code in {"prompt_cache_binding_missing", "prompt_cache_binding_mismatch"}:
        return 409
    return 400


def _apply_request_prompt_cache_binding(
    *,
    llm: Any,
    gen_kwargs: Dict[str, Any],
    binding: Optional[Union[Dict[str, Any], str]],
    loaded_runtime: Optional[_GatewayLoadedRuntime] = None,
) -> None:
    if binding is None:
        return
    if isinstance(binding, str):
        binding_obj: Dict[str, Any] = {"binding_id": binding.strip()}
    elif isinstance(binding, dict):
        binding_obj = dict(binding)
    else:
        raise HTTPException(
            status_code=400,
            detail={
                "error": {
                    "message": "prompt_cache_binding must be an object or binding_id string.",
                    "type": "invalid_request_error",
                    "code": "prompt_cache_binding_invalid",
                }
            },
        )

    binding_key = binding_obj.get("key")
    current_key = gen_kwargs.get("prompt_cache_key")
    if isinstance(binding_key, str) and binding_key.strip():
        binding_key_s = binding_key.strip()
        if isinstance(current_key, str) and current_key.strip() and current_key.strip() != binding_key_s:
            raise HTTPException(
                status_code=400,
                detail={
                    "error": {
                        "message": "prompt_cache_key and prompt_cache_binding.key must match.",
                        "type": "invalid_request_error",
                        "code": "prompt_cache_binding_key_mismatch",
                    }
                },
            )
        gen_kwargs["prompt_cache_key"] = binding_key_s
        current_key = binding_key_s

    try:
        if loaded_runtime is not None:
            _run_loaded_gateway_runtime(
                loaded_runtime,
                lambda: llm.prompt_cache_validate_binding(current_key, binding_obj),
            )
        else:
            llm.prompt_cache_validate_binding(current_key, binding_obj)
    except Exception as e:
        payload = _gateway_prompt_cache_error_payload(llm, e, operation="binding")
        raise HTTPException(
            status_code=_prompt_cache_binding_status(e),
            detail={
                "error": {
                    "message": payload.get("error") or str(e),
                    "type": "invalid_request_error",
                    "code": payload.get("code") or "prompt_cache_binding_invalid",
                    "details": payload,
                }
            },
        ) from e


def detect_target_format(
    model: str,
    request: ChatCompletionRequest,
    http_request: Request
) -> SyntaxFormat:
    """
    Detect the target format for tool call syntax conversion.

    Args:
        model: Model identifier
        request: Chat completion request
        http_request: HTTP request object

    Returns:
        Target syntax format
    """
    # Explicit format override
    if request.agent_format:
        try:
            return SyntaxFormat(request.agent_format.lower())
        except ValueError:
            logger.warning(f"Invalid agent_format '{request.agent_format}', using auto-detection")

    # Auto-detect from headers and model
    user_agent = http_request.headers.get("user-agent", "")
    return auto_detect_format(model, user_agent)

def convert_to_abstractcore_messages(openai_messages: List[ChatMessage]) -> List[Dict[str, Any]]:
    """Convert OpenAI messages to AbstractCore format."""
    messages = []
    for msg in openai_messages:
        message_dict = {"role": msg.role}

        if msg.content is not None:
            message_dict["content"] = msg.content
        if msg.tool_calls:
            message_dict["tool_calls"] = msg.tool_calls
        if msg.tool_call_id:
            message_dict["tool_call_id"] = msg.tool_call_id
        if msg.name:
            message_dict["name"] = msg.name

        messages.append(message_dict)

    return messages

def create_syntax_rewriter(target_format: SyntaxFormat, model_name: str) -> ToolCallSyntaxRewriter:
    """Create appropriate syntax rewriter for target format."""
    if target_format == SyntaxFormat.PASSTHROUGH:
        return create_passthrough_rewriter()
    elif target_format == SyntaxFormat.CODEX:
        return create_codex_rewriter(model_name)
    elif target_format == SyntaxFormat.OPENAI:
        return create_openai_rewriter(model_name)
    else:
        return ToolCallSyntaxRewriter(target_format, model_name=model_name)


def _normalize_model_residency_task(task: Optional[str]) -> str:
    raw = str(task or "").strip().lower().replace("-", "_")
    if not raw:
        return "text_generation"
    aliases = {
        "text": "text_generation",
        "chat": "text_generation",
        "chat_completion": "text_generation",
        "chat_completions": "text_generation",
        "llm": "text_generation",
        "completion": "text_generation",
        "text_generation": "text_generation",
        "image": "image_generation",
        "vision": "image_generation",
        "image_generation": "image_generation",
        "text_to_image": "image_generation",
        "t2i": "image_generation",
        "image_to_image": "image_to_image",
        "i2i": "image_to_image",
        "image_edit": "image_to_image",
        "edit_image": "image_to_image",
        "image_upscale": "image_upscale",
        "upscale": "image_upscale",
        "upscale_image": "image_upscale",
        "super_resolution": "image_upscale",
        "video": "video_generation",
        "video_generation": "video_generation",
        "text_to_video": "text_to_video",
        "t2v": "text_to_video",
        "image_to_video": "image_to_video",
        "i2v": "image_to_video",
        "speech": "tts",
        "voice": "tts",
        "audio_speech": "tts",
        "text_to_speech": "tts",
        "tts": "tts",
        "transcription": "stt",
        "transcriptions": "stt",
        "speech_to_text": "stt",
        "audio_transcription": "stt",
        "audio_transcriptions": "stt",
        "stt": "stt",
        "music": "music_generation",
        "music_generation": "music_generation",
        "text_to_music": "music_generation",
        "t2m": "music_generation",
    }
    if raw not in aliases:
        raise HTTPException(
            status_code=400,
            detail={"error": {"message": f"Unknown model residency task: {task}", "type": "invalid_request"}},
        )
    return aliases[raw]


_VISION_RESIDENCY_TASKS = {
    "image_generation",
    "image_to_image",
    "image_upscale",
    "video_generation",
    "text_to_video",
    "image_to_video",
}


def _model_residency_payload(req: BaseModel) -> Dict[str, Any]:
    data = req.model_dump(exclude_none=True)
    options = data.get("options")
    if not isinstance(options, dict):
        data["options"] = {}
    return data


def _capability_residency_core_for_request(task: str, http_request: Request) -> Any:
    if task in {"tts", "stt"}:
        from .audio_endpoints import _audio_catalog_core, _get_capability_core

        base_url = None
        provider_key = _provider_api_key_from_request(http_request)
        if provider_key:
            return _audio_catalog_core(http_request, base_url=base_url, api_key=provider_key)
        return _get_capability_core()
    from .capability_generation import create_capability_generation_core

    return create_capability_generation_core()


def _capability_residency_error(exc: Exception, *, task: str) -> HTTPException:
    if isinstance(exc, HTTPException):
        return exc
    detail = {
        "error": {
            "message": str(exc),
            "type": "capability_unavailable" if isinstance(exc, CapabilityUnavailableError) else "residency_error",
            "task": task,
        }
    }
    if isinstance(exc, CapabilityUnavailableError):
        return HTTPException(status_code=501, detail=detail)
    return HTTPException(status_code=500, detail=detail)


def _model_residency_bool(value: Any) -> Optional[bool]:
    if isinstance(value, bool):
        return value
    if isinstance(value, (int, float)):
        return bool(value)
    if isinstance(value, str):
        raw = value.strip().lower()
        if raw in {"1", "true", "yes", "on"}:
            return True
        if raw in {"0", "false", "no", "off"}:
            return False
    return None


def _model_residency_loaded_new(runtime: Dict[str, Any]) -> bool:
    for key in ("loaded_new", "created_new", "created", "warmed_new", "preloaded_new"):
        if key in runtime:
            return bool(_model_residency_bool(runtime.get(key)))

    details = runtime.get("details")
    if isinstance(details, dict):
        for key in ("loaded_new", "created_new", "created", "warmed_new", "preloaded_new"):
            if key in details:
                return bool(_model_residency_bool(details.get(key)))

        loaded = _model_residency_bool(runtime.get("loaded"))
        engine_cached = _model_residency_bool(details.get("engine_cached"))
        if loaded is True and engine_cached is not None:
            return not engine_cached

    return False


def _normalize_model_residency_loaded_fields(runtime: Dict[str, Any]) -> Dict[str, Any]:
    """Normalize runtime records so `loaded` is the only loaded-state boolean."""
    out = dict(runtime or {}) if isinstance(runtime, dict) else {}
    loaded = _model_residency_bool(out.get("loaded"))
    resident = _model_residency_bool(out.get("resident"))
    if loaded is None and resident is not None:
        loaded = resident
    if loaded is not None:
        out["loaded"] = bool(loaded)
    out.pop("resident", None)
    state = out.get("state")
    if isinstance(state, str) and state.strip().lower() == "resident":
        out["state"] = "loaded"
    return out


def _capability_residency_load(task: str, payload: Dict[str, Any], http_request: Request) -> Dict[str, Any]:
    if task in _VISION_RESIDENCY_TASKS:
        from . import vision_endpoints as _vision_endpoints

        provider_key = _provider_api_key_from_request(http_request)
        try:
            runtime = _vision_endpoints.load_server_vision_loaded_model(payload, api_key=provider_key)
        except Exception as e:
            raise _capability_residency_error(e, task=task) from e
        runtime_out = _normalize_model_residency_loaded_fields(dict(runtime))
        return {
            "ok": True,
            "success": True,
            "loaded_new": _model_residency_loaded_new(runtime_out),
            "runtime": runtime_out,
            "affected_models": [runtime_out],
        }

    if task in {"tts", "stt"}:
        core = _capability_residency_core_for_request(task, http_request)
        target = core.voice if task == "tts" else core.audio
        method = getattr(target, "load_resident_model", None)
        if not callable(method):
            raise HTTPException(
                status_code=501,
                detail={"error": {"message": f"{task} capability does not expose load_resident_model.", "type": "not_implemented"}},
            )
        try:
            runtime = method(payload)
        except Exception as e:
            raise _capability_residency_error(e, task=task) from e
        runtime_out = _normalize_model_residency_loaded_fields(dict(runtime))
        return {
            "ok": True,
            "success": True,
            "loaded_new": _model_residency_loaded_new(runtime_out),
            "runtime": runtime_out,
            "affected_models": [runtime_out],
        }

    raise HTTPException(
        status_code=501,
        detail={"error": {"message": f"Model residency task {task!r} is not implemented yet.", "type": "not_implemented"}},
    )


def _capability_residency_loaded(task: Optional[str], filters: Dict[str, Any], http_request: Request) -> list[Dict[str, Any]]:
    records: list[Dict[str, Any]] = []
    if task is None:
        from . import vision_endpoints as _vision_endpoints

        records.extend(_vision_endpoints.list_server_vision_loaded_models(filters))
        tasks = ["tts", "stt"]
    else:
        tasks = [task]
    for item_task in tasks:
        if item_task in _VISION_RESIDENCY_TASKS:
            from . import vision_endpoints as _vision_endpoints

            records.extend(_vision_endpoints.list_server_vision_loaded_models(filters))
            continue
        if item_task in {"tts", "stt"}:
            try:
                core = _capability_residency_core_for_request(item_task, http_request)
                target = core.voice if item_task == "tts" else core.audio
                method = getattr(target, "list_loaded_models", None)
                if not callable(method):
                    method = getattr(target, "list_resident_models", None)
                if callable(method):
                    records.extend(
                        [
                            _normalize_model_residency_loaded_fields(dict(entry))
                            for entry in list(method(filters) or [])
                            if isinstance(entry, dict)
                        ]
                    )
            except Exception:
                continue
    return records


def _capability_residency_unload(task: str, payload: Dict[str, Any], http_request: Request) -> Dict[str, Any]:
    if task in _VISION_RESIDENCY_TASKS:
        from . import vision_endpoints as _vision_endpoints

        try:
            runtime = _vision_endpoints.unload_server_vision_loaded_model(payload)
        except Exception as e:
            raise _capability_residency_error(e, task=task) from e
        runtime_out = _normalize_model_residency_loaded_fields(dict(runtime))
        return {
            "ok": True,
            "success": True,
            "runtime": runtime_out,
            "unloaded": bool(runtime_out.get("unloaded", True)),
            "affected_models": [runtime_out],
        }

    if task in {"tts", "stt"}:
        core = _capability_residency_core_for_request(task, http_request)
        target = core.voice if task == "tts" else core.audio
        method = getattr(target, "unload_resident_model", None)
        if not callable(method):
            raise HTTPException(
                status_code=501,
                detail={"error": {"message": f"{task} capability does not expose unload_resident_model.", "type": "not_implemented"}},
            )
        try:
            runtime = method(payload)
        except Exception as e:
            raise _capability_residency_error(e, task=task) from e
        runtime_out = _normalize_model_residency_loaded_fields(dict(runtime))
        return {
            "ok": True,
            "success": True,
            "runtime": runtime_out,
            "unloaded": bool(runtime_out.get("unloaded", True)),
            "affected_models": [runtime_out],
        }

    raise HTTPException(
        status_code=501,
        detail={"error": {"message": f"Model residency task {task!r} is not implemented yet.", "type": "not_implemented"}},
    )


class LoadedModelRequest(BaseModel):
    task: Optional[str] = Field(
        default=None,
        description="Residency task. Omit for backward-compatible text generation runtimes.",
        example="text_generation",
    )
    provider: Optional[str] = Field(default=None, description="Provider/backend name to load.", example="mlx")
    model: Optional[str] = Field(default=None, description="Model identifier to keep warm.", example="mlx-community/Qwen3.6-27B-4bit")
    base_url: Optional[str] = Field(
        default=None,
        description="Optional provider base URL override. Useful for OpenAI-compatible local runtimes.",
        example=None,
    )
    timeout_s: Optional[float] = Field(
        default=None,
        description="Optional provider timeout in seconds for the loaded runtime.",
        example=7200.0,
    )
    options: Dict[str, Any] = Field(default_factory=dict, description="Task/backend-specific residency options.")
    pin: Optional[bool] = Field(default=True, description="Whether the model should be explicitly kept loaded when supported.")
    ttl_s: Optional[float] = Field(default=None, description="Optional future TTL hint in seconds. Currently advisory.")

    class Config:
        json_schema_extra = {
            "examples": [
                {
                    "summary": "Load Gateway Runtime",
                    "value": {
                        "provider": "mlx",
                        "model": "mlx-community/Qwen3.6-27B-4bit",
                        "base_url": None,
                        "timeout_s": 7200.0,
                    },
                },
                {
                    "summary": "Load Image Runtime",
                    "value": {
                        "task": "image_generation",
                        "provider": "diffusers",
                        "model": "runwayml/stable-diffusion-v1-5",
                        "pin": True,
                    },
                },
                {
                    "summary": "Load Cloned TTS Runtime",
                    "value": {
                        "task": "tts",
                        "provider": "cloned",
                        "model": "omnivoice",
                        "options": {"voice": "my_voice"},
                    },
                },
            ]
        }


class UnloadModelRequest(BaseModel):
    task: Optional[str] = Field(default=None, description="Residency task. Omit for backward-compatible text runtime unload when runtime_id/provider+model matches text.")
    runtime_id: Optional[str] = Field(default=None, description="Specific loaded runtime identifier returned by `/acore/models/load`.")
    provider: Optional[str] = Field(default=None, description="Provider name for runtime selection when `runtime_id` is omitted.", example="mlx")
    model: Optional[str] = Field(default=None, description="Model identifier for runtime selection when `runtime_id` is omitted.", example="mlx-community/Qwen3.6-27B-4bit")
    base_url: Optional[str] = Field(default=None, description="Optional provider base URL used when the runtime was loaded.", example=None)
    options: Dict[str, Any] = Field(default_factory=dict, description="Task/backend-specific unload selectors.")

    class Config:
        json_schema_extra = {
            "examples": [
                {
                    "summary": "Unload Gateway Runtime",
                    "value": {
                        "provider": "mlx",
                        "model": "mlx-community/Qwen3.6-27B-4bit",
                        "base_url": None,
                    },
                }
            ]
        }

# ============================================================================
# Endpoints
# ============================================================================

@app.get("/", include_in_schema=False)
async def server_root():
    """Small browser landing page for the server root."""
    return HTMLResponse(
        """<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <title>AbstractCore Server</title>
  <style>
    :root { color-scheme: dark; --bg:#0b1020; --panel:#121a30; --text:#e8edff; --muted:#9ca8cc; --line:#2b3860; --accent:#35d0ff; }
    * { box-sizing: border-box; }
    body { margin:0; min-height:100vh; display:grid; place-items:center; font:15px/1.55 ui-sans-serif,-apple-system,BlinkMacSystemFont,"Segoe UI",sans-serif; color:var(--text); background:var(--bg); }
    main { width:min(720px, calc(100vw - 32px)); border:1px solid var(--line); border-radius:18px; background:var(--panel); padding:28px; }
    h1 { margin:0 0 8px; font-size:2rem; }
    p { margin:0 0 18px; color:var(--muted); }
    nav { display:flex; flex-wrap:wrap; gap:10px; }
    a { display:inline-flex; align-items:center; border:1px solid var(--line); border-radius:999px; padding:10px 14px; color:var(--text); text-decoration:none; font-weight:700; }
    a:hover { border-color:var(--accent); color:var(--accent); }
  </style>
</head>
<body>
  <main>
    <h1>AbstractCore Server</h1>
    <p>OpenAI-compatible API gateway is running.</p>
    <nav>
      <a href="/docs">Swagger UI</a>
      <a href="/docs-lite">Docs Lite</a>
      <a href="/redoc">ReDoc</a>
      <a href="/openapi.json">OpenAPI JSON</a>
      <a href="/health">Health</a>
    </nav>
  </main>
</body>
</html>"""
    )


@app.get("/favicon.ico", include_in_schema=False)
async def favicon():
    """Return an empty favicon response to avoid browser startup noise."""
    return Response(status_code=204)


@app.get("/health", tags=["health"])
async def health_check():
    """Health check endpoint."""
    return {
        "status": "healthy",
        "version": __version__,
        "features": [
            "multi-agent-syntax-support",
            "auto-format-detection",
            "universal-tool-calls",
            "abstractcore-integration"
        ]
    }


@app.get(
    "/acore/models/loaded",
    tags=["runtime"],
    summary="List Loaded Gateway Runtimes",
    description="List task-aware provider/model runtimes currently kept warm inside the gateway process.",
)
def acore_models_loaded(
    http_request: Request,
    task: Optional[str] = Query(None, description="Optional task filter. Omit to include text, image, TTS, and STT residency records.", example="image_generation"),
    provider: Optional[str] = Query(None, description="Optional provider filter.", example="mlx"),
    model: Optional[str] = Query(None, description="Optional exact model filter.", example="mlx-community/Qwen3.6-27B-4bit"),
):
    task_filter = _normalize_model_residency_task(task) if task else None
    provider_filter = str(provider or "").strip().lower()
    model_filter = str(model or "").strip()
    runtimes: list[Dict[str, Any]] = []
    if task_filter in {None, "text_generation"}:
        with _GATEWAY_RUNTIME_LOCK:
            runtimes.extend(
                _gateway_text_residency_record(runtime)
                for runtime in _GATEWAY_LOADED_RUNTIMES.values()
                if (not provider_filter or runtime.provider == provider_filter)
                and (not model_filter or runtime.model == model_filter)
            )
    if task_filter != "text_generation":
        filters = {"provider": provider_filter, "model": model_filter}
        if task_filter:
            filters["task"] = task_filter
        runtimes.extend(_capability_residency_loaded(task_filter, filters, http_request))
    models = sorted(runtimes, key=lambda item: str(item.get("runtime_id") or item.get("load_id") or ""))
    return {
        "ok": True,
        "success": True,
        "operation": "list_loaded",
        "object": "list",
        "data": models,
        "models": models,
        "affected_models": models,
    }


@app.post(
    "/acore/models/load",
    tags=["runtime"],
    summary="Load Gateway Runtime",
    description=(
        "Load a task-specific provider/model into the gateway runtime registry so subsequent text, image, "
        "voice, and transcription calls can reuse warm in-process providers when supported."
    ),
)
def acore_models_load(req: LoadedModelRequest, http_request: Request):
    task = _normalize_model_residency_task(req.task)
    payload = _model_residency_payload(req)
    payload["task"] = task
    if task != "text_generation":
        return _capability_residency_load(task, payload, http_request)

    provider = str(req.provider or "").strip().lower()
    model = str(req.model or "").strip()
    if not provider or not model:
        raise HTTPException(
            status_code=400,
            detail={"error": {"message": "provider and model are required", "type": "invalid_request"}},
        )

    provider_kwargs: Dict[str, Any] = {}
    provider_api_key = _provider_api_key_from_request(http_request)
    if provider_api_key:
        provider_kwargs["api_key"] = provider_api_key
    _guard_unauthenticated_server_provider_key_use(
        provider,
        explicit_provider_key=bool(provider_api_key),
        http_request=http_request,
    )

    if isinstance(req.base_url, str) and req.base_url.strip():
        base_url = req.base_url.strip()
        if not _server_allows_request_base_url(base_url):
            raise HTTPException(
                status_code=403,
                detail={
                    "error": {
                        "message": (
                            "Request-level base_url overrides are restricted for security. "
                            "By default only loopback URLs are allowed. "
                            "To allow additional hosts/prefixes, set ABSTRACTCORE_SERVER_BASE_URL_ALLOWLIST."
                        ),
                        "type": "forbidden",
                    }
                },
            )

        parsed = urllib.parse.urlsplit(base_url)
        host = str(parsed.hostname or "").strip()
        if host and not _is_loopback_host(host):
            env_var = _provider_api_key_env_var(provider)
            if _server_has_provider_api_key(provider) and not provider_api_key:
                raise HTTPException(
                    status_code=403,
                    detail={
                        "error": {
                            "message": (
                                "Refusing request-level base_url override without an explicit provider key because "
                                f"the server has {env_var} set. Provide the provider key via "
                                "X-AbstractCore-Provider-API-Key."
                            ),
                            "type": "forbidden",
                        }
                    },
                )
        provider_kwargs["base_url"] = base_url

    if req.timeout_s is not None:
        provider_kwargs["timeout"] = req.timeout_s

    runtime, runtime_cache_loaded_new = _ensure_loaded_gateway_runtime(
        provider=provider,
        model=model,
        provider_kwargs=provider_kwargs,
        explicit_provider_key_hash=_provider_key_fingerprint(provider_api_key),
    )
    before_record = _gateway_text_residency_record(runtime)
    load_options = dict(req.options or {})
    load_options.setdefault("pin", req.pin)
    if req.ttl_s is not None:
        load_options.setdefault("ttl_s", req.ttl_s)
    provider_load_result: Any = None
    if before_record.get("loaded") is not True:
        try:
            provider_load_result = _gateway_load_provider_residency(runtime, load_options)
        except Exception as exc:
            if runtime_cache_loaded_new:
                _drop_loaded_gateway_runtime(runtime)
            raise HTTPException(
                status_code=502,
                detail={
                    "error": {
                        "message": f"Provider model load failed: {exc}",
                        "type": "provider_load_failed",
                    }
                },
            ) from exc
    runtime_record = _gateway_text_residency_record(runtime)
    provider_loaded_new = bool(before_record.get("loaded") is not True and runtime_record.get("loaded") is True)
    loaded_new = bool((runtime_cache_loaded_new or provider_loaded_new) and runtime_record.get("loaded") is True)
    if provider_load_result is not None:
        runtime_record["provider_load_result"] = provider_load_result
    return {
        "ok": True,
        "success": True,
        "loaded_new": loaded_new,
        "provider_loaded_new": provider_loaded_new,
        "runtime_cache_loaded_new": bool(runtime_cache_loaded_new),
        "runtime": runtime_record,
        "affected_models": [runtime_record],
        "prompt_cache_capabilities": _gateway_prompt_cache_capabilities_dict(runtime.llm),
    }


@app.post(
    "/acore/models/unload",
    tags=["runtime"],
    summary="Unload Gateway Runtime",
    description="Unload a previously loaded task-specific provider/model runtime from the gateway process.",
)
def acore_models_unload(req: UnloadModelRequest, http_request: Request):
    task = _normalize_model_residency_task(req.task) if req.task else None
    payload = _model_residency_payload(req)
    if isinstance(payload.get("options"), dict):
        for k, v in dict(payload["options"]).items():
            payload.setdefault(k, v)
    if task and task != "text_generation":
        payload["task"] = task
        return _capability_residency_unload(task, payload, http_request)

    runtime_id = str(req.runtime_id or "").strip() or None
    provider = str(req.provider or "").strip().lower() or None
    model = str(req.model or "").strip() or None
    if runtime_id is None and (provider is None or model is None):
        raise HTTPException(
            status_code=400,
            detail={
                "error": {
                    "message": "Provide runtime_id or provider+model to unload a gateway runtime.",
                    "type": "invalid_request",
                }
            },
        )
    runtime = _get_loaded_gateway_runtime(
        provider=provider,
        model=model,
        base_url=req.base_url,
        explicit_provider_key_hash=_provider_key_fingerprint(_provider_api_key_from_request(http_request)),
        runtime_id=runtime_id,
    )
    if runtime is not None:
        _unload_loaded_gateway_runtime(runtime)
        return {
            "ok": True,
            "success": True,
            "runtime": {
                **_gateway_runtime_to_dict(runtime),
                "task": "text_generation",
                "state": "unloaded",
                "loaded": False,
                "pinned": False,
                "isolation": "in_process",
                "health": "ok",
                "error": None,
            },
            "unloaded": True,
            "affected_models": [
                {
                    **_gateway_runtime_to_dict(runtime),
                    "task": "text_generation",
                    "state": "unloaded",
                    "loaded": False,
                    "pinned": False,
                    "isolation": "in_process",
                    "health": "ok",
                    "error": None,
                }
            ],
        }
    if task is None:
        for candidate_task in ("image_generation", "video_generation", "tts", "stt"):
            try:
                payload["task"] = candidate_task
                return _capability_residency_unload(candidate_task, payload, http_request)
            except HTTPException as exc:
                if exc.status_code not in {404, 501}:
                    raise
                continue
    raise HTTPException(
        status_code=404,
        detail={"error": {"message": "Loaded model runtime not found.", "type": "not_found"}},
    )


class PromptCacheProxyBase(BaseModel):
    """Proxy configuration for forwarding AbstractCore prompt-cache control-plane calls."""

    runtime_id: Optional[str] = Field(
        default=None,
        description=(
            "Specific loaded gateway runtime identifier returned by `/acore/models/load`. "
            "When supplied, the prompt-cache or bloc control-plane call targets that local runtime."
        ),
    )
    base_url: Optional[str] = Field(
        default=None,
        description=(
            "Upstream base URL for an AbstractEndpoint instance. Can include an OpenAI-style `/v1` suffix "
            "(it will be stripped when proxying `/acore/prompt_cache/*`). "
            "When `runtime_id` or `provider` + `model` is supplied, `base_url` is used to disambiguate the local runtime key."
        ),
        example="http://localhost:8001/v1",
    )
    provider: Optional[str] = Field(
        default=None,
        description=(
            "Gateway-local provider name for a previously loaded runtime. "
            "Use together with `model` when `base_url` is omitted."
        ),
        example="mlx",
    )
    model: Optional[str] = Field(
        default=None,
        description=(
            "Gateway-local model identifier for a previously loaded runtime. "
            "Use together with `provider` when `base_url` is omitted."
        ),
        example="mlx-community/Qwen3.6-27B-4bit",
    )
    api_key: Optional[str] = Field(
        default=None,
        description="Deprecated/disabled for HTTP request bodies. Use X-AbstractCore-Provider-API-Key for provider overrides.",
        example=None,
    )


class PromptCacheSetProxyRequest(PromptCacheProxyBase):
    key: str = Field(..., description="Prompt-cache key to select or create on the upstream AbstractEndpoint.", example="project-default")
    make_default: bool = Field(default=True, description="Whether the upstream endpoint should make this key the default cache key.")
    ttl_s: Optional[float] = Field(default=None, description="Optional upstream cache TTL in seconds.", example=3600)


class PromptCacheUpdateProxyRequest(PromptCacheProxyBase):
    key: str = Field(..., description="Prompt-cache key to update on the upstream AbstractEndpoint.", example="project-default")
    prompt: Optional[str] = Field(default=None, description="Plain prompt text to prepare/cache upstream.")
    messages: Optional[List[Dict[str, Any]]] = Field(default=None, description="Chat messages to prepare/cache upstream.")
    system_prompt: Optional[str] = Field(default=None, description="Optional system prompt to include in the prepared cache state.")
    tools: Optional[List[Dict[str, Any]]] = Field(default=None, description="Optional tool schemas to include in the prepared cache state.")
    thinking: Optional[Union[bool, str]] = Field(
        default=None,
        description="Optional unified thinking/reasoning control to apply while preparing the cache state.",
        example="off",
    )
    add_generation_prompt: bool = Field(default=False, description="Whether to ask the upstream endpoint to add a generation prompt marker.")
    ttl_s: Optional[float] = Field(default=None, description="Optional upstream cache TTL in seconds.", example=3600)


class PromptCacheForkProxyRequest(PromptCacheProxyBase):
    from_key: str = Field(..., description="Existing upstream cache key to fork from.", example="project-default")
    to_key: str = Field(..., description="New upstream cache key to create.", example="project-branch")
    make_default: bool = Field(default=False, description="Whether the forked cache key should become the upstream default.")
    ttl_s: Optional[float] = Field(default=None, description="Optional upstream cache TTL in seconds.", example=3600)


class PromptCacheClearProxyRequest(PromptCacheProxyBase):
    key: Optional[str] = Field(default=None, description="Specific upstream cache key to clear. Omit to clear upstream default/all cache state, depending on backend support.")


class PromptCachePrepareModulesProxyRequest(PromptCacheProxyBase):
    namespace: str = Field(..., description="Namespace for the prepared module cache.", example="abstractcore.tools")
    modules: List[Dict[str, Any]] = Field(..., description="Module descriptors/payloads to prepare on the upstream endpoint.")
    make_default: bool = Field(default=False, description="Whether the prepared module cache should become the upstream default.")
    ttl_s: Optional[float] = Field(default=None, description="Optional upstream cache TTL in seconds.", example=3600)
    version: int = Field(default=1, description="Payload version for forward-compatible module preparation.")


class BlocProxyBase(PromptCacheProxyBase):
    """Proxy configuration for forwarding bloc control-plane calls to an AbstractEndpoint."""


class BlocUpsertTextProxyRequest(BlocProxyBase):
    path: str = Field(..., description="Logical source path for the extracted text snapshot.", example="/tmp/orbit.txt")
    content: str = Field(..., description="Extracted text content to persist in the upstream bloc store.")
    sha256: Optional[str] = Field(default=None, description="Optional caller-supplied bloc sha256. Defaults to a hash of `content` when omitted.")
    content_sha256: Optional[str] = Field(default=None, description="Optional content hash override. Defaults to a hash of `content` when omitted.")
    media_type: str = Field(default="text", description="Logical media type for the stored bloc.", example="text")
    size_bytes: Optional[int] = Field(default=None, description="Optional original source size in bytes.")
    mtime_ns: Optional[int] = Field(default=None, description="Optional source modification time in nanoseconds.")
    format: Optional[str] = Field(default=None, description="Optional source format identifier.", example="text/plain")
    estimated_tokens: Optional[int] = Field(default=None, description="Optional estimated token count for the stored content.")
    relpath_base: Optional[str] = Field(default=None, description="Optional base path used to derive a stable relative path in the record.")
    summary: Optional[str] = Field(default=None, description="Optional explicit summary override.")
    keywords: Optional[List[str]] = Field(default=None, description="Optional explicit keywords override.")

    class Config:
        json_schema_extra = {
            "examples": [
                {
                    "summary": "Persist Text Bloc",
                    "value": {
                        "provider": "mlx",
                        "model": "mlx-community/Qwen3.6-27B-4bit",
                        "path": "/tmp/orbit.txt",
                        "content": "Document title: Orbit Notes\n\nThe launch window is Tuesday at 14:30 UTC.\n",
                        "media_type": "text",
                        "format": "text/plain",
                    },
                }
            ]
        }


class BlocKVEnsureProxyRequest(BlocProxyBase):
    sha256: Optional[str] = Field(default=None, description="Bloc sha256 selector.")
    bloc_id: Optional[int] = Field(default=None, description="Stable integer bloc selector.")
    artifact_path: Optional[str] = Field(default=None, description="Optional explicit artifact path override.")
    force_rebuild: bool = Field(default=False, description="Force artifact rebuild even if a valid one already exists.")
    debug: bool = Field(default=False, description="Return verbose bloc KV verification metadata.")

    class Config:
        json_schema_extra = {
            "examples": [
                {
                    "summary": "Ensure Bloc KV Artifact",
                    "value": {
                        "provider": "mlx",
                        "model": "mlx-community/Qwen3.6-27B-4bit",
                        "sha256": "abababababababababababababababababababababababababababababababab",
                        "force_rebuild": False,
                    },
                }
            ]
        }


class BlocKVLoadProxyRequest(BlocProxyBase):
    sha256: Optional[str] = Field(default=None, description="Bloc sha256 selector.")
    bloc_id: Optional[int] = Field(default=None, description="Stable integer bloc selector.")
    artifact_path: Optional[str] = Field(default=None, description="Optional explicit artifact path override.")
    stable_cache_key: Optional[str] = Field(default=None, description="Optional long-lived cache key to keep loaded upstream.")
    key: Optional[str] = Field(default=None, description="Target upstream cache key to load or fork into.")
    make_default: bool = Field(default=False, description="Whether the loaded or forked key should become the upstream default.")
    force_rebuild: bool = Field(default=False, description="Force artifact rebuild before loading.")
    debug: bool = Field(default=False, description="Return verbose bloc KV verification metadata.")

    class Config:
        json_schema_extra = {
            "examples": [
                {
                    "summary": "Load Bloc KV Artifact",
                    "value": {
                        "provider": "mlx",
                        "model": "mlx-community/Qwen3.6-27B-4bit",
                        "sha256": "abababababababababababababababababababababababababababababababab",
                        "stable_cache_key": "stable:orbit",
                        "key": "work:orbit",
                        "make_default": False,
                    },
                }
            ]
        }


class BlocDeleteProxyRequest(BlocProxyBase):
    sha256: Optional[str] = Field(default=None, description="Bloc sha256 selector.")
    bloc_id: Optional[int] = Field(default=None, description="Stable integer bloc selector.")
    delete_kv: bool = Field(default=True, description="Also delete KV artifacts under the bloc directory.")
    clear_loaded: bool = Field(default=False, description="Clear live prompt-cache keys bound to this bloc before deleting.")
    force: bool = Field(default=False, description="Delete even if live keys are detected. Prefer clear_loaded for normal use.")
    dry_run: bool = Field(default=False, description="Report what would be deleted without deleting files.")

    class Config:
        json_schema_extra = {
            "examples": [
                {
                    "summary": "Delete Bloc Safely",
                    "value": {
                        "provider": "mlx",
                        "model": "mlx-community/Qwen3.6-27B-4bit",
                        "sha256": "abababababababababababababababababababababababababababababababab",
                        "delete_kv": True,
                        "clear_loaded": True,
                    },
                }
            ]
        }


class BlocKVDeleteProxyRequest(BlocProxyBase):
    sha256: Optional[str] = Field(default=None, description="Bloc sha256 selector.")
    bloc_id: Optional[int] = Field(default=None, description="Stable integer bloc selector.")
    artifact_path: Optional[str] = Field(default=None, description="Optional explicit artifact path selector.")
    clear_loaded: bool = Field(default=False, description="Clear live prompt-cache keys bound to this artifact before deleting.")
    force: bool = Field(default=False, description="Delete even if live keys are detected. Prefer clear_loaded for normal use.")
    dry_run: bool = Field(default=False, description="Report what would be deleted without deleting files.")
    debug: bool = Field(default=False, description="Return verbose safety metadata.")

    class Config:
        json_schema_extra = {
            "examples": [
                {
                    "summary": "Delete Bloc KV Artifact Safely",
                    "value": {
                        "provider": "mlx",
                        "model": "mlx-community/Qwen3.6-27B-4bit",
                        "sha256": "abababababababababababababababababababababababababababababababab",
                        "clear_loaded": True,
                        "debug": True,
                    },
                }
            ]
        }


class BlocKVPruneProxyRequest(BlocProxyBase):
    sha256: Optional[str] = Field(default=None, description="Optional bloc sha256 filter.")
    bloc_id: Optional[int] = Field(default=None, description="Optional stable integer bloc filter.")
    clear_loaded: bool = Field(default=False, description="Clear live prompt-cache keys bound to matching artifacts before deleting.")
    force: bool = Field(default=False, description="Delete even if live keys are detected. Prefer clear_loaded for normal use.")
    dry_run: bool = Field(default=False, description="Report matching artifacts without deleting files.")
    debug: bool = Field(default=False, description="Return verbose safety metadata.")

    class Config:
        json_schema_extra = {
            "examples": [
                {
                    "summary": "Dry-run Prune Bloc KV Artifacts",
                    "value": {
                        "provider": "mlx",
                        "model": "mlx-community/Qwen3.6-27B-4bit",
                        "sha256": "abababababababababababababababababababababababababababababababab",
                        "dry_run": True,
                        "debug": True,
                    },
                }
            ]
        }


def _normalize_control_plane_base_url(base_url: str) -> str:
    u = str(base_url or "").strip().rstrip("/")
    if u.endswith("/v1"):
        u = u[:-3]
    return u.rstrip("/")


def _proxy_prompt_cache_request(
    *,
    http_request: Optional[Request] = None,
    base_url: Optional[str],
    api_key: Optional[str],
    method: str,
    path: str,
    json_body: Optional[Dict[str, Any]] = None,
    timeout_s: float = 30.0,
) -> Dict[str, Any]:
    if not isinstance(base_url, str) or not base_url.strip():
        return {
            "supported": False,
            "error": "base_url is required to proxy prompt cache control plane calls (use AbstractEndpoint)",
        }
    _reject_body_api_key(api_key)

    base_url_s = base_url.strip()
    if not _server_allows_request_base_url(base_url_s):
        return {
            "supported": False,
            "error": (
                "Request-level base_url overrides are restricted for security. "
                "By default only loopback URLs are allowed. "
                "To allow additional hosts/prefixes, set ABSTRACTCORE_SERVER_BASE_URL_ALLOWLIST."
            ),
        }

    upstream_root = _normalize_control_plane_base_url(base_url_s)
    url = f"{upstream_root}{path}"

    headers: Dict[str, str] = {}
    upstream_api_key = _provider_api_key_from_request(http_request) if http_request is not None else None
    if upstream_api_key:
        headers["Authorization"] = f"Bearer {upstream_api_key}"

    try:
        with httpx.Client(timeout=timeout_s) as client:
            if method.upper() == "GET":
                resp = client.get(url, headers=headers)
            else:
                resp = client.post(url, headers=headers, json=json_body or {})
    except Exception as e:
        return {"supported": False, "error": str(e)}

    try:
        payload = resp.json()
    except Exception:
        payload = {"error": resp.text}

    if resp.status_code >= 400:
        return {
            "supported": False,
            "status_code": int(resp.status_code),
            "error": payload,
            "upstream": url,
        }

    if isinstance(payload, dict):
        return payload
    return {"supported": True, "data": payload}


def _proxy_endpoint_control_request(
    *,
    http_request: Optional[Request] = None,
    base_url: Optional[str],
    api_key: Optional[str],
    method: str,
    path: str,
    json_body: Optional[Dict[str, Any]] = None,
    query_params: Optional[Dict[str, Any]] = None,
    timeout_s: float = 30.0,
) -> Any:
    if not isinstance(base_url, str) or not base_url.strip():
        return JSONResponse(
            status_code=400,
            content={"ok": False, "error": "base_url is required to proxy AbstractEndpoint bloc control-plane calls"},
        )
    _reject_body_api_key(api_key)

    base_url_s = base_url.strip()
    if not _server_allows_request_base_url(base_url_s):
        return JSONResponse(
            status_code=403,
            content={
                "ok": False,
                "error": (
                    "Request-level base_url overrides are restricted for security. "
                    "By default only loopback URLs are allowed. "
                    "To allow additional hosts/prefixes, set ABSTRACTCORE_SERVER_BASE_URL_ALLOWLIST."
                ),
            },
        )

    upstream_root = _normalize_control_plane_base_url(base_url_s)
    url = f"{upstream_root}{path}"

    headers: Dict[str, str] = {}
    upstream_api_key = _provider_api_key_from_request(http_request) if http_request is not None else None
    if upstream_api_key:
        headers["Authorization"] = f"Bearer {upstream_api_key}"

    params = {k: v for k, v in (query_params or {}).items() if v is not None}

    try:
        with httpx.Client(timeout=timeout_s) as client:
            if method.upper() == "GET":
                resp = client.get(url, headers=headers, params=params)
            else:
                resp = client.post(url, headers=headers, params=params, json=json_body or {})
    except Exception as e:
        return JSONResponse(status_code=502, content={"ok": False, "error": str(e)})

    try:
        payload = resp.json()
    except Exception:
        payload = {"error": resp.text}

    if resp.status_code >= 400:
        if isinstance(payload, dict):
            content = dict(payload)
            content.setdefault("upstream", url)
        else:
            content = {"ok": False, "error": payload, "upstream": url}
        return JSONResponse(status_code=int(resp.status_code), content=content)

    if isinstance(payload, dict):
        return payload
    return {"ok": True, "data": payload}


def _local_bloc_error(operation: str, error: Union[Exception, str], *, status_code: int) -> JSONResponse:
    return JSONResponse(
        status_code=int(status_code),
        content={
            "ok": False,
            "operation": operation,
            "error": str(error),
        },
    )


def _resolve_local_bloc_selector(*, sha256: Optional[str], bloc_id: Optional[int]):
    if isinstance(sha256, str) and sha256.strip():
        rec = _SERVER_BLOC_STORE.get(sha256.strip().lower())
        if rec is not None:
            return rec
    if bloc_id is not None:
        try:
            rec = _SERVER_BLOC_STORE.get_by_bloc_id(int(bloc_id))
        except Exception:
            rec = None
        if rec is not None:
            return rec
    return None


def _optional_local_runtime_for_bloc_control(
    *,
    http_request: Optional[Request],
    runtime_id: Optional[str],
    provider: Optional[str],
    model: Optional[str],
    base_url: Optional[str],
) -> Optional[_GatewayLoadedRuntime]:
    explicit_provider_key_hash = _provider_key_fingerprint(_provider_api_key_from_request(http_request)) if http_request is not None else None
    return _get_loaded_gateway_runtime(
        provider=str(provider or "").strip().lower() or None,
        model=str(model or "").strip() or None,
        base_url=base_url,
        explicit_provider_key_hash=explicit_provider_key_hash,
        runtime_id=runtime_id,
    )


def _live_bloc_bindings_across_loaded_runtimes(record_sha256: Optional[str] = None) -> List[Dict[str, Any]]:
    live: List[Dict[str, Any]] = []
    with _GATEWAY_RUNTIME_LOCK:
        runtimes = list(_GATEWAY_LOADED_RUNTIMES.values())
    for runtime in runtimes:
        try:
            bindings = _run_loaded_gateway_runtime(
                runtime,
                lambda runtime=runtime: find_bloc_kv_live_bindings(
                    provider=runtime.llm,
                    sha256=record_sha256,
                ),
            )
        except Exception:
            bindings = []
        for binding in bindings:
            item = dict(binding)
            item.setdefault("runtime_id", runtime.runtime_id)
            item.setdefault("provider", runtime.provider)
            item.setdefault("model", runtime.model)
            live.append(item)
    return live


def _clear_live_bloc_bindings_across_loaded_runtimes(record_sha256: Optional[str] = None) -> List[str]:
    cleared: List[str] = []
    with _GATEWAY_RUNTIME_LOCK:
        runtimes = list(_GATEWAY_LOADED_RUNTIMES.values())
    for runtime in runtimes:
        try:
            keys = _run_loaded_gateway_runtime(
                runtime,
                lambda runtime=runtime: [
                    str(binding.get("key"))
                    for binding in find_bloc_kv_live_bindings(provider=runtime.llm, sha256=record_sha256)
                    if isinstance(binding.get("key"), str) and str(binding.get("key")).strip()
                ],
            )
            for key in keys:
                _run_loaded_gateway_runtime(runtime, lambda runtime=runtime, key=key: runtime.llm.prompt_cache_clear(key))
                cleared.append(key)
        except Exception:
            continue
    return cleared


@app.get(
    "/acore/prompt_cache/stats",
    tags=["prompt-cache"],
    summary="Prompt Cache Stats",
    description="Inspect prompt-cache stats on either a loaded gateway runtime (`provider` + `model`) or an upstream AbstractEndpoint (`base_url`).",
)
def acore_prompt_cache_stats(
    http_request: Request,
    runtime_id: Optional[str] = Query(None, description="Specific loaded gateway runtime identifier."),
    base_url: Optional[str] = Query(None, description="Upstream AbstractEndpoint base_url (optionally including /v1)"),
    provider: Optional[str] = Query(None, description="Gateway-local provider name for a loaded runtime."),
    model: Optional[str] = Query(None, description="Gateway-local model identifier for a loaded runtime."),
    api_key: Optional[str] = Query(None, description="Deprecated/disabled. Use X-AbstractCore-Provider-API-Key for provider overrides."),
):
    _reject_query_api_key(api_key)
    local_requested = bool((runtime_id and runtime_id.strip()) or (str(provider or "").strip() and str(model or "").strip()))
    if local_requested:
        provider_s = str(provider or "").strip().lower()
        model_s = str(model or "").strip()
        runtime = _resolve_local_runtime_for_control(
            provider=provider_s,
            model=model_s,
            base_url=base_url,
            runtime_id=runtime_id,
            http_request=http_request,
        )
        llm = runtime.llm
        caps = _gateway_prompt_cache_capabilities_dict(llm)
        if not caps.get("supports_stats") or not hasattr(llm, "get_prompt_cache_stats"):
            return {
                "supported": False,
                "operation": "stats",
                "code": "prompt_cache_unsupported",
                "error": "provider does not expose prompt cache stats",
                "capabilities": caps,
            }
        try:
            with runtime.lock:
                _touch_loaded_gateway_runtime(runtime)
                return {
                    "supported": True,
                    "operation": "stats",
                    "capabilities": caps,
                    "stats": llm.get_prompt_cache_stats(),
                }
        except Exception as e:
            return _gateway_prompt_cache_error_payload(llm, e, operation="stats")
    if not isinstance(base_url, str) or not base_url.strip():
        return {
            "supported": False,
            "error": "base_url or provider+model is required for prompt cache control plane calls",
        }
    return _proxy_prompt_cache_request(
        http_request=http_request,
        base_url=base_url,
        api_key=api_key,
        method="GET",
        path="/acore/prompt_cache/stats",
        json_body=None,
    )


@app.get(
    "/acore/prompt_cache/capabilities",
    tags=["prompt-cache"],
    summary="Prompt Cache Capabilities",
    description="Inspect prompt-cache capabilities on either a loaded gateway runtime (`provider` + `model`) or an upstream AbstractEndpoint (`base_url`).",
)
def acore_prompt_cache_capabilities(
    http_request: Request,
    runtime_id: Optional[str] = Query(None, description="Specific loaded gateway runtime identifier."),
    base_url: Optional[str] = Query(None, description="Upstream AbstractEndpoint base_url (optionally including /v1)"),
    provider: Optional[str] = Query(None, description="Gateway-local provider name for a loaded runtime."),
    model: Optional[str] = Query(None, description="Gateway-local model identifier for a loaded runtime."),
    api_key: Optional[str] = Query(None, description="Deprecated/disabled. Use X-AbstractCore-Provider-API-Key for provider overrides."),
):
    _reject_query_api_key(api_key)
    local_requested = bool((runtime_id and runtime_id.strip()) or (str(provider or "").strip() and str(model or "").strip()))
    if local_requested:
        provider_s = str(provider or "").strip().lower()
        model_s = str(model or "").strip()
        runtime = _resolve_local_runtime_for_control(
            provider=provider_s,
            model=model_s,
            base_url=base_url,
            runtime_id=runtime_id,
            http_request=http_request,
        )
        caps = _gateway_prompt_cache_capabilities_dict(runtime.llm)
        return {
            "supported": bool(caps.get("supported")),
            "operation": "capabilities",
            "capabilities": caps,
        }
    if not isinstance(base_url, str) or not base_url.strip():
        return {
            "supported": False,
            "error": "base_url or provider+model is required for prompt cache control plane calls",
        }
    return _proxy_prompt_cache_request(
        http_request=http_request,
        base_url=base_url,
        api_key=api_key,
        method="GET",
        path="/acore/prompt_cache/capabilities",
        json_body=None,
    )


@app.post(
    "/acore/prompt_cache/set",
    tags=["prompt-cache"],
    summary="Set Prompt Cache Key",
    description="Select or create a prompt-cache key on a loaded gateway runtime (`provider` + `model`) or proxy the same operation to an upstream AbstractEndpoint (`base_url`).",
)
def acore_prompt_cache_set(req: PromptCacheSetProxyRequest, http_request: Request):
    _reject_body_api_key(req.api_key)
    local_requested = bool((req.runtime_id and req.runtime_id.strip()) or (str(req.provider or "").strip() and str(req.model or "").strip()))
    if local_requested:
        provider_s = str(req.provider or "").strip().lower()
        model_s = str(req.model or "").strip()
        runtime = _resolve_local_runtime_for_control(provider=provider_s, model=model_s, base_url=req.base_url, runtime_id=req.runtime_id, http_request=http_request)
        llm = runtime.llm
        caps = _gateway_prompt_cache_capabilities_dict(llm)
        if not caps.get("supports_set") or not hasattr(llm, "prompt_cache_set"):
            return {
                "supported": False,
                "operation": "set",
                "code": "prompt_cache_unsupported",
                "error": "provider does not support prompt cache control plane",
                "capabilities": caps,
            }
        try:
            ok = _run_loaded_gateway_runtime(
                runtime,
                lambda: llm.prompt_cache_set(req.key, make_default=req.make_default, ttl_s=req.ttl_s),
            )
            return {"supported": True, "operation": "set", "ok": bool(ok), "capabilities": caps}
        except Exception as e:
            return _gateway_prompt_cache_error_payload(llm, e, operation="set")
    if not isinstance(req.base_url, str) or not req.base_url.strip():
        return {
            "supported": False,
            "error": "base_url or provider+model is required for prompt cache control plane calls",
        }
    body = req.model_dump(exclude_none=True)
    body.pop("runtime_id", None)
    base_url = body.pop("base_url", None)
    body.pop("provider", None)
    body.pop("model", None)
    api_key = body.pop("api_key", None)
    return _proxy_prompt_cache_request(
        http_request=http_request,
        base_url=base_url,
        api_key=api_key,
        method="POST",
        path="/acore/prompt_cache/set",
        json_body=body,
    )


@app.post(
    "/acore/prompt_cache/update",
    tags=["prompt-cache"],
    summary="Update Prompt Cache",
    description="Prepare prompt/messages/tools into a loaded gateway runtime (`provider` + `model`) or proxy the same operation to an upstream AbstractEndpoint (`base_url`).",
)
def acore_prompt_cache_update(req: PromptCacheUpdateProxyRequest, http_request: Request):
    _reject_body_api_key(req.api_key)
    local_requested = bool((req.runtime_id and req.runtime_id.strip()) or (str(req.provider or "").strip() and str(req.model or "").strip()))
    if local_requested:
        provider_s = str(req.provider or "").strip().lower()
        model_s = str(req.model or "").strip()
        runtime = _resolve_local_runtime_for_control(provider=provider_s, model=model_s, base_url=req.base_url, runtime_id=req.runtime_id, http_request=http_request)
        llm = runtime.llm
        caps = _gateway_prompt_cache_capabilities_dict(llm)
        if not caps.get("supports_update") or not hasattr(llm, "prompt_cache_update"):
            return {
                "supported": False,
                "operation": "update",
                "code": "prompt_cache_unsupported",
                "error": "provider does not support prompt cache control plane",
                "capabilities": caps,
            }
        try:
            ok = _run_loaded_gateway_runtime(
                runtime,
                lambda: llm.prompt_cache_update(
                    req.key,
                    prompt=req.prompt or "",
                    messages=req.messages,
                    system_prompt=req.system_prompt,
                    tools=req.tools,
                    thinking=req.thinking,
                    add_generation_prompt=bool(req.add_generation_prompt),
                    ttl_s=req.ttl_s,
                ),
            )
            return {"supported": True, "operation": "update", "ok": bool(ok), "capabilities": caps}
        except Exception as e:
            return _gateway_prompt_cache_error_payload(llm, e, operation="update")
    if not isinstance(req.base_url, str) or not req.base_url.strip():
        return {
            "supported": False,
            "error": "base_url or provider+model is required for prompt cache control plane calls",
        }
    body = req.model_dump(exclude_none=True)
    body.pop("runtime_id", None)
    base_url = body.pop("base_url", None)
    body.pop("provider", None)
    body.pop("model", None)
    api_key = body.pop("api_key", None)
    return _proxy_prompt_cache_request(
        http_request=http_request,
        base_url=base_url,
        api_key=api_key,
        method="POST",
        path="/acore/prompt_cache/update",
        json_body=body,
    )


@app.post(
    "/acore/prompt_cache/fork",
    tags=["prompt-cache"],
    summary="Fork Prompt Cache",
    description="Fork one prompt-cache key into another on a loaded gateway runtime (`provider` + `model`) or an upstream AbstractEndpoint (`base_url`).",
)
def acore_prompt_cache_fork(req: PromptCacheForkProxyRequest, http_request: Request):
    _reject_body_api_key(req.api_key)
    local_requested = bool((req.runtime_id and req.runtime_id.strip()) or (str(req.provider or "").strip() and str(req.model or "").strip()))
    if local_requested:
        provider_s = str(req.provider or "").strip().lower()
        model_s = str(req.model or "").strip()
        runtime = _resolve_local_runtime_for_control(provider=provider_s, model=model_s, base_url=req.base_url, runtime_id=req.runtime_id, http_request=http_request)
        llm = runtime.llm
        caps = _gateway_prompt_cache_capabilities_dict(llm)
        if not caps.get("supports_fork") or not hasattr(llm, "prompt_cache_fork"):
            return {
                "supported": False,
                "operation": "fork",
                "code": "prompt_cache_unsupported",
                "error": "provider does not support prompt cache control plane",
                "capabilities": caps,
            }
        try:
            ok = _run_loaded_gateway_runtime(
                runtime,
                lambda: llm.prompt_cache_fork(
                    req.from_key,
                    req.to_key,
                    make_default=bool(req.make_default),
                    ttl_s=req.ttl_s,
                ),
            )
            return {"supported": True, "operation": "fork", "ok": bool(ok), "capabilities": caps}
        except Exception as e:
            return _gateway_prompt_cache_error_payload(llm, e, operation="fork")
    if not isinstance(req.base_url, str) or not req.base_url.strip():
        return {
            "supported": False,
            "error": "base_url or provider+model is required for prompt cache control plane calls",
        }
    body = req.model_dump(exclude_none=True)
    body.pop("runtime_id", None)
    base_url = body.pop("base_url", None)
    body.pop("provider", None)
    body.pop("model", None)
    api_key = body.pop("api_key", None)
    return _proxy_prompt_cache_request(
        http_request=http_request,
        base_url=base_url,
        api_key=api_key,
        method="POST",
        path="/acore/prompt_cache/fork",
        json_body=body,
    )


@app.post(
    "/acore/prompt_cache/clear",
    tags=["prompt-cache"],
    summary="Clear Prompt Cache",
    description="Clear prompt-cache state on a loaded gateway runtime (`provider` + `model`) or an upstream AbstractEndpoint (`base_url`).",
)
def acore_prompt_cache_clear(req: PromptCacheClearProxyRequest, http_request: Request):
    _reject_body_api_key(req.api_key)
    local_requested = bool((req.runtime_id and req.runtime_id.strip()) or (str(req.provider or "").strip() and str(req.model or "").strip()))
    if local_requested:
        provider_s = str(req.provider or "").strip().lower()
        model_s = str(req.model or "").strip()
        runtime = _resolve_local_runtime_for_control(provider=provider_s, model=model_s, base_url=req.base_url, runtime_id=req.runtime_id, http_request=http_request)
        llm = runtime.llm
        caps = _gateway_prompt_cache_capabilities_dict(llm)
        if not caps.get("supports_clear") or not hasattr(llm, "prompt_cache_clear"):
            return {
                "supported": False,
                "operation": "clear",
                "code": "prompt_cache_unsupported",
                "error": "provider does not support prompt cache control plane",
                "capabilities": caps,
            }
        try:
            ok = _run_loaded_gateway_runtime(runtime, lambda: llm.prompt_cache_clear(req.key))
            return {"supported": True, "operation": "clear", "ok": bool(ok), "capabilities": caps}
        except Exception as e:
            return _gateway_prompt_cache_error_payload(llm, e, operation="clear")
    if not isinstance(req.base_url, str) or not req.base_url.strip():
        return {
            "supported": False,
            "error": "base_url or provider+model is required for prompt cache control plane calls",
        }
    body = req.model_dump(exclude_none=True)
    body.pop("runtime_id", None)
    base_url = body.pop("base_url", None)
    body.pop("provider", None)
    body.pop("model", None)
    api_key = body.pop("api_key", None)
    return _proxy_prompt_cache_request(
        http_request=http_request,
        base_url=base_url,
        api_key=api_key,
        method="POST",
        path="/acore/prompt_cache/clear",
        json_body=body,
    )


@app.post(
    "/acore/prompt_cache/prepare_modules",
    tags=["prompt-cache"],
    summary="Prepare Prompt Cache Modules",
    description="Prepare module/tool context on a loaded gateway runtime (`provider` + `model`) or an upstream AbstractEndpoint (`base_url`).",
)
def acore_prompt_cache_prepare_modules(req: PromptCachePrepareModulesProxyRequest, http_request: Request):
    _reject_body_api_key(req.api_key)
    local_requested = bool((req.runtime_id and req.runtime_id.strip()) or (str(req.provider or "").strip() and str(req.model or "").strip()))
    if local_requested:
        provider_s = str(req.provider or "").strip().lower()
        model_s = str(req.model or "").strip()
        runtime = _resolve_local_runtime_for_control(provider=provider_s, model=model_s, base_url=req.base_url, runtime_id=req.runtime_id, http_request=http_request)
        llm = runtime.llm
        caps = _gateway_prompt_cache_capabilities_dict(llm)
        if not caps.get("supports_prepare_modules") or not hasattr(llm, "prompt_cache_prepare_modules"):
            return {
                "supported": False,
                "operation": "prepare_modules",
                "code": "prompt_cache_unsupported",
                "error": "provider does not support prompt cache module preparation",
                "capabilities": caps,
            }
        try:
            return _run_loaded_gateway_runtime(
                runtime,
                lambda: llm.prompt_cache_prepare_modules(
                    namespace=req.namespace,
                    modules=req.modules,
                    make_default=bool(req.make_default),
                    ttl_s=req.ttl_s,
                    version=int(req.version),
                ),
            )
        except Exception as e:
            return _gateway_prompt_cache_error_payload(llm, e, operation="prepare_modules")
    if not isinstance(req.base_url, str) or not req.base_url.strip():
        return {
            "supported": False,
            "error": "base_url or provider+model is required for prompt cache control plane calls",
        }
    body = req.model_dump(exclude_none=True)
    body.pop("runtime_id", None)
    base_url = body.pop("base_url", None)
    body.pop("provider", None)
    body.pop("model", None)
    api_key = body.pop("api_key", None)
    return _proxy_prompt_cache_request(
        http_request=http_request,
        base_url=base_url,
        api_key=api_key,
        method="POST",
        path="/acore/prompt_cache/prepare_modules",
        json_body=body,
    )


@app.post(
    "/acore/blocs/upsert_text",
    tags=["memory-blocs"],
    summary="Persist Text Bloc",
    description="Persist extracted text into the gateway-local bloc store, or proxy the same operation to an upstream AbstractEndpoint when `base_url` is provided.",
)
def acore_blocs_upsert_text(req: BlocUpsertTextProxyRequest, http_request: Request):
    _reject_body_api_key(req.api_key)
    if not isinstance(req.base_url, str) or not req.base_url.strip():
        try:
            content = str(req.content or "")
            if not content:
                return _local_bloc_error("upsert_text", "content is required", status_code=400)
            path = str(req.path or "").strip()
            if not path:
                return _local_bloc_error("upsert_text", "path is required", status_code=400)
            content_sha256 = (
                str(req.content_sha256).strip().lower()
                if isinstance(req.content_sha256, str) and req.content_sha256.strip()
                else hashlib.sha256(content.encode("utf-8")).hexdigest()
            )
            sha256 = (
                str(req.sha256).strip().lower()
                if isinstance(req.sha256, str) and req.sha256.strip()
                else hashlib.sha256(content.encode("utf-8")).hexdigest()
            )
            rec = _SERVER_BLOC_STORE.upsert(
                file_meta={
                    "path": path,
                    "media_type": str(req.media_type or "text"),
                    "size_bytes": int(req.size_bytes) if req.size_bytes is not None else len(content.encode("utf-8")),
                    "mtime_ns": int(req.mtime_ns) if req.mtime_ns is not None else time.time_ns(),
                    "sha256": sha256,
                    "content_sha256": content_sha256,
                    "format": req.format,
                    "content_length": len(content),
                    "estimated_tokens": int(req.estimated_tokens) if req.estimated_tokens is not None else None,
                },
                content=content,
                relpath_base=Path(os.path.expanduser(req.relpath_base)) if isinstance(req.relpath_base, str) and req.relpath_base.strip() else None,
                summary=req.summary,
                keywords=req.keywords,
            )
            _SERVER_BLOC_STORE.ensure_bloc_ids()
            rec = _SERVER_BLOC_STORE.get(rec.sha256) or rec
            return {"ok": True, "operation": "upsert_text", "record": rec.to_dict()}
        except Exception as e:
            return _local_bloc_error("upsert_text", e, status_code=500)
    body = req.model_dump(exclude_none=True)
    base_url = body.pop("base_url", None)
    body.pop("provider", None)
    body.pop("model", None)
    api_key = body.pop("api_key", None)
    return _proxy_endpoint_control_request(
        http_request=http_request,
        base_url=base_url,
        api_key=api_key,
        method="POST",
        path="/acore/blocs/upsert_text",
        json_body=body,
    )


@app.get(
    "/acore/blocs/record",
    tags=["memory-blocs"],
    summary="Get Bloc Record",
    description="Fetch a bloc record from the gateway-local bloc store, or proxy the same operation to an upstream AbstractEndpoint when `base_url` is provided.",
)
def acore_bloc_record(
    http_request: Request,
    base_url: Optional[str] = Query(None, description="Upstream AbstractEndpoint base_url (optionally including /v1)"),
    sha256: Optional[str] = Query(None, description="Bloc sha256 selector."),
    bloc_id: Optional[int] = Query(None, description="Stable integer bloc selector."),
    api_key: Optional[str] = Query(None, description="Deprecated/disabled. Use X-AbstractCore-Provider-API-Key for provider overrides."),
):
    _reject_query_api_key(api_key)
    if not isinstance(base_url, str) or not base_url.strip():
        rec = _resolve_local_bloc_selector(sha256=sha256, bloc_id=bloc_id)
        if rec is None:
            return _local_bloc_error("record", "bloc not found", status_code=404)
        return {"ok": True, "operation": "record", "record": rec.to_dict()}
    return _proxy_endpoint_control_request(
        http_request=http_request,
        base_url=base_url,
        api_key=api_key,
        method="GET",
        path="/acore/blocs/record",
        query_params={"sha256": sha256, "bloc_id": bloc_id},
    )


@app.get(
    "/acore/blocs",
    tags=["memory-blocs"],
    summary="List Bloc Records",
    description="List gateway-local bloc records, or proxy the same operation to an upstream AbstractEndpoint when `base_url` is provided.",
)
def acore_blocs_list(
    http_request: Request,
    base_url: Optional[str] = Query(None, description="Upstream AbstractEndpoint base_url (optionally including /v1)"),
    sha256: Optional[str] = Query(None, description="Optional bloc sha256 filter."),
    bloc_id: Optional[int] = Query(None, description="Optional stable integer bloc filter."),
    api_key: Optional[str] = Query(None, description="Deprecated/disabled. Use X-AbstractCore-Provider-API-Key for provider overrides."),
):
    _reject_query_api_key(api_key)
    if not isinstance(base_url, str) or not base_url.strip():
        if sha256 or bloc_id is not None:
            rec = _resolve_local_bloc_selector(sha256=sha256, bloc_id=bloc_id)
            records = [rec.to_dict()] if rec is not None else []
        else:
            records = [rec.to_dict() for rec in _SERVER_BLOC_STORE.list()]
        return {"ok": True, "operation": "list", "records": records}
    return _proxy_endpoint_control_request(
        http_request=http_request,
        base_url=base_url,
        api_key=api_key,
        method="GET",
        path="/acore/blocs",
        query_params={"sha256": sha256, "bloc_id": bloc_id},
    )


@app.post(
    "/acore/blocs/delete",
    tags=["memory-blocs"],
    summary="Delete Bloc",
    description="Delete one gateway-local bloc, with optional live prompt-cache safety checks, or proxy to an upstream AbstractEndpoint.",
)
def acore_bloc_delete(req: BlocDeleteProxyRequest, http_request: Request):
    _reject_body_api_key(req.api_key)
    if isinstance(req.base_url, str) and req.base_url.strip():
        body = req.model_dump(exclude_none=True)
        base_url = body.pop("base_url", None)
        body.pop("runtime_id", None)
        body.pop("provider", None)
        body.pop("model", None)
        api_key = body.pop("api_key", None)
        return _proxy_endpoint_control_request(
            http_request=http_request,
            base_url=base_url,
            api_key=api_key,
            method="POST",
            path="/acore/blocs/delete",
            json_body=body,
        )

    rec = _resolve_local_bloc_selector(sha256=req.sha256, bloc_id=req.bloc_id)
    if rec is None:
        return _local_bloc_error("delete", "bloc not found", status_code=404)
    try:
        live = _live_bloc_bindings_across_loaded_runtimes(rec.sha256) if req.delete_kv else []
        if live and not (req.clear_loaded or req.force):
            return JSONResponse(
                status_code=409,
                content={"ok": False, "operation": "delete", "error": "bloc has loaded KV artifacts in live prompt-cache keys", "live_bindings": live},
            )
        if live and req.clear_loaded and not req.dry_run:
            _clear_live_bloc_bindings_across_loaded_runtimes(rec.sha256)
        result = delete_bloc(
            store=_SERVER_BLOC_STORE,
            sha256=rec.sha256,
            delete_kv=bool(req.delete_kv),
            clear_loaded=bool(req.clear_loaded),
            force=bool(req.force),
            dry_run=bool(req.dry_run),
        )
        payload = result.to_dict()
        if live:
            payload["live_bindings"] = live
        return {"ok": True, "operation": "delete", "result": payload}
    except BlocKVArtifactInUseError as e:
        return JSONResponse(
            status_code=409,
            content={"ok": False, "operation": "delete", "error": str(e), "live_bindings": e.live_bindings},
        )
    except Exception as e:
        return _local_bloc_error("delete", e, status_code=500)


@app.get(
    "/acore/blocs/kv/manifest",
    tags=["memory-blocs"],
    summary="Get Bloc KV Manifest",
    description="Inspect a bloc KV manifest on a loaded gateway runtime (`provider` + `model`) or on an upstream AbstractEndpoint (`base_url`).",
)
def acore_bloc_kv_manifest(
    http_request: Request,
    runtime_id: Optional[str] = Query(None, description="Specific loaded gateway runtime identifier."),
    base_url: Optional[str] = Query(None, description="Upstream AbstractEndpoint base_url (optionally including /v1)"),
    provider: Optional[str] = Query(None, description="Gateway-local provider name for a loaded runtime."),
    model: Optional[str] = Query(None, description="Gateway-local model identifier for a loaded runtime."),
    sha256: Optional[str] = Query(None, description="Bloc sha256 selector."),
    bloc_id: Optional[int] = Query(None, description="Stable integer bloc selector."),
    artifact_path: Optional[str] = Query(None, description="Optional explicit upstream artifact path override."),
    api_key: Optional[str] = Query(None, description="Deprecated/disabled. Use X-AbstractCore-Provider-API-Key for provider overrides."),
):
    _reject_query_api_key(api_key)
    local_requested = bool((runtime_id and runtime_id.strip()) or (str(provider or "").strip() and str(model or "").strip()))
    if local_requested:
        provider_s = str(provider or "").strip().lower()
        model_s = str(model or "").strip()
        rec = _resolve_local_bloc_selector(sha256=sha256, bloc_id=bloc_id)
        if rec is None:
            return _local_bloc_error("kv_manifest", "bloc not found", status_code=404)
        runtime = _resolve_local_runtime_for_control(provider=provider_s, model=model_s, base_url=base_url, runtime_id=runtime_id, http_request=http_request)
        try:
            with runtime.lock:
                _touch_loaded_gateway_runtime(runtime)
                manifest = read_bloc_kv_manifest(
                    provider=runtime.llm,
                    store=_SERVER_BLOC_STORE,
                    model=model_s,
                    record=rec,
                    artifact_path=artifact_path,
                )
            if manifest is None:
                return _local_bloc_error("kv_manifest", "manifest not found", status_code=404)
            return {"ok": True, "operation": "kv_manifest", "manifest": manifest.to_dict()}
        except Exception as e:
            return _local_bloc_error("kv_manifest", e, status_code=500)
    return _proxy_endpoint_control_request(
        http_request=http_request,
        base_url=base_url,
        api_key=api_key,
        method="GET",
        path="/acore/blocs/kv/manifest",
        query_params={"sha256": sha256, "bloc_id": bloc_id, "artifact_path": artifact_path},
    )


@app.get(
    "/acore/blocs/kv/list",
    tags=["memory-blocs"],
    summary="List Bloc KV Artifacts",
    description="List manifest-backed bloc KV artifacts locally or on an upstream AbstractEndpoint.",
)
def acore_bloc_kv_list(
    http_request: Request,
    base_url: Optional[str] = Query(None, description="Upstream AbstractEndpoint base_url (optionally including /v1)"),
    provider: Optional[str] = Query(None, description="Optional provider filter."),
    model: Optional[str] = Query(None, description="Optional model filter."),
    sha256: Optional[str] = Query(None, description="Optional bloc sha256 filter."),
    bloc_id: Optional[int] = Query(None, description="Optional stable integer bloc filter."),
    api_key: Optional[str] = Query(None, description="Deprecated/disabled. Use X-AbstractCore-Provider-API-Key for provider overrides."),
):
    _reject_query_api_key(api_key)
    if not isinstance(base_url, str) or not base_url.strip():
        artifacts = list_bloc_kv_artifacts(
            store=_SERVER_BLOC_STORE,
            sha256=sha256,
            bloc_id=bloc_id,
            provider=provider,
            model=model,
        )
        return {"ok": True, "operation": "kv_list", "artifacts": artifacts}
    return _proxy_endpoint_control_request(
        http_request=http_request,
        base_url=base_url,
        api_key=api_key,
        method="GET",
        path="/acore/blocs/kv/list",
        query_params={"sha256": sha256, "bloc_id": bloc_id, "provider": provider, "model": model},
    )


@app.post(
    "/acore/blocs/kv/ensure",
    tags=["memory-blocs"],
    summary="Ensure Bloc KV Artifact",
    description="Compile or validate a provider-backed bloc KV artifact on a loaded gateway runtime (`provider` + `model`) or proxy the same operation to an upstream AbstractEndpoint (`base_url`).",
)
def acore_bloc_kv_ensure(req: BlocKVEnsureProxyRequest, http_request: Request):
    _reject_body_api_key(req.api_key)
    local_requested = bool((req.runtime_id and req.runtime_id.strip()) or (str(req.provider or "").strip() and str(req.model or "").strip()))
    if local_requested:
        provider_s = str(req.provider or "").strip().lower()
        model_s = str(req.model or "").strip()
        rec = _resolve_local_bloc_selector(sha256=req.sha256, bloc_id=req.bloc_id)
        if rec is None:
            return _local_bloc_error("kv_ensure", "bloc not found", status_code=404)
        runtime = _resolve_local_runtime_for_control(provider=provider_s, model=model_s, base_url=req.base_url, runtime_id=req.runtime_id, http_request=http_request)
        try:
            result = _run_loaded_gateway_runtime(
                runtime,
                lambda: ensure_bloc_kv_artifact(
                    provider=runtime.llm,
                    store=_SERVER_BLOC_STORE,
                    model=model_s,
                    record=rec,
                    artifact_path=req.artifact_path,
                    force_rebuild=bool(req.force_rebuild),
                    debug=bool(req.debug),
                ),
            )
            artifact = {
                "artifact_path": str(result.artifact_path),
                "manifest_path": str(result.manifest_path),
                "compiled": bool(result.compiled),
                "rebuilt": bool(result.rebuilt),
                "source_cache_key": result.source_cache_key,
                "binding_id": result.binding_id,
                "prompt_cache_binding": result.prompt_cache_binding,
                "manifest": result.manifest.to_dict(),
            }
            if result.debug is not None:
                artifact["debug"] = result.debug
            return {
                "ok": True,
                "operation": "kv_ensure",
                "artifact": artifact,
            }
        except Exception as e:
            return _local_bloc_error("kv_ensure", e, status_code=500)
    body = req.model_dump(exclude_none=True)
    body.pop("runtime_id", None)
    base_url = body.pop("base_url", None)
    body.pop("provider", None)
    body.pop("model", None)
    api_key = body.pop("api_key", None)
    return _proxy_endpoint_control_request(
        http_request=http_request,
        base_url=base_url,
        api_key=api_key,
        method="POST",
        path="/acore/blocs/kv/ensure",
        json_body=body,
    )


@app.post(
    "/acore/blocs/kv/load",
    tags=["memory-blocs"],
    summary="Load Bloc KV Artifact",
    description="Load or fork a provider-backed bloc KV artifact into a prompt-cache key on a loaded gateway runtime (`provider` + `model`) or an upstream AbstractEndpoint (`base_url`).",
)
def acore_bloc_kv_load(req: BlocKVLoadProxyRequest, http_request: Request):
    _reject_body_api_key(req.api_key)
    local_requested = bool((req.runtime_id and req.runtime_id.strip()) or (str(req.provider or "").strip() and str(req.model or "").strip()))
    if local_requested:
        provider_s = str(req.provider or "").strip().lower()
        model_s = str(req.model or "").strip()
        rec = _resolve_local_bloc_selector(sha256=req.sha256, bloc_id=req.bloc_id)
        if rec is None:
            return _local_bloc_error("kv_load", "bloc not found", status_code=404)
        runtime = _resolve_local_runtime_for_control(provider=provider_s, model=model_s, base_url=req.base_url, runtime_id=req.runtime_id, http_request=http_request)
        try:
            result = _run_loaded_gateway_runtime(
                runtime,
                lambda: load_bloc_kv_artifact(
                    provider=runtime.llm,
                    store=_SERVER_BLOC_STORE,
                    model=model_s,
                    record=rec,
                    artifact_path=req.artifact_path,
                    stable_cache_key=req.stable_cache_key,
                    key=req.key,
                    make_default=bool(req.make_default),
                    force_rebuild=bool(req.force_rebuild),
                    debug=bool(req.debug),
                ),
            )
            artifact = {
                "artifact_path": str(result.artifact_path),
                "manifest_path": str(result.manifest_path),
                "compiled": bool(result.compiled),
                "loaded": bool(result.loaded),
                "reloaded_stable_key": bool(result.reloaded_stable_key),
                "key": result.key,
                "stable_cache_key": result.stable_cache_key,
                "forked_from": result.forked_from,
                "binding_id": result.binding_id,
                "prompt_cache_binding": result.prompt_cache_binding,
                "manifest": result.manifest.to_dict(),
            }
            if result.debug is not None:
                artifact["debug"] = result.debug
            return {
                "ok": True,
                "operation": "kv_load",
                "artifact": artifact,
            }
        except Exception as e:
            return _local_bloc_error("kv_load", e, status_code=500)
    body = req.model_dump(exclude_none=True)
    body.pop("runtime_id", None)
    base_url = body.pop("base_url", None)
    body.pop("provider", None)
    body.pop("model", None)
    api_key = body.pop("api_key", None)
    return _proxy_endpoint_control_request(
        http_request=http_request,
        base_url=base_url,
        api_key=api_key,
        method="POST",
        path="/acore/blocs/kv/load",
        json_body=body,
    )


@app.post(
    "/acore/blocs/kv/delete",
    tags=["memory-blocs"],
    summary="Delete Bloc KV Artifact",
    description="Delete one bloc KV artifact, with optional live prompt-cache safety checks, locally or on an upstream AbstractEndpoint.",
)
def acore_bloc_kv_delete(req: BlocKVDeleteProxyRequest, http_request: Request):
    _reject_body_api_key(req.api_key)
    if isinstance(req.base_url, str) and req.base_url.strip():
        body = req.model_dump(exclude_none=True)
        base_url = body.pop("base_url", None)
        body.pop("runtime_id", None)
        body.pop("provider", None)
        body.pop("model", None)
        api_key = body.pop("api_key", None)
        return _proxy_endpoint_control_request(
            http_request=http_request,
            base_url=base_url,
            api_key=api_key,
            method="POST",
            path="/acore/blocs/kv/delete",
            json_body=body,
        )

    provider_s = str(req.provider or "").strip().lower()
    model_s = str(req.model or "").strip()
    runtime = _optional_local_runtime_for_bloc_control(
        http_request=http_request,
        runtime_id=req.runtime_id,
        provider=provider_s,
        model=model_s,
        base_url=req.base_url,
    )
    try:
        if runtime is not None:
            result = _run_loaded_gateway_runtime(
                runtime,
                lambda: delete_bloc_kv_artifact(
                    store=_SERVER_BLOC_STORE,
                    provider=runtime.llm,
                    sha256=req.sha256,
                    bloc_id=req.bloc_id,
                    provider_name=provider_s or None,
                    model=model_s or None,
                    artifact_path=req.artifact_path,
                    clear_loaded=bool(req.clear_loaded),
                    force=bool(req.force),
                    dry_run=bool(req.dry_run),
                    debug=bool(req.debug),
                ),
            )
        else:
            rec_for_live = _resolve_local_bloc_selector(sha256=req.sha256, bloc_id=req.bloc_id)
            live_sha = rec_for_live.sha256 if rec_for_live is not None else None
            live = _live_bloc_bindings_across_loaded_runtimes(live_sha)
            if live and not (req.clear_loaded or req.force):
                return JSONResponse(
                    status_code=409,
                    content={"ok": False, "operation": "kv_delete", "error": "matching bloc KV artifact may be loaded in a live prompt-cache key", "live_bindings": live},
                )
            if live and req.clear_loaded and not req.dry_run:
                _clear_live_bloc_bindings_across_loaded_runtimes(live_sha)
            result = delete_bloc_kv_artifact(
                store=_SERVER_BLOC_STORE,
                sha256=req.sha256,
                bloc_id=req.bloc_id,
                provider_name=provider_s or None,
                model=model_s or None,
                artifact_path=req.artifact_path,
                clear_loaded=False,
                force=bool(req.force),
                dry_run=bool(req.dry_run),
                debug=bool(req.debug),
            )
        return {"ok": True, "operation": "kv_delete", "result": result.to_dict()}
    except BlocKVArtifactInUseError as e:
        return JSONResponse(
            status_code=409,
            content={"ok": False, "operation": "kv_delete", "error": str(e), "live_bindings": e.live_bindings},
        )
    except Exception as e:
        status = 404 if "not found" in str(e).lower() else 500
        return _local_bloc_error("kv_delete", e, status_code=status)


@app.post(
    "/acore/blocs/kv/prune",
    tags=["memory-blocs"],
    summary="Prune Bloc KV Artifacts",
    description="Delete matching bloc KV artifacts by filter, with optional live prompt-cache safety checks, locally or on an upstream AbstractEndpoint.",
)
def acore_bloc_kv_prune(req: BlocKVPruneProxyRequest, http_request: Request):
    _reject_body_api_key(req.api_key)
    if isinstance(req.base_url, str) and req.base_url.strip():
        body = req.model_dump(exclude_none=True)
        base_url = body.pop("base_url", None)
        body.pop("runtime_id", None)
        api_key = body.pop("api_key", None)
        return _proxy_endpoint_control_request(
            http_request=http_request,
            base_url=base_url,
            api_key=api_key,
            method="POST",
            path="/acore/blocs/kv/prune",
            json_body=body,
        )

    provider_s = str(req.provider or "").strip().lower()
    model_s = str(req.model or "").strip()
    runtime = _optional_local_runtime_for_bloc_control(
        http_request=http_request,
        runtime_id=req.runtime_id,
        provider=provider_s,
        model=model_s,
        base_url=req.base_url,
    )
    try:
        if runtime is not None:
            results = _run_loaded_gateway_runtime(
                runtime,
                lambda: prune_bloc_kv_artifacts(
                    store=_SERVER_BLOC_STORE,
                    provider=runtime.llm,
                    sha256=req.sha256,
                    bloc_id=req.bloc_id,
                    provider_name=provider_s or None,
                    model=model_s or None,
                    clear_loaded=bool(req.clear_loaded),
                    force=bool(req.force),
                    dry_run=bool(req.dry_run),
                    debug=bool(req.debug),
                ),
            )
        else:
            rec_for_live = _resolve_local_bloc_selector(sha256=req.sha256, bloc_id=req.bloc_id)
            live_sha = rec_for_live.sha256 if rec_for_live is not None else None
            live = _live_bloc_bindings_across_loaded_runtimes(live_sha)
            if live and not (req.clear_loaded or req.force):
                return JSONResponse(
                    status_code=409,
                    content={"ok": False, "operation": "kv_prune", "error": "matching bloc KV artifacts may be loaded in live prompt-cache keys", "live_bindings": live},
                )
            if live and req.clear_loaded and not req.dry_run:
                _clear_live_bloc_bindings_across_loaded_runtimes(live_sha)
            results = prune_bloc_kv_artifacts(
                store=_SERVER_BLOC_STORE,
                sha256=req.sha256,
                bloc_id=req.bloc_id,
                provider_name=provider_s or None,
                model=model_s or None,
                clear_loaded=False,
                force=bool(req.force),
                dry_run=bool(req.dry_run),
                debug=bool(req.debug),
            )
        return {"ok": True, "operation": "kv_prune", "results": [result.to_dict() for result in results]}
    except BlocKVArtifactInUseError as e:
        return JSONResponse(
            status_code=409,
            content={"ok": False, "operation": "kv_prune", "error": str(e), "live_bindings": e.live_bindings},
        )
    except Exception as e:
        return _local_bloc_error("kv_prune", e, status_code=500)


class _ServerCapabilityOwner:
    config: Dict[str, Any] = {}
    model: str = "abstractcore-server"

    def generate(self, *args: Any, **kwargs: Any) -> Any:
        _ = args, kwargs
        raise RuntimeError("Server capability discovery owner does not expose text generation.")


def _server_capability_registry() -> CapabilityRegistry:
    try:
        from .audio_endpoints import _get_capability_core

        core = _get_capability_core()
        capabilities = getattr(core, "capabilities", None)
        if capabilities is not None:
            return capabilities
    except Exception:
        pass
    return CapabilityRegistry(_ServerCapabilityOwner())


def _capability_error_response(error: Exception, *, operation: str) -> JSONResponse:
    if isinstance(error, CapabilityUnavailableError):
        return JSONResponse(status_code=501, content={"ok": False, "operation": operation, "error": str(error)})
    return JSONResponse(status_code=500, content={"ok": False, "operation": operation, "error": str(error)})


class CapabilityDefaultRouteRequest(BaseModel):
    provider: Optional[str] = Field(default=None, description="Default provider/backend id for this capability route.")
    model: Optional[str] = Field(default=None, description="Default model id for this capability route.")
    base_url: Optional[str] = Field(default=None, description="Optional upstream provider base URL for this capability route.")
    options: Optional[Dict[str, Any]] = Field(default=None, description="Optional plugin/provider parameters, such as voice/profile/language.")

    class Config:
        json_schema_extra = {
            "examples": [
                {
                    "provider": "mlx-gen",
                    "model": "AbstractFramework/flux.2-klein-9b-4bit",
                    "base_url": None,
                    "options": {},
                }
            ]
        }


def _capability_defaults_payload() -> Dict[str, Any]:
    from ..config.manager import ConfigurationManager

    manager = ConfigurationManager()
    return {
        "ok": True,
        "version": 1,
        "authority": "abstractcore.server",
        "writable": True,
        "source": "abstractcore.server",
        "config_file": str(manager.config_file),
        "routes": manager.list_capability_defaults(),
        "errors": [],
    }


@app.get(
    "/v1/config/capability-defaults",
    tags=["configuration"],
    summary="List Capability Routing Defaults",
    description="List explicit capability routing defaults from this AbstractCore execution host.",
)
def list_capability_defaults():
    return _capability_defaults_payload()


@app.put(
    "/v1/config/capability-defaults/{kind}/{modality}",
    tags=["configuration"],
    summary="Set Capability Routing Default",
    description="Persist one capability routing default on this AbstractCore execution host.",
)
def set_capability_default(
    req: CapabilityDefaultRouteRequest,
    kind: str = FastAPIPath(..., description="Capability route kind: input, output, embedding, or rerank."),
    modality: str = FastAPIPath(..., description="Capability modality: text, image, video, voice, sound, music, or scene3d."),
):
    from ..config.manager import ConfigurationManager

    manager = ConfigurationManager()
    ok = manager.set_capability_default(
        kind,
        modality,
        provider=req.provider,
        model=req.model,
        base_url=req.base_url,
        options=dict(req.options or {}) if isinstance(req.options, dict) else {},
    )
    if not ok:
        raise HTTPException(status_code=400, detail=f"Failed to set capability default {kind}.{modality}")
    return _capability_defaults_payload()


@app.delete(
    "/v1/config/capability-defaults/{kind}/{modality}",
    tags=["configuration"],
    summary="Clear Capability Routing Default",
    description="Clear one persisted capability routing default on this AbstractCore execution host.",
)
def clear_capability_default(
    kind: str = FastAPIPath(..., description="Capability route kind: input, output, embedding, or rerank."),
    modality: str = FastAPIPath(..., description="Capability modality: text, image, video, voice, sound, music, or scene3d."),
):
    from ..config.manager import ConfigurationManager

    manager = ConfigurationManager()
    ok = manager.clear_capability_default(kind, modality)
    if not ok:
        raise HTTPException(status_code=400, detail=f"Failed to clear capability default {kind}.{modality}")
    return _capability_defaults_payload()


@app.get(
    "/v1/capabilities",
    tags=["capabilities"],
    summary="List Capability Plugins",
    description="Inspect optional capability plugin availability and registered backend metadata.",
)
def list_capability_plugins():
    registry = _server_capability_registry()
    try:
        return {
            "ok": True,
            "operation": "capabilities",
            "status": registry.status(),
            "backends": [info.to_dict() for info in registry.list_backend_infos()],
        }
    except Exception as e:
        return _capability_error_response(e, operation="capabilities")


@app.get(
    "/v1/capabilities/{capability}/providers",
    tags=["capabilities"],
    summary="List Capability Providers",
    description="List normalized provider availability for one optional capability plugin.",
)
def list_capability_providers(
    capability: str = FastAPIPath(..., description="Capability id such as voice, audio, vision, or music."),
    task: Optional[str] = Query(None, description="Optional canonical task filter."),
):
    registry = _server_capability_registry()
    try:
        return {
            "ok": True,
            "operation": "capability_providers",
            "capability": capability,
            "task": task,
            "providers": registry.available_providers(capability, task=task),
        }
    except Exception as e:
        return _capability_error_response(e, operation="capability_providers")


@app.get(
    "/v1/capabilities/{capability}/models",
    tags=["capabilities"],
    summary="List Capability Models",
    description="List normalized model availability for one optional capability plugin.",
)
def list_capability_models(
    capability: str = FastAPIPath(..., description="Capability id such as voice, audio, vision, or music."),
    task: Optional[str] = Query(None, description="Optional canonical task filter."),
    provider: Optional[str] = Query(None, description="Optional provider filter."),
):
    registry = _server_capability_registry()
    try:
        return {
            "ok": True,
            "operation": "capability_models",
            "capability": capability,
            "task": task,
            "provider": provider,
            "models": registry.list_models(capability, task=task, provider=provider),
        }
    except Exception as e:
        return _capability_error_response(e, operation="capability_models")


@app.get(
    "/v1/audio/music/providers",
    tags=["audio"],
    summary="List Music Providers",
    description="List normalized provider availability for the music capability plugin.",
)
def list_music_providers(task: Optional[str] = Query("text_to_music", description="Optional music task filter.")):
    registry = _server_capability_registry()
    try:
        return {
            "ok": True,
            "operation": "music_providers",
            "capability": "music",
            "task": task,
            "providers": registry.available_providers("music", task=task),
        }
    except Exception as e:
        return _capability_error_response(e, operation="music_providers")


@app.get(
    "/v1/audio/music/models",
    tags=["audio"],
    summary="List Music Models",
    description="List normalized model availability for the music capability plugin.",
)
def list_music_models(
    task: Optional[str] = Query("text_to_music", description="Optional music task filter."),
    provider: Optional[str] = Query(None, description="Optional provider filter."),
):
    registry = _server_capability_registry()
    try:
        return {
            "ok": True,
            "operation": "music_models",
            "capability": "music",
            "task": task,
            "provider": provider,
            "models": registry.list_models("music", task=task, provider=provider),
        }
    except Exception as e:
        return _capability_error_response(e, operation="music_models")


@app.get("/v1/models", tags=["models"])
async def list_models(
    http_request: Request,
    provider: Optional[str] = Query(
        None,
        description="Filter by provider (e.g., 'ollama', 'openai', 'anthropic', 'lmstudio')",
    ),
    input_type: Optional[ModelInputCapability] = Query(
        None,
        description="Filter by input capability: 'text', 'image', 'audio', 'video'"
    ),
    output_type: Optional[ModelOutputCapability] = Query(
        None,
        description="Filter by output capability: 'text', 'embeddings'"
    ),
    capability_route: Optional[List[str]] = Query(
        None,
        description=(
            "Precise capability route filter. Repeat or comma-separate values, "
            "for example capability_route=input.image&capability_route=output.text."
        ),
    ),
    api_key: Optional[str] = Query(
        None,
        description=(
            "Optional upstream provider API key override. Prefer `X-AbstractCore-Provider-API-Key`; "
            "this query parameter is supported for tooling/Swagger UI convenience and is redacted from server logs."
        ),
    ),
    base_url: Optional[str] = Query(
        None,
        description="Optional upstream base URL override (useful for OpenAI-compatible providers).",
    ),
):
    """
    List available models from AbstractCore providers.

    Returns a list of all available models, optionally filtered by provider and/or capabilities.

    **Filtering System:**
    - `input_type`: Filter by what INPUT the model can process (text, image, audio, video)
    - `output_type`: Filter by what OUTPUT the model generates (text, embeddings)
    - `capability_route`: Precise route filter using `<kind>.<modality>` keys
      such as `input.image`, `output.text`, or `embedding.text`

    **Examples:**
    - `/v1/models` - All models from all providers
    - `/v1/models?output_type=embeddings` - Only embedding models
    - `/v1/models?capability_route=embedding.text` - Only text embedding models
    - `/v1/models?capability_route=input.image,output.text` - Vision text models
    - `/v1/models?input_type=text&output_type=text` - Text-only models that generate text
    - `/v1/models?input_type=image` - Models that can analyze images
    - `/v1/models?provider=ollama&input_type=image` - Ollama vision models only
    """
    try:
        models_data = []

        # Use the capability enums directly
        input_capabilities = [input_type] if input_type else None
        output_capabilities = [output_type] if output_type else None
        try:
            capability_routes = normalize_capability_route_filter(capability_route)
        except ValueError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc
        api_key_s = str(api_key).strip() if isinstance(api_key, str) and str(api_key).strip() else ""
        if api_key_s and not provider:
            raise HTTPException(
                status_code=400,
                detail={
                    "error": {
                        "message": "api_key requires provider=... so the override target is unambiguous.",
                        "type": "invalid_request",
                    }
                },
            )

        if provider:
            provider_norm = provider.lower().strip()
            from ..providers.registry import is_provider_available

            # Match prior behavior: unknown providers return an empty list (200 OK).
            if not is_provider_available(provider_norm):
                return {"object": "list", "data": []}

            provider_kwargs: Dict[str, Any] = {}

            candidate_key = api_key_s or (_provider_api_key_from_request(http_request) or "")

            if candidate_key:
                provider_kwargs["api_key"] = candidate_key
            _guard_unauthenticated_server_provider_key_use(
                provider_norm,
                explicit_provider_key=bool(candidate_key),
                http_request=http_request,
            )
            if isinstance(base_url, str) and base_url.strip():
                base_url_s = base_url.strip()
                if not _server_allows_request_base_url(base_url_s):
                    raise HTTPException(
                        status_code=403,
                        detail={
                            "error": {
                                "message": (
                                    "Request-level base_url overrides are restricted for security. "
                                    "By default only loopback URLs are allowed. "
                                    "To allow additional hosts/prefixes, set "
                                    "ABSTRACTCORE_SERVER_BASE_URL_ALLOWLIST."
                                ),
                                "type": "forbidden",
                            }
                        },
                    )

                parsed = urllib.parse.urlsplit(base_url_s)
                host = str(parsed.hostname or "").strip()
                if host and not _is_loopback_host(host):
                    env_var = _provider_api_key_env_var(provider_norm)
                    if _server_has_provider_api_key(provider_norm) and not candidate_key:
                        raise HTTPException(
                            status_code=403,
                            detail={
                                "error": {
                                    "message": (
                                        "Refusing request-level base_url override without an explicit provider key "
                                        f"because the server has {env_var} set. Provide the provider key via "
                                        "X-AbstractCore-Provider-API-Key."
                                    ),
                                    "type": "forbidden",
                                }
                            },
                        )

                provider_kwargs["base_url"] = base_url_s

            # Get models from specific provider with optional filtering
            try:
                models = get_models_from_provider(
                    provider_norm,
                    input_capabilities=input_capabilities,
                    output_capabilities=output_capabilities,
                    capability_routes=capability_routes,
                    raise_on_error=True,
                    **provider_kwargs,
                )
            except Exception as e:
                raise HTTPException(
                    status_code=502,
                    detail=f"Failed to list models for provider '{provider_norm}': {e}",
                ) from e
            for model in models:
                model_id = f"{provider_norm}/{model}"
                models_data.append({
                    "id": model_id,
                    "object": "model",
                    "owned_by": provider_norm,
                    "created": int(time.time()),
                    "permission": [{"allow_create_engine": False, "allow_sampling": True}]
                })

            filter_parts = []
            if input_type:
                filter_parts.append(f"input_type={input_type.value}")
            if output_type:
                filter_parts.append(f"output_type={output_type.value}")
            if capability_routes:
                filter_parts.append(f"capability_route={','.join(capability_routes)}")

            filter_msg = f" ({', '.join(filter_parts)})" if filter_parts else ""
            logger.info(f"Listed {len(models_data)} models for provider {provider}{filter_msg}")
        else:
            # Get models from all providers using centralized registry
            from ..providers.registry import list_available_providers
            providers = list_available_providers()
            for prov in providers:
                if not _server_auth_enabled() and _server_has_provider_api_key(prov):
                    logger.warning(
                        "Skipping provider model discovery because server-held API key requires server auth",
                        provider=prov,
                    )
                    continue
                models = get_models_from_provider(
                    prov, 
                    input_capabilities=input_capabilities,
                    output_capabilities=output_capabilities,
                    capability_routes=capability_routes,
                )
                for model in models:
                    model_id = f"{prov}/{model}"
                    models_data.append({
                        "id": model_id,
                        "object": "model",
                        "owned_by": prov,
                        "created": int(time.time()),
                        "permission": [{"allow_create_engine": False, "allow_sampling": True}]
                    })

            filter_parts = []
            if input_type:
                filter_parts.append(f"input_type={input_type.value}")
            if output_type:
                filter_parts.append(f"output_type={output_type.value}")
            if capability_routes:
                filter_parts.append(f"capability_route={','.join(capability_routes)}")

            filter_msg = f" ({', '.join(filter_parts)})" if filter_parts else ""
            logger.info(f"Listed {len(models_data)} models from all providers{filter_msg}")

        return {
            "object": "list",
            "data": sorted(models_data, key=lambda x: x["id"])
        }

    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Failed to list models: {_redact_text(str(e))}")
        return {
            "object": "list",
            "data": []
        }

@app.get("/providers", tags=["providers"])
async def list_providers(
    include_models: bool = Query(
        False,
        description="Include model lists for each provider. Set to true for full information (slower)."
    )
):
    """
    List all available AbstractCore providers and their capabilities.

    Returns comprehensive information about all registered LLM providers, including:
    - Provider name, display name, and type
    - Number of available models and sample models (if include_models=True)
    - Current availability status and detailed error information
    - Provider description and supported features
    - Authentication requirements and installation instructions
    - Local vs. cloud provider designation

    **Query Parameters:**
    - `include_models` (bool, default=False): Include model lists for each provider.
      Set to `true` for full information (slower).

    **Performance:**
    - `include_models=false`: Metadata only (very fast, ~15ms) - **DEFAULT**
    - `include_models=true`: Full information including model lists (slower, ~800ms)

    **Supported Providers:**
    - **OpenAI**: Commercial API with GPT-4, GPT-3.5, and embedding models
    - **Anthropic**: Commercial API with Claude 3 family models
    - **Ollama**: Local LLM server for running open-source models
    - **LMStudio**: Local model development and testing platform
    - **MLX**: Apple Silicon optimized local inference
    - **HuggingFace**: Access to HuggingFace models (transformers and embeddings)

    **Use Cases:**
    - Fast provider discovery: `GET /providers` (default, very fast)
    - Full provider information: `GET /providers?include_models=true`
    - Build dynamic provider selection UIs
    - Monitor provider status and troubleshoot issues
    - Get installation instructions for missing dependencies

    **Returns:** A list of provider objects with comprehensive metadata.
    """
    try:
        from ..providers.registry import get_all_providers_with_models, get_all_providers_status, list_available_providers

        if include_models and not _server_auth_enabled():
            keyed_providers = [p for p in list_available_providers() if _server_has_provider_api_key(p)]
            if keyed_providers:
                raise HTTPException(
                    status_code=401,
                    detail={
                        "error": {
                            "message": (
                                "Provider model discovery would use server-held API keys, but inbound server auth is "
                                "not configured. Set ABSTRACTCORE_AUTH_TOKEN "
                                "or omit include_models=true."
                            ),
                            "type": "authentication_error",
                        }
                    },
                    headers={"WWW-Authenticate": "Bearer"},
                )

        # Get providers with models (available providers)
        available_providers = get_all_providers_with_models(include_models=include_models)

        # Optionally include all providers (even those with issues) for debugging
        # Uncomment the next line if you want to see providers with errors too:
        # all_providers = get_all_providers_status()

        logger.info(f"Listed {len(available_providers)} available providers with models")

        return {
            "providers": available_providers,
            "total_providers": len(available_providers),
            "registry_version": "2.0",  # Indicate this is using the new registry system
            "note": "Provider information from centralized AbstractCore registry"
        }

    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Failed to list providers: {_redact_text(str(e))}")
        return {
            "providers": [],
            "total_providers": 0,
            "error": _redact_text(str(e)),
            "registry_version": "2.0"
        }

@app.post(
    "/v1/responses",
    tags=["responses"],
    summary="Create Response",
    description=(
        "Create a model response using either the OpenAI Responses API request shape "
        "or the legacy AbstractCore chat-completions request shape."
    ),
    openapi_extra={
        "requestBody": {
            "description": (
                "JSON body in either OpenAI Responses format (`model` plus `input`) "
                "or legacy chat-completions format (`model` plus `messages`)."
            )
        }
    },
)
async def create_response(
    http_request: Request,
    request_body: Annotated[
        Dict[str, Any],
        Body(
            ...,
            description=(
                "Responses request body. Accepts either the OpenAI Responses API shape "
                "with `input` content items, or the legacy chat-completions shape with "
                "`messages`. The examples below show both supported formats."
            ),
            examples={
                "openai_format": {
                    "summary": "OpenAI Responses API Format",
                    "description": "Use input array with input_text and input_file objects",
                    "value": {
                        "model": "gpt-4o",
                        "input": [
                            {
                                "role": "user",
                                "content": [
                                    {"type": "input_text", "text": "Analyze this document"},
                                    {"type": "input_file", "file_url": "https://example.com/doc.pdf"}
                                ]
                            }
                        ],
                        "stream": False
                    }
                },
                "legacy_format": {
                    "summary": "Legacy Format (Backward Compatible)",
                    "description": "Use messages array like standard chat completions",
                    "value": {
                        "model": "openai/gpt-4",
                        "messages": [
                            {"role": "user", "content": "Tell me a story"}
                        ],
                        "stream": False
                    }
                },
                "file_analysis": {
                    "summary": "Document Analysis",
                    "description": "Analyze files using OpenAI format",
                    "value": {
                        "model": "openai/gpt-4",
                        "input": [
                            {
                                "role": "user",
                                "content": [
                                    {"type": "input_text", "text": "What's the key information in this CSV?"},
                                    {"type": "input_file", "file_url": "data:text/csv;base64,RGF0ZSxQcm9kdWN0LFNhbGVzCjIwMjQtMDEtMDEsUHJvZHVjdCBBLDEwMDAwCjIwMjQtMDEtMDIsUHJvZHVjdCBCLDE1MDAwCjIwMjQtMDEtMDMsUHJvZHVjdCBDLDI1MDAw"}
                                ]
                            }
                        ]
                    }
                },
                "streaming_example": {
                    "summary": "Streaming Response",
                    "description": "Enable streaming for real-time responses",
                    "value": {
                        "model": "lmstudio/qwen/qwen3-next-80b",
                        "input": [
                            {
                                "role": "user",
                                "content": [
                                    {"type": "input_text", "text": "Analyze the letter and provide a summary of the key points."},
                                    {"type": "input_file", "file_url": "https://www.berkshirehathaway.com/letters/2024ltr.pdf"}
                                ]
                            }
                        ],
                        "stream": True
                    }
                }
            }
        )
    ]
):
    """
    OpenAI Responses API + Backward Compatibility

    Supports both OpenAI's responses format and our legacy format for seamless migration.
    Streaming can be enabled by setting "stream": true for real-time interaction.

    **OpenAI Format (input_file support):**
    ```json
    {
      "model": "gpt-4o",
      "input": [
        {
          "role": "user",
          "content": [
            {"type": "input_text", "text": "Analyze this document"},
            {"type": "input_file", "file_url": "https://example.com/doc.pdf"}
          ]
        }
      ]
    }
    ```

    **Legacy Format (backward compatibility):**
    ```json
    {
      "model": "openai/gpt-4",
      "messages": [
        {"role": "user", "content": "Tell me a story"}
      ]
    }
    ```

    **Key Features:**
    - **100% OpenAI Compatible**: Supports input_file with file_url
    - **Universal File Support**: PDF, DOCX, XLSX, CSV, images, and more
    - **Multi-Provider**: Works with all providers (OpenAI, Anthropic, Ollama, etc.)
    - **Optional Streaming**: Set "stream": true for real-time responses
    - **Backward Compatible**: Existing clients continue to work

    **Returns:**
    - OpenAI format (`input`): OpenAI Responses `object: "response"` payload, or Responses SSE events when streaming.
    - Legacy format (`messages`): Chat Completions payload, or chat-completions SSE chunks when streaming.
    """
    try:
        # Use the parsed request body directly
        request_data = request_body
        _reject_body_api_key(request_data.get("api_key") if isinstance(request_data, dict) else None)

        # Detect OpenAI responses format vs legacy format
        if "input" in request_data:
            # OpenAI Responses API format
            logger.info("📡 OpenAI Responses API format detected")

            # Parse as OpenAI format
            openai_request = OpenAIResponsesRequest(**request_data)

            # Convert to internal format
            chat_request = convert_openai_responses_to_chat_completion(openai_request)
            openai_output_format: Literal["chat_completions", "responses"] = "responses"

        elif "messages" in request_data:
            # Legacy format (backward compatibility)
            logger.info("📡 Legacy responses format detected")

            # Parse as ChatCompletionRequest
            chat_request = ChatCompletionRequest(**request_data)
            openai_request = None
            openai_output_format = "chat_completions"

        else:
            raise HTTPException(
                status_code=400,
                detail={"error": {"message": "Request must contain either 'input' (OpenAI format) or 'messages' (legacy format)", "type": "invalid_request"}}
            )

        # Respect user's streaming preference (defaults to False)

        # Process using our standard pipeline
        provider, model = parse_model_string(chat_request.model)

        logger.info(
            "📡 Responses API Request",
            provider=provider,
            model=model,
            format="openai" if "input" in request_data else "legacy",
            messages=len(chat_request.messages)
        )

        return await process_chat_completion(
            provider,
            model,
            chat_request,
            http_request,
            openai_output_format=openai_output_format,
            openai_responses_request=openai_request,
        )

    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Responses API error: {_redact_text(str(e))}")
        raise HTTPException(
            status_code=400,
            detail={"error": {"message": _redact_text(str(e)), "type": "processing_error"}}
        )

_DIRECT_EMBEDDING_PROVIDERS = set(_list_direct_embedding_providers())
_SUPPORTED_EMBEDDING_PROVIDERS = set(_list_embedding_providers())


@app.get("/v1/embeddings/providers", tags=["embeddings"])
async def list_embedding_providers(
    provider: Optional[str] = Query(
        None,
        description="Optional embedding provider filter, e.g. huggingface, lmstudio, openai, or vllm.",
    ),
):
    """List supported text embedding providers from the AbstractCore embeddings catalog."""
    from ..embeddings.models import list_provider_details

    provider_text = str(provider or "").strip().lower()
    provider_details = list_provider_details(provider_text or None)
    providers = [
        str(item.get("provider") or item.get("id") or "").strip()
        for item in provider_details
        if isinstance(item, dict) and str(item.get("provider") or item.get("id") or "").strip()
    ]
    return {
        "kind": "embedding_providers",
        "scope": "embedding.text",
        "provider": provider_text or None,
        "providers": providers,
        "available_providers": providers,
        "embedding_providers": providers,
        "provider_details": provider_details,
        "models": [],
        "embedding_models": [],
        "models_by_provider": {},
        "embedding_models_by_provider": {},
        "provider_models": [],
        "available": bool(providers),
        "error": None if providers or not provider_text else f"Unsupported embedding provider: {provider_text}",
    }


def _normalize_embedding_inputs(value: Union[str, List[str]]) -> List[str]:
    if isinstance(value, str):
        inputs = [value]
    else:
        inputs = list(value)

    if not inputs:
        raise HTTPException(
            status_code=400,
            detail={"error": {"message": "Embedding input array must not be empty.", "type": "invalid_request"}},
        )
    for idx, text in enumerate(inputs):
        if not isinstance(text, str):
            raise HTTPException(
                status_code=400,
                detail={
                    "error": {
                        "message": f"Embedding input at index {idx} must be a string.",
                        "type": "invalid_request",
                    }
                },
            )
        if not text.strip():
            raise HTTPException(
                status_code=400,
                detail={
                    "error": {
                        "message": f"Embedding input at index {idx} must not be empty.",
                        "type": "invalid_request",
                    }
                },
            )
    return inputs


def _embedding_dimensions(value: Optional[int]) -> Optional[int]:
    if value is None or value == 0:
        return None
    if value < 0:
        raise HTTPException(
            status_code=400,
            detail={"error": {"message": "dimensions must be a positive integer, 0, or null.", "type": "invalid_request"}},
        )
    return int(value)


def _embedding_provider_kwargs(
    *,
    provider: str,
    request: EmbeddingRequest,
    http_request: Request,
) -> Dict[str, Any]:
    provider_kwargs: Dict[str, Any] = {}

    provider_api_key = _provider_api_key_from_request(http_request)
    if provider_api_key:
        provider_kwargs["api_key"] = provider_api_key

    _guard_unauthenticated_server_provider_key_use(
        provider,
        explicit_provider_key=bool(provider_api_key),
        http_request=http_request,
    )

    if isinstance(request.base_url, str) and request.base_url.strip():
        base_url = request.base_url.strip()
        if not _server_allows_request_base_url(base_url):
            raise HTTPException(
                status_code=403,
                detail={
                    "error": {
                        "message": (
                            "Request-level base_url overrides are restricted for security. "
                            "By default only loopback URLs are allowed. "
                            "To allow additional hosts/prefixes, set ABSTRACTCORE_SERVER_BASE_URL_ALLOWLIST."
                        ),
                        "type": "forbidden",
                    }
                },
            )

        parsed = urllib.parse.urlsplit(base_url)
        host = str(parsed.hostname or "").strip()
        if host and not _is_loopback_host(host):
            env_var = _provider_api_key_env_var(provider)
            if _server_has_provider_api_key(provider) and not provider_api_key:
                raise HTTPException(
                    status_code=403,
                    detail={
                        "error": {
                            "message": (
                                "Refusing request-level base_url override without an explicit provider key because "
                                f"the server has {env_var} set. Provide the provider key via "
                                "X-AbstractCore-Provider-API-Key. "
                                f"(provider={provider})"
                            ),
                            "type": "forbidden",
                        }
                    },
                )

        provider_kwargs["base_url"] = base_url

    return provider_kwargs


def _create_direct_embedding_provider(provider: str, model: str, provider_kwargs: Dict[str, Any]):
    if provider == "openai":
        from ..providers.openai_provider import OpenAIProvider

        return OpenAIProvider(model=model, **provider_kwargs)
    if provider == "openrouter":
        from ..providers.openrouter_provider import OpenRouterProvider

        return OpenRouterProvider(model=model, validate_model=False, **provider_kwargs)
    if provider == "portkey":
        from ..providers.portkey_provider import PortkeyProvider

        return PortkeyProvider(model=model, **provider_kwargs)
    if provider == "openai-compatible":
        from ..providers.openai_compatible_provider import OpenAICompatibleProvider

        return OpenAICompatibleProvider(model=model, validate_model=False, **provider_kwargs)
    if provider == "vllm":
        from ..providers.vllm_provider import VLLMProvider

        return VLLMProvider(model=model, validate_model=False, **provider_kwargs)
    if provider == "lmstudio":
        from ..providers.lmstudio_provider import LMStudioProvider

        return LMStudioProvider(model=model, validate_model=False, **provider_kwargs)
    raise ValueError(f"Unsupported direct embedding provider: {provider}")


def _provider_exception_status(exc: Exception) -> int:
    if isinstance(exc, AuthenticationError):
        return 401
    if isinstance(exc, RateLimitError):
        return 429
    if isinstance(exc, ModelNotFoundError):
        return 404
    if isinstance(exc, InvalidRequestError):
        return 400
    if isinstance(exc, ProviderAPIError):
        return 502
    text = str(exc).lower()
    if "api key" in text or "authentication" in text or "unauthorized" in text:
        return 401
    return 500


@app.post("/v1/embeddings", tags=["embeddings"])
async def create_embeddings(request: EmbeddingRequest, http_request: Request):
    """
    Create embedding vectors representing the input text.

    Creates an embedding vector representing the input text. Embeddings are useful for:
    - Semantic search
    - Document similarity
    - Clustering and classification
    - Retrieval-Augmented Generation (RAG)

    **Supported Providers:**
    - **HuggingFace**: Local sentence-transformers models with ONNX acceleration
    - **Ollama**: Embedding models via Ollama API
    - **LMStudio**: Embedding models via LMStudio API

    **Model Format:** Use `provider/model` format:
    - `huggingface/sentence-transformers/all-MiniLM-L6-v2`
    - `ollama/granite-embedding:278m`
    - `lmstudio/text-embedding-all-minilm-l6-v2`

    **To see available embedding models:** `GET /v1/models?output_type=embeddings`

    **Returns:** A list of embedding objects containing the embedding vector and metadata.
    """
    try:
        # Parse provider and model
        provider, model = parse_model_string(request.model)

        logger.info(
            "🔢 Embedding Request",
            provider=provider,
            model=model,
            input_type=type(request.input).__name__,
            input_count=len(request.input) if isinstance(request.input, list) else 1
        )

        # Validate provider
        provider_lower = provider.lower()
        if provider_lower not in _SUPPORTED_EMBEDDING_PROVIDERS:
            raise HTTPException(
                status_code=400,
                detail={
                    "error": {
                        "message": (
                            f"Embedding provider '{provider}' not supported. Supported providers: "
                            f"{', '.join(sorted(_SUPPORTED_EMBEDDING_PROVIDERS))}. "
                            "Anthropic does not expose a native embeddings API; use another embeddings provider."
                        ),
                        "type": "unsupported_provider"
                    }
                }
            )

        _reject_body_api_key(request.api_key)
        provider_kwargs = _embedding_provider_kwargs(provider=provider_lower, request=request, http_request=http_request)
        inputs = _normalize_embedding_inputs(request.input)
        dimensions = _embedding_dimensions(request.dimensions)

        encoding_format = str(request.encoding_format or "float").strip().lower()
        if encoding_format not in {"float", "base64"}:
            raise HTTPException(
                status_code=400,
                detail={
                    "error": {
                        "message": "encoding_format must be either 'float' or 'base64'.",
                        "type": "invalid_request",
                    }
                },
            )

        if provider_lower in _DIRECT_EMBEDDING_PROVIDERS:
            request_kwargs: Dict[str, Any] = {"encoding_format": encoding_format}
            if dimensions is not None:
                request_kwargs["dimensions"] = dimensions
            if request.user:
                request_kwargs["user"] = request.user

            direct_provider = _create_direct_embedding_provider(provider_lower, model, provider_kwargs)
            result = direct_provider.embed(
                input_text=request.input if isinstance(request.input, str) else inputs,
                **request_kwargs,
            )
            if not isinstance(result, dict):
                raise ValueError(f"Invalid response from {provider_lower} embedding provider")
            result["model"] = f"{provider_lower}/{model}"

            logger.info(
                "✅ Remote embeddings generated",
                provider=provider_lower,
                count=len(result.get("data", [])) if isinstance(result.get("data"), list) else None,
                model=result.get("model"),
            )
            return result

        if encoding_format != "float":
            raise HTTPException(
                status_code=400,
                detail={
                    "error": {
                        "message": f"encoding_format='base64' is only supported for OpenAI-compatible embedding providers, not {provider_lower}.",
                        "type": "unsupported_parameter",
                    }
                },
            )

        # Route local/native providers through EmbeddingManager. strict=True is
        # important for the HTTP server: provider/model failures must surface as
        # errors, not silent zero-vector fallbacks.
        from ..embeddings.manager import EmbeddingManager

        embedder = EmbeddingManager(
            model=model,
            provider=provider_lower,
            output_dims=dimensions,
            strict=True,
            provider_kwargs=provider_kwargs,
        )

        # Generate embeddings
        embeddings = embedder.embed_batch(inputs)

        # Convert to OpenAI format
        embedding_objects = []
        for i, embedding in enumerate(embeddings):
            embedding_objects.append({
                "object": "embedding",
                "embedding": embedding,
                "index": i
            })

        # Calculate usage using centralized token utilities
        # Calculate total tokens using centralized utility
        from ..utils.token_utils import TokenUtils
        model_name = getattr(embedder, 'model_name', None)
        total_tokens = sum(TokenUtils.estimate_tokens(text, model_name) for text in inputs)

        response = {
            "object": "list",
            "data": embedding_objects,
            "model": f"{provider}/{model}",
            "usage": {
                "prompt_tokens": total_tokens,
                "total_tokens": total_tokens
            }
        }

        logger.info(
            "✅ Embeddings generated",
            provider=provider_lower,
            count=len(embedding_objects),
            dimensions=len(embeddings[0]) if embeddings else 0,
            total_tokens=total_tokens
        )

        return response

    except HTTPException:
        # Re-raise HTTP exceptions as-is
        raise
    except Exception as e:
        logger.error(f"❌ Embedding generation failed: {_redact_text(str(e))}", exc_info=True)
        raise HTTPException(
            status_code=_provider_exception_status(e),
            detail={"error": {"message": _redact_text(str(e)), "type": "embedding_error"}}
        )

# ============================================================================
# Media Processing Utilities
# ============================================================================

def handle_base64_image(data_url: str) -> str:
    """
    Process base64 data URL and save to temporary file.

    Args:
        data_url: Base64 data URL (e.g., "data:image/jpeg;base64,..." or "data:application/pdf;base64,...")

    Returns:
        Path to temporary file
    """
    try:
        # Parse data URL
        if not data_url.startswith("data:"):
            raise ValueError("Invalid data URL format")

        # Extract media type and base64 data
        header, data = data_url.split(",", 1)
        media_type = header.split(";")[0].split(":")[1]

        # Determine file extension for all supported media types
        ext_map = {
            # Images
            "image/jpeg": ".jpg",
            "image/jpg": ".jpg",
            "image/png": ".png",
            "image/gif": ".gif",
            "image/webp": ".webp",
            "image/bmp": ".bmp",
            "image/tiff": ".tiff",
            # Documents
            "application/pdf": ".pdf",
            "application/vnd.openxmlformats-officedocument.wordprocessingml.document": ".docx",
            "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet": ".xlsx",
            "application/vnd.openxmlformats-officedocument.presentationml.presentation": ".pptx",
            # Data files
            "text/csv": ".csv",
            "text/tab-separated-values": ".tsv",
            "application/json": ".json",
            "application/xml": ".xml",
            "text/xml": ".xml",
            "text/plain": ".txt",
            "text/markdown": ".md",
            # Generic fallback
            "application/octet-stream": ".bin"
        }
        extension = ext_map.get(media_type, ".bin")

        # Decode base64 data
        file_data = base64.b64decode(data)

        # Save to temporary file with request-specific prefix for better isolation
        import hashlib
        data_hash = hashlib.md5(data[:100].encode() if len(data) > 100 else data.encode()).hexdigest()[:8]
        request_id = uuid.uuid4().hex[:8]
        prefix = f"abstractcore_b64_{data_hash}_{request_id}_"

        with tempfile.NamedTemporaryFile(delete=False, suffix=extension, prefix=prefix) as temp_file:
            temp_file.write(file_data)
            temp_file_path = temp_file.name

        # Log the temporary file creation for debugging
        logger.debug(f"Processed base64 media to temporary file: {temp_file_path} (size: {len(file_data)} bytes)")
        return temp_file_path

    except Exception as e:
        logger.error(f"Failed to process base64 media: {_redact_text(str(e))}")
        raise HTTPException(
            status_code=400,
            detail={"error": {"message": f"Invalid base64 media data: {_redact_text(str(e))}", "type": "media_error"}}
        )

def download_file_temporarily(url: str) -> str:
    """
    Download file from URL to temporary file (supports images, documents, data files).

    Args:
        url: HTTP(S) URL to file

    Returns:
        Path to temporary file
    """
    # Security: prevent SSRF by default. Operators can allow specific hosts/prefixes
    # via ABSTRACTCORE_SERVER_URL_FETCH_ALLOWLIST.
    _require_public_url(url, allowlist_env="ABSTRACTCORE_SERVER_URL_FETCH_ALLOWLIST")

    max_bytes = 10 * 1024 * 1024  # 10MB per file
    max_redirects = 5

    # Determine extension from content-type or URL
    ext_map = {
        # Images
        "image/jpeg": ".jpg",
        "image/jpg": ".jpg",
        "image/png": ".png",
        "image/gif": ".gif",
        "image/webp": ".webp",
        "image/bmp": ".bmp",
        "image/tiff": ".tiff",
        # Documents
        "application/pdf": ".pdf",
        "application/vnd.openxmlformats-officedocument.wordprocessingml.document": ".docx",
        "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet": ".xlsx",
        "application/vnd.openxmlformats-officedocument.presentationml.presentation": ".pptx",
        # Data files
        "text/csv": ".csv",
        "text/tab-separated-values": ".tsv",
        "application/json": ".json",
        "application/xml": ".xml",
        "text/xml": ".xml",
        "text/plain": ".txt",
        "text/markdown": ".md",
        # Generic fallback
        "application/octet-stream": ".bin",
    }

    def _ext_from_url(u: str) -> str:
        try:
            p = urllib.parse.urlsplit(u)
            path = str(p.path or "").lower()
        except Exception:
            path = ""
        if path.endswith(".pdf"):
            return ".pdf"
        if path.endswith(".jpg") or path.endswith(".jpeg"):
            return ".jpg"
        if path.endswith(".png"):
            return ".png"
        if path.endswith(".docx"):
            return ".docx"
        if path.endswith(".xlsx"):
            return ".xlsx"
        if path.endswith(".csv"):
            return ".csv"
        if path.endswith(".md"):
            return ".md"
        if path.endswith(".txt"):
            return ".txt"
        if path.endswith(".json"):
            return ".json"
        if path.endswith(".xml"):
            return ".xml"
        return ".bin"

    # Browser-ish headers to reduce 403s on commodity hosting.
    headers = {
        "User-Agent": "Mozilla/5.0 (AbstractCore Server)",
        "Accept": "*/*",
    }

    current_url = str(url).strip()
    try:
        with httpx.Client(timeout=30.0, follow_redirects=False) as client:
            for _ in range(max_redirects + 1):
                # Re-validate each hop (redirects are common SSRF bypasses).
                _require_public_url(current_url, allowlist_env="ABSTRACTCORE_SERVER_URL_FETCH_ALLOWLIST")

                with client.stream("GET", current_url, headers=headers) as resp:
                    status = int(getattr(resp, "status_code", 0) or 0)
                    if status in {301, 302, 303, 307, 308}:
                        loc = resp.headers.get("location")
                        if not loc:
                            raise ValueError(f"Redirect without Location header (status={status})")
                        current_url = urllib.parse.urljoin(current_url, str(loc))
                        continue

                    if status >= 400:
                        body = ""
                        try:
                            body = (resp.text or "").strip()
                        except Exception:
                            body = ""
                        raise ValueError(f"HTTP {status}: {body[:200] or '(empty)'}")

                    content_len = resp.headers.get("content-length")
                    if content_len:
                        try:
                            n = int(str(content_len).strip())
                        except Exception:
                            n = -1
                        if n > max_bytes:
                            raise ValueError("File too large (max 10MB)")

                    content_type = str(resp.headers.get("content-type", "") or "").split(";", 1)[0].strip().lower()
                    extension = ext_map.get(content_type) or _ext_from_url(current_url)

                    import hashlib

                    url_hash = hashlib.md5(current_url.encode("utf-8")).hexdigest()[:8]
                    request_id = uuid.uuid4().hex[:8]
                    prefix = f"abstractcore_file_{url_hash}_{request_id}_"

                    total = 0
                    with tempfile.NamedTemporaryFile(delete=False, suffix=extension, prefix=prefix) as temp_file:
                        for chunk in resp.iter_bytes():
                            if not chunk:
                                continue
                            total += len(chunk)
                            if total > max_bytes:
                                raise ValueError("File too large (max 10MB)")
                            temp_file.write(chunk)
                        temp_file_path = temp_file.name

                    logger.info(
                        f"Downloaded file to temporary file: {temp_file_path} "
                        f"(size: {total} bytes, type: {content_type or 'unknown'})"
                    )
                    return temp_file_path

            raise ValueError("Too many redirects")
    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Failed to download file from URL {_redact_url(url)}: {_redact_text(str(e))}")
        raise HTTPException(
            status_code=400,
            detail={"error": {"message": f"Failed to download file: {_redact_text(str(e))}", "type": "media_error"}},
        ) from e

def download_image_temporarily(url: str) -> str:
    """
    Download image from URL to temporary file (backward compatibility wrapper).

    Args:
        url: HTTP(S) URL to image

    Returns:
        Path to temporary file
    """
    return download_file_temporarily(url)

def process_image_url_object(image_url_obj: Dict[str, Any]) -> Optional[str]:
    """
    Process OpenAI image_url object and return local file path.

    Args:
        image_url_obj: Image URL object with 'url' field

    Returns:
        Local file path or None if processing failed
    """
    url = str(image_url_obj.get("url", "") or "").strip()
    if not url:
        return None

    if url.startswith("data:"):
        return handle_base64_image(url)

    if url.startswith(("http://", "https://")):
        # SSRF protection is enforced in download_file_temporarily().
        return download_image_temporarily(url)

    # Local file paths are an explicit operator choice for HTTP requests.
    return _resolve_local_media_path(url)

def process_file_url_object(file_url_obj: Dict[str, Any]) -> Optional[str]:
    """
    Process OpenAI file_url object and return local file path.

    Simplified format (consistent with image_url):
    {"url": "https://example.com/file.pdf"} or
    {"url": "/local/path/file.pdf"} or
    {"url": "data:application/pdf;base64,..."}

    Args:
        file_url_obj: File URL object with 'url' field (same as image_url)

    Returns:
        Local file path or None if processing failed
    """
    try:
        # Reuse existing image URL processing logic - works perfectly for any file type
        return process_image_url_object(file_url_obj)

    except HTTPException:
        raise
    except Exception as e:
        logger.error(f"Failed to process file URL object: {_redact_text(str(e))}")
        return None

def process_message_content(message: ChatMessage) -> Tuple[str, List[str]]:
    """
    Extract media files from message content and return clean text + media list.

    Supports both OpenAI formats:
    - content as string: "Analyze this @image.jpg"
    - content as array: [{"type": "text", "text": "..."}, {"type": "image_url", "image_url": {...}}, {"type": "file", "file_url": {...}}]

    Args:
        message: ChatMessage with content to process

    Returns:
        Tuple of (clean_text, media_file_paths)
    """
    if message.content is None:
        return "", []

    if isinstance(message.content, str):
        # Legacy format: optional @filename extraction (dangerous in HTTP server contexts).
        #
        # By default, treat "@file" as plain text. Operators can opt-in by setting:
        # - ABSTRACTCORE_SERVER_MEDIA_ROOT (safe-root), and/or
        # - ABSTRACTCORE_SERVER_ALLOW_LOCAL_FILES=1 (unsafe).
        if _server_media_root() is None and not _server_allows_local_files():
            return str(message.content or ""), []

        clean_text, candidates = MessagePreprocessor.parse_file_attachments(
            str(message.content or ""),
            validate_existence=False,
            verbose=False,
        )
        resolved_files: List[str] = []
        for f in candidates:
            resolved_files.append(_resolve_local_media_path(f))
        return clean_text, resolved_files

    elif isinstance(message.content, list):
        # OpenAI array format: extract image_url objects
        text_parts = []
        media_files = []

        for item in message.content:
            if isinstance(item, dict):
                item_type = item.get("type")
                if item_type == "text" and item.get("text"):
                    text_parts.append(item["text"])
                elif item_type == "image_url" and item.get("image_url"):
                    media_file = process_image_url_object(item["image_url"])
                    if media_file:
                        media_files.append(media_file)
                elif item_type == "file" and item.get("file_url"):
                    media_file = process_file_url_object(item["file_url"])
                    if media_file:
                        media_files.append(media_file)
            elif hasattr(item, 'type'):
                # Pydantic ContentItem object
                if item.type == "text" and item.text:
                    text_parts.append(item.text)
                elif item.type == "image_url" and item.image_url:
                    media_file = process_image_url_object(item.image_url)
                    if media_file:
                        media_files.append(media_file)
                elif item.type == "file" and item.file_url:
                    media_file = process_file_url_object(item.file_url)
                    if media_file:
                        media_files.append(media_file)

        return " ".join(text_parts), media_files

    return str(message.content), []

def adapt_prompt_for_media_types(text: str, media_files: List[str]) -> str:
    """
    Intelligently adapt prompts based on attached media file types.

    Fixes common mismatches like "What is in this image?" when sending documents.

    Args:
        text: Original text content
        media_files: List of media file paths

    Returns:
        Adapted text content
    """
    if not media_files or not text:
        return text

    # Analyze media file types
    image_extensions = {'.jpg', '.jpeg', '.png', '.gif', '.webp', '.bmp', '.tiff'}
    document_extensions = {'.pdf', '.docx', '.xlsx', '.pptx'}
    data_extensions = {'.csv', '.tsv', '.json', '.xml'}
    text_extensions = {'.txt', '.md'}

    has_images = False
    has_documents = False
    has_data = False
    has_text = False

    for file_path in media_files:
        ext = os.path.splitext(file_path)[1].lower()
        if ext in image_extensions:
            has_images = True
        elif ext in document_extensions:
            has_documents = True
        elif ext in data_extensions:
            has_data = True
        elif ext in text_extensions:
            has_text = True

    # Common prompt adaptations
    adapted_text = text

    # Fix "What is in this image?" when not dealing with images
    if "what is in this image" in text.lower():
        if has_documents and not has_images:
            adapted_text = text.replace("What is in this image?", "What is in this document?")
            adapted_text = adapted_text.replace("what is in this image?", "what is in this document?")
            adapted_text = adapted_text.replace("What is in this image", "What is in this document")
            adapted_text = adapted_text.replace("what is in this image", "what is in this document")
        elif has_data and not has_images:
            adapted_text = text.replace("What is in this image?", "What data is in this file?")
            adapted_text = adapted_text.replace("what is in this image?", "what data is in this file?")
            adapted_text = adapted_text.replace("What is in this image", "What data is in this file")
            adapted_text = adapted_text.replace("what is in this image", "what data is in this file")
        elif has_text and not has_images:
            adapted_text = text.replace("What is in this image?", "What is in this text file?")
            adapted_text = adapted_text.replace("what is in this image?", "what is in this text file?")
            adapted_text = adapted_text.replace("What is in this image", "What is in this text file")
            adapted_text = adapted_text.replace("what is in this image", "what is in this text file")

    # Fix "What is in this document?" when dealing with images
    elif "what is in this document" in text.lower() and has_images and not (has_documents or has_data or has_text):
        adapted_text = text.replace("What is in this document?", "What is in this image?")
        adapted_text = adapted_text.replace("what is in this document?", "what is in this image?")
        adapted_text = adapted_text.replace("What is in this document", "What is in this image")
        adapted_text = adapted_text.replace("what is in this document", "what is in this image")

    # Handle mixed content with specific naming
    if adapted_text != text:
        # Count media types for better description
        total_files = len(media_files)
        if total_files > 1:
            types = []
            if has_images:
                types.append("image(s)")
            if has_documents:
                types.append("document(s)")
            if has_data:
                types.append("data file(s)")
            if has_text:
                types.append("text file(s)")

            if len(types) > 1:
                adapted_text = adapted_text.replace("this document", f"these {' and '.join(types)}")
                adapted_text = adapted_text.replace("this image", f"these {' and '.join(types)}")
                adapted_text = adapted_text.replace("this file", f"these {' and '.join(types)}")

    if adapted_text != text:
        logger.info(f"Adapted prompt for media types: '{text}' → '{adapted_text}'")

    return adapted_text

def validate_media_files(files: List[str]) -> None:
    """
    Validate media files for security and size limits.

    Args:
        files: List of file paths to validate

    Raises:
        HTTPException: If validation fails
    """
    ALLOWED_EXTENSIONS = {'.jpg', '.jpeg', '.png', '.gif', '.webp', '.bmp', '.tiff',
                         '.pdf', '.docx', '.xlsx', '.pptx', '.csv', '.tsv', '.txt', '.md',
                         '.json', '.xml'}

    total_size = 0
    max_total_size = 32 * 1024 * 1024  # 32MB total limit

    for file_path in files:
        if not os.path.exists(file_path):
            raise HTTPException(
                status_code=400,
                detail={"error": {"message": f"File not found: {file_path}", "type": "file_not_found"}}
            )

        # Check extension
        ext = os.path.splitext(file_path)[1].lower()
        if ext not in ALLOWED_EXTENSIONS:
            raise HTTPException(
                status_code=400,
                detail={"error": {"message": f"File type {ext} not allowed", "type": "invalid_file_type"}}
            )

        # Check individual file size (10MB per file)
        file_size = os.path.getsize(file_path)
        if file_size > 10 * 1024 * 1024:
            raise HTTPException(
                status_code=400,
                detail={"error": {"message": f"File too large: {file_path} (max 10MB per file)", "type": "file_too_large"}}
            )

        total_size += file_size

        # Check total size across all files
        if total_size > max_total_size:
            raise HTTPException(
                status_code=400,
                detail={"error": {"message": "Total file size exceeds 32MB limit", "type": "total_size_exceeded"}}
            )

@app.post("/v1/chat/completions", tags=["chat"])
async def chat_completions(request: ChatCompletionRequest, http_request: Request):
    """
    Create a model response for the given chat conversation with optional media attachments.

    Given a list of messages comprising a conversation, the model will return a response.
    This endpoint supports streaming, tool calling, media attachments, and multiple providers.

    **Key Features:**
    - Multi-provider support (OpenAI, Anthropic, Ollama, LMStudio, etc.)
    - Streaming responses with server-sent events
    - Tool/function calling with automatic syntax conversion
    - Media attachments (images, documents, data files)
    - OpenAI Vision API compatible format

    **Provider Format:** Use `provider/model` format in the model field:
    - `openai/gpt-4` - OpenAI GPT-4
    - `ollama/llama3:latest` - Ollama LLaMA 3
    - `anthropic/claude-3-opus-20240229` - Anthropic Claude 3 Opus

    **Media Attachments:** Support for OpenAI Vision API compatible format:
    - String content: "Analyze this @image.jpg" (AbstractCore @filename syntax)
    - Array content: [{"type": "text", "text": "..."}, {"type": "image_url", "image_url": {"url": "data:image/jpeg;base64,..."}}]
    - Supported formats: Images (PNG, JPEG, GIF, WEBP), Documents (PDF, DOCX, XLSX, PPTX), Data (CSV, TSV, TXT, MD)
    - Size limits: 10MB per file, 32MB total per request

    **To see available text models:** `GET /v1/models?input_type=text&output_type=text`

    **Returns:** A chat completion object, or a stream of chat completion chunks if streaming is enabled.
    """
    provider, model = parse_model_string(request.model)
    return await process_chat_completion(provider, model, request, http_request)

@app.post("/{provider}/v1/chat/completions", tags=["chat"])
async def provider_chat_completions(
    provider: Annotated[
        str,
        FastAPIPath(
            description="Provider route prefix, e.g. `openai`, `anthropic`, `ollama`, `openrouter`, or `openai-compatible`."
        ),
    ],
    request: ChatCompletionRequest,
    http_request: Request
):
    """
    Provider-specific chat completions endpoint.

    Same functionality as `/v1/chat/completions` but allows specifying the provider in the URL path.
    Useful when you want explicit provider routing or when the model name doesn't include the provider prefix.

    **Examples:**
    - `POST /ollama/v1/chat/completions` with `"model": "llama3:latest"`
    - `POST /openai/v1/chat/completions` with `"model": "gpt-4"`

    **Note:** Provider in the URL takes precedence over provider in the model name.
    """
    _, model = parse_model_string(request.model)
    return await process_chat_completion(provider, model, request, http_request)


def _extract_trace_metadata(http_request: Request) -> Dict[str, Any]:
    """Extract trace metadata from request headers (schema-safe)."""
    meta: Dict[str, Any] = {}

    raw = (
        http_request.headers.get("x-abstractcore-trace-metadata")
        or http_request.headers.get("x-abstract-trace-metadata")
    )
    if raw:
        try:
            parsed = json.loads(raw)
            if isinstance(parsed, dict):
                meta.update(parsed)
        except Exception:
            # Ignore invalid metadata payloads; tracing is best-effort.
            pass

    header_map = {
        "actor_id": "x-abstractcore-actor-id",
        "session_id": "x-abstractcore-session-id",
        "run_id": "x-abstractcore-run-id",
        "parent_run_id": "x-abstractcore-parent-run-id",
    }
    for key, header in header_map.items():
        val = http_request.headers.get(header)
        if val is not None and key not in meta:
            meta[key] = val

    # Never log or return these directly; they are for internal correlation only.
    return meta


async def process_chat_completion(
    provider: str,
    model: str,
    request: ChatCompletionRequest,
    http_request: Request,
    *,
    openai_output_format: Literal["chat_completions", "responses"] = "chat_completions",
    openai_responses_request: Optional["OpenAIResponsesRequest"] = None,
):
    """
    Core chat completion processing with syntax rewriting support.
    """
    request_id = uuid.uuid4().hex[:8]

    try:
        if openai_output_format == "responses" and openai_responses_request is None:
            raise HTTPException(
                status_code=400,
                detail={"error": {"message": "Internal error: missing OpenAIResponsesRequest for responses output.", "type": "server_error"}},
            )

        logger.info(
            "📥 Chat Completion Request",
            request_id=request_id,
            provider=provider,
            model=model,
            messages=len(request.messages),
            has_tools=bool(request.tools),
            stream=request.stream,
            openai_output_format=openai_output_format,
        )

        _reject_body_api_key(request.api_key)
        provider_api_key = _provider_api_key_from_request(http_request)

        # Detect target format for tool call syntax
        target_format = detect_target_format(f"{provider}/{model}", request, http_request)
        user_agent_raw = http_request.headers.get("user-agent", "")
        user_agent = str(user_agent_raw or "")
        if len(user_agent) > 50:
            #[WARNING:TRUNCATION] bounded user-agent capture for request logs
            user_agent = user_agent[:50].rstrip() + "…"
        logger.info(
            "🎯 Target Format Detected",
            request_id=request_id,
            target_format=target_format.value,
            user_agent=user_agent,
        )

        # Process media from messages
        all_media_files = []
        processed_messages = []

        for message in request.messages:
            clean_text, media_files = process_message_content(message)
            all_media_files.extend(media_files)

            # Adapt prompt based on media file types to avoid confusion
            if media_files:
                adapted_text = adapt_prompt_for_media_types(clean_text, media_files)
            else:
                adapted_text = clean_text

            # Create processed message with adapted text
            processed_message = message.model_copy()
            processed_message.content = adapted_text
            processed_messages.append(processed_message)

        # Validate media files if any were found
        if all_media_files:
            validate_media_files(all_media_files)
            #[WARNING:TRUNCATION] bounded filename preview for request logs
            files_preview = [os.path.basename(f) for f in all_media_files[:5]]
            logger.info(
                "📎 Media Files Processed",
                request_id=request_id,
                file_count=len(all_media_files),
                files=files_preview,
                files_truncated=len(all_media_files) > 5,
            )

        # Create LLM instance
        # Prepare provider-specific kwargs
        provider_kwargs = {}
        trace_metadata = _extract_trace_metadata(http_request)
        if trace_metadata:
            # Enable trace capture (trace_id) without retaining full trace buffers by default.
            provider_kwargs["enable_tracing"] = True
            provider_kwargs.setdefault("max_traces", 0)
        if provider_api_key:
            provider_kwargs["api_key"] = provider_api_key
            logger.debug("🔑 Explicit provider API key supplied", request_id=request_id, provider=provider)
        _guard_unauthenticated_server_provider_key_use(
            provider,
            explicit_provider_key=bool(provider_api_key),
            http_request=http_request,
        )
        if isinstance(request.base_url, str) and request.base_url.strip():
            base_url = request.base_url.strip()
            if not _server_allows_request_base_url(base_url):
                raise HTTPException(
                    status_code=403,
                    detail={
                        "error": {
                            "message": (
                                "Request-level base_url overrides are restricted for security. "
                                "By default only loopback URLs are allowed. "
                                "To allow additional hosts/prefixes, set "
                                "ABSTRACTCORE_SERVER_BASE_URL_ALLOWLIST."
                            ),
                            "type": "forbidden",
                        }
                    },
                )

            # Prevent accidental provider key exfiltration when the server has an env API key.
            # If a client sets a non-loopback base_url and does not provide an explicit provider key,
            # providers may fall back to server env keys and send them to the supplied endpoint.
            parsed = urllib.parse.urlsplit(base_url)
            host = str(parsed.hostname or "").strip()
            if host and not _is_loopback_host(host):
                env_var = _provider_api_key_env_var(provider)
                explicit_key = bool(provider_api_key)
                if _server_has_provider_api_key(provider) and not explicit_key:
                    raise HTTPException(
                        status_code=403,
                        detail={
                            "error": {
                                "message": (
                                    "Refusing request-level base_url override without an explicit provider key because "
                                    f"the server has {env_var} set. Provide the provider key via "
                                    "X-AbstractCore-Provider-API-Key. "
                                    f"(provider={provider})"
                                ),
                                "type": "forbidden",
                            }
                        },
                    )

            provider_kwargs["base_url"] = base_url
            logger.info("🔗 Custom Base URL", request_id=request_id, base_url=_redact_url(base_url))
        if request.timeout_s is not None:
            # Orchestrator policy: allow the caller to specify the provider timeout.
            # Note: BaseProvider treats non-positive values as "unlimited".
            provider_kwargs["timeout"] = request.timeout_s

        provider_normalized = provider.strip().lower()
        unload_after_requested = bool(getattr(request, "unload_after", False))
        allow_unsafe_unload_after = _parse_bool_env("ABSTRACTCORE_ALLOW_UNSAFE_UNLOAD_AFTER")
        if unload_after_requested and provider_normalized == "ollama" and not allow_unsafe_unload_after:
            raise HTTPException(
                status_code=403,
                detail={
                    "error": {
                        "message": (
                            "unload_after=true is disabled for provider 'ollama' because it can unload shared server "
                            "state and disrupt other clients. Set ABSTRACTCORE_ALLOW_UNSAFE_UNLOAD_AFTER=1 to enable."
                        ),
                        "type": "forbidden",
                    }
                },
            )

        loaded_runtime = _get_loaded_gateway_runtime(
            provider=provider_normalized,
            model=model,
            base_url=provider_kwargs.get("base_url"),
            explicit_provider_key_hash=_provider_key_fingerprint(provider_api_key),
        )
        llm = loaded_runtime.llm if loaded_runtime is not None else create_llm(provider, model=model, **provider_kwargs)
        ollama_key: Optional[Tuple[str, str, str]] = None
        if loaded_runtime is None and provider_normalized == "ollama":
            base_url_for_key = provider_kwargs.get("base_url")
            if not isinstance(base_url_for_key, str) or not base_url_for_key.strip():
                base_url_for_key = request.base_url
            ollama_key = _ollama_inflight_key(provider, base_url_for_key, model)
            _ollama_inflight_enter(ollama_key)

        # Convert messages
        messages = convert_to_abstractcore_messages(processed_messages)

        # Create syntax rewriter
        syntax_rewriter = create_syntax_rewriter(target_format, f"{provider}/{model}")

        # Prepare generation parameters
        gen_kwargs = {
            "prompt": "",  # Empty when using messages
            "messages": messages,
            "media": all_media_files if all_media_files else None,  # Add media files
            "temperature": request.temperature,
            "max_tokens": request.max_tokens,
            "stream": request.stream,
            "tools": request.tools,
            "tool_choice": request.tool_choice if request.tools else None,
            "execute_tools": False,  # Server mode - don't execute tools
        }
        if trace_metadata:
            gen_kwargs["trace_metadata"] = trace_metadata

        # Add optional parameters
        if request.thinking is not None:
            gen_kwargs["thinking"] = request.thinking
        if request.stop:
            gen_kwargs["stop"] = request.stop
        if request.seed:
            gen_kwargs["seed"] = request.seed
        if request.frequency_penalty:
            gen_kwargs["frequency_penalty"] = request.frequency_penalty
        if request.presence_penalty:
            gen_kwargs["presence_penalty"] = request.presence_penalty
        if isinstance(request.prompt_cache_key, str) and request.prompt_cache_key.strip():
            gen_kwargs["prompt_cache_key"] = request.prompt_cache_key.strip()
        if isinstance(request.prompt_cache_retention, str) and request.prompt_cache_retention.strip():
            gen_kwargs["prompt_cache_retention"] = request.prompt_cache_retention.strip()
        _apply_request_prompt_cache_binding(
            llm=llm,
            gen_kwargs=gen_kwargs,
            binding=request.prompt_cache_binding,
            loaded_runtime=loaded_runtime,
        )

        # Generate response
        # Only cleanup files created by this request (with our specific prefixes)
        temp_files_to_cleanup = [
            f for f in all_media_files
            if f.startswith("/tmp/") and (
                "abstractcore_img_" in f or
                "abstractcore_file_" in f or
                "abstractcore_b64_" in f or
                "temp" in f
            )
        ]

        try:
            if request.stream:
                return StreamingResponse(
                    (
                        generate_streaming_responses_response(
                            llm,
                            gen_kwargs,
                            provider,
                            model,
                            syntax_rewriter,
                            request_id,
                            openai_responses_request,  # type: ignore[arg-type]
                            temp_files_to_cleanup,
                            unload_after=unload_after_requested,
                            ollama_key=ollama_key,
                            allow_unsafe_unload_after=allow_unsafe_unload_after,
                            loaded_runtime=loaded_runtime,
                        )
                        if openai_output_format == "responses"
                        else generate_streaming_response(
                            llm,
                            gen_kwargs,
                            provider,
                            model,
                            syntax_rewriter,
                            request_id,
                            temp_files_to_cleanup,
                            unload_after=unload_after_requested,
                            ollama_key=ollama_key,
                            allow_unsafe_unload_after=allow_unsafe_unload_after,
                            loaded_runtime=loaded_runtime,
                        )
                    ),
                    media_type="text/event-stream",
                    headers={"Cache-Control": "no-cache", "Connection": "keep-alive"}
                )
            else:
                if loaded_runtime is not None:
                    response = _run_loaded_gateway_runtime(
                        loaded_runtime,
                        lambda: llm.generate(**gen_kwargs),
                    )
                else:
                    response = llm.generate(**gen_kwargs)
                if openai_output_format == "responses":
                    openai_response = convert_to_openai_responses_response(
                        response,
                        provider,
                        model,
                        syntax_rewriter,
                        request_id,
                        openai_responses_request,  # type: ignore[arg-type]
                    )
                else:
                    openai_response = convert_to_openai_response(
                        response, provider, model, syntax_rewriter, request_id
                    )
                trace_id = None
                if hasattr(response, "metadata") and isinstance(getattr(response, "metadata"), dict):
                    trace_id = response.metadata.get("trace_id")
                if trace_id:
                    return JSONResponse(
                        content=openai_response,
                        headers={"X-AbstractCore-Trace-Id": str(trace_id)},
                    )
                return openai_response
        finally:
            if not request.stream:
                if loaded_runtime is not None and unload_after_requested:
                    _best_effort_unload_loaded_gateway_runtime(loaded_runtime, request_id=request_id)
                elif provider_normalized == "ollama" and ollama_key is not None:
                    should_unload = _ollama_inflight_exit(ollama_key, unload_after_requested=unload_after_requested)
                    if should_unload and allow_unsafe_unload_after:
                        _best_effort_unload(llm, request_id=request_id, provider=provider, model=model)
                    elif should_unload:
                        logger.warning(
                            "⚠️ Unload requested but disabled by server policy",
                            request_id=request_id,
                            provider=provider,
                            model=model,
                        )
                elif unload_after_requested:
                    _best_effort_unload(llm, request_id=request_id, provider=provider, model=model)

            # Cleanup temporary files (base64 and downloaded images) with delay to avoid race conditions
            def delayed_cleanup():
                """Cleanup temporary files after a short delay to avoid race conditions"""
                time.sleep(1)  # Short delay to ensure generation is complete
                for temp_file in temp_files_to_cleanup:
                    try:
                        if os.path.exists(temp_file):
                            # Additional check: only delete files created by this session
                            if ("abstractcore_img_" in temp_file or "abstractcore_file_" in temp_file or "abstractcore_b64_" in temp_file):
                                os.unlink(temp_file)
                                logger.debug(f"Cleaned up temporary file: {temp_file}")
                            else:
                                logger.debug(f"Skipped cleanup of non-AbstractCore file: {temp_file}")
                    except Exception as e:
                        logger.warning(f"Failed to cleanup temporary file {temp_file}: {e}")

            # Run cleanup in background thread to avoid blocking response
            cleanup_thread = threading.Thread(target=delayed_cleanup, daemon=True)
            cleanup_thread.start()

    except HTTPException:
        raise
    except Exception as e:
        logger.error(
            "❌ Chat completion failed",
            request_id=request_id,
            error=str(e),
            error_type=type(e).__name__
        )
        raise HTTPException(
            status_code=500,
            detail={"error": {"message": str(e), "type": "server_error"}}
        )

def generate_streaming_response(
    llm,
    gen_kwargs: Dict[str, Any],
    provider: str,
    model: str,
    syntax_rewriter: ToolCallSyntaxRewriter,
    request_id: str,
    temp_files_to_cleanup: List[str] = None,
    *,
    unload_after: bool = False,
    ollama_key: Optional[Tuple[str, str, str]] = None,
    allow_unsafe_unload_after: bool = False,
    loaded_runtime: Optional[_GatewayLoadedRuntime] = None,
) -> Iterator[str]:
    """Generate OpenAI-compatible streaming response with syntax rewriting."""
    provider_normalized = provider.strip().lower()
    try:
        chat_id = f"chatcmpl-{uuid.uuid4().hex[:8]}"
        created_time = int(time.time())
        has_tool_calls = False

        def _iter_chunks() -> Iterator[Any]:
            if loaded_runtime is not None:
                for item in _stream_loaded_gateway_runtime(
                    loaded_runtime,
                    lambda: llm.generate(**gen_kwargs),
                ):
                    yield item
                return
            for item in llm.generate(**gen_kwargs):
                yield item

        for chunk in _iter_chunks():
            # Content streaming
            if hasattr(chunk, 'content') and chunk.content:
                content = chunk.content

                # For OpenAI/Codex format: only clean if content contains tool calls
                # For other formats: apply syntax rewriting
                if syntax_rewriter.target_format in [SyntaxFormat.OPENAI, SyntaxFormat.CODEX]:
                    # Only clean content if it contains tool call patterns
                    # This prevents stripping spaces from regular text chunks
                    if any(pattern in content for pattern in ['<function_call>', '<tool_call>', '<|tool_call|>', '```tool_code']):
                        content = syntax_rewriter.remove_tool_call_patterns(content)
                elif syntax_rewriter.target_format != SyntaxFormat.PASSTHROUGH:
                    # Apply format-specific rewriting for non-OpenAI formats
                    content = syntax_rewriter.rewrite_content(content)

                # Only send content if it's meaningful (not just whitespace)
                if content.strip():
                    openai_chunk = {
                        "id": chat_id,
                        "object": "chat.completion.chunk",
                        "created": created_time,
                        "model": f"{provider}/{model}",
                        "choices": [{
                            "index": 0,
                            "delta": {"content": content},
                            "finish_reason": None
                        }]
                    }
                    yield f"data: {json.dumps(openai_chunk)}\n\n"

            # Tool calls
            if hasattr(chunk, 'tool_calls') and chunk.tool_calls:
                has_tool_calls = True
                # OpenAI/Codex clients expect structured tool_calls deltas.
                if syntax_rewriter.target_format in [SyntaxFormat.OPENAI, SyntaxFormat.CODEX]:
                    openai_tool_calls = syntax_rewriter.convert_to_openai_format(chunk.tool_calls)

                    for i, openai_tool_call in enumerate(openai_tool_calls):
                        tool_chunk = {
                            "id": chat_id,
                            "object": "chat.completion.chunk",
                            "created": created_time,
                            "model": f"{provider}/{model}",
                            "choices": [{
                                "index": 0,
                                "delta": {
                                    "tool_calls": [{
                                        "index": i,  # Proper indexing for multiple tools
                                        "id": openai_tool_call["id"],
                                        "type": "function",
                                        "function": openai_tool_call["function"]
                                    }]
                                },
                                "finish_reason": "tool_calls"  # Critical for Codex
                            }]
                        }
                        yield f"data: {json.dumps(tool_chunk)}\n\n"
                # Tag-based clients parse tool calls from assistant content.
                elif syntax_rewriter.target_format != SyntaxFormat.PASSTHROUGH:
                    tool_text = syntax_rewriter.rewrite_content("", detected_tool_calls=chunk.tool_calls)
                    if tool_text and tool_text.strip():
                        openai_chunk = {
                            "id": chat_id,
                            "object": "chat.completion.chunk",
                            "created": created_time,
                            "model": f"{provider}/{model}",
                            "choices": [{
                                "index": 0,
                                "delta": {"content": tool_text},
                                "finish_reason": None
                            }]
                        }
                        yield f"data: {json.dumps(openai_chunk)}\n\n"

        # Final chunk
        final_chunk = {
            "id": chat_id,
            "object": "chat.completion.chunk",
            "created": created_time,
            "model": f"{provider}/{model}",
            "choices": [{
                "index": 0,
                "delta": {},
                "finish_reason": "tool_calls" if has_tool_calls else "stop"
            }]
        }
        yield f"data: {json.dumps(final_chunk)}\n\n"
        yield "data: [DONE]\n\n"

        logger.info(
            "✅ Streaming completed",
            request_id=request_id,
            has_tool_calls=has_tool_calls
        )

        # Cleanup temporary files for streaming with delay to avoid race conditions
        if temp_files_to_cleanup:
            import threading

            def delayed_streaming_cleanup():
                """Cleanup temporary files after streaming completes"""
                time.sleep(2)  # Longer delay for streaming to ensure all chunks are sent
                for temp_file in temp_files_to_cleanup:
                    try:
                        if os.path.exists(temp_file):
                            # Additional check: only delete files created by this session
                            if ("abstractcore_img_" in temp_file or "abstractcore_file_" in temp_file or "abstractcore_b64_" in temp_file):
                                os.unlink(temp_file)
                                logger.debug(f"Cleaned up temporary file during streaming: {temp_file}")
                            else:
                                logger.debug(f"Skipped cleanup of non-AbstractCore streaming file: {temp_file}")
                    except Exception as cleanup_error:
                        logger.warning(f"Failed to cleanup temporary file {temp_file}: {cleanup_error}")

            # Run cleanup in background thread
            cleanup_thread = threading.Thread(target=delayed_streaming_cleanup, daemon=True)
            cleanup_thread.start()

    except Exception as e:
        logger.error(
            "❌ Streaming failed",
            request_id=request_id,
            error=str(e)
        )
        error_chunk = {"error": {"message": str(e), "type": "server_error"}}
        yield f"data: {json.dumps(error_chunk)}\n\n"
    finally:
        if loaded_runtime is not None and unload_after:
            _best_effort_unload_loaded_gateway_runtime(loaded_runtime, request_id=request_id)
        elif provider_normalized == "ollama" and ollama_key is not None:
            try:
                should_unload = _ollama_inflight_exit(ollama_key, unload_after_requested=unload_after)
            except Exception as e:
                logger.warning(
                    "⚠️ Failed to update in-flight unload state",
                    request_id=request_id,
                    provider=provider,
                    model=model,
                    error=str(e),
                    error_type=type(e).__name__,
                )
                should_unload = False

            if should_unload and allow_unsafe_unload_after:
                _best_effort_unload(llm, request_id=request_id, provider=provider, model=model)
            elif should_unload:
                logger.warning(
                    "⚠️ Unload requested but disabled by server policy",
                    request_id=request_id,
                    provider=provider,
                    model=model,
                )
        elif unload_after:
            _best_effort_unload(llm, request_id=request_id, provider=provider, model=model)

def convert_to_openai_response(
    response,
    provider: str,
    model: str,
    syntax_rewriter: ToolCallSyntaxRewriter,
    request_id: str
) -> Dict[str, Any]:
    """Convert AbstractCore response to OpenAI format with syntax rewriting."""

    if response is None:
        logger.warning("Received None response", request_id=request_id)
        return {
            "id": f"chatcmpl-{uuid.uuid4().hex[:8]}",
            "object": "chat.completion",
            "created": int(time.time()),
            "model": f"{provider}/{model}",
            "choices": [{
                "index": 0,
                "message": {"role": "assistant", "content": "Error: No response generated"},
                "finish_reason": "error"
            }],
            "usage": {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}
        }

    # Apply syntax rewriting to content
    content = response.content if hasattr(response, 'content') else str(response)
    tool_calls = getattr(response, "tool_calls", None) if hasattr(response, "tool_calls") else None

    # For OpenAI/Codex format: only clean if content contains tool calls
    # For other formats: apply syntax rewriting
    if syntax_rewriter.target_format in [SyntaxFormat.OPENAI, SyntaxFormat.CODEX]:
        # Only clean content if it contains tool call patterns
        # This prevents stripping spaces from regular text
        if any(pattern in content for pattern in ['<function_call>', '<tool_call>', '<|tool_call|>', '```tool_code']):
            content = syntax_rewriter.remove_tool_call_patterns(content)
    elif syntax_rewriter.target_format != SyntaxFormat.PASSTHROUGH:
        # Apply format-specific rewriting for non-OpenAI formats.
        # Prefer structured tool_calls when present (content may be cleaned upstream).
        if tool_calls:
            content = syntax_rewriter.rewrite_content(content, detected_tool_calls=tool_calls)
        else:
            content = syntax_rewriter.rewrite_content(content)

    response_dict = {
        "id": f"chatcmpl-{uuid.uuid4().hex[:8]}",
        "object": "chat.completion",
        "created": int(time.time()),
        "model": f"{provider}/{model}",
        "choices": [{
            "index": 0,
            "message": {"role": "assistant", "content": content},
            "finish_reason": "stop"
        }],
        "usage": {
            "prompt_tokens": getattr(response, 'usage', {}).get('prompt_tokens', 0) if hasattr(response, 'usage') else 0,
            "completion_tokens": getattr(response, 'usage', {}).get('completion_tokens', 0) if hasattr(response, 'usage') else 0,
            "total_tokens": getattr(response, 'usage', {}).get('total_tokens', 0) if hasattr(response, 'usage') else 0
        }
    }

    # Add tool calls if present
    if tool_calls:
        openai_tool_calls = syntax_rewriter.convert_to_openai_format(tool_calls)
        response_dict["choices"][0]["message"]["tool_calls"] = openai_tool_calls
        response_dict["choices"][0]["finish_reason"] = "tool_calls"

        logger.info(
            "🔧 Tool calls converted",
            request_id=request_id,
            tool_count=len(tool_calls),
            target_format=syntax_rewriter.target_format.value
        )

    return response_dict

# ============================================================================
# Server Runner
# ============================================================================

def _resolve_external_host(bind_host: str) -> str:
    """Best-effort external host for display when binding to 0.0.0.0."""
    if bind_host in {"0.0.0.0", "::", "[::]"}:
        try:
            with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
                sock.connect(("8.8.8.8", 80))
                return sock.getsockname()[0]
        except Exception:
            return bind_host
    return bind_host

def run_server(host: str = "0.0.0.0", port: int = 8000):
    """Run the server"""
    import uvicorn
    uvicorn.run(app, host=host, port=port, log_level="error")

# ============================================================================
# Server Runner Function
# ============================================================================

def run_server_with_args(argv: Optional[List[str]] = None, *, prog: Optional[str] = None):
    """Run the server with argument parsing for CLI usage."""
    parser = argparse.ArgumentParser(
        prog=prog,
        description="AbstractCore Server - Universal LLM Gateway with Media Processing",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  abstractcore serve                                  # Start server with defaults
  abstractcore serve --host 127.0.0.1 --port 8080    # Custom host/port
  abstractcore serve --reload                        # Development auto-reload
  python -m abstractcore.server.app                    # Start server with defaults
  python -m abstractcore.server.app --debug           # Start with debug logging
  python -m abstractcore.server.app --host 127.0.0.1 --port 8080  # Custom host/port
  python -m abstractcore.server.app --debug --port 8080           # Debug on custom port

Environment Variables:
  ABSTRACTCORE_AUTH_TOKEN=...  # Server auth token for non-health endpoints
  ABSTRACTCORE_DEBUG=true    # Enable debug mode (equivalent to --debug)
  HOST=127.0.0.1            # Server host (overridden by --host)
  PORT=8080                  # Server port (overridden by --port)

Debug Mode:
  The --debug flag enables verbose logging and better error reporting, including:
  - Detailed HTTP request/response logging
  - Full error traces for 422 Unprocessable Entity errors
  - Media processing diagnostics
  - Provider initialization details
        """
    )

    parser.add_argument(
        '--debug',
        action='store_true',
        help='Enable debug logging and show detailed diagnostics (overrides centralized config)'
    )
    parser.add_argument(
        '--host',
        default=os.getenv("HOST", "0.0.0.0"),
        help='Host to bind the server to (default: 0.0.0.0)'
    )
    parser.add_argument(
        '--port',
        type=int,
        default=int(os.getenv("PORT", "8000")),
        help='Port to bind the server to (default: 8000)'
    )
    parser.add_argument(
        '--reload',
        action='store_true',
        help='Enable uvicorn auto-reload (development only)'
    )

    args = parser.parse_args(argv)

    # Reconfigure logging if debug mode is requested (--debug overrides config defaults)
    if args.debug:
        reconfigure_for_debug()
        print("🐛 Debug mode enabled - detailed logging active")

    logger.info(
        "🚀 Starting AbstractCore Server",
        host=args.host,
        port=args.port,
        debug=debug_mode,
        version=__version__
    )

    # Print access URLs (outside logging)
    internal_url = f"http://127.0.0.1:{args.port}"
    external_host = _resolve_external_host(args.host)
    external_url = f"http://{external_host}:{args.port}"
    print(f"Internal URL: {internal_url}")
    print(f"External URL: {external_url}")

    # Enhanced uvicorn configuration for debug mode
    uvicorn_config = {
        "app": "abstractcore.server.app:app" if args.reload else app,
        "host": args.host,
        "port": args.port,
        "log_level": "debug" if debug_mode else "error",
        "reload": bool(args.reload),
    }

    # In debug mode, enable more detailed uvicorn logging
    if debug_mode:
        uvicorn_config.update({
            "access_log": True,
            "use_colors": True,
        })

    import uvicorn
    uvicorn.run(**uvicorn_config)

# ============================================================================
# Startup
# ============================================================================

if __name__ == "__main__":
    run_server_with_args()
