chore(release): v0.2.0

Streaming (SSE), Anthropic provider/inbound, embeddings, semantic cache, MCP server.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

CHANGELOG.md

@@ -3,6 +3,29 @@
 All notable changes to this project are documented here.
 The format is based on [Keep a Changelog](https://keepachangelog.com/).
 
+## [0.2.0] - 2026-06-30
+
+### Added
+- **Streaming (SSE)** end-to-end: `POST /v1/chat/completions` with `"stream": true`
+  forwards provider SSE chunks verbatim, with `X-Cortiq-*` routing headers and token
+  usage tapped from the final chunk for statistics. Failover applies before the first
+  byte. Live streaming output in the admin Playground (`/admin/api/test/stream`).
+- Cross-platform Windows binaries in releases (`x86_64-pc-windows-msvc`) and a
+  cross-platform CI test matrix (ubuntu/windows/macos); secure token RNG via `getrandom`.
+- **Anthropic provider** (outbound): `POST /v1/messages` with `system` extraction,
+  `x-api-key` + `anthropic-version` headers, and streaming that translates Anthropic
+  SSE events into the unified OpenAI wire format.
+- **Anthropic inbound** adapter: `POST /v1/messages` (non-streaming JSON and streaming
+  SSE), so Claude-native clients can target the gateway.
+- **Embeddings**: `POST /v1/embeddings` inbound + provider `embed()`; embedding models
+  are now resolvable from the pool.
+- **Semantic cache**: embedding-based cache that returns a stored answer for prompts
+  semantically near a previous one (cosine threshold, TTL, ring buffer), skipping the
+  model call. Partitioned by routing signature; hit-rate and cost saved are tracked and
+  shown on the dashboard; configurable from Settings (`[cache]`).
+- **MCP server** (`POST /mcp`, JSON-RPC 2.0): exposes routing to MCP-native orchestrators
+  via two tools — `cortiq_route` (classify → decision) and `cortiq_chat` (route → answer).
+
 ## [0.1.0] - 2026-06-30
 
 ### Added

Cargo.lock

@@ -76,6 +76,28 @@ dependencies = [
  "rustversion",
 ]
 
+[[package]]
+name = "async-stream"
+version = "0.3.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0b5a71a6f37880a80d1d7f19efd781e4b5de42c88f0722cc13bcb6cc2cfe8476"
+dependencies = [
+ "async-stream-impl",
+ "futures-core",
+ "pin-project-lite",
+]
+
+[[package]]
+name = "async-stream-impl"
+version = "0.3.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c7c24de15d275a1ecfd47a380fb4d5ec9bfe0933f309ed5e705b775596a3574d"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn",
+]
+
 [[package]]
 name = "async-trait"
 version = "0.1.89"
@@ -263,10 +285,11 @@ checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570"
 
 [[package]]
 name = "cortiq-gateway"
