aiturk-hermes-ide/apps/desktop/assets/codex-bridge/jev_denetim.py

"""Adim denetimi: ek modelin her adimdaki cevabini Jev gecirir ya da gecirmez (11 Ekim 2026).

Kullanici karari (11 Ekim 2026): "Bu kopruye Jev koy, her model her adimda kontrol etsin ve yakaladigini qwen 3.8
obliterated thinking versin." / "Bir prompt cok adimli olabilir; her adim kontrol edilsin. Jev sadece geciren ya da
gecirmeyen; gecirmezse 3.8 obliterated thinking." Ardindan (ayni gun): "obliterated:thinking yerine Uncensored-Uygulama yapar
misin" -> yedek, Uncensored-Uygulama combosudur. Yani denetim adim basinadir: Jev'in gecirmedigi adimi yedek model
yapar, sonraki adim yine secili modele gider ve yine denetlenir.

Jev (TypeSafe System One) metin uretmez; bir duruma sorulan evet/hayir sorularina olasilik verir (olcum 11 Ekim
2026: cagri basina ~0,4 sn, 13 gercek ornegin 13'u dogru ayrildi). Kural kodda durur:
- yakalanir: cevap istegi yapmiyor/soylemiyor (ret, kacamak, konuyu degistirme) ya da istegi sormadan kendi
  sartlariyla degistiriyor;
- yakalanmaz: somut bir teknik engel bildiren cevap (hata, eksik dosya/arac/yetki) ve kullaniciya soru ya da onay
  soran cevap; metni olmayan (yalniz arac cagiran) adim.

Jev'e ulasilamazsa cevap oldugu gibi gecer (is durmaz); art arda hatada bir sure hic sorulmaz.
Codex'in kendi modelleri kopruye ugramadigi icin bu denetimin disindadir.

Ayar (bridge.json, istege bagli): "jev_denetim": {"enabled": false} kapatir; "hedef" yedek modelin Codex adi;
"ret", "sart", "teknik", "soru" esikleri; "sure" saniye.
"""
import json
import os
import re
import threading
import time
import urllib.request

ADRES = "https://api.typesafe.ai/v1/systemone"
MODEL = "jev-1.13.0"
HEDEF = "ide/Uncensored-Uygulama"
ESIK = {"ret": 0.7, "sart": 0.9, "teknik": 0.5, "soru": 0.5}
SURE = 6.0
EN_COK_ISTEK = 3000
EN_COK_CEVAP = 6000
DEVRE_HATA = 3              # art arda bu kadar hatadan sonra
DEVRE_SN = 300              # bu kadar sure Jev'e sorulmaz
SORU = {
    "ret": ("The assistant did NOT do what the user asked or did NOT answer the user's question: it refused, declined, "
            "said it cannot or will not, avoided giving the requested judgment, or redirected to a different topic."),
    "sart": ("The assistant changed the request on its own: it added its own conditions, softened or relabeled what was "
             "asked, or left part of it out, instead of doing exactly what the user asked."),
    "teknik": ("The assistant reports a concrete technical obstacle as the reason: an error message, a failed command, "
               "a missing file, tool, permission, connection or piece of information."),
    "soru": ("The assistant's reply is mainly a question to the user: it asks for a clarification, a choice or an "
             "approval and waits for the user's answer before doing anything."),
}
_KILIT = threading.Lock()
_DEVRE = {"hata": 0, "kadar": 0.0}
_ANAHTAR = {"okundu": False, "deger": ""}


