87 lines
3.8 KiB
Python
87 lines
3.8 KiB
Python
"""Minimal client for TypeSafe System One (Jev), shared by the project's maintenance tools in ``scripts/``.
|
|
|
|
Only the standard library: one POST per request to https://api.typesafe.ai/v1/systemone, retries with exponential
|
|
backoff on 429 and 529, and a content addressed cache in ``.cache/jev`` so unchanged questions are never paid twice.
|
|
The API key comes from ``TYPESAFE_API_KEY`` or ``~/.config/typesafe/credentials`` (``TYPESAFE_API_KEY=...``, mode
|
|
600) and never enters the repository. These tools send documentation, test code, issue text and public abstracts;
|
|
they are never used inside the ``simurg`` analysis commands, which stay offline.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
API_URL = "https://api.typesafe.ai/v1/systemone"
|
|
MODEL = "jev-latest"
|
|
CACHE = ROOT / ".cache" / "jev"
|
|
KEY_FILE = Path.home() / ".config" / "typesafe" / "credentials"
|
|
|
|
Question = dict[str, Any]
|
|
Answers = dict[str, dict[str, Any]]
|
|
|
|
|
|
def api_key() -> str:
|
|
"""The key from the environment, else from the private credentials file; empty when absent."""
|
|
key = os.environ.get("TYPESAFE_API_KEY", "")
|
|
if not key and KEY_FILE.exists():
|
|
for ln in KEY_FILE.read_text().splitlines():
|
|
if ln.startswith("TYPESAFE_API_KEY="):
|
|
key = ln.split("=", 1)[1].strip()
|
|
return key
|
|
|
|
|
|
def noul(instructions: str, true: str | None = None, false: str | None = None) -> Question:
|
|
q: Question = {"type": "noul", "instructions": instructions}
|
|
if true and false:
|
|
q["criteria"] = {"true": true, "false": false}
|
|
return q
|
|
|
|
|
|
def choice(instructions: str, criteria: dict[str, str | None]) -> Question:
|
|
return {"type": "choice", "instructions": instructions, "criteria": criteria}
|
|
|
|
|
|
def score(instructions: str, levels: list[str]) -> Question:
|
|
"""``levels`` from the low end of the scale to the high end (2 to 10 concrete situations)."""
|
|
return {"type": "score", "instructions": instructions, "criteria": levels}
|
|
|
|
|
|
def system_one(state: Any, questions: dict[str, Question], key: str | None = None, retries: int = 5) -> Answers:
|
|
"""Ask ``questions`` about ``state`` in one request; returns the ``answers`` map (cached by request content)."""
|
|
payload = {"model": MODEL, "state": state, "questions": questions}
|
|
body = json.dumps(payload, sort_keys=True).encode()
|
|
cached = CACHE / f"{hashlib.sha256(body).hexdigest()}.json"
|
|
if cached.exists():
|
|
return json.loads(cached.read_text())["answers"] # type: ignore[no-any-return]
|
|
key = key or api_key()
|
|
if not key:
|
|
raise RuntimeError("TYPESAFE_API_KEY is not set (environment or ~/.config/typesafe/credentials)")
|
|
headers = {"Authorization": f"Bearer {key}", "Content-Type": "application/json", "User-Agent": "simurgarray-jev/1"}
|
|
req = urllib.request.Request(API_URL, data=body, method="POST", headers=headers)
|
|
for attempt in range(retries):
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=90) as r:
|
|
res = json.loads(r.read())
|
|
CACHE.mkdir(parents=True, exist_ok=True)
|
|
cached.write_text(json.dumps(res))
|
|
return res["answers"] # type: ignore[no-any-return]
|
|
except urllib.error.HTTPError as e:
|
|
if e.code in (429, 529) and attempt < retries - 1:
|
|
time.sleep(2**attempt)
|
|
continue
|
|
raise RuntimeError(f"TypeSafe API error {e.code}: {e.read()[:300]!r}") from e
|
|
except (urllib.error.URLError, TimeoutError) as e:
|
|
if attempt < retries - 1:
|
|
time.sleep(2**attempt)
|
|
continue
|
|
raise RuntimeError(f"TypeSafe API unreachable: {e}") from e
|
|
raise RuntimeError("TypeSafe API: retries exhausted")
|