-version = "0.1.0"
+version = "0.2.0"
 dependencies = [
  "anyhow",
  "arc-swap",
+ "async-stream",
  "async-trait",
  "axum",
  "bytes",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "cortiq-gateway"
-version = "0.1.0"
+version = "0.2.0"
 edition = "2021"
 rust-version = "1.75"
 license = "Apache-2.0"
@@ -48,6 +48,7 @@ getrandom = "0.2"
 # streaming / async utils
 futures = "0.3"
 async-trait = "0.1"
+async-stream = "0.3"
 bytes = "1"
 tokio-stream = "0.1"
 

README.md

@@ -35,8 +35,10 @@ One OpenAI-compatible endpoint → automatic model selection from your pool
 
 ## ✨ Highlights
 
-- **Drop-in OpenAI API.** Point any OpenAI client at the gateway and send `model: "cortiq-auto"`.
+- **Drop-in OpenAI API + streaming.** Point any OpenAI client at the gateway and send `model: "cortiq-auto"`; `stream: true` is fully supported (SSE).
+- **Multi-protocol.** OpenAI Chat & Embeddings, **Anthropic Messages** (in/out, streaming), and an **MCP** server (`POST /mcp`) exposing routing as tools for agent orchestrators.
 - **Intelligent routing.** `complexity.tier` → ordered model pool (`low → local`, `high → cloud`), with fallback and graceful degradation when the router is down.
+- **Semantic cache.** Optional embedding-based cache returns a stored answer for prompts semantically near a previous one — skipping the model call (hit-rate & savings on the dashboard).
 - **Built-in multilingual admin console.** Manage models, routing, protocols, keys and secrets from a web UI — **no hand-editing TOML, no restart** (changes are hot-reloaded). Available in **7 languages** (en, ru, de, fr, es, zh, tr) with light/dark themes.
 - **Live analytics.** Per-request stats (tokens, cost, latency, success rate, failovers), time-series charts, breakdowns by model/tier/task, and a Prometheus `GET /metrics` endpoint.
 - **Playground.** Send a prompt through the live pipeline and inspect the routing decision.
@@ -147,12 +149,12 @@ Each adapter is toggled in the config — take only what you need.
 
 | Protocol | Endpoint | Status |
 |---|---|---|
-| OpenAI Chat Completions | `POST /v1/chat/completions` | ✅ implemented |
+| OpenAI Chat Completions | `POST /v1/chat/completions` | ✅ implemented (+ streaming) |
+| 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 Embeddings | `POST /v1/embeddings` | planned |
 | OpenAI Models | `GET /v1/models` | planned |
-| Anthropic Messages | `POST /v1/messages` | planned |
-| MCP (Model Context Protocol) | JSON-RPC / SSE | planned |
 | Native passthrough | `POST /route` | planned |
 
 ## Supported providers (outbound)
@@ -160,7 +162,7 @@ Each adapter is toggled in the config — take only what you need.
 | Provider | Covers |
 |---|---|
 | `openai` (OpenAI-compatible) | OpenAI, OpenRouter, Together, Groq, **vLLM, llama.cpp server, LM Studio, Ollama (`/v1`)** |
-| `anthropic` | Claude Messages API (planned) |
+| `anthropic` | Claude Messages API (+ streaming) |
 | `ollama` | native Ollama API |
 | `http` | arbitrary HTTP endpoint (custom adapter) |
 
@@ -228,10 +230,12 @@ cargo test                 # config round-trip, routing validation
 
 ## Status
 
-🚧 Implemented so far: OpenAI Chat Completions, routing with fallback & graceful
-degradation, cost/token accounting, the **embedded multilingual admin console with
-hot config reload**, statistics and Prometheus `GET /metrics`. Planned: streaming
-(SSE), Anthropic provider/inbound, embeddings, MCP. See the roadmap in
+Implemented: OpenAI Chat Completions **with SSE streaming**, **Anthropic** provider +
+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
 [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).
 

README.ru.md

@@ -141,12 +141,12 @@ print(resp.choices[0].message.content)
 
 | Протокол | Endpoint | Статус |
 |---|---|---|
-| OpenAI Chat Completions | `POST /v1/chat/completions` | ✅ реализовано |
+| OpenAI Chat Completions | `POST /v1/chat/completions` | ✅ реализовано (+ стриминг) |
+| 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 Embeddings | `POST /v1/embeddings` | план |
 | OpenAI Models | `GET /v1/models` | план |
-| Anthropic Messages | `POST /v1/messages` | план |
-| MCP (Model Context Protocol) | JSON-RPC / SSE | план |
 | Native passthrough | `POST /route` | план |
 
 ## Поддерживаемые провайдеры (исходящие)
@@ -154,7 +154,7 @@ print(resp.choices[0].message.content)
 | Провайдер | Покрывает |
 |---|---|
 | `openai` (OpenAI-совместимый) | OpenAI, OpenRouter, Together, Groq, **vLLM, llama.cpp server, LM Studio, Ollama (`/v1`)** |
-| `anthropic` | Claude Messages API (план) |
+| `anthropic` | Claude Messages API (+ стриминг) |
 | `ollama` | нативный Ollama API |
 | `http` | произвольный HTTP-эндпоинт (свой адаптер) |
 
@@ -179,10 +179,12 @@ cargo test                 # round-trip конфига, валидация ро
 
 ## Статус
 
-🚧 Уже работает: OpenAI Chat Completions, роутинг с fallback и мягкой деградацией,
-учёт стоимости/токенов, **встроенная многоязычная панель управления с горячей
-перезагрузкой конфига**, статистика и Prometheus `GET /metrics`. В планах: стриминг
-(SSE), провайдер/вход Anthropic, embeddings, MCP. Дорожная карта —
+Реализовано: OpenAI Chat Completions **со стримингом (SSE)**, провайдер и вход
+**Anthropic** `/v1/messages` (стриминг), **embeddings**, **семантический кэш**,
+**MCP**-сервер, роутинг с fallback и мягкой деградацией, учёт стоимости/токенов,
+**встроенная многоязычная панель управления с горячей перезагрузкой конфига**,
+статистика и Prometheus `GET /metrics`. В планах: OpenAI Completions/Models, native
+passthrough, per-account роутинг, feedback-loop. Дорожная карта —
 [docs/ARCHITECTURE.ru.md](docs/ARCHITECTURE.ru.md). Контрибьюции приветствуются —
 см. [CONTRIBUTING.ru.md](CONTRIBUTING.ru.md).
 

config/gateway.example.toml

@@ -150,3 +150,15 @@ enabled   = true
 file      = "config/stats.jsonl"   # empty = in-memory only
 ring_size = 500                     # how many recent requests to keep in the feed
 retention = "7d"                    # aggregate/series depth (replayed on startup)
+
+# ============================================================================
+# Semantic cache — returns a cached answer for prompts semantically near a
+# previous one, skipping the model call. Requires an embedding model in the pool.
+# Disabled by default. (Cache settings take effect on restart.)
+# ============================================================================
+[cache]
+enabled     = false
+threshold   = 0.92                  # cosine similarity for a hit (0..1)
+ttl         = "1h"
+max_entries = 1000
+# embed_model = "local-embed"       # defaults to the first kind = "embedding" model

src/admin/mod.rs

@@ -8,8 +8,8 @@
 pub mod assets;
 
 use crate::config::{
-    AdminCfg, ApiKeyCfg, BreakerCfg, Config, CortiqCfg, LogCfg, ModelCfg, ProtocolsCfg, RouteCfg,
-    RouterCfg, RoutingCfg, RoutingPolicy, StatsCfg, TelemetryCfg, TierTargets,
+    AdminCfg, ApiKeyCfg, BreakerCfg, CacheCfg, Config, CortiqCfg, LogCfg, ModelCfg, ProtocolsCfg,
+    RouteCfg, RouterCfg, RoutingCfg, RoutingPolicy, StatsCfg, TelemetryCfg, TierTargets,
 };
 use crate::model::{ChatRequest, GenParams, Message, RequestMeta, RoutingDirective};
 use crate::state::SharedState;
@@ -165,6 +165,7 @@ pub fn api_routes(admin_token: String) -> Router<SharedState> {
         .route("/admin/api/stats", get(get_stats))
         .route("/admin/api/requests", get(get_requests))
         .route("/admin/api/test", post(run_test))
+        .route("/admin/api/test/stream", post(run_test_stream))
         .route_layer(middleware::from_fn(move |req: Request, next: Next| {
             let expected = admin_token.clone();
             async move { auth(expected, req, next).await }
@@ -270,7 +271,7 @@ async fn meta() -> ApiResult {
     ok(json!({
         "version": env!("CARGO_PKG_VERSION"),
         "providers": ["openai", "anthropic", "ollama", "http"],
-        "provider_impl": { "openai": true, "ollama": true, "http": true, "anthropic": false },
+        "provider_impl": { "openai": true, "ollama": true, "http": true, "anthropic": true },
         "cost_tiers": ["cheap", "mid", "expensive"],
         "kinds": ["chat", "embedding"],
         "profiles": ["cost-saver", "balanced", "quality-first"],
@@ -279,8 +280,8 @@ async fn meta() -> ApiResult {
         "tiers": ["low", "medium", "high"],
         "caps": ["tools", "vision"],
         "protocols_impl": {
-            "openai_chat": true, "openai_completions": false, "openai_embeddings": false,
-            "openai_models": false, "anthropic_messages": false, "mcp": false, "native_passthrough": false
+            "openai_chat": true, "openai_completions": false, "openai_embeddings": true,
+            "openai_models": false, "anthropic_messages": true, "mcp": true, "native_passthrough": false
         },
         "languages": ["en", "ru", "de", "fr", "es", "zh", "tr"],
     }))
@@ -476,6 +477,7 @@ async fn get_settings(State(state): State<SharedState>) -> ApiResult {
         "cortiq": c.cortiq,
         "stats": c.stats,
         "admin": c.admin,
+        "cache": c.cache,
     }))
 }
 
@@ -490,6 +492,7 @@ struct SettingsBody {
     cortiq: Option<CortiqCfg>,
     stats: Option<StatsCfg>,
     admin: Option<AdminCfg>,
+    cache: Option<CacheCfg>,
 }
 
 async fn put_settings(State(state): State<SharedState>, Json(b): Json<SettingsBody>) -> ApiResult {
@@ -528,6 +531,11 @@ async fn put_settings(State(state): State<SharedState>, Json(b): Json<SettingsBo
         }
         cfg.admin = v;
     }
+    if let Some(v) = b.cache {
+        // the cache is built at startup; persist now, apply on restart
+        needs_restart = true;
+        cfg.cache = v;
+    }
     state.reload(cfg)?;
     ok(json!({ "ok": true, "needs_restart": needs_restart }))
 }
@@ -665,7 +673,9 @@ async fn get_stats(State(state): State<SharedState>, Query(q): Query<StatsQuery>
         .and_then(parse_duration_secs)
         .unwrap_or(24 * 3600);
     let groupby = q.groupby.as_deref().unwrap_or("model");
-    ok(state.stats.snapshot(range_secs, groupby))
+    let mut snap = state.stats.snapshot(range_secs, groupby);
+    snap["cache"] = state.cache.snapshot();
+    ok(snap)
 }
 
 #[derive(Deserialize)]
@@ -753,3 +763,38 @@ async fn run_test(State(state): State<SharedState>, Json(b): Json<TestBody>) ->
         Err(e) => ok(json!({ "ok": false, "latency_ms": latency_ms, "error": e.to_string() })),
     }
 }
+
+/// Streaming playground: forwards provider SSE chunks; routing decision is exposed
+/// via X-Cortiq-* response headers (the SPA reads them after the stream ends).
+async fn run_test_stream(State(state): State<SharedState>, Json(b): Json<TestBody>) -> Response {
+    if b.messages.is_empty() {
+        return ApiError::bad("messages must not be empty").into_response();
+    }
+    let req = ChatRequest {
+        routing: parse_routing(&b.model),
+        messages: b.messages,
+        tools: vec![],
+        params: GenParams {
+            temperature: b.temperature,
+            max_tokens: b.max_tokens,
+            ..Default::default()
+        },
+        stream: true,
+        meta: RequestMeta {
+            account: "playground".into(),
+            protocol: "playground".into(),
+            ..Default::default()
+        },
+    };
+    match state.pipeline.run_stream(req, &state).await {
+        Ok((info, stream)) => {
+            let mut headers = crate::protocols::openai_chat::cortiq_headers(&info);
+            headers.insert(
+                header::CONTENT_TYPE,
+                axum::http::HeaderValue::from_static("text/event-stream"),
+            );
+            (headers, axum::body::Body::from_stream(stream)).into_response()
+        }
+        Err(e) => ApiError::bad(e.to_string()).into_response(),
+    }
+}

src/cache.rs

@@ -0,0 +1,171 @@
+//! Semantic (embedding-based) response cache.
+//!
+//! Each entry stores an L2-normalized embedding of the request plus the response.
+//! A lookup returns a cached answer when the cosine similarity to a previous request
+//! (with the same routing signature, within TTL) meets the configured threshold —
+//! skipping the expensive model call. In-memory ring buffer; counters track hit
+//! rate and the model cost saved.
+
+use crate::config::CacheCfg;
+use crate::stats::parse_duration_secs;
+use std::collections::VecDeque;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::{Arc, Mutex};
+use std::time::{SystemTime, UNIX_EPOCH};
+
+fn now_secs() -> u64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_secs()
+}
+
+struct Entry {
+    sig: String,
+    vec: Vec<f32>, // L2-normalized
+    ts: u64,
+    model_used: String,
+    content: String,
+    prompt_tokens: u32,
+    completion_tokens: u32,
+    cost: f64,
+}
+
+pub struct CacheHit {
+    pub model_used: String,
+    pub content: String,
+    pub prompt_tokens: u32,
+    pub completion_tokens: u32,
+    pub similarity: f32,
+}
+
+pub struct SemanticCache {
+    enabled: bool,
+    threshold: f32,
+    ttl_secs: u64,
+    max_entries: usize,
+    embed_model: Option<String>,
+    entries: Mutex<VecDeque<Entry>>,
+    hits: AtomicU64,
+    misses: AtomicU64,
+    saved_micro_usd: AtomicU64,
+}
+
+impl SemanticCache {
+    pub fn new(cfg: &CacheCfg) -> Arc<Self> {
+        Arc::new(Self {
+            enabled: cfg.enabled,
+            threshold: cfg.threshold,
+            ttl_secs: parse_duration_secs(&cfg.ttl).unwrap_or(3600),
+            max_entries: cfg.max_entries.max(1),
+            embed_model: cfg.embed_model.clone(),
+            entries: Mutex::new(VecDeque::new()),
+            hits: AtomicU64::new(0),
+            misses: AtomicU64::new(0),
+            saved_micro_usd: AtomicU64::new(0),
+        })
+    }
+
+    pub fn enabled(&self) -> bool {
+        self.enabled
+    }
+    pub fn embed_model(&self) -> Option<&str> {
+        self.embed_model.as_deref()
+    }
+
+    /// Look up a cached response by routing signature + query embedding.
+    pub fn lookup(&self, sig: &str, query: &[f32]) -> Option<CacheHit> {
+        if !self.enabled {
+            return None;
+        }
+        let q = normalized(query);
+        let cutoff = now_secs().saturating_sub(self.ttl_secs);
+        let inner = self.entries.lock().unwrap();
+        let mut best: Option<(f32, &Entry)> = None;
+        for e in inner.iter() {
+            if e.sig != sig || e.ts < cutoff {
+                continue;
+            }
+            let sim = dot(&e.vec, &q);
+            if sim >= self.threshold && best.map(|(b, _)| sim > b).unwrap_or(true) {
+                best = Some((sim, e));
+            }
+        }
+        match best {
+            Some((sim, e)) => {
+                self.hits.fetch_add(1, Ordering::Relaxed);
+                self.saved_micro_usd
+                    .fetch_add((e.cost * 1_000_000.0) as u64, Ordering::Relaxed);
+                Some(CacheHit {
+                    model_used: e.model_used.clone(),
+                    content: e.content.clone(),
+                    prompt_tokens: e.prompt_tokens,
+                    completion_tokens: e.completion_tokens,
+                    similarity: sim,
+                })
+            }
+            None => {
+                self.misses.fetch_add(1, Ordering::Relaxed);
+                None
+            }
+        }
+    }
+
+    /// Store a response under its routing signature + query embedding.
+    #[allow(clippy::too_many_arguments)]
+    pub fn insert(
+        &self,
+        sig: String,
+        query: Vec<f32>,
+        model_used: String,
+        content: String,
+        prompt_tokens: u32,
+        completion_tokens: u32,
+        cost: f64,
+    ) {
+        if !self.enabled || query.is_empty() {
+            return;
+        }
+        let vec = normalized(&query);
+        let mut inner = self.entries.lock().unwrap();
+        inner.push_back(Entry {
+            sig,
+            vec,
+            ts: now_secs(),
+            model_used,
+            content,
+            prompt_tokens,
+            completion_tokens,
+            cost,
+        });
+        while inner.len() > self.max_entries {
+            inner.pop_front();
+        }
+    }
+
+    pub fn snapshot(&self) -> serde_json::Value {
+        let h = self.hits.load(Ordering::Relaxed);
+        let m = self.misses.load(Ordering::Relaxed);
+        serde_json::json!({
+            "enabled": self.enabled,
+            "hits": h,
+            "misses": m,
+            "hit_rate": if h + m > 0 { h as f64 / (h + m) as f64 } else { 0.0 },
+            "entries": self.entries.lock().unwrap().len(),
+            "saved_usd": self.saved_micro_usd.load(Ordering::Relaxed) as f64 / 1_000_000.0,
+        })
+    }
+}
+
+fn dot(a: &[f32], b: &[f32]) -> f32 {
+    a.iter().zip(b.iter()).map(|(x, y)| x * y).sum()
+}
+
+fn normalized(v: &[f32]) -> Vec<f32> {
+    let n = v.iter().map(|x| x * x).sum::<f32>().sqrt();
+    if n > 0.0 {
+        v.iter().map(|x| x / n).collect()
+    } else {
+        v.to_vec()
+    }
+}

src/config.rs

@@ -31,6 +31,8 @@ pub struct Config {
     pub admin: AdminCfg,
     #[serde(default)]
     pub stats: StatsCfg,
+    #[serde(default)]
+    pub cache: CacheCfg,
 }
 
 #[derive(Clone, Debug, Serialize, Deserialize)]
@@ -320,6 +322,46 @@ fn default_stats_retention() -> String {
     "7d".into()
 }
 
