Spaces:

Erinaldordso
/

Minecraft-Friends-AI-ZeroGPU

Paused

Minecraft-Friends-AI-ZeroGPU / gemini_voyager_proxy.py

Harden embedding startup and AltoClef OpenCode runtime

46e3de3 verified about 1 month ago

17.2 kB

"""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")

	"""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")