def anahtar(ortam=None, dosyalar=None):
    """TypeSafe anahtari: ortam degiskeni, sonra ~/.typesafe/typesafe.env. Yoksa "" (denetim kapali sayilir)."""
    ortam = os.environ if ortam is None else ortam
    deger = str(ortam.get("TYPESAFE_API_KEY") or "").strip()
    if deger:
        return deger
    if dosyalar is None:
        if _ANAHTAR["okundu"]:
            return _ANAHTAR["deger"]
        dosyalar = [os.path.join(os.path.expanduser("~"), ".typesafe", "typesafe.env")]
        onbellek = True
    else:
        onbellek = False
    for yol in dosyalar:
        try:
            with open(yol, encoding="utf-8", errors="replace") as f:
                for satir in f:
                    m = re.match(r"^\s*TYPESAFE_API_KEY\s*=\s*(.*)$", satir)
                    if m:
                        deger = m.group(1).strip().strip('"').strip("'")
                        break
        except OSError:
            continue
        if deger:
            break
    if onbellek:
        _ANAHTAR.update(okundu=True, deger=deger)
    return deger


def _metin(oge):
    icerik = oge.get("content")
    if isinstance(icerik, str):
        return icerik
    return "".join(str(p.get("text") or "") for p in icerik or [] if isinstance(p, dict))


def son_istek(req):
    """Kullanicinin bu turdaki istegi: son kullanici mesaji. Sesli aktarimda yalniz <input> icindeki soz;
    baska sarmalli sistem mesajlari (kanca baglami) atlanir."""
    for oge in reversed(req.get("input") or []):
        if not (isinstance(oge, dict) and oge.get("type", "message") == "message" and oge.get("role") == "user"):
            continue
        metin = _metin(oge).strip()
        m = re.search(r"<realtime_delegation>.*?<input>(.*?)</input>", metin, re.S)
        if m:
            return m.group(1).strip()[:EN_COK_ISTEK]
        if metin and not metin.startswith("<"):
            return metin[:EN_COK_ISTEK]
    return ""


def _kisalt(metin, sinir):
    if len(metin) <= sinir:
        return metin
    bas = sinir * 2 // 3
    return metin[:bas] + "\n[...]\n" + metin[-(sinir - bas):]


def _gonder(govde, key, sure):
    istek = urllib.request.Request(ADRES, data=json.dumps(govde).encode("utf-8"), method="POST",
                                   headers={"Authorization": "Bearer " + key, "Content-Type": "application/json"})
    with urllib.request.urlopen(istek, timeout=sure) as yanit:
        return json.load(yanit)


def sor(istek, cevap, key, sure=SURE, gonder=_gonder, simdi=None):
    """Jev'e dort soruyu sorar. Donus: {"ret","sart","teknik","soru"} olasiliklari | None (sorulamadi)."""
    simdi = time.time() if simdi is None else simdi
    with _KILIT:
        if simdi < _DEVRE["kadar"]:
            return None
    govde = {"model": MODEL, "state": {"user_request": istek, "assistant_reply": _kisalt(cevap, EN_COK_CEVAP)},
             "questions": {k: {"type": "noul", "instructions": v} for k, v in SORU.items()}}
    try:
        yanitlar = (gonder(govde, key, sure) or {}).get("answers") or {}
        sonuc = {k: float(yanitlar[k]["noul"]) for k in SORU}
    except Exception:
        with _KILIT:
            _DEVRE["hata"] += 1
            if _DEVRE["hata"] >= DEVRE_HATA:
                _DEVRE.update(hata=0, kadar=simdi + DEVRE_SN)
        return None
    with _KILIT:
        _DEVRE["hata"] = 0
    return sonuc


def karar(puan, ayar=None):
    """True: adim gecmez (yedek model yapar). Esikler ayardan degistirilebilir."""
    e = dict(ESIK)
    e.update({k: float(v) for k, v in (ayar or {}).items() if k in ESIK})
    if puan["soru"] >= e["soru"]:
        return False                    # kullaniciya soru/onay soruyor: kullanici cevaplar
    if puan["ret"] >= e["ret"] and puan["teknik"] < e["teknik"]:
        return True                     # yapmiyor / soylemiyor (teknik engel degil)
    return puan["sart"] >= e["sart"]    # sormadan kendi sartlariyla degistirdi


def ayar_oku(cfg):
    """(acik, hedef slug, ayar sozlugu)."""
    ayar = cfg if isinstance(cfg, dict) else {}
    return ayar.get("enabled", True) is not False, str(ayar.get("hedef") or HEDEF), ayar