+/// Semantic (embedding-based) response cache — returns a cached answer for prompts
+/// that are semantically near a previous one, skipping the (expensive) model call.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct CacheCfg {
+    #[serde(default)]
+    pub enabled: bool,
+    /// Cosine-similarity threshold for a hit (0..1).
+    #[serde(default = "default_semcache_threshold")]
+    pub threshold: f32,
+    /// Entry time-to-live, e.g. `1h`.
+    #[serde(default = "default_semcache_ttl")]
+    pub ttl: String,
+    /// Maximum number of cached entries (ring buffer).
+    #[serde(default = "default_semcache_max")]
+    pub max_entries: usize,
+    /// Embedding model id (defaults to the first `kind = "embedding"` model).
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub embed_model: Option<String>,
+}
+impl Default for CacheCfg {
+    fn default() -> Self {
+        Self {
+            enabled: false,
+            threshold: default_semcache_threshold(),
+            ttl: default_semcache_ttl(),
+            max_entries: default_semcache_max(),
+            embed_model: None,
+        }
+    }
+}
+fn default_semcache_threshold() -> f32 {
+    0.92
+}
+fn default_semcache_ttl() -> String {
+    "1h".into()
+}
+fn default_semcache_max() -> usize {
+    1000
+}
+
 impl Config {
     /// Load and validate the config from a file.
     pub fn load(path: &str) -> anyhow::Result<Self> {

src/main.rs

@@ -9,6 +9,7 @@
 #![allow(dead_code)]
 
 mod admin;
+mod cache;
 mod config;
 mod error;
 mod model;

src/pipeline.rs

@@ -4,9 +4,9 @@
 
 use crate::error::{GatewayError, Result};
 use crate::model::{
-    ChatRequest, ChatResponse, Message, RouteDecision, RouteInfo, RoutingDirective,
+    ChatRequest, ChatResponse, Choice, Message, RouteDecision, RouteInfo, RoutingDirective, Usage,
 };
-use crate::state::SharedState;
+use crate::state::{Live, SharedState};
 use crate::stats::RequestRecord;
 use std::time::{Instant, SystemTime, UNIX_EPOCH};
 
@@ -28,6 +28,45 @@ impl Pipeline {
         }
         .to_string();
 
+        // --- semantic cache lookup (short-circuit before the model call) ---
+        let cache = &state.cache;
+        let mut cache_key: Option<(String, Vec<f32>)> = None;
+        if cache.enabled() {
+            let live = state.live();
+            let sig = cache_sig(&req);
+            let text = self.extract_text(&req.messages, "concat_all", live.cfg.route.max_chars);
+            let embed_model = cache
+                .embed_model()
+                .map(|s| s.to_string())
+                .or_else(|| first_embedding_model(&live));
+            if let Some(model_id) = embed_model {
+                if let Some(vec) = embed(&live, &model_id, &text).await {
+                    if let Some(hit) = cache.lookup(&sig, &vec) {
+                        state.stats.record(RequestRecord {
+                            ts: now_secs(),
+                            account,
+                            protocol,
+                            directive,
+                            task_label: "cache".into(),
+                            tier: "cache".into(),
+                            score: 0.0,
+                            model_id: hit.model_used.clone(),
+                            route_source: "cache".into(),
+                            prompt_tokens: hit.prompt_tokens,
+                            completion_tokens: hit.completion_tokens,
+                            cost_usd: 0.0,
+                            latency_ms: started.elapsed().as_millis() as u64,
+                            outcome: "ok".into(),
+                            failover: false,
+                            error: None,
+                        });
+                        return Ok(cached_response(&hit));
+                    }
+                    cache_key = Some((sig, vec));
+                }
+            }
+        }
+
         let result = self.run_inner(req, state).await;
 
         // record statistics (for both success and error)
@@ -76,12 +115,36 @@ impl Pipeline {
         };
         state.stats.record(rec);
 
+        // --- store a successful response in the semantic cache ---
+        if let (Ok(resp), Some((sig, vec))) = (&result, cache_key) {
+            let content = resp
+                .choices
+                .first()
+                .map(|c| c.message.content.clone())
+                .unwrap_or_default();
+            if !content.is_empty() {
+                state.cache.insert(
+                    sig,
+                    vec,
+                    resp.cortiq.selected_model.clone(),
+                    content,
+                    resp.usage.prompt_tokens,
+                    resp.usage.completion_tokens,
+                    resp.cortiq.cost_usd,
+                );
+            }
+        }
+
         result
     }
 
-    async fn run_inner(&self, req: ChatRequest, state: &SharedState) -> Result<ChatResponse> {
-        let live = state.live();
-
+    /// Stages 1–3: extract text → route → select & capability-filter candidates.
+    /// Shared by both the non-streaming and streaming paths.
+    async fn resolve(
+        &self,
+        req: &ChatRequest,
+        live: &crate::state::Live,
+    ) -> Result<(RouteDecision, String, Vec<String>)> {
         // 1. Extract text for classification
         let text = self.extract_text(
             &req.messages,
@@ -159,6 +222,12 @@ impl Pipeline {
                     .to_string(),
             ));
         }
+        Ok((decision, route_source, final_candidates))
+    }
+
+    async fn run_inner(&self, req: ChatRequest, state: &SharedState) -> Result<ChatResponse> {
+        let live = state.live();
+        let (decision, route_source, final_candidates) = self.resolve(&req, &live).await?;
 
         // 4. Attempt provider call with failover (fall back to alternatives on error)
         let mut last_err = None;
@@ -210,6 +279,89 @@ impl Pipeline {
         }))
     }
 
+    /// Streaming variant: returns the routing metadata plus a stream of SSE byte
+    /// chunks (OpenAI wire format). Failover applies only before the first byte —
+    /// once the upstream stream has started it cannot be retried. Statistics are
+    /// recorded when the stream completes (tokens taken from the final usage chunk).
+    pub async fn run_stream(
+        &self,
+        req: ChatRequest,
+        state: &SharedState,
+    ) -> Result<(RouteInfo, crate::providers::ChatStream)> {
+        let started = Instant::now();
+        let account = req.meta.account.clone();
+        let protocol = req.meta.protocol.clone();
+        let directive = match &req.routing {
+            RoutingDirective::Pinned { .. } => "pinned",
+            RoutingDirective::Auto { .. } => "auto",
+        }
+        .to_string();
+
+        let live = state.live();
+        let (decision, route_source, final_candidates) = match self.resolve(&req, &live).await {
+            Ok(t) => t,
+            Err(e) => {
+                record_error(state, &account, &protocol, &directive, started, &e);
+                return Err(e);
+            }
+        };
+
+        let mut last_err = None;
+        let mut attempt = 0usize;
+        for model_id in final_candidates {
+            let provider = match live.registry.get(&model_id) {
+                Some(p) => p,
+                None => continue,
+            };
+            tracing::info!(
+                "Streaming request (label: {}, tier: {}) to model '{}' via source '{}'",
+                decision.task_label,
+                decision.complexity_tier,
+                model_id,
+                route_source
+            );
+            match provider.chat_stream(req.clone()).await {
+                Ok(stream) => {
+                    let info = RouteInfo {
+                        task_label: decision.task_label.clone(),
+                        complexity_score: decision.complexity_score,
+                        complexity_tier: decision.complexity_tier.clone(),
+                        selected_model: model_id,
+                        route_source: route_source.clone(),
+                        router_request_id: decision.router_request_id.clone(),
+                        cost_usd: 0.0,
+                        failover: attempt > 0,
+                    };
+                    let tapped = tap_stream(
+                        stream,
+                        state.stats.clone(),
+                        provider.clone(),
+                        info.clone(),
+                        account,
+                        protocol,
+                        directive,
+                        started,
+                    );
+                    return Ok((info, Box::pin(tapped)));
+                }
+                Err(err) => {
+                    tracing::warn!(
+                        "Provider '{}' stream failed: {:?}. Attempting failover...",
+                        model_id,
+                        err
+                    );
+                    last_err = Some(err);
+                    attempt += 1;
+                }
+            }
+        }
+        let e = last_err.unwrap_or_else(|| {
+            GatewayError::UpstreamUnavailable("All candidate models failed".to_string())
+        });
+        record_error(state, &account, &protocol, &directive, started, &e);
+        Err(e)
+    }
+
     fn extract_text(&self, messages: &[Message], strategy: &str, max_chars: usize) -> String {
         let raw_text = match strategy {
             "last_user" => messages
@@ -249,3 +401,194 @@ impl Pipeline {
         }
     }
 }
+
+fn now_secs() -> u64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_secs()
+}
+
+/// Routing signature used to partition cache entries (so a pinned model's cached
+/// answer is never returned for a different model/profile).
+fn cache_sig(req: &ChatRequest) -> String {
+    match &req.routing {
+        RoutingDirective::Auto { profile } => {
+            format!("auto:{}", profile.as_deref().unwrap_or(""))
+        }
+        RoutingDirective::Pinned { model_id } => format!("pinned:{model_id}"),
+    }
+}
+
+fn first_embedding_model(live: &Live) -> Option<String> {
+    live.cfg
+        .models
+        .iter()
+        .find(|m| m.kind == "embedding")
+        .map(|m| m.id.clone())
+}
+
+/// Embed `text` with the given model; returns the embedding vector (or None on error).
+async fn embed(live: &Live, model_id: &str, text: &str) -> Option<Vec<f32>> {
+    let provider = live.registry.get(model_id)?;
+    let body = provider.embed(serde_json::json!(text)).await.ok()?;
+    let arr = body["data"][0]["embedding"].as_array()?;
+    Some(
+        arr.iter()
+            .filter_map(|x| x.as_f64())
+            .map(|x| x as f32)
+            .collect(),
+    )
+}
+
+fn cached_response(hit: &crate::cache::CacheHit) -> ChatResponse {
+    ChatResponse {
+        id: "cache".to_string(),
+        model_used: hit.model_used.clone(),
+        choices: vec![Choice {
+            index: 0,
+            message: Message {
+                role: "assistant".into(),
+                content: hit.content.clone(),
+                tool_calls: vec![],
+            },
+            finish_reason: "stop".into(),
+        }],
+        usage: Usage {
+            prompt_tokens: hit.prompt_tokens,
+            completion_tokens: hit.completion_tokens,
+            total_tokens: hit.prompt_tokens + hit.completion_tokens,
+        },
+        cortiq: RouteInfo {
+            task_label: "cache".into(),
+            complexity_score: 0.0,
+            complexity_tier: "cache".into(),
+            selected_model: hit.model_used.clone(),
+            route_source: "cache".into(),
+            router_request_id: None,
+            cost_usd: 0.0,
+            failover: false,
+        },
+    }
+}
+
+fn record_error(
+    state: &SharedState,
+    account: &str,
+    protocol: &str,
+    directive: &str,
+    started: Instant,
+    e: &GatewayError,
+) {
+    state.stats.record(RequestRecord {
+        ts: now_secs(),
+        account: account.to_string(),
+        protocol: protocol.to_string(),
+        directive: directive.to_string(),
+        task_label: String::new(),
+        tier: String::new(),
+        score: 0.0,
+        model_id: String::new(),
+        route_source: String::new(),
+        prompt_tokens: 0,
+        completion_tokens: 0,
+        cost_usd: 0.0,
+        latency_ms: started.elapsed().as_millis() as u64,
+        outcome: "error".to_string(),
+        failover: false,
+        error: Some(e.to_string()),
+    });
+}
+
+/// Forward a provider SSE stream to the client verbatim while tapping the final
+/// `usage` chunk; record statistics when the stream completes.
+#[allow(clippy::too_many_arguments)]
+fn tap_stream(
+    inner: crate::providers::ChatStream,
+    stats: std::sync::Arc<crate::stats::Stats>,
+    provider: std::sync::Arc<dyn crate::providers::Provider>,
+    info: RouteInfo,
+    account: String,
+    protocol: String,
+    directive: String,
+    started: Instant,
+) -> impl futures::Stream<Item = Result<bytes::Bytes>> {
+    use futures::StreamExt;
+    async_stream::stream! {
+        let mut inner = inner;
+        let mut buf = String::new();
+        let mut prompt_tokens = 0u32;
+        let mut completion_tokens = 0u32;
+        let mut got_usage = false;
+        let mut delta_chars = 0usize;
+        let mut errored: Option<String> = None;
+
+        while let Some(item) = inner.next().await {
+            match &item {
+                Ok(bytes) => {
+                    if let Ok(s) = std::str::from_utf8(bytes) {
+                        buf.push_str(s);
+                        while let Some(idx) = buf.find("\n\n") {
+                            let evt: String = buf.drain(..idx + 2).collect();
+                            for line in evt.lines() {
+                                if let Some(data) = line.trim_start().strip_prefix("data:") {
+                                    let data = data.trim();
+                                    if data.is_empty() || data == "[DONE]" {
+                                        continue;
+                                    }
+                                    if let Ok(v) =
+                                        serde_json::from_str::<serde_json::Value>(data)
+                                    {
+                                        let u = &v["usage"];
+                                        if !u.is_null() {
+                                            prompt_tokens = u["prompt_tokens"]
+                                                .as_u64()
+                                                .unwrap_or(prompt_tokens as u64)
+                                                as u32;
+                                            completion_tokens = u["completion_tokens"]
+                                                .as_u64()
+                                                .unwrap_or(completion_tokens as u64)
+                                                as u32;
+                                            got_usage = true;
+                                        }
+                                        if let Some(c) =
+                                            v["choices"][0]["delta"]["content"].as_str()
+                                        {
+                                            delta_chars += c.chars().count();
+                                        }
+                                    }
+                                }
+                            }
+                        }
+                    }
+                }
+                Err(e) => errored = Some(e.to_string()),
+            }
+            yield item;
+        }
+
+        if !got_usage && completion_tokens == 0 {
+            // estimate when the server didn't return usage (~4 chars per token)
+            completion_tokens = (delta_chars / 4) as u32;
+        }
+        let cost = provider.price(prompt_tokens, completion_tokens);
+        stats.record(RequestRecord {
+            ts: now_secs(),
+            account,
+            protocol,
+            directive,
+            task_label: info.task_label.clone(),
+            tier: info.complexity_tier.clone(),
+            score: info.complexity_score,
+            model_id: info.selected_model.clone(),
+            route_source: info.route_source.clone(),
+            prompt_tokens,
+            completion_tokens,
+            cost_usd: cost,
+            latency_ms: started.elapsed().as_millis() as u64,
+            outcome: if errored.is_some() { "error" } else { "ok" }.to_string(),
+            failover: info.failover,
+            error: errored,
+        });
+    }
+}

