feat(ift): EventHub AI Router stack for Zed

FastAPI gateway with A/B/C lane orchestration, LiteLLM proxy config,
Swarm stack (postgres, redis, VPN off-by-default), deploy/smoke/audit scripts.

router/Dockerfile

@@ -0,0 +1,22 @@
+FROM python:3.12-slim
+
+WORKDIR /app
+
+RUN apt-get update && apt-get install -y --no-install-recommends curl \
+    && rm -rf /var/lib/apt/lists/*
+
+COPY router/requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY router/*.py ./
+COPY config /app/config
+
+ENV CONFIG_DIR=/app/config
+ENV PYTHONUNBUFFERED=1
+
+EXPOSE 8000
+
+HEALTHCHECK --interval=30s --timeout=5s --retries=3 \
+  CMD curl -sf http://127.0.0.1:8000/health || exit 1
+
+CMD ["uvicorn", "router:app", "--host", "0.0.0.0", "--port", "8000"]

router/metrics.py

@@ -0,0 +1,31 @@
+"""Prometheus metrics for gateway tier/lane routing."""
+
+from __future__ import annotations
+
+from prometheus_client import CONTENT_TYPE_LATEST, Counter, Histogram, generate_latest
+
+REQUESTS = Counter(
+    "ai_router_requests_total",
+    "Total routed requests",
+    ["tier", "lane", "model", "status"],
+)
+ESCALATIONS = Counter(
+    "ai_router_escalations_total",
+    "Lane escalations",
+    ["from_lane", "to_lane", "reason"],
+)
+CLASSIFY = Counter(
+    "ai_router_classify_total",
+    "Classification results",
+    ["tier"],
+)
+DURATION = Histogram(
+    "ai_router_request_duration_seconds",
+    "Request duration",
+    ["tier", "model"],
+    buckets=(0.1, 0.25, 0.5, 1, 2, 5, 10, 30, 60, 120, 180),
+)
+
+
+def metrics_payload() -> tuple[bytes, str]:
+    return generate_latest(), CONTENT_TYPE_LATEST

router/orchestrator.py

@@ -0,0 +1,300 @@
+"""Lane orchestration A/B/C with Redis session context."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+import os
+import re
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import Any
+
+from rules_loader import load_model_matrix, load_orchestration, load_routing_rules
+
+log = logging.getLogger("orchestrator")
+
+LANE_PREFIX = {"A": "a", "B": "b", "C": "c"}
+TIER_SUFFIX = {
+    "SIMPLE": "simple",
+    "MEDIUM_OPS": "medium-ops",
+    "MEDIUM_CODE": "medium-code",
+    "COMPLEX": "complex",
+    "REASONING": "reasoning",
+    "VISION_OCR": "vision-ocr",
+    "VISION_UI": "vision",
+}
+
+
+class Tier(str, Enum):
+    SIMPLE = "SIMPLE"
+    MEDIUM_OPS = "MEDIUM_OPS"
+    MEDIUM_CODE = "MEDIUM_CODE"
+    COMPLEX = "COMPLEX"
+    REASONING = "REASONING"
+    VISION_OCR = "VISION_OCR"
+    VISION_UI = "VISION_UI"
+
+
+@dataclass
+class SessionCtx:
+    escalation_level: int = 0
+    last_prompt_hash: str = ""
+    turn_count: int = 0
+    budget_pct: float = 0.0
+
+
+@dataclass
+class RouteDecision:
+    tier: Tier
+    lane: str
+    model: str
+    quality_mode: str
+    escalation_level: int
+    confidence: float
+    delegated_internal: bool = False
+
+
+class SessionStore:
+    def __init__(self, redis_url: str | None) -> None:
+        self._redis = None
+        if redis_url:
+            try:
+                import redis
+
+                self._redis = redis.from_url(redis_url, decode_responses=True)
+                self._redis.ping()
+            except Exception as exc:  # noqa: BLE001
+                log.warning("Redis unavailable (%s), using in-memory sessions", exc)
+        self._memory: dict[str, SessionCtx] = {}
+        orch = load_orchestration()
+        self._prefix = orch.get("redis", {}).get("key_prefix", "ai-router:session:")
+        self._ttl = int(orch.get("redis", {}).get("ttl_sec", 1800))
+
+    def get(self, session_id: str) -> SessionCtx:
+        if not session_id:
+            return SessionCtx()
+        key = f"{self._prefix}{session_id}"
+        if self._redis:
+            raw = self._redis.get(key)
+            if raw:
+                data = json.loads(raw)
+                return SessionCtx(**data)
+            return SessionCtx()
+        return self._memory.get(session_id, SessionCtx())
+
+    def save(self, session_id: str, ctx: SessionCtx) -> None:
+        if not session_id:
+            return
+        key = f"{self._prefix}{session_id}"
+        payload = json.dumps(
+            {
+                "escalation_level": ctx.escalation_level,
+                "last_prompt_hash": ctx.last_prompt_hash,
+                "turn_count": ctx.turn_count,
+                "budget_pct": ctx.budget_pct,
+            }
+        )
+        if self._redis:
+            self._redis.setex(key, self._ttl, payload)
+        else:
+            self._memory[session_id] = ctx
+
+
+class Classifier:
+    def __init__(self) -> None:
+        rules = load_routing_rules()
+        gw = rules.get("gateway", {})
+        self._simple_re = re.compile(
+            gw.get("simple_patterns", [r"^(hello)\\b"])[0],
+            re.IGNORECASE,
+        )
+        self._complex = self._kw_re(gw.get("complex_keywords", []))
+        self._reasoning = self._kw_re(gw.get("reasoning_keywords", []))
+        self._medium_ops = self._kw_re(gw.get("medium_ops_keywords", []))
+        self._ocr = self._kw_re(gw.get("ocr_keywords", []))
+        self._escalation = self._kw_re(gw.get("escalation_keywords", []))
+        wt = gw.get("word_thresholds", {})
+        self._simple_max = int(wt.get("simple_max_words", 12))
+        self._simple_q_max = int(wt.get("simple_question_max_words", 25))
+        self._complex_min = int(wt.get("complex_min_words", 400))
+        self._low_conf = float(rules.get("gateway", {}).get("confidence", {}).get("low_threshold", 0.6))
+        self._code_block = re.compile(r"```[\s\S]*?```|`[^`]+`")
+
+    @staticmethod
+    def _kw_re(keywords: list[str]) -> re.Pattern[str]:
+        if not keywords:
+            return re.compile(r"(?!x)x")
+        escaped = [re.escape(k) for k in keywords]
+        return re.compile("|".join(escaped), re.IGNORECASE)
+
+    def wants_escalation(self, text: str, prompt_hash: str, prev_hash: str) -> bool:
+        if self._escalation.search(text):
+            return True
+        return bool(prompt_hash and prompt_hash == prev_hash)
+
+    def classify(
+        self,
+        messages: list[dict[str, Any]],
+        *,
+        has_image: bool,
+        text: str,
+    ) -> tuple[Tier, float]:
+        if has_image:
+            if self._ocr.search(text) or len(text.split()) < 30:
+                return Tier.VISION_OCR, 0.95
+            return Tier.VISION_UI, 0.9
+
+        text = text.strip()
+        if not text:
+            return Tier.SIMPLE, 0.9
+
+        words = len(text.split())
+        if words <= self._simple_max and self._simple_re.search(text):
+            return Tier.SIMPLE, 0.92
+        if self._reasoning.search(text):
+            return Tier.REASONING, 0.88
+        if self._complex.search(text):
+            return Tier.COMPLEX, 0.88
+        if self._code_block.search(text):
+            return Tier.MEDIUM_CODE, 0.85
+        if self._medium_ops.search(text):
+            return Tier.MEDIUM_OPS, 0.82
+        if words <= self._simple_q_max and "?" in text and not self._code_block.search(text):
+            return Tier.SIMPLE, 0.75
+        if words >= self._complex_min:
+            return Tier.COMPLEX, 0.7
+        return Tier.MEDIUM_OPS, 0.55
+
+
+class Orchestrator:
+    def __init__(self, session_store: SessionStore) -> None:
+        self.sessions = session_store
+        self.classifier = Classifier()
+        self._orch = load_orchestration()
+        self._matrix = load_model_matrix()
+        self._models: dict[str, dict] = self._matrix.get("models", {})
+        self._no_esc = {
+            name for name, cfg in self._models.items() if cfg.get("no_escalation")
+        }
+        self.default_quality = os.environ.get(
+            "DEFAULT_QUALITY_MODE",
+            self._orch.get("default_quality_mode", "auto"),
+        )
+
+    def _lane_from_mode(self, quality_mode: str) -> str | None:
+        if quality_mode == "auto":
+            return None
+        return self._orch.get("quality_mode_map", {}).get(quality_mode)
+
+    def _start_lane(self, tier: Tier) -> str:
+        return self._orch.get("start_lanes", {}).get(tier.value, "A")
+
+    def _bump_lane(self, lane: str, levels: int) -> str:
+        order: list[str] = self._orch.get("lane_order", ["A", "B", "C"])
+        try:
+            idx = order.index(lane)
+        except ValueError:
+            return lane
+        return order[min(idx + levels, len(order) - 1)]
+
+    def _cap_lane(self, lane: str, budget_pct: float) -> str:
+        caps = self._orch.get("budget_caps", {})
+        hard = float(caps.get("hard_pct", 95))
+        warn = float(caps.get("warn_pct", 80))
+        if budget_pct >= hard:
+            return caps.get("hard_max_lane", "A")
+        if budget_pct >= warn:
+            max_lane = caps.get("warn_max_lane", "B")
+            order = self._orch.get("lane_order", ["A", "B", "C"])
+            if order.index(lane) > order.index(max_lane):
+                return max_lane
+        return lane
+
+    def _model_name(self, lane: str, tier: Tier) -> str:
+        prefix = LANE_PREFIX.get(lane, "a")
+        suffix = TIER_SUFFIX[tier]
+        return f"{prefix}-{suffix}"
+
+    def resolve(
+        self,
+        messages: list[dict[str, Any]],
+        *,
+        quality_mode: str | None = None,
+        session_id: str = "",
+        text: str = "",
+        has_image: bool = False,
+        token_estimate: int = 0,
+    ) -> RouteDecision:
+        mode = quality_mode or self.default_quality
+        tier, confidence = self.classifier.classify(messages, has_image=has_image, text=text)
+        ctx = self.sessions.get(session_id)
+        prompt_hash = hashlib.sha256(text.encode()).hexdigest()[:16]
+
+        if mode != "auto":
+            lane = self._lane_from_mode(mode) or "A"
+            escalation = 0
+        else:
+            lane = self._start_lane(tier)
+            escalation = ctx.escalation_level
+            if tier != Tier.VISION_OCR and self.classifier.wants_escalation(
+                text, prompt_hash, ctx.last_prompt_hash
+            ):
+                escalation += 1
+            min_tokens = int(
+                self._orch.get("escalation", {}).get("context_tokens_min_lane_b", 32000)
+            )
+            if token_estimate > min_tokens:
+                order = self._orch.get("lane_order", ["A", "B", "C"])
+                if order.index(lane) < order.index("B"):
+                    lane = "B"
+            lane = self._bump_lane(lane, escalation)
+            lane = self._cap_lane(lane, ctx.budget_pct)
+
+        model = self._model_name(lane, tier)
+        if model not in self._models:
+            log.warning("model %s missing from matrix, fallback smart-router-internal", model)
+            return RouteDecision(
+                tier=tier,
+                lane=lane,
+                model="smart-router-internal",
+                quality_mode=mode,
+                escalation_level=escalation,
+                confidence=confidence,
+                delegated_internal=True,
+            )
+
+        delegated = confidence < self.classifier._low_conf and tier in (
+            Tier.MEDIUM_OPS,
+            Tier.SIMPLE,
+        )
+
+        return RouteDecision(
+            tier=tier,
+            lane=lane,
+            model="smart-router-internal" if delegated else model,
+            quality_mode=mode,
+            escalation_level=escalation,
+            confidence=confidence,
+            delegated_internal=delegated,
+        )
+
+    def after_request(
+        self,
+        session_id: str,
+        *,
+        prompt_hash: str,
+        success: bool,
+        escalate: bool,
+    ) -> None:
+        if not session_id:
+            return
+        ctx = self.sessions.get(session_id)
+        ctx.last_prompt_hash = prompt_hash
+        ctx.turn_count += 1
+        if success and not escalate:
+            ctx.escalation_level = 0
+        elif escalate:
+            ctx.escalation_level = min(ctx.escalation_level + 1, 2)
+        self.sessions.save(session_id, ctx)

router/requirements.txt

@@ -0,0 +1,6 @@
+fastapi>=0.115.0
+uvicorn[standard]>=0.32.0
+httpx>=0.27.0
+pyyaml>=6.0.2
+prometheus-client>=0.21.0
+redis>=5.2.0

router/router.py

@@ -0,0 +1,288 @@
+"""
+FastAPI AI router gateway — Zed entrypoint with tier/lane orchestration.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+import os
+import time
+from typing import Any
+
+import httpx
+from fastapi import FastAPI, Header, HTTPException, Request, Response
+from fastapi.responses import JSONResponse, StreamingResponse
+
+from metrics import CLASSIFY, DURATION, ESCALATIONS, REQUESTS, metrics_payload
+from orchestrator import Orchestrator, SessionStore, Tier
+from rules_loader import reload_configs
+
+logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
+log = logging.getLogger("ai-router")
+
+LITELLM_URL = os.environ.get("LITELLM_INTERNAL_URL", "http://litellm:4000").rstrip("/")
+LITELLM_KEY = os.environ.get("LITELLM_MASTER_KEY", "")
+ROUTER_API_KEY = os.environ.get("ROUTER_API_KEY", "")
+DEFAULT_MODEL = os.environ.get("DEFAULT_LITELLM_MODEL", "smart-router")
+REDIS_URL = os.environ.get("REDIS_URL", "redis://redis:6379/0")
+
+session_store = SessionStore(REDIS_URL or None)
+orchestrator = Orchestrator(session_store)
+
+app = FastAPI(title="EventHub AI Router", version="2.0.0")
+
+
+def _extract_text(messages: list[dict[str, Any]]) -> str:
+    parts: list[str] = []
+    for msg in messages:
+        content = msg.get("content", "")
+        if isinstance(content, str):
+            parts.append(content)
+        elif isinstance(content, list):
+            for block in content:
+                if isinstance(block, dict) and block.get("type") == "text":
+                    parts.append(str(block.get("text", "")))
+    return "\n".join(parts)
+
+
+def _has_image(messages: list[dict[str, Any]]) -> bool:
+    for msg in messages:
+        content = msg.get("content", "")
+        if not isinstance(content, list):
+            continue
+        for block in content:
+            if isinstance(block, dict) and block.get("type") in ("image_url", "image"):
+                return True
+    return False
+
+
+def _estimate_tokens(text: str) -> int:
+    return max(1, len(text) // 4)
+
+
+def _auth_or_403(authorization: str | None) -> None:
+    if not ROUTER_API_KEY:
+        return
+    if not authorization or not authorization.startswith("Bearer "):
+        raise HTTPException(status_code=401, detail="Missing Bearer token")
+    token = authorization.removeprefix("Bearer ").strip()
+    if token != ROUTER_API_KEY:
+        raise HTTPException(status_code=403, detail="Invalid API key")
+
+
+def _quality_mode(header: str | None, body: dict[str, Any]) -> str | None:
+    if header:
+        mode = header.strip().lower()
+        if mode in ("auto", "economy", "balanced", "max"):
+            return mode
+    meta = body.get("metadata") or {}
+    if isinstance(meta, dict):
+        mode = str(meta.get("quality_mode", "")).lower()
+        if mode in ("auto", "economy", "balanced", "max"):
+            return mode
+    return None
+
+
+def _session_id(body: dict[str, Any]) -> str:
+    meta = body.get("metadata") or {}
+    if isinstance(meta, dict) and meta.get("session_id"):
+        return str(meta["session_id"])
+    if body.get("user"):
+        return str(body["user"])
+    return ""
+
+
+def _router_meta(decision, *, requested: str) -> dict[str, Any]:
+    return {
+        "tier": decision.tier.value,
+        "lane": decision.lane,
+        "model": decision.model,
+        "escalation_level": decision.escalation_level,
+        "quality_mode": decision.quality_mode,
+        "confidence": round(decision.confidence, 3),
+        "requested_model": requested,
+        "delegated_internal": decision.delegated_internal,
+    }
+
+
+@app.get("/health")
+async def health() -> dict[str, str]:
+    return {"status": "ok"}
+
+
+@app.get("/metrics")
+async def metrics() -> Response:
+    body, content_type = metrics_payload()
+    return Response(content=body, media_type=content_type)
+
+
+@app.post("/admin/reload-config")
+async def admin_reload(authorization: str | None = Header(default=None)) -> dict[str, str]:
+    _auth_or_403(authorization)
+    reload_configs()
+    global orchestrator  # noqa: PLW0603
+    orchestrator = Orchestrator(session_store)
+    return {"status": "reloaded"}
+
+
+@app.get("/v1/models")
+async def list_models(authorization: str | None = Header(default=None)) -> dict[str, Any]:
+    _auth_or_403(authorization)
+    async with httpx.AsyncClient(timeout=30.0) as client:
+        resp = await client.get(
+            f"{LITELLM_URL}/v1/models",
+            headers={"Authorization": f"Bearer {LITELLM_KEY}"},
+        )
+    if resp.status_code >= 400:
+        raise HTTPException(status_code=resp.status_code, detail=resp.text)
+    return resp.json()
+
+
+@app.post("/classify")
+async def classify_debug(
+    request: Request,
+    authorization: str | None = Header(default=None),
+    x_ai_quality: str | None = Header(default=None, alias="X-AI-Quality"),
+) -> dict[str, Any]:
+    _auth_or_403(authorization)
+    body = await request.json()
+    messages = body.get("messages") or []
+    text = _extract_text(messages)
+    decision = orchestrator.resolve(
+        messages,
+        quality_mode=_quality_mode(x_ai_quality, body),
+        session_id=_session_id(body),
+        text=text,
+        has_image=_has_image(messages),
+        token_estimate=_estimate_tokens(text),
+    )
+    CLASSIFY.labels(tier=decision.tier.value).inc()
+    return {
+        "tier": decision.tier.value,
+        "lane": decision.lane,
+        "model": decision.model,
+        "x_router_meta": _router_meta(decision, requested=body.get("model", DEFAULT_MODEL)),
+    }
+
+
+async def _forward_litellm(
+    forward: dict[str, Any],
+    *,
+    stream: bool,
+    decision_meta: dict[str, Any],
+) -> Any:
+    headers = {
+        "Authorization": f"Bearer {LITELLM_KEY}",
+        "Content-Type": "application/json",
+    }
+    start = time.perf_counter()
+    tier = decision_meta.get("tier", "UNKNOWN")
+    model = forward.get("model", DEFAULT_MODEL)
+
+    async with httpx.AsyncClient(timeout=180.0) as client:
+        if stream:
+            req = client.build_request(
+                "POST",
+                f"{LITELLM_URL}/v1/chat/completions",
+                headers=headers,
+                content=json.dumps(forward),
+            )
+            resp = await client.send(req, stream=True)
+
+            async def event_stream():
+                async for chunk in resp.aiter_bytes():
+                    yield chunk
+
+            REQUESTS.labels(tier=tier, lane=decision_meta["lane"], model=model, status=str(resp.status_code)).inc()
+            DURATION.labels(tier=tier, model=model).observe(time.perf_counter() - start)
+            return StreamingResponse(
+                event_stream(),
+                status_code=resp.status_code,
+                media_type=resp.headers.get("content-type", "text/event-stream"),
+                headers={"X-Router-Meta": json.dumps(decision_meta)},
+            )
+
+        resp = await client.post(
+            f"{LITELLM_URL}/v1/chat/completions",
+            headers=headers,
+            json=forward,
+        )
+
+    status = str(resp.status_code)
+    REQUESTS.labels(tier=tier, lane=decision_meta["lane"], model=model, status=status).inc()
+    DURATION.labels(tier=tier, model=model).observe(time.perf_counter() - start)
+
+    if resp.status_code >= 400:
+        return JSONResponse(status_code=resp.status_code, content=resp.json())
+
+    data = resp.json()
+    if isinstance(data, dict):
+        data["x_router_meta"] = decision_meta
+    return data
+
+
+@app.post("/v1/chat/completions")
+async def chat_completions(
+    request: Request,
+    authorization: str | None = Header(default=None),
+    x_ai_quality: str | None = Header(default=None, alias="X-AI-Quality"),
+) -> Any:
+    _auth_or_403(authorization)
+
+    body: dict[str, Any] = await request.json()
+    messages = body.get("messages") or []
+    if not isinstance(messages, list):
+        raise HTTPException(status_code=400, detail="messages must be a list")
+
+    text = _extract_text(messages)
+    has_image = _has_image(messages)
+    session_id = _session_id(body)
+    prompt_hash = hashlib.sha256(text.encode()).hexdigest()[:16]
+
+    requested_model = body.get("model", DEFAULT_MODEL)
+    if requested_model in ("smart-router", "auto", ""):
+        decision = orchestrator.resolve(
+            messages,
+            quality_mode=_quality_mode(x_ai_quality, body),
+            session_id=session_id,
+            text=text,
+            has_image=has_image,
+            token_estimate=_estimate_tokens(text),
+        )
+        target_model = decision.model
+        meta = _router_meta(decision, requested=requested_model)
+    else:
+        target_model = requested_model
+        meta = {"selected_model": target_model, "requested_model": requested_model}
+
+    forward = dict(body)
+    forward["model"] = target_model
+    forward.setdefault("metadata", {})
+    if isinstance(forward["metadata"], dict):
+        forward["metadata"]["semantic_tier"] = meta.get("tier")
+        forward["metadata"]["session_id"] = session_id or forward["metadata"].get("session_id")
+
+    log.info(
+        "route tier=%s lane=%s model=%s requested=%s",
+        meta.get("tier"),
+        meta.get("lane"),
+        target_model,
+        requested_model,
+    )
+
+    stream = bool(body.get("stream", False))
+    result = await _forward_litellm(forward, stream=stream, decision_meta=meta)
+
+    escalate = isinstance(result, JSONResponse) and result.status_code >= 429
+    orchestrator.after_request(
+        session_id,
+        prompt_hash=prompt_hash,
+        success=not escalate,
+        escalate=escalate,
+    )
+    if escalate and meta.get("lane") != "C":
+        ESCALATIONS.labels(from_lane=meta.get("lane", "?"), to_lane="next", reason="http_error").inc()
+
+    return result

router/rules_loader.py

@@ -0,0 +1,39 @@
+"""Load routing YAML configs from CONFIG_DIR."""
+
+from __future__ import annotations
+
+import os
+from functools import lru_cache
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+CONFIG_DIR = Path(os.environ.get("CONFIG_DIR", "/app/config"))
+
+
+@lru_cache(maxsize=1)
+def load_routing_rules() -> dict[str, Any]:
+    path = CONFIG_DIR / "routing_rules.yaml"
+    with path.open(encoding="utf-8") as f:
+        return yaml.safe_load(f) or {}
+
+
+@lru_cache(maxsize=1)
+def load_model_matrix() -> dict[str, Any]:
+    path = CONFIG_DIR / "model_matrix.yaml"
+    with path.open(encoding="utf-8") as f:
+        return yaml.safe_load(f) or {}
+
+
+@lru_cache(maxsize=1)
+def load_orchestration() -> dict[str, Any]:
+    path = CONFIG_DIR / "orchestration.yaml"
+    with path.open(encoding="utf-8") as f:
+        return yaml.safe_load(f) or {}
+
+
+def reload_configs() -> None:
+    load_routing_rules.cache_clear()
+    load_model_matrix.cache_clear()
+    load_orchestration.cache_clear()