"""OpenAI-compatible local adapter backed by Gemini CLI for Voyager. Circuit breaker per key: quota -> cooldown (exponential), auth -> disabled. """ from __future__ import annotations import json import os import subprocess import threading import time import urllib.error import urllib.request import uuid from dataclasses import dataclass, field from enum import Enum from typing import Any from fastapi import FastAPI, HTTPException, Request from fastapi.responses import JSONResponse, StreamingResponse from starlette.concurrency import run_in_threadpool MODEL = os.environ.get("GEMINI_VOYAGER_MODEL", "gemini-3.5-flash-lite") if MODEL != "gemini-3.5-flash-lite": raise RuntimeError(f"Gemini Voyager brain lock rejected model: {MODEL}") CLI = os.environ["GEMINI_BOT_CLI"] WORKSPACE = os.environ["GEMINI_BOT_WORKSPACE"] HOME = os.environ["GEMINI_BOT_HOME"] SYSTEM_MD = os.environ["GEMINI_BOT_SYSTEM_MD"] RAW_KEYS = [os.environ.get(f"GEMINI_BOT_KEY_{i}") for i in range(1, 7)] RAW_KEYS = [key for key in RAW_KEYS if key] EMBED_BASE = os.environ.get( "GEMINI_VOYAGER_EMBED_BASE", "https://erinaldordso-mine.hf.space/v1" ) EMBED_RETRIES = max(1, int(os.environ.get("GEMINI_EMBED_RETRIES", "8"))) EMBED_BACKOFF = max(0.1, float(os.environ.get("GEMINI_EMBED_BACKOFF", "1"))) EMBED_RETRYABLE = frozenset({429, 500, 502, 503, 504}) # Verified against @google/gemini-cli@0.61.0: ModelConfigService merges # modelConfigs.customOverrides and passes generateContentConfig straight to the # @google/genai SDK, and the CLI exposes no per-request temperature flag. TEMPERATURE_ENFORCEMENT = ( "settings.json modelConfigs.customOverrides[0].generateContentConfig.temperature=0 " "for model " + MODEL + " (gemini-cli 0.61.0 has no per-request temperature flag)" ) LOCK = threading.Lock() STATE_LOCK = threading.Lock() class KeyState(Enum): HEALTHY = "healthy" QUOTA_LIMITED = "quota_limited" AUTH_FAILED = "auth_failed" TRANSIENT_ERROR = "transient_error" MODEL_UNAVAILABLE = "model_unavailable" @dataclass class KeyStatus: key_index: int state: KeyState = KeyState.HEALTHY failures: int = 0 consecutive_failures: int = 0 last_failure_at: float = 0.0 last_failure_category: str = "" cooldown_until: float = 0.0 total_calls: int = 0 total_successes: int = 0 def is_available(self) -> bool: if self.state == KeyState.AUTH_FAILED: return False if self.state == KeyState.QUOTA_LIMITED: return time.time() >= self.cooldown_until if self.state == KeyState.MODEL_UNAVAILABLE: return False return True def record_success(self) -> None: self.total_calls += 1 self.total_successes += 1 self.consecutive_failures = 0 if self.state in (KeyState.QUOTA_LIMITED, KeyState.TRANSIENT_ERROR): self.state = KeyState.HEALTHY self.cooldown_until = 0.0 def record_failure(self, category: str) -> None: self.total_calls += 1 self.failures += 1 self.consecutive_failures += 1 self.last_failure_at = time.time() self.last_failure_category = category if category == "authentication_or_permission": self.state = KeyState.AUTH_FAILED elif category == "quota_or_rate_limit": self.state = KeyState.QUOTA_LIMITED # Exponential backoff: 30s, 60s, 120s, 240s, 480s, max 900s base = 30 max_backoff = 900 backoff = min(base * (2 ** min(self.consecutive_failures - 1, 5)), max_backoff) self.cooldown_until = time.time() + backoff elif category == "model_unavailable": self.state = KeyState.MODEL_UNAVAILABLE else: self.state = KeyState.TRANSIENT_ERROR # Short backoff for transient self.cooldown_until = time.time() + min(30 * self.consecutive_failures, 120) def to_dict(self) -> dict[str, Any]: return { "key_index": self.key_index, "state": self.state.value, "failures": self.failures, "consecutive_failures": self.consecutive_failures, "last_failure_at": self.last_failure_at or None, "last_failure_category": self.last_failure_category or None, "cooldown_until": self.cooldown_until if self.cooldown_until > time.time() else None, "cooldown_remaining": max(0, int(self.cooldown_until - time.time())) if self.cooldown_until > time.time() else 0, "total_calls": self.total_calls, "total_successes": self.total_successes, "success_rate": round(self.total_successes / max(1, self.total_calls), 3), "available": self.is_available(), } KEY_STATUS = [KeyStatus(i) for i in range(len(RAW_KEYS))] state = { "calls": 0, "failures": 0, "selected_key": 0, "last_elapsed_s": None, "prompt_tokens": 0, "completion_tokens": 0, "last_error": None, "last_failure": None, "last_failure_at": None, "last_failed_key_slot": None, "failures_by_key_slot": [0] * len(RAW_KEYS), "last_failure_category": None, "failure_category_by_key_slot": [None] * len(RAW_KEYS), "last_requested_temperature": None, "effective_temperature": TEMPERATURE_ENFORCEMENT, } def _failure_category(error: Any, stderr: str = "") -> str: """Classify CLI errors without publishing messages, prompts or credentials.""" fields = [error.get(k, "") for k in ("type", "status", "message")] if isinstance(error, dict) else [] detail = (" ".join(map(str, fields)) + " " + str(stderr or "")[:4000]).lower() if any(token in detail for token in ("quota", "resource_exhausted", "rate limit", "too many requests", "429")): return "quota_or_rate_limit" if any(token in detail for token in ("invalid api key", "api key is invalid", "api_key_invalid", "unauthorized", "permission_denied", "401", "403")): return "authentication_or_permission" if any(token in detail for token in ("model not found", "model unavailable", "not supported", "404")): return "model_unavailable" if any(token in detail for token in ("timeout", "timed out", "connection reset", "503", "502")): return "transient_transport" return "unclassified" def _content(value: Any) -> str: if isinstance(value, str): return value if isinstance(value, list): parts = [] for part in value: if isinstance(part, dict) and isinstance(part.get("text"), str): parts.append(part["text"]) return "\n".join(parts) return str(value or "") def _prompt(messages: list[dict[str, Any]]) -> str: rendered = [] for message in messages: role = str(message.get("role", "user")).upper() rendered.append(f"[{role}]\n{_content(message.get('content'))}") return ( "You are the language model inside the Voyager Minecraft agent. " "Follow the supplied system and user instructions exactly. " "Return only the response format requested by the final prompt; no wrapper commentary.\n\n" + "\n\n".join(rendered) ) def _usage(stats: dict[str, Any]) -> tuple[int, int]: prompt = completion = 0 models = stats.get("models") if isinstance(stats, dict) else {} if isinstance(models, dict): for value in models.values(): tokens = value.get("tokens", {}) if isinstance(value, dict) else {} prompt += int(tokens.get("prompt", tokens.get("input", tokens.get("inputTokens", 0))) or 0) completion += int(tokens.get("candidates", tokens.get("output", tokens.get("outputTokens", 0))) or 0) if prompt == 0 and completion == 0 and isinstance(stats, dict): tokens = stats.get("tokens", {}) if isinstance(stats.get("tokens"), dict) else {} prompt = int(tokens.get("prompt", tokens.get("input", tokens.get("inputTokens", 0))) or 0) completion = int(tokens.get("candidates", tokens.get("output", tokens.get("outputTokens", 0))) or 0) return prompt, completion def _select_available_key() -> int | None: """Select next available key using round-robin with state awareness.""" with STATE_LOCK: available = [(i, ks) for i, ks in enumerate(KEY_STATUS) if ks.is_available()] if not available: return None # Prefer healthy keys, then quota-limited (after cooldown) healthy = [i for i, ks in available if ks.state == KeyState.HEALTHY] if healthy: # Round-robin among healthy start = state["selected_key"] % len(healthy) for offset in range(len(healthy)): idx = (start + offset) % len(healthy) state["selected_key"] = healthy[idx] return healthy[idx] # Fall back to first available (quota after cooldown) idx = available[0][0] state["selected_key"] = idx return idx def _call_cli(prompt: str) -> tuple[str, int, int]: if not RAW_KEYS: raise RuntimeError("Gemini credentials unavailable") with LOCK: last = "Gemini CLI failed" max_tries = min(3, len(RAW_KEYS)) tried = 0 while tried < max_tries: slot = _select_available_key() if slot is None: # No keys available - all in cooldown or disabled with STATE_LOCK: # Find earliest cooldown expiry next_ready = min( (ks.cooldown_until for ks in KEY_STATUS if ks.cooldown_until > time.time()), default=time.time() + 60 ) wait = int(next_ready - time.time()) + 1 raise RuntimeError(f"All Gemini keys unavailable. Next key ready in ~{wait}s") ks = KEY_STATUS[slot] with STATE_LOCK: state["calls"] += 1 env = { "PATH": os.environ.get("PATH", "/usr/bin:/bin"), "HOME": HOME, "GEMINI_CLI_HOME": HOME, "GEMINI_CLI_TRUST_WORKSPACE": "true", "GEMINI_SYSTEM_MD": SYSTEM_MD, "GEMINI_TELEMETRY_ENABLED": "false", "GEMINI_TELEMETRY_LOG_PROMPTS": "false", "LANG": "C.UTF-8", "GEMINI_API_KEY": RAW_KEYS[slot], } started = time.monotonic() proc = subprocess.run( [CLI, "--model", MODEL, "--output-format", "json", "--approval-mode", "plan", "--prompt", prompt], cwd=WORKSPACE, env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, timeout=180, check=False, ) elapsed = round(time.monotonic() - started, 1) try: envelope = json.loads(proc.stdout or "{}") except ValueError: envelope = {} if proc.returncode == 0 and isinstance(envelope.get("response"), str): p_tokens, c_tokens = _usage(envelope.get("stats") or {}) ks.record_success() with STATE_LOCK: state["last_elapsed_s"] = elapsed state["prompt_tokens"] += p_tokens state["completion_tokens"] += c_tokens state["last_error"] = None return envelope["response"], p_tokens, c_tokens code = envelope.get("error", {}).get("code") if isinstance(envelope.get("error"), dict) else None safe_code = str(code) if isinstance(code, (str, int)) and str(code).isalnum() and len(str(code)) <= 32 else None last = f"Gemini CLI exit {proc.returncode}" + (f" API {safe_code}" if safe_code else "") category = _failure_category(envelope.get("error"), proc.stderr) ks.record_failure(category) with STATE_LOCK: state["failures"] += 1 state["last_error"] = last state["last_failure"] = last state["last_failure_at"] = int(time.time()) state["last_failed_key_slot"] = slot + 1 state["failures_by_key_slot"][slot] += 1 state["last_failure_category"] = category state["failure_category_by_key_slot"][slot] = category tried += 1 raise RuntimeError(last) def _chat_payload(body: dict[str, Any]) -> dict[str, Any]: requested = body.get("temperature") # Gemini CLI 0.61.0 has no per-request temperature flag. Sampling is fixed # for this model by modelConfigs.customOverrides in the workspace # settings.json written by gemini_bot_stack.py. Record what was asked for # so a mismatch is visible in /gemini-bot/status instead of silent. with STATE_LOCK: state["last_requested_temperature"] = requested state["effective_temperature"] = TEMPERATURE_ENFORCEMENT if isinstance(requested, (int, float)) and float(requested) != 0.0: print( f"[gemini-proxy] requested temperature={requested} but {TEMPERATURE_ENFORCEMENT}", flush=True, ) answer, prompt_tokens, completion_tokens = _call_cli(_prompt(body.get("messages") or [])) return { "id": "chatcmpl-" + uuid.uuid4().hex[:16], "object": "chat.completion", "created": int(time.time()), "model": MODEL, "choices": [{"index": 0, "message": {"role": "assistant", "content": answer}, "finish_reason": "stop"}], "usage": {"prompt_tokens": prompt_tokens, "completion_tokens": completion_tokens, "total_tokens": prompt_tokens + completion_tokens}, } api = FastAPI(title="Gemini Voyager Adapter", version="2.0.0") @api.get("/health") def health(): with STATE_LOCK: snapshot = dict(state) key_states = [ks.to_dict() for ks in KEY_STATUS] return { "ready": any(ks.is_available() for ks in KEY_STATUS), "model": MODEL, "decision_brain": "gemini-cli-only", "embedding_backend": "local-only", "configured_key_count": len(RAW_KEYS), "healthy_key_count": sum(1 for ks in KEY_STATUS if ks.state == KeyState.HEALTHY), "keys": key_states, **snapshot, } @api.get("/v1/models") def models(): return {"object": "list", "data": [{"id": MODEL, "object": "model", "owned_by": "gemini-cli"}]} @api.post("/v1/chat/completions") async def chat(request: Request): body = await request.json() try: payload = await run_in_threadpool(_chat_payload, body) except Exception as exc: raise HTTPException(status_code=503, detail=str(exc)[:160]) from None if not body.get("stream"): return JSONResponse(payload) content = payload["choices"][0]["message"]["content"] chunk = {"id": payload["id"], "object": "chat.completion.chunk", "created": payload["created"], "model": MODEL, "choices": [{"index": 0, "delta": {"role": "assistant", "content": content}, "finish_reason": "stop"}]} async def events(): yield "data: " + json.dumps(chunk, ensure_ascii=False) + "\n\n" yield "data: [DONE]\n\n" return StreamingResponse(events(), media_type="text/event-stream") def _forward_embeddings(body: bytes) -> dict[str, Any]: last_error: Exception | None = None for attempt in range(EMBED_RETRIES): headers = {"Content-Type": "application/json"} api_key = os.environ.get("API_KEY", "").strip() if api_key: headers["Authorization"] = f"Bearer {api_key}" upstream = urllib.request.Request( EMBED_BASE.rstrip("/") + "/embeddings", data=body, headers=headers, method="POST" ) try: with urllib.request.urlopen(upstream, timeout=120) as response: return json.loads(response.read().decode("utf-8")) except urllib.error.HTTPError as exc: last_error = exc if exc.code not in EMBED_RETRYABLE: raise except (urllib.error.URLError, TimeoutError, OSError) as exc: last_error = exc if attempt + 1 < EMBED_RETRIES: time.sleep(min(EMBED_BACKOFF * (2 ** attempt), 8.0)) assert last_error is not None raise last_error @api.post("/v1/embeddings") async def embeddings(request: Request): body = await request.body() try: payload = await run_in_threadpool(_forward_embeddings, body) return JSONResponse(payload) except urllib.error.HTTPError as exc: status = exc.code if 400 <= exc.code < 500 and exc.code not in EMBED_RETRYABLE else 503 raise HTTPException(status_code=status, detail=f"embedding backend unavailable: HTTP {exc.code}") from None except Exception as exc: raise HTTPException(status_code=503, detail=f"embedding backend unavailable: {type(exc).__name__}") from None if __name__ == "__main__": import uvicorn uvicorn.run(api, host="127.0.0.1", port=int(os.environ.get("GEMINI_VOYAGER_PROXY_PORT", "7863")), log_level="warning")