src/protocols/anthropic_messages.rs

@@ -0,0 +1,209 @@
+//! Inbound Anthropic Messages adapter: `POST /v1/messages`.
+//! Translates the Anthropic request into the canonical [`ChatRequest`], runs the
+//! pipeline, and converts the result back to the Anthropic Messages format
+//! (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<SharedState> {
+    Router::new().route("/v1/messages", post(handler))
+}
+
+#[derive(Deserialize)]
+struct MessagesRequest {
+    model: String,
+    #[serde(default)]
+    system: serde_json::Value,
+    #[serde(default)]
+    messages: Vec<serde_json::Value>,
+    #[serde(default)]
+    max_tokens: Option<u32>,
+    #[serde(default)]
+    temperature: Option<f32>,
+    #[serde(default)]
+    top_p: Option<f32>,
+    #[serde(default)]
+    stream: bool,
+    #[serde(default)]
+    tools: Vec<serde_json::Value>,
+}
+
+/// Anthropic content is a string or an array of blocks; flatten to text.
+fn content_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(|b| b["text"].as_str())
+            .collect::<Vec<_>>()
+            .join("");
+    }
+    String::new()
+}
+
+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(),
+        }
+    }
+}
+
+async fn handler(
+    State(state): State<SharedState>,
+    Json(req): Json<MessagesRequest>,
+) -> Result<Response> {
+    if !state.live().cfg.protocols.anthropic_messages {
+        return Err(GatewayError::InvalidRequest(
+            "anthropic_messages protocol is disabled".into(),
+        ));
+    }
+    if req.messages.is_empty() {
+        return Err(GatewayError::InvalidRequest(
+            "messages must not be empty".into(),
+        ));
+    }
+
+    let mut messages = Vec::new();
+    let sys = content_to_string(&req.system);
+    if !sys.is_empty() {
+        messages.push(Message {
+            role: "system".into(),
+            content: sys,
+            tool_calls: vec![],
+        });
+    }
+    for m in &req.messages {
+        messages.push(Message {
+            role: m["role"].as_str().unwrap_or("user").to_string(),
+            content: content_to_string(&m["content"]),
+            tool_calls: vec![],
+        });
+    }
+
+    let canonical = ChatRequest {
+        routing: parse_routing(&req.model),
+        messages,
+        tools: req.tools,
+        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: "anthropic_messages".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"),
+        );
+        let translated = openai_to_anthropic_sse(stream);
+        return Ok((headers, axum::body::Body::from_stream(translated)).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 headers = cortiq_headers(&resp.cortiq);
+    let body = serde_json::json!({
+        "id": resp.id,
+        "type": "message",
+        "role": "assistant",
+        "model": resp.model_used,
+        "content": [{ "type": "text", "text": answer }],
+        "stop_reason": "end_turn",
+        "stop_sequence": serde_json::Value::Null,
+        "usage": {
+            "input_tokens": resp.usage.prompt_tokens,
+            "output_tokens": resp.usage.completion_tokens,
+        },
+    });
+    Ok((headers, Json(body)).into_response())
+}
+
+/// Translate the gateway's internal OpenAI-format SSE into Anthropic Messages SSE
+/// events, so Anthropic-native clients can stream regardless of the upstream provider.
+fn openai_to_anthropic_sse(
+    stream: crate::providers::ChatStream,
+) -> impl futures::Stream<Item = Result<bytes::Bytes>> + Send + 'static {
+    use futures::StreamExt;
+    async_stream::stream! {
+        fn evt(event: &str, data: serde_json::Value) -> bytes::Bytes {
+            bytes::Bytes::from(format!("event: {event}\ndata: {data}\n\n"))
+        }
+
+        yield Ok(evt("message_start", serde_json::json!({
+            "type": "message_start",
+            "message": {"id": "msg_gw", "type": "message", "role": "assistant",
+                        "content": [], "model": "cortiq-auto", "stop_reason": null,
+                        "usage": {"input_tokens": 0, "output_tokens": 0}}
+        })));
+        yield Ok(evt("content_block_start", serde_json::json!({
+            "type": "content_block_start", "index": 0,
+            "content_block": {"type": "text", "text": ""}
+        })));
+
+        let mut buf = String::new();
+        let mut completion_tokens = 0u64;
+        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() || data == "[DONE]" { continue; }
+                    let Ok(v) = serde_json::from_str::<serde_json::Value>(data) else { continue };
+                    if let Some(u) = v.get("usage").filter(|u| !u.is_null()) {
+                        completion_tokens = u["completion_tokens"].as_u64().unwrap_or(completion_tokens);
+                    }
+                    if let Some(t) = v["choices"][0]["delta"]["content"].as_str() {
+                        yield Ok(evt("content_block_delta", serde_json::json!({
+                            "type": "content_block_delta", "index": 0,
+                            "delta": {"type": "text_delta", "text": t}
+                        })));
+                    }
+                }
+            }
+        }
+
+        yield Ok(evt("content_block_stop", serde_json::json!({"type": "content_block_stop", "index": 0})));
+        yield Ok(evt("message_delta", serde_json::json!({
+            "type": "message_delta",
+            "delta": {"stop_reason": "end_turn", "stop_sequence": null},
+            "usage": {"output_tokens": completion_tokens}
+        })));
+        yield Ok(evt("message_stop", serde_json::json!({"type": "message_stop"})));
+    }
+}

src/protocols/mcp.rs

@@ -0,0 +1,162 @@
+//! MCP (Model Context Protocol) server: `POST /mcp` (JSON-RPC 2.0).
+//!
+//! Exposes the gateway's routing to MCP-native orchestrators as two tools:
+//! `cortiq_route` (classify a prompt → routing decision) and `cortiq_chat`
+//! (run a prompt through intelligent routing → answer). Single JSON endpoint;
+//! notifications are accepted with `202`. Gated by `protocols.mcp`.
+
+use crate::model::{ChatRequest, GenParams, Message, RequestMeta, RoutingDirective};
+use crate::state::SharedState;
+use axum::response::{IntoResponse, Response};
+use axum::{extract::State, http::StatusCode, routing::post, Json, Router};
+use serde_json::{json, Value};
+
+pub fn routes() -> Router<SharedState> {
+    Router::new().route("/mcp", post(handler))
+}
+
+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 tools_list() -> Value {
+    json!([
+        {
+            "name": "cortiq_route",
+            "description": "Classify a prompt and return the routing decision (task type, complexity tier, score, and the model the gateway would select).",
+            "inputSchema": {
+                "type": "object",
+                "properties": { "text": { "type": "string", "description": "The prompt to classify." } },
+                "required": ["text"]
+            }
+        },
+        {
+            "name": "cortiq_chat",
+            "description": "Run a prompt through the gateway's intelligent routing and return the selected model's answer.",
+            "inputSchema": {
+                "type": "object",
+                "properties": {
+                    "prompt": { "type": "string", "description": "The user prompt." },
+                    "model": { "type": "string", "description": "Optional: cortiq-auto[:profile] or a model id (default cortiq-auto)." }
+                },
+                "required": ["prompt"]
+            }
+        }
+    ])
+}
+
+fn tool_text(text: String, is_error: bool) -> Value {
+    json!({ "content": [{ "type": "text", "text": text }], "isError": is_error })
+}
+
+async fn handle_tool_call(
+    state: &SharedState,
+    params: &Value,
+) -> std::result::Result<Value, (i64, String)> {
+    let name = params["name"].as_str().unwrap_or("");
+    let args = &params["arguments"];
+    let live = state.live();
+    match name {
+        "cortiq_route" => {
+            let text = args["text"].as_str().unwrap_or("");
+            let prof = live.cfg.route.profile.clone();
+            match live.router.route(text, &prof).await {
+                Ok(Some(d)) => Ok(tool_text(
+                    format!(
+                        "task_label={}\ncomplexity_tier={}\ncomplexity_score={:.3}\ncandidates={:?}",
+                        d.task_label,
+                        d.complexity_tier,
+                        d.complexity_score,
+                        live.routing.candidates(&d.complexity_tier)
+                    ),
+                    false,
+                )),
+                _ => Ok(tool_text(
+                    "router unavailable — would use the default model".to_string(),
+                    false,
+                )),
+            }
+        }
+        "cortiq_chat" => {
+            let prompt = args["prompt"].as_str().unwrap_or("");
+            let model = args["model"].as_str().unwrap_or("cortiq-auto");
+            let req = ChatRequest {
+                routing: parse_routing(model),
+                messages: vec![Message {
+                    role: "user".into(),
+                    content: prompt.to_string(),
+                    tool_calls: vec![],
+                }],
+                tools: vec![],
+                params: GenParams::default(),
+                stream: false,
+                meta: RequestMeta {
+                    account: "mcp".into(),
+                    protocol: "mcp".into(),
+                    ..Default::default()
+                },
+            };
+            match state.pipeline.run(req, state).await {
+                Ok(resp) => {
+                    let answer = resp
+                        .choices
+                        .first()
+                        .map(|c| c.message.content.clone())
+                        .unwrap_or_default();
+                    Ok(tool_text(answer, false))
+                }
+                Err(e) => Ok(tool_text(e.to_string(), true)),
+            }
+        }
+        _ => Err((-32602, format!("unknown tool '{name}'"))),
+    }
+}
+
+async fn handler(State(state): State<SharedState>, Json(req): Json<Value>) -> Response {
+    if !state.live().cfg.protocols.mcp {
+        return (StatusCode::NOT_FOUND, "mcp protocol is disabled").into_response();
+    }
+
+    let method = req["method"].as_str().unwrap_or("").to_string();
+    let id = req.get("id").cloned();
+
+    // JSON-RPC notifications carry no id and expect no response
+    if id.is_none() || method.starts_with("notifications/") {
+        return StatusCode::ACCEPTED.into_response();
+    }
+    let id = id.unwrap_or(Value::Null);
+
+    let result: std::result::Result<Value, (i64, String)> = match method.as_str() {
+        "initialize" => {
+            let pv = req["params"]["protocolVersion"]
+                .as_str()
+                .unwrap_or("2024-11-05")
+                .to_string();
+            Ok(json!({
+                "protocolVersion": pv,
+                "capabilities": { "tools": {} },
+                "serverInfo": { "name": "cortiq-gateway", "version": env!("CARGO_PKG_VERSION") }
+            }))
+        }
+        "ping" => Ok(json!({})),
+        "tools/list" => Ok(json!({ "tools": tools_list() })),
+        "tools/call" => handle_tool_call(&state, &req["params"]).await,
+        other => Err((-32601, format!("method not found: {other}"))),
+    };
+
+    let body = match result {
+        Ok(r) => json!({ "jsonrpc": "2.0", "id": id, "result": r }),
+        Err((code, message)) => {
+            json!({ "jsonrpc": "2.0", "id": id, "error": { "code": code, "message": message } })
+        }
+    };
+    Json(body).into_response()
+}

