"""OpenAI-compatible shim so graphify's --backend openai can reach Gemini via Tier-1 OAuth CLI (ADR-114), falling back to local Ollama on any failure. Not a general-purpose proxy — built for graphify's semantic-extraction calls specifically (single system+user message pair, optional inline images). Usage (pointed at by graphify): OPENAI_BASE_URL=http://127.0.0.1:8793/v1 OPENAI_API_KEY=unused \ graphify . --update --backend openai --model gemini-2.5-flash """ import base64 import json import logging import os import signal import subprocess import tempfile import threading import time import urllib.error import urllib.request from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer log = logging.getLogger("graphify-gemini-bridge") logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") PORT = 8793 GCP_PROJECT = "gemini-pwa-360dlm" GEMINI_MODEL_DEFAULT = "gemini-2.5-flash" GEMINI_TIMEOUT_S = 240 OLLAMA_URL = "http://127.0.0.1:11434/v1/chat/completions" OLLAMA_TEXT_MODEL = "gemma3:4b" OLLAMA_VISION_MODEL = "llava" OLLAMA_TIMEOUT_S = 600 # Observed twice: llava (CPU inference) and the gemini-cli node process fight # for the same 2 cores, gemini calls slow past their own timeout, more chunks # fall back to ollama, and the box swaps hard (~150MB free, other tenants on # this VPS affected). Default OFF until that contention is resolved for real # (e.g. dedicated core pinning or a smaller/quantized fallback model). OLLAMA_FALLBACK_ENABLED = os.environ.get("BRIDGE_OLLAMA_FALLBACK", "false").lower() == "true" # Serialize Gemini CLI calls: unknown free-tier daily quota, and graphify # fires up to --max-concurrency (default 4) requests in parallel. One at a # time avoids burning quota on a burst; Ollama absorbs the rest in parallel. _gemini_lock = threading.Lock() def _gemini_env(): env = {**os.environ, "HOME": "/root", "NO_BROWSER": "true", "GEMINI_CLI_TRUST_WORKSPACE": "true", "GOOGLE_CLOUD_PROJECT": GCP_PROJECT} env.pop("GEMINI_API_KEY", None) # a stray key silently bypasses OAuth (same guard as gemini-proxy) return env def _extract_text_and_images(messages): """Flatten system+user messages into one prompt string; pull out the first inline image (data URI), if any — matches graphify's _openai_content shape: content is either a string or a list of {type: text|image_url}.""" parts = [] image_b64 = None image_mime = None for msg in messages: content = msg.get("content") if isinstance(content, str): parts.append(content) elif isinstance(content, list): for block in content: if block.get("type") == "text": parts.append(block.get("text", "")) elif block.get("type") == "image_url" and image_b64 is None: url = block.get("image_url", {}).get("url", "") if url.startswith("data:"): header, _, b64data = url.partition(",") image_mime = header.split(";")[0].removeprefix("data:") or "image/png" image_b64 = b64data return "\n\n".join(p for p in parts if p), image_b64, image_mime _EXT = {"image/jpeg": ".jpg", "image/png": ".png", "image/webp": ".webp"} _STDIN_INSTRUCTION = "Process the appended content exactly as instructed within it." def _call_gemini_cli(model, prompt, image_b64, image_mime): """graphify chunks (up to a 60k-token budget) can blow past the OS ARG_MAX if passed as a -p argument ('Argument list too long') — including image chunks, which carry accompanying text and are NOT reliably small. @file image references only parse when literally inside -p, so -p stays short (just the @file tag + a fixed instruction) and the real prompt — text or image task text — always goes via stdin, which the CLI appends to -p.""" tmp_path = None try: if image_b64: fd, tmp_path = tempfile.mkstemp(suffix=_EXT.get(image_mime, ".png"), dir="/tmp") with os.fdopen(fd, "wb") as f: f.write(base64.b64decode(image_b64)) cli_args = ["gemini", "-m", model, "-p", f"@{tmp_path} {_STDIN_INSTRUCTION}"] else: cli_args = ["gemini", "-m", model, "-p", _STDIN_INSTRUCTION] stdin_data = prompt with _gemini_lock: # gemini CLI (node) spawns a heavier grandchild worker process. # subprocess.run's timeout only SIGKILLs the direct child, leaving # the grandchild orphaned and still running — observed to pile up # 1GB+-RSS zombies, one per timed-out chunk, until the box swapped # solid. start_new_session puts the whole tree in its own process # group so a timeout can kill all of it via killpg. proc = subprocess.Popen( cli_args, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, env=_gemini_env(), cwd="/tmp", start_new_session=True, ) try: stdout, stderr = proc.communicate(input=stdin_data, timeout=GEMINI_TIMEOUT_S) returncode = proc.returncode except subprocess.TimeoutExpired: try: os.killpg(os.getpgid(proc.pid), signal.SIGKILL) except ProcessLookupError: pass proc.wait() raise RuntimeError(f"gemini-cli timed out after {GEMINI_TIMEOUT_S}s (process group killed)") if returncode != 0: raise RuntimeError(f"gemini-cli exit {returncode}: {(stderr or stdout)[-300:]}") out = stdout.strip() if not out: raise ValueError("gemini-cli produced empty output") return out finally: if tmp_path and os.path.exists(tmp_path): os.unlink(tmp_path) # This is a shared 2-core/8GB VPS running postgres, every 360lm PWA, and other # concurrent Claude sessions — a big num_ctx on gemma3/llava (CPU inference) # has been observed to pull 5+ GB RSS and swap the box. Cap well below # graphify's own 131072 ollama ceiling; this is a degraded fallback path # (Gemini CLI should handle the overwhelming majority of calls), not the # primary path, so a lower ceiling trading some truncation risk for host # stability is the right tradeoff. Serialize fallback calls too — one gemini # + one ollama call at a time, never a burst of either. _OLLAMA_NUM_CTX_CAP = 16384 _ollama_lock = threading.Lock() def _estimate_num_ctx(messages): """Ollama defaults num_ctx to 2048 and silently truncates larger prompts (graphify's own ollama backend derives this; forwarding via --backend openai bypasses that, so the shim must do it itself).""" chars = sum(len(json.dumps(m.get("content", ""))) for m in messages) estimated = chars // 4 + 400 return min(max(estimated + 2000, 8192), _OLLAMA_NUM_CTX_CAP) def _call_ollama(body, has_image): body = { **body, "model": OLLAMA_VISION_MODEL if has_image else OLLAMA_TEXT_MODEL, "options": {"num_ctx": _estimate_num_ctx(body.get("messages", []))}, } req = urllib.request.Request( OLLAMA_URL, data=json.dumps(body).encode(), method="POST", headers={"Content-Type": "application/json"}, ) with _ollama_lock: with urllib.request.urlopen(req, timeout=OLLAMA_TIMEOUT_S) as resp: return json.loads(resp.read()) def _openai_response(model, content): return { "id": "gemini-bridge", "object": "chat.completion", "created": int(time.time()), "model": model, "choices": [{ "index": 0, "message": {"role": "assistant", "content": content}, "finish_reason": "stop", }], "usage": {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}, } class Handler(BaseHTTPRequestHandler): def log_message(self, fmt, *args): log.info("%s - %s", self.address_string(), fmt % args) def do_POST(self): if self.path.rstrip("/") != "/v1/chat/completions": self.send_response(404) self.end_headers() return length = int(self.headers.get("Content-Length", 0)) body = json.loads(self.rfile.read(length) or b"{}") messages = body.get("messages", []) model = body.get("model") or GEMINI_MODEL_DEFAULT prompt, image_b64, image_mime = _extract_text_and_images(messages) has_image = image_b64 is not None try: content = _call_gemini_cli(model, prompt, image_b64, image_mime) log.info("tier=gemini-cli/oauth model=%s image=%s chars=%d", model, has_image, len(content)) result = _openai_response(model, content) except Exception as e: if not OLLAMA_FALLBACK_ENABLED: log.warning("gemini-cli failed (%s: %s) -> fallback disabled, skipping chunk", type(e).__name__, e) try: self.send_response(502) self.send_header("Content-Type", "application/json") self.end_headers() self.wfile.write(json.dumps({"error": {"message": f"gemini-cli failed, fallback disabled: {e}"}}).encode()) except (BrokenPipeError, ConnectionResetError): pass return log.warning("gemini-cli failed (%s: %s) -> falling back to ollama", type(e).__name__, e) try: result = _call_ollama(body, has_image) log.info("tier=ollama model=%s image=%s", result.get("model"), has_image) except (urllib.error.URLError, TimeoutError, OSError) as e2: log.error("ollama fallback also failed: %s: %s", type(e2).__name__, e2) try: self.send_response(502) self.send_header("Content-Type", "application/json") self.end_headers() self.wfile.write(json.dumps({"error": {"message": f"both tiers failed: {e2}"}}).encode()) except (BrokenPipeError, ConnectionResetError): pass # client (graphify's SDK) already gave up and disconnected return payload = json.dumps(result).encode() try: self.send_response(200) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(payload))) self.end_headers() self.wfile.write(payload) except (BrokenPipeError, ConnectionResetError): log.warning("client disconnected before response could be sent") def do_GET(self): if self.path.rstrip("/") == "/healthz": self.send_response(200) self.send_header("Content-Type", "application/json") self.end_headers() self.wfile.write(b'{"ok": true}') return self.send_response(404) self.end_headers() if __name__ == "__main__": server = ThreadingHTTPServer(("127.0.0.1", PORT), Handler) log.info("graphify-gemini-bridge listening on 127.0.0.1:%d (tier1=gemini-cli/oauth, fallback=ollama)", PORT) server.serve_forever()