From 6a25d8ac0889601521b84c4d6502782c5c86a465 Mon Sep 17 00:00:00 2001 From: infosave2007 Date: Tue, 30 Jun 2026 17:42:45 +0300 Subject: [PATCH] feat: finish remaining inbound protocols (models, completions, native /route) - GET /v1/models: pool + virtual cortiq-auto - POST /v1/completions (legacy): non-stream + streaming text_completion - POST /route: native passthrough returning the routing decision - all inbound protocol adapters now implemented; admin meta + README/CHANGELOG updated - bump to 0.2.1 --- CHANGELOG.md | 11 ++ Cargo.lock | 2 +- Cargo.toml | 2 +- README.md | 9 +- README.ru.md | 10 +- src/admin/mod.rs | 4 +- src/protocols/mod.rs | 7 +- src/protocols/native_passthrough.rs | 86 +++++++++++++ src/protocols/openai_completions.rs | 183 ++++++++++++++++++++++++++++ src/protocols/openai_models.rs | 32 +++++ 10 files changed, 331 insertions(+), 15 deletions(-) create mode 100644 src/protocols/native_passthrough.rs create mode 100644 src/protocols/openai_completions.rs create mode 100644 src/protocols/openai_models.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index e8371f1..87b71c0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,17 @@ All notable changes to this project are documented here. The format is based on [Keep a Changelog](https://keepachangelog.com/). +## [0.2.1] - 2026-06-30 + +### Added +- **OpenAI Models** (`GET /v1/models`): lists the pool + the virtual `cortiq-auto`. +- **OpenAI Completions (legacy)** (`POST /v1/completions`): non-streaming and streaming + (`text_completion`), routed through the same pipeline. +- **Native passthrough** (`POST /route`): returns the routing decision (task, complexity, + candidate models) without calling a model. + +All inbound protocol adapters are now implemented. + ## [0.2.0] - 2026-06-30 ### Added diff --git a/Cargo.lock b/Cargo.lock index 32048bc..e4d2aad 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -285,7 +285,7 @@ checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570" [[package]] name = "cortiq-gateway" -version = "0.2.0" +version = "0.2.1" dependencies = [ "anyhow", "arc-swap", diff --git a/Cargo.toml b/Cargo.toml index 7f05fb7..1f87055 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "cortiq-gateway" -version = "0.2.0" +version = "0.2.1" edition = "2021" rust-version = "1.75" license = "Apache-2.0" diff --git a/README.md b/README.md index 0616228..04b2c7f 100644 --- a/README.md +++ b/README.md @@ -153,9 +153,9 @@ Each adapter is toggled in the config — take only what you need. | OpenAI Embeddings | `POST /v1/embeddings` | ✅ implemented | | Anthropic Messages | `POST /v1/messages` | ✅ implemented (+ streaming) | | MCP (Model Context Protocol) | `POST /mcp` (JSON-RPC) | ✅ implemented | -| OpenAI Completions (legacy) | `POST /v1/completions` | planned | -| OpenAI Models | `GET /v1/models` | planned | -| Native passthrough | `POST /route` | planned | +| OpenAI Completions (legacy) | `POST /v1/completions` | ✅ implemented (+ streaming) | +| OpenAI Models | `GET /v1/models` | ✅ implemented | +| Native passthrough | `POST /route` | ✅ implemented | ## Supported providers (outbound) @@ -234,8 +234,7 @@ Implemented: OpenAI Chat Completions **with SSE streaming**, **Anthropic** provi inbound `/v1/messages` (streaming), **embeddings**, a **semantic cache**, **MCP** server, routing with fallback & graceful degradation, cost/token accounting, the **embedded multilingual admin console with hot config reload**, statistics and Prometheus -`GET /metrics`. Planned: OpenAI Completions/Models, native passthrough, per-account -routing, feedback loop. See the roadmap in +`GET /metrics`. Planned: per-account routing tables, router feedback loop. See the roadmap in [docs/ARCHITECTURE.md](https://github.com/infosave2007/cortiq-gateway/blob/master/docs/ARCHITECTURE.md). Contributions welcome — see [CONTRIBUTING.md](https://github.com/infosave2007/cortiq-gateway/blob/master/CONTRIBUTING.md). diff --git a/README.ru.md b/README.ru.md index b9a71a1..2d48554 100644 --- a/README.ru.md +++ b/README.ru.md @@ -145,9 +145,9 @@ print(resp.choices[0].message.content) | OpenAI Embeddings | `POST /v1/embeddings` | ✅ реализовано | | Anthropic Messages | `POST /v1/messages` | ✅ реализовано (+ стриминг) | | MCP (Model Context Protocol) | `POST /mcp` (JSON-RPC) | ✅ реализовано | -| OpenAI Completions (legacy) | `POST /v1/completions` | план | -| OpenAI Models | `GET /v1/models` | план | -| Native passthrough | `POST /route` | план | +| OpenAI Completions (legacy) | `POST /v1/completions` | ✅ реализовано (+ стриминг) | +| OpenAI Models | `GET /v1/models` | ✅ реализовано | +| Native passthrough | `POST /route` | ✅ реализовано | ## Поддерживаемые провайдеры (исходящие) @@ -183,8 +183,8 @@ cargo test # round-trip конфига, валидация ро **Anthropic** `/v1/messages` (стриминг), **embeddings**, **семантический кэш**, **MCP**-сервер, роутинг с fallback и мягкой деградацией, учёт стоимости/токенов, **встроенная многоязычная панель управления с горячей перезагрузкой конфига**, -статистика и Prometheus `GET /metrics`. В планах: OpenAI Completions/Models, native -passthrough, per-account роутинг, feedback-loop. Дорожная карта — +статистика и Prometheus `GET /metrics`. В планах: per-account таблицы роутинга, +feedback-loop в роутер. Дорожная карта — [docs/ARCHITECTURE.ru.md](docs/ARCHITECTURE.ru.md). Контрибьюции приветствуются — см. [CONTRIBUTING.ru.md](CONTRIBUTING.ru.md). diff --git a/src/admin/mod.rs b/src/admin/mod.rs index 5c79fc1..8916466 100644 --- a/src/admin/mod.rs +++ b/src/admin/mod.rs @@ -280,8 +280,8 @@ async fn meta() -> ApiResult { "tiers": ["low", "medium", "high"], "caps": ["tools", "vision"], "protocols_impl": { - "openai_chat": true, "openai_completions": false, "openai_embeddings": true, - "openai_models": false, "anthropic_messages": true, "mcp": true, "native_passthrough": false + "openai_chat": true, "openai_completions": true, "openai_embeddings": true, + "openai_models": true, "anthropic_messages": true, "mcp": true, "native_passthrough": true }, "languages": ["en", "ru", "de", "fr", "es", "zh", "tr"], })) diff --git a/src/protocols/mod.rs b/src/protocols/mod.rs index 8794ae6..4d7193c 100644 --- a/src/protocols/mod.rs +++ b/src/protocols/mod.rs @@ -3,8 +3,11 @@ pub mod anthropic_messages; pub mod mcp; +pub mod native_passthrough; pub mod openai_chat; +pub mod openai_completions; pub mod openai_embeddings; +pub mod openai_models; use crate::state::SharedState; use axum::Router; @@ -15,8 +18,10 @@ use axum::Router; pub fn build_router() -> Router { Router::new() .merge(openai_chat::routes()) + .merge(openai_completions::routes()) .merge(openai_embeddings::routes()) + .merge(openai_models::routes()) .merge(anthropic_messages::routes()) .merge(mcp::routes()) - // TODO: openai_completions, openai_models, native_passthrough. + .merge(native_passthrough::routes()) } diff --git a/src/protocols/native_passthrough.rs b/src/protocols/native_passthrough.rs new file mode 100644 index 0000000..a0ab8a6 --- /dev/null +++ b/src/protocols/native_passthrough.rs @@ -0,0 +1,86 @@ +//! Native passthrough adapter: `POST /route`. +//! Returns the gateway's routing decision (task type, complexity, candidate models) +//! without calling a model — direct access for clients that route themselves. + +use crate::error::{GatewayError, Result}; +use crate::state::SharedState; +use axum::{extract::State, response::IntoResponse, routing::post, Json, Router}; +use serde::Deserialize; + +pub fn routes() -> Router { + Router::new().route("/route", post(handler)) +} + +#[derive(Deserialize)] +struct RouteRequest { + /// `{ "input": { "text": "..." } }` (router-style) — optional. + #[serde(default)] + input: serde_json::Value, + /// `{ "text": "..." }` — convenience shorthand. + #[serde(default)] + text: Option, + #[serde(default)] + profile: Option, +} + +async fn handler( + State(state): State, + Json(req): Json, +) -> Result { + let live = state.live(); + if !live.cfg.protocols.native_passthrough { + return Err(GatewayError::InvalidRequest( + "native_passthrough protocol is disabled".into(), + )); + } + + let text = req + .text + .clone() + .or_else(|| req.input["text"].as_str().map(|s| s.to_string())) + .or_else(|| req.input.as_str().map(|s| s.to_string())) + .unwrap_or_default(); + if text.is_empty() { + return Err(GatewayError::InvalidRequest( + "missing input text (use {\"text\": ...} or {\"input\": {\"text\": ...}})".into(), + )); + } + + let profile = req + .profile + .clone() + .unwrap_or_else(|| live.cfg.route.profile.clone()); + + let (decision, source, candidates) = match live.router.route(&text, &profile).await { + Ok(Some(d)) => { + let cands = live.routing.candidates(&d.complexity_tier); + (Some(d), "router", cands) + } + _ => ( + None, + "fallback", + vec![live.routing.default_model().to_string()], + ), + }; + + let selected = candidates.first().cloned().unwrap_or_default(); + let decision_json = match &decision { + Some(d) => serde_json::json!({ + "task_label": d.task_label, + "complexity": { "score": d.complexity_score, "tier": d.complexity_tier }, + "router_request_id": d.router_request_id, + }), + None => serde_json::json!({ + "task_label": "degraded", + "complexity": { "score": 0.5, "tier": "degraded" }, + "router_request_id": serde_json::Value::Null, + }), + }; + + Ok(Json(serde_json::json!({ + "decision": decision_json, + "source": source, + "candidates": candidates, + "selected_model": selected, + }))) +} diff --git a/src/protocols/openai_completions.rs b/src/protocols/openai_completions.rs new file mode 100644 index 0000000..92e9b5f --- /dev/null +++ b/src/protocols/openai_completions.rs @@ -0,0 +1,183 @@ +//! Inbound OpenAI Completions (legacy) adapter: `POST /v1/completions`. +//! Converts the legacy `prompt` into a single user message, runs the pipeline, and +//! returns the legacy `text_completion` shape (non-streaming JSON or streaming SSE). + +use crate::error::{GatewayError, Result}; +use crate::model::{ChatRequest, GenParams, Message, RequestMeta, RoutingDirective}; +use crate::protocols::openai_chat::cortiq_headers; +use crate::state::SharedState; +use axum::response::{IntoResponse, Response}; +use axum::{extract::State, routing::post, Json, Router}; +use serde::Deserialize; + +pub fn routes() -> Router { + Router::new().route("/v1/completions", post(handler)) +} + +#[derive(Deserialize)] +struct CompletionsRequest { + model: String, + prompt: serde_json::Value, + #[serde(default)] + temperature: Option, + #[serde(default)] + max_tokens: Option, + #[serde(default)] + top_p: Option, + #[serde(default)] + stream: bool, +} + +fn parse_routing(model: &str) -> RoutingDirective { + if let Some(rest) = model.strip_prefix("cortiq-auto") { + RoutingDirective::Auto { + profile: rest.strip_prefix(':').map(|p| p.to_string()), + } + } else { + RoutingDirective::Pinned { + model_id: model.to_string(), + } + } +} + +fn prompt_to_string(v: &serde_json::Value) -> String { + if let Some(s) = v.as_str() { + return s.to_string(); + } + if let Some(arr) = v.as_array() { + return arr + .iter() + .filter_map(|x| x.as_str()) + .collect::>() + .join("\n"); + } + String::new() +} + +async fn handler( + State(state): State, + Json(req): Json, +) -> Result { + if !state.live().cfg.protocols.openai_completions { + return Err(GatewayError::InvalidRequest( + "openai_completions protocol is disabled".into(), + )); + } + let prompt = prompt_to_string(&req.prompt); + if prompt.is_empty() { + return Err(GatewayError::InvalidRequest( + "prompt must not be empty".into(), + )); + } + + let canonical = ChatRequest { + routing: parse_routing(&req.model), + messages: vec![Message { + role: "user".into(), + content: prompt, + tool_calls: vec![], + }], + tools: vec![], + params: GenParams { + temperature: req.temperature, + max_tokens: req.max_tokens, + top_p: req.top_p, + stop: Vec::new(), + passthrough: Default::default(), + }, + stream: req.stream, + meta: RequestMeta { + protocol: "openai_completions".into(), + ..Default::default() + }, + }; + + if req.stream { + let (info, stream) = state.pipeline.run_stream(canonical, &state).await?; + let mut headers = cortiq_headers(&info); + headers.insert( + axum::http::header::CONTENT_TYPE, + axum::http::HeaderValue::from_static("text/event-stream"), + ); + return Ok(( + headers, + axum::body::Body::from_stream(chat_to_completions_sse(stream)), + ) + .into_response()); + } + + let resp = state.pipeline.run(canonical, &state).await?; + let answer = resp + .choices + .first() + .map(|c| c.message.content.clone()) + .unwrap_or_default(); + let finish = resp + .choices + .first() + .map(|c| c.finish_reason.clone()) + .unwrap_or_else(|| "stop".into()); + let headers = cortiq_headers(&resp.cortiq); + let body = serde_json::json!({ + "id": resp.id, + "object": "text_completion", + "created": std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs(), + "model": resp.model_used, + "choices": [{ "text": answer, "index": 0, "logprobs": serde_json::Value::Null, "finish_reason": finish }], + "usage": { + "prompt_tokens": resp.usage.prompt_tokens, + "completion_tokens": resp.usage.completion_tokens, + "total_tokens": resp.usage.total_tokens, + }, + }); + Ok((headers, Json(body)).into_response()) +} + +/// Translate chat `chat.completion.chunk` SSE into legacy `text_completion` SSE +/// (`delta.content` → `text`). +fn chat_to_completions_sse( + stream: crate::providers::ChatStream, +) -> impl futures::Stream> + Send + 'static { + use futures::StreamExt; + async_stream::stream! { + let mut buf = String::new(); + futures::pin_mut!(stream); + while let Some(item) = stream.next().await { + let bytes = match item { Ok(b) => b, Err(e) => { yield Err(e); break; } }; + if let Ok(s) = std::str::from_utf8(&bytes) { + buf.push_str(s); + } + while let Some(idx) = buf.find("\n\n") { + let chunk: String = buf.drain(..idx + 2).collect(); + for line in chunk.lines() { + let Some(data) = line.trim_start().strip_prefix("data:") else { continue }; + let data = data.trim(); + if data.is_empty() { continue; } + if data == "[DONE]" { + yield Ok(bytes::Bytes::from("data: [DONE]\n\n")); + continue; + } + let Ok(v) = serde_json::from_str::(data) else { continue }; + if let Some(u) = v.get("usage").filter(|u| !u.is_null()) { + yield Ok(bytes::Bytes::from(format!( + "data: {}\n\n", + serde_json::json!({"object": "text_completion", "choices": [], "usage": u}) + ))); + } + if let Some(t) = v["choices"][0]["delta"]["content"].as_str() { + yield Ok(bytes::Bytes::from(format!( + "data: {}\n\n", + serde_json::json!({ + "object": "text_completion", + "choices": [{"text": t, "index": 0, "finish_reason": null}] + }) + ))); + } + } + } + } + } +} diff --git a/src/protocols/openai_models.rs b/src/protocols/openai_models.rs new file mode 100644 index 0000000..43326f1 --- /dev/null +++ b/src/protocols/openai_models.rs @@ -0,0 +1,32 @@ +//! Inbound OpenAI Models adapter: `GET /v1/models`. +//! Lists the configured pool (plus the virtual `cortiq-auto`) for client discovery. + +use crate::error::{GatewayError, Result}; +use crate::state::SharedState; +use axum::{extract::State, response::IntoResponse, routing::get, Json, Router}; + +pub fn routes() -> Router { + Router::new().route("/v1/models", get(handler)) +} + +async fn handler(State(state): State) -> Result { + let live = state.live(); + if !live.cfg.protocols.openai_models { + return Err(GatewayError::InvalidRequest( + "openai_models protocol is disabled".into(), + )); + } + let mut data = vec![serde_json::json!({ + "id": "cortiq-auto", + "object": "model", + "owned_by": "cortiq", + })]; + for m in &live.cfg.models { + data.push(serde_json::json!({ + "id": m.id, + "object": "model", + "owned_by": m.provider, + })); + } + Ok(Json(serde_json::json!({ "object": "list", "data": data }))) +}