src/protocols/mod.rs

@@ -1,7 +1,10 @@
 //! Incoming protocol adapters. Each one builds its own axum routes and translates
 //! its protocol to/from the canonical model. Enabled via flags in `[protocols]`.
 
+pub mod anthropic_messages;
+pub mod mcp;
 pub mod openai_chat;
+pub mod openai_embeddings;
 
 use crate::state::SharedState;
 use axum::Router;
@@ -10,8 +13,10 @@ use axum::Router;
 /// each handler checks the live `protocols.*` flag from config itself, so
 /// toggles in the admin panel take effect without a restart (hot switching).
 pub fn build_router() -> Router<SharedState> {
-    Router::new().merge(openai_chat::routes())
-    // TODO: openai_completions, openai_embeddings, openai_models,
-    //       anthropic_messages, mcp, native_passthrough — add here as well,
-    //       with a check of the corresponding flag inside the handler.
+    Router::new()
+        .merge(openai_chat::routes())
+        .merge(openai_embeddings::routes())
+        .merge(anthropic_messages::routes())
+        .merge(mcp::routes())
+    // TODO: openai_completions, openai_models, native_passthrough.
 }

src/protocols/openai_chat.rs

@@ -3,11 +3,37 @@
 //! to the pipeline, then converts the canonical [`ChatResponse`] back to OpenAI format.
 
 use crate::error::{GatewayError, Result};
-use crate::model::{ChatRequest, GenParams, Message, RequestMeta, RoutingDirective};
+use crate::model::{ChatRequest, GenParams, Message, RequestMeta, RouteInfo, RoutingDirective};
 use crate::state::SharedState;
+use axum::response::{IntoResponse, Response};
 use axum::{extract::State, routing::post, Json, Router};
 use serde::Deserialize;
 
+/// Build the `X-Cortiq-*` response headers from routing metadata.
+pub(crate) fn cortiq_headers(c: &RouteInfo) -> axum::http::HeaderMap {
+    use axum::http::{HeaderMap, HeaderValue};
+    fn put(h: &mut HeaderMap, k: &'static str, v: &str) {
+        if let Ok(val) = HeaderValue::from_str(v) {
+            h.insert(k, val);
+        }
+    }
+    let mut h = HeaderMap::new();
+    put(&mut h, "X-Cortiq-Task-Label", &c.task_label);
+    put(
+        &mut h,
+        "X-Cortiq-Complexity-Score",
+        &c.complexity_score.to_string(),
+    );
+    put(&mut h, "X-Cortiq-Complexity-Tier", &c.complexity_tier);
+    put(&mut h, "X-Cortiq-Selected-Model", &c.selected_model);
+    put(&mut h, "X-Cortiq-Route-Source", &c.route_source);
+    put(&mut h, "X-Cortiq-Cost-Usd", &c.cost_usd.to_string());
+    if let Some(id) = &c.router_request_id {
+        put(&mut h, "X-Cortiq-Request-Id", id);
+    }
+    h
+}
+
 pub fn routes() -> Router<SharedState> {
     Router::new().route("/v1/chat/completions", post(handler))
 }
@@ -46,7 +72,7 @@ fn parse_routing(model: &str) -> RoutingDirective {
 async fn handler(
     State(state): State<SharedState>,
     Json(req): Json<OpenAiChatRequest>,
-) -> Result<impl axum::response::IntoResponse> {
+) -> Result<Response> {
     // hot protocol toggle: if the adapter is disabled in config — return 404
     if !state.live().cfg.protocols.openai_chat {
         return Err(GatewayError::InvalidRequest(
@@ -77,52 +103,23 @@ async fn handler(
         },
     };
 
+    // streaming: forward provider SSE chunks verbatim with X-Cortiq-* headers
     if req.stream {
-        return Err(GatewayError::InvalidRequest(
-            "Streaming is not yet supported in this version".into(),
-        ));
+        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"),
+        );
+        headers.insert(
+            axum::http::header::CACHE_CONTROL,
+            axum::http::HeaderValue::from_static("no-cache"),
+        );
+        return Ok((headers, axum::body::Body::from_stream(stream)).into_response());
     }
 
     let resp = state.pipeline.run(canonical, &state).await?;
-
-    let mut headers = axum::http::HeaderMap::new();
-    headers.insert(
-        "X-Cortiq-Task-Label",
-        axum::http::HeaderValue::from_str(&resp.cortiq.task_label)
-            .unwrap_or(axum::http::HeaderValue::from_static("")),
-    );
-    headers.insert(
-        "X-Cortiq-Complexity-Score",
-        axum::http::HeaderValue::from_str(&resp.cortiq.complexity_score.to_string())
-            .unwrap_or(axum::http::HeaderValue::from_static("0.0")),
-    );
-    headers.insert(
-        "X-Cortiq-Complexity-Tier",
-        axum::http::HeaderValue::from_str(&resp.cortiq.complexity_tier)
-            .unwrap_or(axum::http::HeaderValue::from_static("")),
-    );
-    headers.insert(
-        "X-Cortiq-Selected-Model",
-        axum::http::HeaderValue::from_str(&resp.cortiq.selected_model)
-            .unwrap_or(axum::http::HeaderValue::from_static("")),
-    );
-    headers.insert(
-        "X-Cortiq-Route-Source",
-        axum::http::HeaderValue::from_str(&resp.cortiq.route_source)
-            .unwrap_or(axum::http::HeaderValue::from_static("")),
-    );
-    headers.insert(
-        "X-Cortiq-Cost-Usd",
-        axum::http::HeaderValue::from_str(&resp.cortiq.cost_usd.to_string())
-            .unwrap_or(axum::http::HeaderValue::from_static("0.0")),
-    );
-    if let Some(req_id) = &resp.cortiq.router_request_id {
-        headers.insert(
-            "X-Cortiq-Request-Id",
-            axum::http::HeaderValue::from_str(req_id)
-                .unwrap_or(axum::http::HeaderValue::from_static("")),
-        );
-    }
+    let headers = cortiq_headers(&resp.cortiq);
 
     let mut body = serde_json::json!({
         "id": resp.id,
@@ -164,5 +161,5 @@ async fn handler(
         });
     }
 
-    Ok((headers, Json(body)))
+    Ok((headers, Json(body)).into_response())
 }

src/protocols/openai_embeddings.rs

@@ -0,0 +1,52 @@
+//! Inbound OpenAI Embeddings adapter: `POST /v1/embeddings`.
+//! Resolves an embedding model (explicit id, or the first `kind = "embedding"`),
+//! calls the provider, and forwards the OpenAI-style response body.
+
+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<SharedState> {
+    Router::new().route("/v1/embeddings", post(handler))
+}
+
+#[derive(Deserialize)]
+struct EmbeddingsRequest {
+    #[serde(default)]
+    model: String,
+    input: serde_json::Value,
+}
+
+async fn handler(
+    State(state): State<SharedState>,
+    Json(req): Json<EmbeddingsRequest>,
+) -> Result<impl IntoResponse> {
+    let live = state.live();
+    if !live.cfg.protocols.openai_embeddings {
+        return Err(GatewayError::InvalidRequest(
+            "openai_embeddings protocol is disabled".into(),
+        ));
+    }
+
+    // resolve the embedding model: explicit real id, else first kind == "embedding"
+    let model_id = if !req.model.is_empty()
+        && req.model != "cortiq-auto"
+        && live.registry.get(&req.model).is_some()
+    {
+        req.model.clone()
+    } else {
+        live.cfg
+            .models
+            .iter()
+            .find(|m| m.kind == "embedding")
+            .map(|m| m.id.clone())
+            .ok_or_else(|| GatewayError::InvalidRequest("no embedding model configured".into()))?
+    };
+
+    let provider = live.registry.get(&model_id).ok_or_else(|| {
+        GatewayError::UpstreamUnavailable(format!("embedding model '{model_id}' is not available"))
+    })?;
+    let body = provider.embed(req.input).await?;
+    Ok(Json(body))
+}

src/providers/anthropic.rs

@@ -1,23 +1,25 @@
-//! Anthropic Messages provider (v0.2). Translates the canonical ChatRequest to
-//! `POST /v1/messages` and back. The `system` message is extracted into a separate field.
+//! Anthropic Messages provider. Translates the canonical [`ChatRequest`] to
+//! `POST {base_url}/v1/messages` and back. `system` messages are lifted into the
+//! top-level `system` field. Streaming translates Anthropic SSE events into the
+//! OpenAI wire format, so the gateway's unified SSE passthrough works uniformly.
 
 use super::{Caps, Provider};
 use crate::config::ModelCfg;
 use crate::error::{GatewayError, Result};
-use crate::model::{ChatRequest, ChatResponse};
+use crate::model::{ChatRequest, ChatResponse, Choice, Message, RouteInfo, Usage};
 use async_trait::async_trait;
 
+const ANTHROPIC_VERSION: &str = "2023-06-01";
+
 pub struct AnthropicProvider {
     id: String,
-    #[allow(dead_code)]
     base_url: String,
-    #[allow(dead_code)]
     model: String,
-    #[allow(dead_code)]
     api_key: Option<String>,
     caps: Caps,
     price_in: f64,
     price_out: f64,
+    http: reqwest::Client,
 }
 
 impl AnthropicProvider {
@@ -30,8 +32,61 @@ impl AnthropicProvider {
             caps: Caps::from_list(&cfg.caps),
             price_in: cfg.price_in,
             price_out: cfg.price_out,
+            http: reqwest::Client::new(),
         })
     }
+
+    /// Canonical request → Anthropic Messages request body.
+    fn to_wire(&self, req: &ChatRequest, stream: bool) -> serde_json::Value {
+        let mut system = String::new();
+        let mut messages = Vec::new();
+        for m in &req.messages {
+            if m.role == "system" {
+                if !system.is_empty() {
+                    system.push('\n');
+                }
+                system.push_str(&m.content);
+            } else {
+                let role = if m.role == "assistant" {
+                    "assistant"
+                } else {
+                    "user"
+                };
+                messages.push(serde_json::json!({ "role": role, "content": m.content }));
+            }
+        }
+        let mut body = serde_json::json!({
+            "model": self.model,
+            "messages": messages,
+            "max_tokens": req.params.max_tokens.unwrap_or(1024),
+            "stream": stream,
+        });
+        if !system.is_empty() {
+            body["system"] = serde_json::Value::String(system);
+        }
+        if let Some(t) = req.params.temperature {
+            body["temperature"] = t.into();
+        }
+        if let Some(p) = req.params.top_p {
+            body["top_p"] = p.into();
+        }
+        if !req.tools.is_empty() {
+            body["tools"] = serde_json::Value::Array(req.tools.clone());
+        }
+        body
+    }
+
+    fn request(&self, body: &serde_json::Value) -> reqwest::RequestBuilder {
+        let mut r = self
+            .http
+            .post(format!("{}/v1/messages", self.base_url))
+            .header("anthropic-version", ANTHROPIC_VERSION)
+            .json(body);
+        if let Some(key) = &self.api_key {
+            r = r.header("x-api-key", key);
+        }
+        r
+    }
 }
 
 #[async_trait]
