admin/Codex-Launcher---Any-AI-Porovider
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

sync: PR #28 - tkinter GUI restore, proxy feature toggles, CROF guard fixes, gRPC test skip

Browse Source
This commit is contained in:
Roman | RyzenAdvanced
2026-05-29 13:14:48 +04:00
Unverified
parent 64049f5c94
commit 2535ed83d0
2 changed files with 337 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files

398
codex_launcher_lib.py
View File

@@ -8,6 +8,7 @@ the tkinter GUI (Windows). No pip dependencies. No GTK/PyGObject imports.
import base64
import collections
import contextlib
import copy
import hashlib
import json
import os
@@ -68,6 +69,9 @@ BGP_POOLS_FILE = CONFIG_DIR / "bgp-pools.json"
LAUNCH_LOG = LOG_DIR / "launcher.log"
OAUTH_SECRETS_PATH = HOME / ".config" / "codex-launcher" / "oauth-secrets.json"
GEMINI_OAUTH_CLIENT_ID = "681255809395-oo8ft2oprdrnp9e3aqf6av3hmdib135j.apps.googleusercontent.com"
GEMINI_OAUTH_CLIENT_SECRET = "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"
if IS_WINDOWS:
PROXY = BIN_DIR / "translate-proxy.py"
CLEANUP = BIN_DIR / "cleanup-codex-stale.py"
@@ -82,6 +86,51 @@ model_provider = ""
model_catalog_json = ""
"""
_MODULE_DIR = Path(__file__).resolve().parent
if str(_MODULE_DIR) not in sys.path:
sys.path.insert(0, str(_MODULE_DIR))
try:
import universal_runtime as _universal_runtime
except (ImportError, ModuleNotFoundError):
_universal_runtime = None
def detect_runtime_environment():
if _universal_runtime is None:
return {"profile": "unknown", "fallback_mode": "builtin"}
return _universal_runtime.detect_environment()
def build_cross_platform_profile(mode="basic", overrides=None):
if _universal_runtime is None:
return {"profile": "legacy", "mode": mode, "overrides": overrides or {}}
return _universal_runtime.build_runtime_profile(mode=mode, overrides=overrides)
def run_doctor_plus():
if _universal_runtime is None:
return {"health": "unknown", "checks": []}
deps = ["python3" if not IS_WINDOWS else "python", "curl"]
return _universal_runtime.doctor_plus(deps, [CONFIG, ENDPOINTS_FILE, PROXY_CONFIG_DIR / "probe"])
def choose_policy_route(routes, policy=None):
if _universal_runtime is None:
return routes[0] if routes else {}
return _universal_runtime.select_policy_route(routes, policy=policy)
def create_session_portability_pack(destination, metadata=None, files=None):
if _universal_runtime is None:
raise RuntimeError("universal runtime unavailable")
return _universal_runtime.export_session_pack(Path(destination), metadata or {}, [Path(p) for p in (files or [])])
def restore_session_portability_pack(bundle_path, destination_dir):
if _universal_runtime is None:
raise RuntimeError("universal runtime unavailable")
return _universal_runtime.import_session_pack(Path(bundle_path), Path(destination_dir))
CHANGELOG = [
("10.13.8", "2026-05-27", [
"Fix: force_finalize skips Gemini call entirely (was hallucinating tool calls without tools)",
@@ -456,7 +505,7 @@ CHANGELOG = [
]
# ═══════════════════════════════════════════════════════════════════════
# Provider presets (17 providers)
# Provider presets (25+ providers)
# ═══════════════════════════════════════════════════════════════════════
PROVIDER_PRESETS = {
@@ -562,6 +611,41 @@ PROVIDER_PRESETS = {
"base_url": "https://openrouter.ai/api/v1",
"models": [],
},
"Perplexity": {
"backend_type": "openai-compat",
"base_url": "https://api.perplexity.ai",
"models": [
"sonar",
"sonar-pro",
"sonar-reasoning-pro",
"sonar-deep-research",
],
},
"Cohere": {
"backend_type": "openai-compat",
"base_url": "https://api.cohere.ai/compatibility/v1",
"models": [],
},
"Hugging Face": {
"backend_type": "openai-compat",
"base_url": "https://router.huggingface.co/v1",
"models": [],
},
"Together AI": {
"backend_type": "openai-compat",
"base_url": "https://api.together.xyz/v1",
"models": [],
},
"Groq": {
"backend_type": "openai-compat",
"base_url": "https://api.groq.com/openai/v1",
"models": [],
},
"Fireworks AI": {
"backend_type": "openai-compat",
"base_url": "https://api.fireworks.ai/inference/v1",
"models": [],
},
"Google Gemini (API Key)": {
"backend_type": "openai-compat",
"base_url": "https://generativelanguage.googleapis.com/v1beta/openai",
@@ -626,6 +710,16 @@ PROVIDER_PRESETS = {
"base_url": "http://localhost:11434/v1",
"models": [],
},
"LM Studio (local)": {
"backend_type": "openai-compat",
"base_url": "http://127.0.0.1:1234/v1",
"models": [],
},
"vLLM / OpenAI-Compatible (self-hosted)": {
"backend_type": "openai-compat",
"base_url": "http://localhost:8000/v1",
"models": [],
},
}
# ═══════════════════════════════════════════════════════════════════════
@@ -861,29 +955,33 @@ def apply_provider_preset(endpoint, preset_name):
def load_endpoints():
if ENDPOINTS_FILE.exists():
try:
return json.loads(ENDPOINTS_FILE.read_text())
except Exception:
pass
return json.loads(ENDPOINTS_FILE.read_text(encoding="utf-8"))
except Exception as exc:
print(f"[lib] failed to load endpoints: {exc}", file=sys.stderr)
return {"default": None, "endpoints": []}
def save_endpoints(data):
ENDPOINTS_FILE.parent.mkdir(parents=True, exist_ok=True)
ENDPOINTS_FILE.write_text(json.dumps(data, indent=2))
tmp = ENDPOINTS_FILE.with_suffix(".json.tmp")
tmp.write_text(json.dumps(data, indent=2))
os.replace(str(tmp), str(ENDPOINTS_FILE))
def load_bgp_pools():
if BGP_POOLS_FILE.exists():
try:
return json.loads(BGP_POOLS_FILE.read_text())
except Exception:
pass
return json.loads(BGP_POOLS_FILE.read_text(encoding="utf-8"))
except Exception as exc:
print(f"[lib] failed to load bgp pools: {exc}", file=sys.stderr)
return {"pools": []}
def save_bgp_pools(data):
BGP_POOLS_FILE.parent.mkdir(parents=True, exist_ok=True)
BGP_POOLS_FILE.write_text(json.dumps(data, indent=2))
tmp = BGP_POOLS_FILE.with_suffix(".json.tmp")
tmp.write_text(json.dumps(data, indent=2))
os.replace(str(tmp), str(BGP_POOLS_FILE))
def get_endpoint(name):
@@ -949,10 +1047,28 @@ def write_secure_text(path, text):
# ═══════════════════════════════════════════════════════════════════════
def backup_config():
if CONFIG.exists():
tmp = CONFIG_BAK.with_suffix(".tmp")
shutil.copy2(str(CONFIG), str(tmp))
os.replace(str(tmp), str(CONFIG_BAK))
if not CONFIG.exists():
return
tmp = CONFIG_BAK.with_suffix(".tmp")
shutil.copy2(str(CONFIG), str(tmp))
os.replace(str(tmp), str(CONFIG_BAK))
ts = time.strftime("%Y%m%d_%H%M%S")
rot = CONFIG.parent / f"config.toml.{ts}.bak"
try:
shutil.copy2(str(CONFIG), str(rot))
_rotate_backups(CONFIG.parent, "config.toml.*.bak", max_backups=10)
except Exception as exc:
print(f"[lib] backup rotation failed: {exc}", file=sys.stderr)
def _rotate_backups(directory, pattern, max_backups=10):
import glob as _glob
files = sorted(_glob.glob(str(directory / pattern)), key=os.path.getmtime, reverse=True)
for old in files[max_backups:]:
try:
os.remove(old)
except Exception:
pass
def restore_config():
@@ -979,7 +1095,7 @@ def recover_config_if_needed(logfn=None):
if not CONFIG_TXN.exists():
return
try:
txn = json.loads(CONFIG_TXN.read_text())
txn = json.loads(CONFIG_TXN.read_text(encoding="utf-8"))
if txn.get("config_existed") and CONFIG_BAK.exists():
restore_config()
if logfn:
@@ -1219,6 +1335,24 @@ def endpoint_model_headers(endpoint):
return headers
def check_provider_latency(endpoint, timeout=5):
bt = endpoint.get("backend_type", "")
if bt in ("native", "codex-default", "gemini-oauth-antigravity"):
return None
base = endpoint.get("base_url", "").strip()
if not base:
return None
url = base.rstrip("/") + "/models"
try:
headers = endpoint_model_headers(endpoint)
req = urllib.request.Request(url, headers=headers, method="GET")
t0 = time.time()
urllib.request.urlopen(req, timeout=timeout)
return time.time() - t0
except Exception:
return None
def fetch_models_for_endpoint(endpoint, timeout=10):
bt = endpoint.get("backend_type", "")
if bt == "gemini-oauth-antigravity":
@@ -1276,9 +1410,16 @@ ANTIGRAVITY_MODELS = [
def load_oauth_secrets():
try:
with open(OAUTH_SECRETS_PATH, encoding="utf-8") as f:
return json.load(f)
data = json.load(f)
except Exception:
return {}
data = {}
for key in ("antigravity", "gemini_cli"):
sec = data.get(key, {})
if not sec.get("client_id"):
data.setdefault(key, {})["client_id"] = GEMINI_OAUTH_CLIENT_ID
if not sec.get("client_secret"):
data.setdefault(key, {})["client_secret"] = GEMINI_OAUTH_CLIENT_SECRET
return data
def save_oauth_secrets(data):
@@ -1462,7 +1603,7 @@ def run_endpoint_doctor(endpoint):
token_path = PROXY_CONFIG_DIR / token_name
if token_path.exists():
try:
td = json.loads(token_path.read_text())
td = json.loads(token_path.read_text(encoding="utf-8"))
exp = td.get("expires_at", 0)
if exp > time.time():
remaining = exp - time.time()
@@ -1531,9 +1672,9 @@ def run_endpoint_doctor(endpoint):
def _load_pid_registry():
if PID_REGISTRY.exists():
try:
return json.loads(PID_REGISTRY.read_text())
except Exception:
pass
return json.loads(PID_REGISTRY.read_text(encoding="utf-8"))
except Exception as exc:
print(f"[lib] failed to load pid registry: {exc}", file=sys.stderr)
return {}
@@ -1576,7 +1717,7 @@ _PROXY_PORT_FILE = PROXY_CONFIG_DIR / ".last-proxy-port"
def _pick_free_port():
saved = None
try:
saved = int(_PROXY_PORT_FILE.read_text().strip())
saved = int(_PROXY_PORT_FILE.read_text(encoding="utf-8").strip())
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.bind(("127.0.0.1", saved))
@@ -1626,8 +1767,8 @@ def start_proxy_for(endpoint, logfn):
discovered = [] if endpoint.get("oauth_provider") == "google-antigravity" else td.get("available_models", [])
if discovered:
model_list = discovered
except Exception:
pass
except Exception as exc:
print(f"[lib] oauth token discovery: {exc}", file=sys.stderr)
pcfg = {
"port": port,
@@ -1639,6 +1780,11 @@ def start_proxy_for(endpoint, logfn):
"reasoning_enabled": endpoint.get("reasoning_enabled", True),
"reasoning_effort": endpoint.get("reasoning_effort", "medium"),
"force_model": endpoint.get("default_model") or "",
"caveman_mode": endpoint.get("caveman_mode", False),
"rtk_compression": endpoint.get("rtk_compression", False),
"auto_compact": endpoint.get("auto_compact", False),
"adaptive_compact": endpoint.get("adaptive_compact", False),
"tool_output_truncation": endpoint.get("tool_output_truncation", True),
"models": [{"id": m, "object": "model", "created": 1700000000, "owned_by": endpoint["name"]}
for m in model_list],
}
@@ -1726,6 +1872,11 @@ def start_bgp_proxy(pool, model, logfn):
"target_url": "http://bgp.placeholder",
"api_key": "",
"bgp_routes": pool.get("routes", []),
"caveman_mode": endpoint.get("caveman_mode", False) if endpoint else False,
"rtk_compression": endpoint.get("rtk_compression", False) if endpoint else False,
"auto_compact": endpoint.get("auto_compact", False) if endpoint else False,
"adaptive_compact": endpoint.get("adaptive_compact", False) if endpoint else False,
"tool_output_truncation": endpoint.get("tool_output_truncation", True) if endpoint else True,
"models": [{"id": m, "object": "model", "created": 1700000000, "owned_by": "bgp"} for m in bgp_ep["models"]],
}
pcfg_path = PROXY_CONFIG_DIR / f"proxy-{safe_name(pool['name'])}-{port}.json"
@@ -1751,6 +1902,12 @@ def detect_codex_cli():
def detect_codex_desktop():
"""Detect Codex Desktop installation.
Returns (path_or_aumid, is_msix) tuple on Windows, path string on Linux.
For MSIX installs, returns the AppUserModelId since the exe cannot be
launched directly via subprocess from WindowsApps.
"""
if IS_WINDOWS:
la = os.environ.get("LOCALAPPDATA", "")
pf = os.environ.get("PROGRAMFILES", "")
@@ -1763,8 +1920,8 @@ def detect_codex_desktop():
]
for p in desktop_paths:
if p.exists():
return str(p)
# MSIX / Microsoft Store install: locate via Get-AppxPackage
return str(p), False
# MSIX / Microsoft Store install
try:
r = subprocess.run(
["powershell", "-NoProfile", "-Command",
@@ -1775,13 +1932,70 @@ def detect_codex_desktop():
if loc:
msix_exe = Path(loc) / "app" / "Codex.exe"
if msix_exe.exists():
return str(msix_exe)
r2 = subprocess.run(
["powershell", "-NoProfile", "-Command",
"(Get-AppxPackage *OpenAI.Codex*).PackageFamilyName"],
capture_output=True, text=True, timeout=10,
)
family = r2.stdout.strip() if r2.returncode == 0 else ""
if family:
return f"{family}!App", True
except Exception:
pass
return None
return None, False
if START_SH and START_SH.exists():
return str(START_SH)
return None
return str(START_SH), False
return None, False
def launch_codex_desktop(desktop_info):
"""Launch Codex Desktop process.
Args:
desktop_info: (path_or_aumid, is_msix) tuple from detect_codex_desktop()
Returns:
subprocess.Popen object or None
"""
path, is_msix = desktop_info
if IS_WINDOWS:
if is_msix:
return subprocess.Popen(
["cmd", "/c", "start", "", f"shell:AppsFolder\\{path}"],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
creationflags=subprocess.CREATE_NEW_PROCESS_GROUP)
return subprocess.Popen(
[path],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
creationflags=subprocess.CREATE_NEW_PROCESS_GROUP)
else:
return subprocess.Popen(
[path], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
preexec_fn=os.setsid)
def is_codex_desktop_running():
"""Check if Codex Desktop (or MSIX Codex) is currently running."""
if IS_WINDOWS:
try:
for name in ("Codex Desktop.exe", "Codex.exe"):
out = subprocess.run(
["tasklist", "/FI", f"IMAGENAME eq {name}", "/FO", "CSV", "/NH"],
capture_output=True, text=True, timeout=5,
)
for line in out.stdout.strip().splitlines():
parts = line.split(",")
if len(parts) >= 2 and parts[1].strip('"').isdigit():
return True
except Exception:
pass
return False
else:
try:
out = subprocess.run(["pgrep", "-f", "/opt/codex-desktop/electron"], capture_output=True, text=True, timeout=5)
return bool(out.stdout.strip())
except Exception:
return False
def check_codex_auth():
@@ -1813,7 +2027,7 @@ def check_codex_auth():
def last_log_lines(n=15):
try:
t = LAUNCH_LOG.read_text()
t = LAUNCH_LOG.read_text(encoding="utf-8")
return "\n".join(t.splitlines()[-n:])
except Exception:
return "(no log file)"
@@ -1824,24 +2038,25 @@ def last_log_lines(n=15):
def kill_existing_desktop(logfn=None):
if IS_WINDOWS:
try:
out = subprocess.run(
["tasklist", "/FI", "IMAGENAME eq Codex Desktop.exe", "/FO", "CSV", "/NH"],
capture_output=True, text=True, timeout=5,
)
for line in out.stdout.strip().splitlines():
parts = line.split(",")
if len(parts) >= 2:
pid_str = parts[1].strip('"')
if pid_str.isdigit():
pid = int(pid_str)
_kill_process_group(pid)
if logfn:
logfn(f"Killed existing Codex Desktop (pid {pid})")
time.sleep(2)
except Exception as e:
if logfn:
logfn(f"Note: could not kill existing Desktop: {e}")
for img in ("Codex Desktop.exe", "Codex.exe"):
try:
out = subprocess.run(
["tasklist", "/FI", f"IMAGENAME eq {img}", "/FO", "CSV", "/NH"],
capture_output=True, text=True, timeout=5,
)
for line in out.stdout.strip().splitlines():
parts = line.split(",")
if len(parts) >= 2:
pid_str = parts[1].strip('"')
if pid_str.isdigit():
pid = int(pid_str)
_kill_process_group(pid)
if logfn:
logfn(f"Killed existing Codex Desktop (pid {pid})")
time.sleep(2)
except Exception as e:
if logfn:
logfn(f"Note: could not kill existing Desktop: {e}")
else:
try:
out = subprocess.run(["pgrep", "-f", "/opt/codex-desktop/electron"], capture_output=True, text=True, timeout=5)
@@ -1929,9 +2144,9 @@ _DIAGNOSTIC_SYSTEM_PROMPT = (
def load_monitoring_config():
if MONITORING_FILE.exists():
try:
return json.loads(MONITORING_FILE.read_text())
except Exception:
pass
return json.loads(MONITORING_FILE.read_text(encoding="utf-8"))
except Exception as exc:
print(f"[lib] failed to load monitoring config: {exc}", file=sys.stderr)
return {
"enabled": False,
"provider_url": "",
@@ -1951,9 +2166,9 @@ def save_monitoring_config(cfg):
def load_incident_store():
if INCIDENT_STORE_FILE.exists():
try:
return json.loads(INCIDENT_STORE_FILE.read_text())
except Exception:
pass
return json.loads(INCIDENT_STORE_FILE.read_text(encoding="utf-8"))
except Exception as exc:
print(f"[lib] failed to load incident store: {exc}", file=sys.stderr)
return {"version": 1, "incidents": {}, "stats": {"ai_calls": 0, "tokens_used": 0}}
@@ -1966,17 +2181,19 @@ def monitoring_log(msg):
try:
with open(str(MONITORING_LOG), "a") as f:
f.write(f"[{time.strftime('%H:%M:%S')}] {msg}\n")
except Exception:
pass
except Exception as exc:
print(f"[lib] monitoring_log write failed: {exc}", file=sys.stderr)
class IncidentStore:
def __init__(self):
self._store = load_incident_store()
self._dirty = False
self._lock = threading.Lock()
def lookup(self, pattern):
inc = self._store.get("incidents", {}).get(pattern)
with self._lock:
inc = self._store.get("incidents", {}).get(pattern)
if inc and inc.get("success_count", 0) > 0:
rate = inc["success_count"] / max(inc["success_count"] + inc.get("fail_count", 0), 1)
if rate > 0.5:
@@ -1984,34 +2201,45 @@ class IncidentStore:
return None
def record(self, pattern, fix, success=True):
incs = self._store.setdefault("incidents", {})
inc = incs.setdefault(pattern, {
"fix": fix, "success_count": 0, "fail_count": 0,
"last_seen": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
"occurrences": 0,
})
inc["last_seen"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
inc["occurrences"] = inc.get("occurrences", 0) + 1
if success:
inc["success_count"] = inc.get("success_count", 0) + 1
else:
inc["fail_count"] = inc.get("fail_count", 0) + 1
self._dirty = True
with self._lock:
new_store = copy.deepcopy(self._store)
incs = new_store.setdefault("incidents", {})
inc = incs.setdefault(pattern, {
"fix": fix, "success_count": 0, "fail_count": 0,
"last_seen": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
"occurrences": 0,
})
inc = dict(inc)
inc["last_seen"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
inc["occurrences"] = inc.get("occurrences", 0) + 1
if success:
inc["success_count"] = inc.get("success_count", 0) + 1
else:
inc["fail_count"] = inc.get("fail_count", 0) + 1
incs[pattern] = inc
self._store = new_store
self._dirty = True
def record_ai_call(self, tokens=0):
stats = self._store.setdefault("stats", {"ai_calls": 0, "tokens_used": 0})
stats["ai_calls"] = stats.get("ai_calls", 0) + 1
stats["tokens_used"] = stats.get("tokens_used", 0) + tokens
self._dirty = True
with self._lock:
new_store = copy.deepcopy(self._store)
stats = dict(new_store.get("stats", {"ai_calls": 0, "tokens_used": 0}))
stats["ai_calls"] = stats.get("ai_calls", 0) + 1
stats["tokens_used"] = stats.get("tokens_used", 0) + tokens
new_store["stats"] = stats
self._store = new_store
self._dirty = True
def flush(self):
if self._dirty:
save_incident_store(self._store)
self._dirty = False
with self._lock:
if self._dirty:
save_incident_store(self._store)
self._dirty = False
@property
def stats(self):
return self._store.get("stats", {"ai_calls": 0, "tokens_used": 0})
with self._lock:
return dict(self._store.get("stats", {"ai_calls": 0, "tokens_used": 0}))
class AIDiagnosticAgent:
@@ -2132,10 +2360,10 @@ class HealthWatcher(threading.Thread):
try:
cfg_path = PROXY_CONFIG_DIR / "proxy-config.json"
if cfg_path.exists():
d = json.loads(cfg_path.read_text())
d = json.loads(cfg_path.read_text(encoding="utf-8"))
return d.get("port")
except Exception:
pass
except Exception as exc:
print(f"[lib] _get_proxy_port: {exc}", file=sys.stderr)
return None
def _check_health(self, port):
@@ -2200,7 +2428,7 @@ class HealthWatcher(threading.Thread):
for log_name in ["cc-debug.log", "proxy.log"]:
log_path = PROXY_CONFIG_DIR / log_name
try:
text = log_path.read_text()
text = log_path.read_text(encoding="utf-8")
lines.extend(text.splitlines()[-20:])
except Exception:
pass
@@ -2250,9 +2478,9 @@ class _LogAnalyzerThread(threading.Thread):
def load_usage_stats():
try:
if _USAGE_STATS_FILE.exists():
return json.loads(_USAGE_STATS_FILE.read_text())
except Exception:
pass
return json.loads(_USAGE_STATS_FILE.read_text(encoding="utf-8"))
except Exception as exc:
print(f"[lib] failed to load usage stats: {exc}", file=sys.stderr)
return {"providers": {}, "updated": None}
# ═══════════════════════════════════════════════════════════════════════

34
translate-proxy.py
View File

@@ -311,6 +311,9 @@ FORCE_MODEL = ""
BGP_ROUTES = []
CAVEMAN_MODE = False
RTK_COMPRESSION = False
AUTO_COMPACT = False
ADAPTIVE_COMPACT = False
TOOL_OUTPUT_TRUNCATION = True
PROMPT_ENHANCER = False
PROMPT_ENHANCER_MODE = "offline"
PROMPT_ENHANCER_MODEL = ""
@@ -1179,7 +1182,7 @@ def _init_runtime():
global MODELS, CC_VERSION, REASONING_ENABLED, REASONING_EFFORT, BGP_ROUTES
global _api_key_pool, PROMPT_ENHANCER
global VISION_FALLBACK_URL, VISION_FALLBACK_MODEL, VISION_FALLBACK_KEY
global CAVEMAN_MODE, RTK_COMPRESSION
global CAVEMAN_MODE, RTK_COMPRESSION, AUTO_COMPACT, ADAPTIVE_COMPACT, TOOL_OUTPUT_TRUNCATION
CONFIG = load_config()
PORT = CONFIG["port"]
@@ -1215,6 +1218,11 @@ def _init_runtime():
BGP_ROUTES = CONFIG.get("bgp_routes", [])
CAVEMAN_MODE = CONFIG.get("caveman_mode", False) or os.environ.get("CAVEMAN_MODE") == "1"
RTK_COMPRESSION = CONFIG.get("rtk_compression", False) or os.environ.get("RTK_COMPRESSION") == "1"
AUTO_COMPACT = CONFIG.get("auto_compact", False) or os.environ.get("AUTO_COMPACT") == "1"
ADAPTIVE_COMPACT = CONFIG.get("adaptive_compact", False) or os.environ.get("ADAPTIVE_COMPACT") == "1"
TOOL_OUTPUT_TRUNCATION = CONFIG.get("tool_output_truncation", True)
if os.environ.get("TOOL_OUTPUT_TRUNCATION") == "0":
TOOL_OUTPUT_TRUNCATION = False
_api_key_pool = None
if API_KEY and "," in API_KEY and not OAUTH_PROVIDER.startswith("google") and BACKEND not in ("codebuff", "freebuff"):
_api_key_pool = APIKeyPool(BACKEND, API_KEY)
@@ -1962,6 +1970,8 @@ def _crof_item_limit(model):
return min(per_model, _CROF_ADAPTIVE["global_item_limit"])
def _crof_compact_for_retry(input_data, model, aggression=0):
if "crof.ai" not in TARGET_URL:
return input_data
limit = _crof_item_limit(model)
if not isinstance(input_data, list) or len(input_data) < 2:
return input_data
@@ -2149,10 +2159,11 @@ def _compact_input(input_data):
compressed_data.append(item)
input_data = compressed_data
if not isinstance(input_data, list) or len(input_data) <= _MAX_INPUT_ITEMS:
# Skip compaction entirely — just do optional tool output truncation
if not AUTO_COMPACT or not isinstance(input_data, list) or len(input_data) <= _MAX_INPUT_ITEMS:
out = []
for item in input_data:
if isinstance(item, dict) and item.get("type") == "function_call_output":
if TOOL_OUTPUT_TRUNCATION and isinstance(item, dict) and item.get("type") == "function_call_output":
o = item.get("output", "")
if len(o) > _MAX_TOOL_OUTPUT_CHARS:
item = dict(item)
@@ -2191,7 +2202,7 @@ def _compact_input(input_data):
return head + tail
for item in tail:
if isinstance(item, dict) and item.get("type") == "function_call_output":
if TOOL_OUTPUT_TRUNCATION and isinstance(item, dict) and item.get("type") == "function_call_output":
o = item.get("output", "")
if len(o) > _MAX_TOOL_OUTPUT_CHARS:
item["output"] = o[:_MAX_TOOL_OUTPUT_CHARS] + f"\n... [truncated {len(o) - _MAX_TOOL_OUTPUT_CHARS} chars]"
@@ -5703,7 +5714,6 @@ class Handler(http.server.BaseHTTPRequestHandler):
raw_types = [i.get("type") for i in raw_input] if isinstance(raw_input, list) else "str"
resolved_types = [i.get("type") for i in input_data] if isinstance(input_data, list) else "str"
print(f"[{_sid}] prev_id={prev_id} raw={raw_types} resolved={resolved_types}", file=sys.stderr)
with open(_log_path, "a", encoding="utf-8") as _lf:
_lf.write(f"\n{'='*60}\n{_ts} [session={_sid}] REQUEST {self.path}\n")
_lf.write(f" prev_id={prev_id}\n")
@@ -5807,7 +5817,7 @@ class Handler(http.server.BaseHTTPRequestHandler):
body["input"] = input_data
compacted = False
if policy.get("compaction") and isinstance(input_data, list) and "claude" not in model.lower():
if ADAPTIVE_COMPACT and policy.get("compaction") and isinstance(input_data, list) and "claude" not in model.lower():
input_data, compacted = _adaptive_compact(input_data, model, policy)
if compacted:
body = dict(body)
@@ -5819,7 +5829,7 @@ class Handler(http.server.BaseHTTPRequestHandler):
body["input"] = input_data
crof_limit = _crof_item_limit(model)
_crof_eligible = True
_crof_eligible = "crof.ai" in TARGET_URL
if _crof_eligible and not compacted and isinstance(input_data, list):
_needs_compact = len(input_data) > crof_limit
max_tok = _get_model_max_tokens(model)
@@ -6771,7 +6781,7 @@ class Handler(http.server.BaseHTTPRequestHandler):
body["input"] = input_data
compacted = False
if policy.get("compaction") and isinstance(input_data, list) and "claude" not in model.lower():
if ADAPTIVE_COMPACT and policy.get("compaction") and isinstance(input_data, list) and "claude" not in model.lower():
input_data, compacted = _adaptive_compact(input_data, model, policy)
if compacted:
body = dict(body)
@@ -7662,9 +7672,10 @@ class Handler(http.server.BaseHTTPRequestHandler):
has_content = False
has_message = False
has_tool_call = False
_last_stream_usage = {}
def _observe_event(event):
nonlocal last_resp_id, last_output, last_status, finish_reason, has_content, has_message, has_tool_call
nonlocal last_resp_id, last_output, last_status, finish_reason, has_content, has_message, has_tool_call, _last_stream_usage
for line in event.strip().split("\n"):
if line.startswith("data: "):
try:
@@ -7677,6 +7688,9 @@ class Handler(http.server.BaseHTTPRequestHandler):
has_tool_call = any(o.get("type") == "function_call" for o in (last_output or []))
has_message = any(o.get("type") == "message" for o in (last_output or []))
has_content = has_message or has_tool_call
resp_usage = d.get("response", {}).get("usage")
if resp_usage:
_last_stream_usage = resp_usage
except Exception:
pass
@@ -9044,7 +9058,7 @@ def main():
allow_reuse_address = True
daemon_threads = True
request_queue_size = 64
BIND_HOST = getattr(_args, "host", None) or os.environ.get("CODEX_HOST", "127.0.0.1")
BIND_HOST = os.environ.get("CODEX_HOST", "127.0.0.1")
SERVER = ReusableHTTPServer((BIND_HOST, PORT), Handler)
print(f"translate-proxy ({BACKEND}) listening on http://{BIND_HOST}:{PORT}", flush=True)
print(f"Target: {TARGET_URL}", flush=True)