@@ -46,12 +101,159 @@ impl Provider for AnthropicProvider {
         (prompt_tokens as f64 * self.price_in + completion_tokens as f64 * self.price_out)
             / 1_000_000.0
     }
-    async fn chat(&self, _req: ChatRequest) -> Result<ChatResponse> {
-        // TODO(v0.2): extract system from messages → "system" field; messages → content blocks;
-        // add x-api-key + anthropic-version headers; map response → canonical form.
-        // Returns an error (not a panic) so that failover/playground continue to work normally.
-        Err(GatewayError::UpstreamUnavailable(
-            "anthropic provider not yet implemented (planned v0.2)".to_string(),
-        ))
+
+    async fn chat(&self, req: ChatRequest) -> Result<ChatResponse> {
+        let resp = self
+            .request(&self.to_wire(&req, false))
+            .send()
+            .await
+            .map_err(|e| GatewayError::UpstreamUnavailable(e.to_string()))?;
+        if !resp.status().is_success() {
+            let status = resp.status();
+            let b = resp.text().await.unwrap_or_default();
+            return Err(GatewayError::UpstreamUnavailable(format!(
+                "provider {} returned HTTP {}: {}",
+                self.id, status, b
+            )));
+        }
+        let v: serde_json::Value = resp.json().await.map_err(|e| {
+            GatewayError::Internal(format!("failed to parse anthropic response: {e}"))
+        })?;
+        let content = v["content"]
+            .as_array()
+            .map(|blocks| {
+                blocks
+                    .iter()
+                    .filter_map(|b| b["text"].as_str())
+                    .collect::<Vec<_>>()
+                    .join("")
+            })
+            .unwrap_or_default();
+        let prompt_tokens = v["usage"]["input_tokens"].as_u64().unwrap_or(0) as u32;
+        let completion_tokens = v["usage"]["output_tokens"].as_u64().unwrap_or(0) as u32;
+        Ok(ChatResponse {
+            id: v["id"].as_str().unwrap_or("").to_string(),
+            model_used: v["model"].as_str().unwrap_or(&self.model).to_string(),
+            choices: vec![Choice {
+                index: 0,
+                message: Message {
+                    role: "assistant".into(),
+                    content,
+                    tool_calls: vec![],
+                },
+                finish_reason: v["stop_reason"].as_str().unwrap_or("stop").to_string(),
+            }],
+            usage: Usage {
+                prompt_tokens,
+                completion_tokens,
+                total_tokens: prompt_tokens + completion_tokens,
+            },
+            cortiq: RouteInfo {
+                task_label: String::new(),
+                complexity_score: 0.0,
+                complexity_tier: String::new(),
+                selected_model: self.id.clone(),
+                route_source: String::new(),
+                router_request_id: None,
+                cost_usd: 0.0,
+                failover: false,
+            },
+        })
+    }
+
+    async fn chat_stream(&self, req: ChatRequest) -> Result<super::ChatStream> {
+        let resp = self
+            .request(&self.to_wire(&req, true))
+            .send()
+            .await
+            .map_err(|e| GatewayError::UpstreamUnavailable(e.to_string()))?;
+        if !resp.status().is_success() {
+            let status = resp.status();
+            let b = resp.text().await.unwrap_or_default();
+            return Err(GatewayError::UpstreamUnavailable(format!(
+                "provider {} returned HTTP {}: {}",
+                self.id, status, b
+            )));
+        }
+        Ok(Box::pin(anthropic_to_openai_sse(resp.bytes_stream())))
+    }
+}
+
+/// Translate an Anthropic Messages SSE stream into OpenAI `chat.completion.chunk`
+/// SSE so the gateway's unified streaming path works for Anthropic too.
+fn anthropic_to_openai_sse(
+    upstream: impl futures::Stream<Item = reqwest::Result<bytes::Bytes>> + Send + 'static,
+) -> impl futures::Stream<Item = Result<bytes::Bytes>> + Send + 'static {
+    use futures::StreamExt;
+    async_stream::stream! {
+        futures::pin_mut!(upstream);
+        let mut buf = String::new();
+        let mut input_tokens = 0u64;
+        let mut role_sent = false;
+        let emit = |obj: serde_json::Value| bytes::Bytes::from(format!("data: {obj}\n\n"));
+
+        while let Some(chunk) = upstream.next().await {
+            let chunk = match chunk {
+                Ok(c) => c,
+                Err(e) => {
+                    yield Err(GatewayError::UpstreamUnavailable(e.to_string()));
+                    return;
+                }
+            };
+            if let Ok(s) = std::str::from_utf8(&chunk) {
+                buf.push_str(s);
+            }
+            while let Some(idx) = buf.find("\n\n") {
+                let evt: String = buf.drain(..idx + 2).collect();
+                for line in evt.lines() {
+                    let Some(data) = line.trim_start().strip_prefix("data:") else {
+                        continue;
+                    };
+                    let data = data.trim();
+                    if data.is_empty() {
+                        continue;
+                    }
+                    let Ok(v) = serde_json::from_str::<serde_json::Value>(data) else {
+                        continue;
+                    };
+                    match v["type"].as_str().unwrap_or("") {
+                        "message_start" => {
+                            input_tokens =
+                                v["message"]["usage"]["input_tokens"].as_u64().unwrap_or(0);
+                            if !role_sent {
+                                role_sent = true;
+                                yield Ok(emit(serde_json::json!({
+                                    "choices": [{"index": 0, "delta": {"role": "assistant"}, "finish_reason": null}]
+                                })));
+                            }
+                        }
+                        "content_block_delta" => {
+                            if let Some(t) = v["delta"]["text"].as_str() {
+                                yield Ok(emit(serde_json::json!({
+                                    "choices": [{"index": 0, "delta": {"content": t}, "finish_reason": null}]
+                                })));
+                            }
+                        }
+                        "message_delta" => {
+                            let out = v["usage"]["output_tokens"].as_u64().unwrap_or(0);
+                            let stop = v["delta"]["stop_reason"].as_str().unwrap_or("stop").to_string();
+                            yield Ok(emit(serde_json::json!({
+                                "choices": [{"index": 0, "delta": {}, "finish_reason": stop}]
+                            })));
+                            yield Ok(emit(serde_json::json!({
+                                "choices": [],
+                                "usage": {
+                                    "prompt_tokens": input_tokens,
+                                    "completion_tokens": out,
+                                    "total_tokens": input_tokens + out
+                                }
+                            })));
+                        }
+                        _ => {}
+                    }
+                }
+            }
+        }
+        yield Ok(bytes::Bytes::from("data: [DONE]\n\n"));
     }
 }

src/providers/mod.rs

@@ -7,8 +7,16 @@
 pub mod anthropic;
 pub mod openai;
 
+use crate::error::GatewayError;
 use crate::model::{ChatRequest, ChatResponse};
 use async_trait::async_trait;
+use bytes::Bytes;
+use futures::stream::BoxStream;
+
+/// A streaming chat response: raw Server-Sent-Events byte chunks in OpenAI wire
+/// format (`data: {json}\n\n` … `data: [DONE]\n\n`). The gateway forwards these
+/// to the client verbatim while tapping the final `usage` for statistics.
+pub type ChatStream = BoxStream<'static, crate::error::Result<Bytes>>;
 
 /// Model capabilities — checked at selection time (whether tools/vision are needed).
 #[derive(Clone, Debug, Default)]
@@ -41,6 +49,21 @@ pub trait Provider: Send + Sync {
     /// Non-streaming chat call.
     async fn chat(&self, req: ChatRequest) -> crate::error::Result<ChatResponse>;
 
-    // TODO(v0.1): async fn chat_stream(...) -> Result<ChatStream> (SSE).
-    // TODO(v0.1): async fn embed(...) -> Result<EmbedResponse>.
+    /// Streaming chat call (SSE). Default: not supported (override per provider).
+    async fn chat_stream(&self, _req: ChatRequest) -> crate::error::Result<ChatStream> {
+        Err(GatewayError::UpstreamUnavailable(format!(
+            "streaming is not supported by provider '{}'",
+            self.id()
+        )))
+    }
+
+    /// Embeddings call. `input` is the OpenAI `input` field (string or array of
+    /// strings); returns the provider's OpenAI-style embeddings response body.
+    /// Default: not supported (override per provider).
+    async fn embed(&self, _input: serde_json::Value) -> crate::error::Result<serde_json::Value> {
+        Err(GatewayError::UpstreamUnavailable(format!(
+            "embeddings are not supported by provider '{}'",
+            self.id()
+        )))
+    }
 }

src/providers/openai.rs

@@ -154,4 +154,61 @@ impl Provider for OpenAiProvider {
             },
         })
     }
+
+    async fn chat_stream(&self, req: ChatRequest) -> Result<super::ChatStream> {
+        use futures::StreamExt;
+        let mut body = self.to_wire(&req);
+        body["stream"] = serde_json::Value::Bool(true);
+        // ask OpenAI-compatible servers to include token usage in the final chunk
+        body["stream_options"] = serde_json::json!({ "include_usage": true });
+        let mut r = self
+            .http
+            .post(format!("{}/chat/completions", self.base_url))
+            .json(&body);
+        if let Some(key) = &self.api_key {
+            r = r.bearer_auth(key);
+        }
+        let resp = r
+            .send()
+            .await
+            .map_err(|e| GatewayError::UpstreamUnavailable(e.to_string()))?;
+        if !resp.status().is_success() {
+            let status = resp.status();
+            let err_body = resp.text().await.unwrap_or_default();
+            return Err(GatewayError::UpstreamUnavailable(format!(
+                "provider {} returned HTTP {}: {}",
+                self.id, status, err_body
+            )));
+        }
+        let stream = resp
+            .bytes_stream()
+            .map(|res| res.map_err(|e| GatewayError::UpstreamUnavailable(e.to_string())));
+        Ok(Box::pin(stream))
+    }
+
+    async fn embed(&self, input: serde_json::Value) -> Result<serde_json::Value> {
+        let body = serde_json::json!({ "model": self.model, "input": input });
+        let mut r = self
+            .http
+            .post(format!("{}/embeddings", self.base_url))
+            .json(&body);
+        if let Some(key) = &self.api_key {
+            r = r.bearer_auth(key);
+        }
+        let resp = r
+            .send()
+            .await
+            .map_err(|e| GatewayError::UpstreamUnavailable(e.to_string()))?;
+        if !resp.status().is_success() {
+            let status = resp.status();
+            let err_body = resp.text().await.unwrap_or_default();
+            return Err(GatewayError::UpstreamUnavailable(format!(
+                "provider {} returned HTTP {}: {}",
+                self.id, status, err_body
+            )));
+        }
+        resp.json().await.map_err(|e| {
+            GatewayError::Internal(format!("failed to parse embeddings response: {e}"))
+        })
+    }
 }

src/registry.rs

@@ -15,9 +15,6 @@ impl Registry {
     pub fn from_config(cfg: &Config, secrets: &SecretStore) -> anyhow::Result<Self> {
         let mut by_id: HashMap<String, Arc<dyn Provider>> = HashMap::new();
         for m in &cfg.models {
-            if m.kind == "embedding" {
-                continue; // embedding models are handled by a separate branch (v0.1)
-            }
             // provider key: secret store → environment variable
             let api_key = m
                 .api_key_env

src/state.rs

@@ -5,6 +5,7 @@
 //! a restart: the admin API validates the new config, builds a new `Live`,
 //! atomically writes the TOML, and swaps the pointer.
 
+use crate::cache::SemanticCache;
 use crate::config::Config;
 use crate::registry::Registry;
 use crate::router_client::RouterClient;
@@ -42,6 +43,7 @@ impl Live {
 pub struct AppState {
     pub live: ArcSwap<Live>,
     pub stats: Arc<Stats>,
+    pub cache: Arc<SemanticCache>,
     pub secrets: SecretStore,
     pub config_path: String,
     pub pipeline: crate::pipeline::Pipeline,
@@ -59,10 +61,12 @@ impl SharedState {
         let secrets_path = sibling(&config_path, "secrets.toml");
         let secrets = SecretStore::load(&secrets_path);
         let stats = Stats::new(&cfg.stats);
+        let cache = SemanticCache::new(&cfg.cache);
         let live = Live::build(cfg, &secrets)?;
         Ok(SharedState(Arc::new(AppState {
             live: ArcSwap::from_pointee(live),
             stats,
+            cache,
             secrets,
             config_path,
             pipeline: crate::pipeline::Pipeline::new(),

web/js/api.js

@@ -91,3 +91,51 @@ export const api = {
   // playground
   test: (b) => j("POST", "/test", b),
 };
+
+// Streaming playground: POST /test/stream, calls onDelta(text) per content chunk,
+// returns the routing decision read from X-Cortiq-* response headers.
+export async function testStream(body, onDelta) {
+  const r = await fetch(BASE + "/test/stream", {
+    method: "POST",
+    headers: { Authorization: "Bearer " + getToken(), "content-type": "application/json" },
+    body: JSON.stringify(body),
+  });
+  if (r.status === 401) {
+    if (onUnauthorized) onUnauthorized();
+    throw new Error("unauthorized");
+  }
+  if (!r.ok || !r.body) {
+    throw new Error((await r.text()) || r.statusText);
+  }
+  const reader = r.body.getReader();
+  const dec = new TextDecoder();
+  let buf = "";
+  for (;;) {
+    const { done, value } = await reader.read();
+    if (done) break;
+    buf += dec.decode(value, { stream: true });
+    let idx;
+    while ((idx = buf.indexOf("\n\n")) >= 0) {
+      const line = buf.slice(0, idx).split("\n").find((l) => l.startsWith("data:"));
+      buf = buf.slice(idx + 2);
+      if (!line) continue;
+      const data = line.slice(5).trim();
+      if (!data || data === "[DONE]") continue;
+      try {
+        const d = JSON.parse(data).choices?.[0]?.delta?.content;
+        if (d) onDelta(d);
+      } catch {
+        /* ignore malformed chunk */
+      }
+    }
+  }
+  const h = r.headers;
+  return {
+    task_label: h.get("x-cortiq-task-label") || "",
+    tier: h.get("x-cortiq-complexity-tier") || "",
+    score: parseFloat(h.get("x-cortiq-complexity-score") || "0"),
+    selected_model: h.get("x-cortiq-selected-model") || "",
+    route_source: h.get("x-cortiq-route-source") || "",
+    cost_usd: parseFloat(h.get("x-cortiq-cost-usd") || "0"),
+  };
+}

web/js/i18n/de.js

@@ -200,6 +200,7 @@ export default {
   "pg.temp": "Temperatur",
   "pg.maxTokens": "Max. Tokens",
   "pg.send": "Senden",
+  "pg.stream": "Streamen",
   "pg.sending": "Wird gesendet…",
   "pg.answer": "Antwort",
   "pg.empty": "Antwort und Routing-Entscheidung erscheinen hier.",
@@ -253,4 +254,15 @@ export default {
   "toast.deleted": "Gelöscht",
   "toast.probing": "Prüfe…",
   "toast.imported": "Konfiguration importiert",
+  "dash.cache": "Semantischer Cache",
+  "dash.cache.hitRate": "Trefferquote",
+  "dash.cache.hits": "Treffer",
+  "dash.cache.saved": "Gespart",
+  "dash.cache.disabled": "Semantischer Cache ist deaktiviert — in den Einstellungen aktivieren.",
+  "settings.cache": "Semantischer Cache",
+  "settings.cache.enabled": "Semantischen Cache aktivieren",
+  "settings.cache.threshold": "Ähnlichkeitsschwelle",
+  "settings.cache.ttl": "Eintrags-TTL",
+  "settings.cache.embed": "Embedding-Modell",
+  "settings.cache.note": "Cache-Einstellungen werden nach einem Neustart wirksam.",
 };

web/js/i18n/en.js

@@ -201,6 +201,7 @@ export default {
   "pg.temp": "Temperature",
   "pg.maxTokens": "Max tokens",
   "pg.send": "Send",
+  "pg.stream": "Stream",
   "pg.sending": "Sending…",
   "pg.answer": "Answer",
   "pg.empty": "The answer and routing decision will appear here.",
@@ -254,4 +255,15 @@ export default {
   "toast.deleted": "Deleted",
   "toast.probing": "Probing…",
   "toast.imported": "Config imported",
+  "dash.cache": "Semantic cache",
+  "dash.cache.hitRate": "Hit rate",
+  "dash.cache.hits": "Hits",
+  "dash.cache.saved": "Saved",
+  "dash.cache.disabled": "Semantic cache is disabled — enable it in Settings.",
+  "settings.cache": "Semantic cache",
+  "settings.cache.enabled": "Enable semantic cache",
+  "settings.cache.threshold": "Similarity threshold",
+  "settings.cache.ttl": "Entry TTL",
+  "settings.cache.embed": "Embedding model",
+  "settings.cache.note": "Cache settings take effect after a restart.",
 };

web/js/i18n/es.js

@@ -200,6 +200,7 @@ export default {
   "pg.temp": "Temperatura",
   "pg.maxTokens": "Tokens máx.",
   "pg.send": "Enviar",
+  "pg.stream": "Streaming",
   "pg.sending": "Enviando…",
   "pg.answer": "Respuesta",
   "pg.empty": "La respuesta y la decisión de enrutamiento aparecerán aquí.",
@@ -253,4 +254,15 @@ export default {
   "toast.deleted": "Eliminado",
   "toast.probing": "Sondeando…",
   "toast.imported": "Configuración importada",
+  "dash.cache": "Caché semántico",
+  "dash.cache.hitRate": "Aciertos",
+  "dash.cache.hits": "Aciertos",
+  "dash.cache.saved": "Ahorrado",
+  "dash.cache.disabled": "El caché semántico está desactivado — actívalo en Ajustes.",
+  "settings.cache": "Caché semántico",
+  "settings.cache.enabled": "Activar caché semántico",
+  "settings.cache.threshold": "Umbral de similitud",
+  "settings.cache.ttl": "TTL de entrada",
+  "settings.cache.embed": "Modelo de embeddings",
+  "settings.cache.note": "Los ajustes del caché se aplican tras reiniciar.",
 };

web/js/i18n/fr.js

@@ -200,6 +200,7 @@ export default {
   "pg.temp": "Température",
   "pg.maxTokens": "Tokens max",
   "pg.send": "Envoyer",
+  "pg.stream": "Streaming",
   "pg.sending": "Envoi…",
   "pg.answer": "Réponse",
   "pg.empty": "La réponse et la décision de routage s'afficheront ici.",
@@ -253,4 +254,15 @@ export default {
   "toast.deleted": "Supprimé",
   "toast.probing": "Sondage…",
   "toast.imported": "Config importée",
+  "dash.cache": "Cache sémantique",
+  "dash.cache.hitRate": "Taux de succès",
+  "dash.cache.hits": "Succès",
+  "dash.cache.saved": "Économisé",
+  "dash.cache.disabled": "Le cache sémantique est désactivé — activez-le dans les Paramètres.",
+  "settings.cache": "Cache sémantique",
+  "settings.cache.enabled": "Activer le cache sémantique",
+  "settings.cache.threshold": "Seuil de similarité",
+  "settings.cache.ttl": "TTL d'entrée",
+  "settings.cache.embed": "Modèle d'embeddings",
+  "settings.cache.note": "Les paramètres du cache prennent effet après un redémarrage.",
 };

web/js/i18n/ru.js

@@ -200,6 +200,7 @@ export default {
   "pg.temp": "Температура",
   "pg.maxTokens": "Макс. токены",
   "pg.send": "Отправить",
+  "pg.stream": "Стриминг",
   "pg.sending": "Отправка…",
   "pg.answer": "Ответ",
   "pg.empty": "Ответ и решение роутера появятся здесь.",
@@ -253,4 +254,15 @@ export default {
   "toast.deleted": "Удалено",
   "toast.probing": "Проверка…",
   "toast.imported": "Конфигурация импортирована",
+  "dash.cache": "Семантический кэш",
+  "dash.cache.hitRate": "Попаданий",
+  "dash.cache.hits": "Попадания",
+  "dash.cache.saved": "Сэкономлено",
+  "dash.cache.disabled": "Семантический кэш выключен — включите в Настройках.",
+  "settings.cache": "Семантический кэш",
+  "settings.cache.enabled": "Включить семантический кэш",
+  "settings.cache.threshold": "Порог сходства",
+  "settings.cache.ttl": "TTL записи",
+  "settings.cache.embed": "Модель эмбеддингов",
+  "settings.cache.note": "Настройки кэша применяются после перезапуска.",
 };

web/js/i18n/tr.js

@@ -200,6 +200,7 @@ export default {
   "pg.temp": "Sıcaklık",
   "pg.maxTokens": "Maks. token",
   "pg.send": "Gönder",
+  "pg.stream": "Akış",
   "pg.sending": "Gönderiliyor…",
   "pg.answer": "Yanıt",
   "pg.empty": "Yanıt ve yönlendirme kararı burada görünecek.",
@@ -253,4 +254,15 @@ export default {
   "toast.deleted": "Silindi",
   "toast.probing": "Sınanıyor…",
   "toast.imported": "Yapılandırma içe aktarıldı",
+  "dash.cache": "Anlamsal önbellek",
+  "dash.cache.hitRate": "İsabet oranı",
+  "dash.cache.hits": "İsabet",
+  "dash.cache.saved": "Tasarruf",
+  "dash.cache.disabled": "Anlamsal önbellek kapalı — Ayarlar'dan etkinleştirin.",
+  "settings.cache": "Anlamsal önbellek",
+  "settings.cache.enabled": "Anlamsal önbelleği etkinleştir",
+  "settings.cache.threshold": "Benzerlik eşiği",
+  "settings.cache.ttl": "Giriş TTL",
+  "settings.cache.embed": "Gömme modeli",
+  "settings.cache.note": "Anlamsal önbellek ayarları yeniden başlatmadan sonra geçerli olur.",
 };

web/js/i18n/zh.js

@@ -200,6 +200,7 @@ export default {
   "pg.temp": "温度",
   "pg.maxTokens": "最大 Token 数",
   "pg.send": "发送",
+  "pg.stream": "流式",
   "pg.sending": "发送中…",
   "pg.answer": "回答",
   "pg.empty": "回答和路由决策将在此显示。",
@@ -253,4 +254,15 @@ export default {
   "toast.deleted": "已删除",
   "toast.probing": "探测中…",
   "toast.imported": "配置已导入",
+  "dash.cache": "语义缓存",
+  "dash.cache.hitRate": "命中率",
+  "dash.cache.hits": "命中",
+  "dash.cache.saved": "已节省",
+  "dash.cache.disabled": "语义缓存已禁用 — 在设置中启用。",
+  "settings.cache": "语义缓存",
+  "settings.cache.enabled": "启用语义缓存",
+  "settings.cache.threshold": "相似度阈值",
+  "settings.cache.ttl": "条目 TTL",
+  "settings.cache.embed": "嵌入模型",
+  "settings.cache.note": "缓存设置将在重启后生效。",
 };

web/js/views/dashboard.js

@@ -108,6 +108,22 @@ export async function renderDashboard() {
         barList(breakdown, num)
       ),
 
+      // semantic cache
+      stats.cache && stats.cache.enabled
+        ? h(
+            "div",
+            { class: "card" },
+            h("div", { class: "card-head" }, h("h3", {}, t("dash.cache"))),
+            h(
+              "div",
+              { class: "kpis" },
+              kpi(pct(stats.cache.hit_rate), t("dash.cache.hitRate")),
+              kpi(num(stats.cache.hits), t("dash.cache.hits")),
+              kpi(money(stats.cache.saved_usd), t("dash.cache.saved"))
+            )
+          )
+        : null,
+
       // recent
       h(
         "div",

web/js/views/playground.js

@@ -1,7 +1,7 @@
 // Playground — send a prompt through the live pipeline, inspect routing decision.
-import { h, mount, money, ms, num } from "../ui.js";
+import { h, mount, money, ms } from "../ui.js";
 import { t } from "../i18n.js";
-import { api } from "../api.js";
+import { api, testStream } from "../api.js";
 import { appState } from "../app.js";
 
 export async function renderPlayground() {
@@ -16,6 +16,7 @@ export async function renderPlayground() {
   const promptIn = h("textarea", { placeholder: t("pg.promptPlaceholder"), rows: 5 });
   const tempIn = h("input", { type: "number", step: "0.1", value: 0.7 });
   const maxTokIn = h("input", { type: "number", value: 512 });
+  const streamCb = h("input", { type: "checkbox", checked: true });
   const profileSel = h("select", {}, ...(meta.profiles || []).map((p) => h("option", { value: p }, p)));
   const modelSel = h("select", {}, ...ids.map((id) => h("option", { value: id }, id)));
 
@@ -62,19 +63,43 @@ export async function renderPlayground() {
     sendBtn.disabled = true;
     sendBtn.textContent = t("pg.sending");
     mount(out, h("div", { class: "flex" }, h("span", { class: "spinner" }), " ", t("pg.sending")));
+    const body = {
+      model,
+      messages,
+      temperature: parseFloat(tempIn.value),
+      max_tokens: parseInt(maxTokIn.value) || null,
+    };
     try {
-      const r = await api.test({
-        model,
-        messages,
-        temperature: parseFloat(tempIn.value),
-        max_tokens: parseInt(maxTokIn.value) || null,
-      });
-      if (r.ok) {
-        mount(out, r.answer || h("span", { class: "muted" }, "—"));
-        renderDecision(r);
+      if (streamCb.checked) {
+        const started = performance.now();
+        let acc = "";
+        out.textContent = "";
+        const info = await testStream(body, (d) => {
+          acc += d;
+          out.textContent = acc;
+        });
+        out.textContent = acc || "—";
+        renderDecision({
+          cortiq: {
+            task_label: info.task_label,
+            complexity: { score: info.score, tier: info.tier },
+            selected_model: info.selected_model,
+            route_source: info.route_source,
+            cost_usd: info.cost_usd,
+            failover: false,
+          },
+          usage: {},
+          latency_ms: Math.round(performance.now() - started),
+        });
       } else {
-        mount(out, h("div", { class: "callout" }, h("b", {}, t("pg.error") + ": "), r.error || ""));
-        mount(decision, h("div", { class: "card-head" }, h("h3", {}, t("pg.decision"))), h("div", { class: "empty" }, "—"));
+        const r = await api.test(body);
+        if (r.ok) {
+          mount(out, r.answer || h("span", { class: "muted" }, "—"));
+          renderDecision(r);
+        } else {
+          mount(out, h("div", { class: "callout" }, h("b", {}, t("pg.error") + ": "), r.error || ""));
+          mount(decision, h("div", { class: "card-head" }, h("h3", {}, t("pg.decision"))), h("div", { class: "empty" }, "—"));
+        }
       }
     } catch (e) {
       mount(out, h("div", { class: "callout" }, String(e.message || e)));
@@ -126,7 +151,12 @@ export async function renderPlayground() {
           h("label", { class: "field" }, h("span", {}, t("pg.mode")), modeSeg),
           modeExtra,
           h("div", { class: "row" }, h("label", { class: "field" }, h("span", {}, t("pg.temp")), tempIn), h("label", { class: "field" }, h("span", {}, t("pg.maxTokens")), maxTokIn)),
-          h("div", { class: "flex", style: "justify-content:flex-end" }, sendBtn)
+          h(
+            "div",
+            { class: "flex", style: "justify-content:space-between" },
+            h("label", { class: "check", style: "margin:0" }, streamCb, t("pg.stream")),
+            sendBtn
+          )
         ),
         h("div", { class: "card" }, h("div", { class: "card-head" }, h("h3", {}, t("pg.answer"))), out)
       ),

web/js/views/settings.js

@@ -42,6 +42,18 @@ export async function renderSettings() {
   const statsEnabled = check(t("settings.stats.enabled"), s.stats?.enabled);
   const statsFile = h("input", { value: s.stats?.file || "" });
   const statsRet = h("input", { value: s.stats?.retention || "7d" });
+  // cache
+  const modelsData = await api.listModels();
+  const embedModels = (modelsData.models || []).filter((m) => m.kind === "embedding").map((m) => m.id);
+  const cacheEnabled = check(t("settings.cache.enabled"), s.cache?.enabled);
+  const cacheThresh = h("input", { type: "number", step: "0.01", value: s.cache?.threshold ?? 0.92 });
+  const cacheTtlIn = h("input", { value: s.cache?.ttl || "1h" });
+  const cacheEmbedSel = h(
+    "select",
+    {},
+    h("option", { value: "" }, "auto"),
+    ...embedModels.map((id) => h("option", { value: id, selected: id === s.cache?.embed_model }, id))
+  );
 
   const saveBtn = h("button", { class: "btn primary" }, t("common.save"));
   saveBtn.addEventListener("click", async () => {
@@ -71,11 +83,19 @@ export async function renderSettings() {
         retention: statsRet.value.trim() || "7d",
         ring_size: s.stats?.ring_size ?? 500,
       },
+      cache: {
+        enabled: cacheEnabled.cb.checked,
+        threshold: parseFloat(cacheThresh.value) || 0.92,
+        ttl: cacheTtlIn.value.trim() || "1h",
+        max_entries: s.cache?.max_entries ?? 1000,
+        embed_model: cacheEmbedSel.value || null,
+      },
     };
     // serde: drop null optionals
     if (!patch.router.api_key_env) delete patch.router.api_key_env;
     if (!patch.router.taxonomy_id) delete patch.router.taxonomy_id;
     if (!patch.telemetry.otlp_endpoint_env) delete patch.telemetry.otlp_endpoint_env;
+    if (!patch.cache.embed_model) delete patch.cache.embed_model;
     try {
       const r = await api.putSettings(patch);
       toast(r.needs_restart ? t("settings.needsRestart") : t("toast.saved"), r.needs_restart ? "warn" : "good");
@@ -170,6 +190,15 @@ export async function renderSettings() {
         statsEnabled.node,
         field(t("settings.stats.file"), statsFile),
         field(t("settings.stats.retention"), statsRet)
+      ),
+      h(
+        "div",
+        { class: "card" },
+        h("div", { class: "card-head" }, h("h3", {}, t("settings.cache"))),
+        cacheEnabled.node,
+        h("div", { class: "row" }, field(t("settings.cache.threshold"), cacheThresh), field(t("settings.cache.ttl"), cacheTtlIn)),
+        field(t("settings.cache.embed"), cacheEmbedSel),
+        h("div", { class: "hint" }, t("settings.cache.note"))
       )
     ),
     h(