diff --git a/jev.py b/jev.py new file mode 100644 index 0000000..fc59f7d --- /dev/null +++ b/jev.py @@ -0,0 +1,462 @@ +#!/usr/bin/env python3 +""" +jev.py -- TypeSafe System One (Jev) client for the STS2 bot. + +Jev is a calibrated classifier, not an agent. It picks ONE option from a +set WE define. It cannot plan, cannot call tools, and cannot do arithmetic. + +Constraints baked into this client (all measured, see DESIGN.md): + * Never route an arithmetic comparison through Jev. It bucketed a max + damage of 18 correctly but answered "lethal vs 19 HP" as 0.79 yes, + which is wrong. Compute numbers in facts.py and pass conclusions in. + * A wrong answer can still carry high confidence (0.79 in that case). + Confidence gates are a safety net, never a proof. + * Question count is nearly free: 1 question = 0.73s, 3 questions on a + full state = 0.90s. Batch every question for a state into ONE call. + * Because answers to questions in one call are independent, never make a + question depend on another question's answer. Use a second call. + +The API key is read from the sops-nix secrets path and is never logged. +""" + +from __future__ import annotations + +import http.client +import json +import os +import time +import urllib.error +import urllib.request +from dataclasses import dataclass, field +from typing import Any + +API_URL = "https://api.typesafe.ai/v1/systemone" +DEFAULT_MODEL = "jev-latest" + +ENV_NAMES = ("TYPESAFEAI_API_KEY", "TYPESAFE_API_KEY") +DEFAULT_KEY_PATHS = ( + "~/.config/secrets/global-env/TYPESAFEAI_API_KEY", + "~/.config/secrets/global-env/TYPESAFE_API_KEY", +) + +RETRYABLE_STATUS = {429, 500, 502, 503, 504} + +# When set, every request and its parsed answers are appended as one JSON +# line. State is NOT traced (it can be huge and combat states are already +# dumped by run.py); questions and answers ARE the decision process. +TRACE_PATH = os.environ.get("JEV_TRACE") + +# A Noul answer carries no confidence field, so it is gated on its distance +# from 0.5. 0.15 means act when noul >= 0.65 or <= 0.35. A wrong call on a +# low-stakes binary (end the turn? use a potion?) is recoverable in this game. +NOUL_MARGIN = 0.15 + + +class JevError(RuntimeError): + """Raised for transport, auth, and protocol failures.""" + + +# -------------------------------------------------------------------------- +# Question builders +# -------------------------------------------------------------------------- + +def noul(instructions: Any, criteria: dict | None = None) -> dict: + """Yes/no question. Returns only P(yes); there is no confidence field.""" + q: dict[str, Any] = {"type": "noul", "instructions": instructions} + if criteria is not None: + q["criteria"] = criteria + return q + + +def choice(instructions: Any, criteria: dict) -> dict: + """Pick one option from a map of option -> description (max 255).""" + return {"type": "choice", "instructions": instructions, "criteria": criteria} + + +def score(instructions: Any, criteria: list) -> dict: + """Position on an ordered list of levels.""" + return {"type": "score", "instructions": instructions, "criteria": criteria} + + +# -------------------------------------------------------------------------- +# Typed answers +# -------------------------------------------------------------------------- + +@dataclass(frozen=True) +class NoulAnswer: + noul: float + + @property + def yes(self) -> bool: + return self.noul >= 0.5 + + @property + def kind(self) -> str: + return "noul" + + +@dataclass(frozen=True) +class ChoiceAnswer: + choice: str + probabilities: dict = field(default_factory=dict) + confidence: float = 0.0 + + @property + def kind(self) -> str: + return "choice" + + def runner_up(self) -> tuple[str, float] | None: + others = [(k, v) for k, v in self.probabilities.items() if k != self.choice] + return max(others, key=lambda kv: kv[1]) if others else None + + +@dataclass(frozen=True) +class ScoreAnswer: + score: float + legend: dict = field(default_factory=dict) + probabilities: dict = field(default_factory=dict) + confidence: float = 0.0 + + @property + def kind(self) -> str: + return "score" + + +Answer = NoulAnswer | ChoiceAnswer | ScoreAnswer + + +@dataclass +class JevResponse: + answers: dict[str, Answer] + model: str + input_tokens: int + output_tokens: int + latency_s: float + + def __getitem__(self, qid: str) -> Answer: + return self.answers[qid] + + def get(self, qid: str) -> Answer | None: + return self.answers.get(qid) + + def summary(self) -> str: + lines = [f"model={self.model} {self.latency_s:.2f}s in={self.input_tokens} out={self.output_tokens}"] + for qid, a in self.answers.items(): + if isinstance(a, NoulAnswer): + lines.append(f" {qid:24s} noul={a.noul:.2f} yes={a.yes}") + elif isinstance(a, ChoiceAnswer): + lines.append(f" {qid:24s} choice={a.choice!r} conf={a.confidence:.2f}") + else: + lines.append(f" {qid:24s} score={a.score:.2f} conf={a.confidence:.2f}") + return "\n".join(lines) + + +# -------------------------------------------------------------------------- +# Client +# -------------------------------------------------------------------------- + +def load_key(key_path: str | None = None) -> str: + """Resolve the API key. Environment first, then the sops-nix path.""" + if key_path: + path = os.path.expanduser(key_path) + if not os.path.exists(path): + raise JevError(f"key file not found: {path}") + with open(path, "r", encoding="utf-8") as fh: + key = fh.read().strip() + if not key: + raise JevError(f"key file is empty: {path}") + return key + + for name in ENV_NAMES: + value = os.environ.get(name) + if value and value.strip(): + return value.strip() + + for candidate in DEFAULT_KEY_PATHS: + path = os.path.expanduser(candidate) + if os.path.exists(path): + with open(path, "r", encoding="utf-8") as fh: + key = fh.read().strip() + if key: + return key + + raise JevError( + "no TypeSafe API key found. Looked at env " + + ", ".join(ENV_NAMES) + + " and paths " + + ", ".join(DEFAULT_KEY_PATHS) + ) + + +class JevClient: + """One client, many batched calls. The key never leaves this object.""" + + def __init__( + self, + model: str = DEFAULT_MODEL, + timeout: float = 45.0, + retries: int = 2, + key_path: str | None = None, + key: str | None = None, + ) -> None: + self._key = key or load_key(key_path) + self.model = model + self.timeout = timeout + self.retries = retries + + def __repr__(self) -> str: # never leak the key + return f"JevClient(model={self.model!r}, timeout={self.timeout}, key=REDACTED)" + + # -- core call --------------------------------------------------------- + + def ask( + self, + state: Any, + questions: dict[str, dict], + model: str | None = None, + ) -> JevResponse: + if not questions: + raise JevError("no questions supplied") + if len(questions) > 255: + raise JevError(f"too many questions: {len(questions)} (max 255)") + + payload = { + "model": model or self.model, + "state": state, + "questions": questions, + } + body = json.dumps(payload).encode("utf-8") + + started = time.monotonic() + last_error: Exception | None = None + + for attempt in range(self.retries + 1): + request = urllib.request.Request( + API_URL, + data=body, + method="POST", + headers={ + "Content-Type": "application/json", + "Authorization": f"Bearer {self._key}", + "User-Agent": "sts2-bot/0.1", + }, + ) + try: + with urllib.request.urlopen(request, timeout=self.timeout) as response: + raw = response.read() + break + except urllib.error.HTTPError as exc: + detail = exc.read().decode("utf-8", "replace") + last_error = JevError(f"HTTP {exc.code}: {detail[:400]}") + if exc.code in RETRYABLE_STATUS and attempt < self.retries: + retry_after = exc.headers.get("retry-after") + delay = float(retry_after) if retry_after else 1.5 * (2 ** attempt) + time.sleep(delay) + continue + raise last_error from exc + except (urllib.error.URLError, TimeoutError) as exc: + last_error = JevError(f"transport failure: {exc}") + if attempt < self.retries: + time.sleep(1.5 * (2 ** attempt)) + continue + raise last_error from exc + except (http.client.HTTPException, OSError) as exc: + # RemoteDisconnected is an HTTPException, NOT a URLError, so it + # escaped the handler above and killed a run mid-fight. + last_error = JevError(f"transport failure: {type(exc).__name__}: {exc}") + if attempt < self.retries: + time.sleep(1.5 * (2 ** attempt)) + continue + raise last_error from exc + else: + raise last_error or JevError("request failed") + + elapsed = time.monotonic() - started + try: + data = json.loads(raw) + except json.JSONDecodeError as exc: + raise JevError(f"non-JSON response: {raw[:200]!r}") from exc + + response = self._parse(data, elapsed) + _trace({ + "model": response.model, + "latency_s": round(elapsed, 3), + "input_tokens": response.input_tokens, + "output_tokens": response.output_tokens, + "questions": questions, + "answers": {qid: _answer_record(a) for qid, a in response.answers.items()}, + }) + return response + + # -- parsing ----------------------------------------------------------- + + @staticmethod + def _parse(data: dict, elapsed: float) -> JevResponse: + answers: dict[str, Answer] = {} + for qid, item in (data.get("answers") or {}).items(): + kind = item.get("type") + if kind == "noul": + answers[qid] = NoulAnswer(noul=float(item.get("noul", 0.0))) + elif kind == "choice": + answers[qid] = ChoiceAnswer( + choice=item.get("choice"), + probabilities=item.get("probabilities") or {}, + confidence=float(item.get("confidence", 0.0)), + ) + elif kind == "score": + answers[qid] = ScoreAnswer( + score=float(item.get("score", 0.0)), + legend=item.get("legend") or {}, + probabilities=item.get("probabilities") or {}, + confidence=float(item.get("confidence", 0.0)), + ) + else: + raise JevError(f"unknown answer type {kind!r} for question {qid!r}") + + usage = data.get("usage") or {} + return JevResponse( + answers=answers, + model=data.get("model", "?"), + input_tokens=int(usage.get("input_tokens", 0)), + output_tokens=int(usage.get("output_tokens", 0)), + latency_s=elapsed, + ) + + +# -------------------------------------------------------------------------- +# Trace log (one JSON line per call, path from $JEV_TRACE) +# -------------------------------------------------------------------------- + +def _answer_record(a: Answer) -> dict: + if isinstance(a, NoulAnswer): + return {"kind": "noul", "noul": a.noul, "yes": a.yes} + if isinstance(a, ChoiceAnswer): + return { + "kind": "choice", + "choice": a.choice, + "probabilities": a.probabilities, + "confidence": a.confidence, + "margin": round(margin(a), 3), + "gated": gate(a), + } + return { + "kind": "score", + "score": a.score, + "confidence": a.confidence, + "probabilities": a.probabilities, + "gated": gate(a), + } + + +def _trace(record: dict) -> None: + if not TRACE_PATH: + return + try: + line = json.dumps({"ts": time.strftime("%H:%M:%S"), **record}, + sort_keys=True, default=str) + with open(TRACE_PATH, "a", encoding="utf-8") as fh: + fh.write(line + "\n") + except OSError: + pass # tracing must never break a run + + +# -------------------------------------------------------------------------- +# Confidence gate +# -------------------------------------------------------------------------- + +def gate(answer: Answer, threshold: float = 0.6) -> bool: + """ + True when the answer is safe to act on automatically. + + Do NOT gate a Choice on `confidence` alone. Confidence measures how peaked + the distribution is, so it falls as the option count rises. Measured case: + 5 cards, Jev picks Bash at 0.61 with Defend at 0.29 -> confidence 0.50. + That is a clear plurality, yet a fixed 0.55 floor would reject it. + + Gate on the margin over the runner-up instead, which is scale-free, plus a + floor on the top probability itself. + + Noul has no confidence field, so it is gated on distance from 0.5. + """ + if isinstance(answer, NoulAnswer): + return abs(answer.noul - 0.5) >= NOUL_MARGIN + if isinstance(answer, ChoiceAnswer): + top = answer.probabilities.get(answer.choice, 0.0) + runner = max( + (v for k, v in answer.probabilities.items() if k != answer.choice), + default=0.0, + ) + return top >= 0.45 and (top - runner) >= 0.20 + return answer.confidence >= threshold + + +def margin(answer: ChoiceAnswer) -> float: + """Top probability minus runner-up. Scale-free, unlike confidence.""" + top = answer.probabilities.get(answer.choice, 0.0) + runner = max( + (v for k, v in answer.probabilities.items() if k != answer.choice), + default=0.0, + ) + return top - runner + + +# -------------------------------------------------------------------------- +# Smoke test +# -------------------------------------------------------------------------- + +def _selftest() -> int: + client = JevClient() + print(f"[ok] key loaded from secrets store (client={client!r})") + + state = { + "combat": { + "energy": 3, + "player": {"hp": 42, "max_hp": 80, "block": 0}, + "enemies": [ + {"id": "JAW_WORM_0", "hp": 12, "intent": "attacking for 11"}, + {"id": "CULTIST_1", "hp": 30, "intent": "buffing strength"}, + ], + "facts": {"lethal_available": True}, # computed in facts.py, not by Jev + "hand": [ + {"name": "Strike", "cost": 1, "text": "Deal 6 damage."}, + {"name": "Bash", "cost": 2, "text": "Deal 8 damage. Apply 2 Vulnerable."}, + {"name": "Defend", "cost": 1, "text": "Gain 5 Block."}, + {"name": "Cleave", "cost": 1, "text": "Deal 8 damage to ALL enemies."}, + ], + } + } + + response = client.ask( + state, + { + "should_prioritize_damage": noul( + "Given the intents in `combat.enemies`, is dealing damage this turn " + "better than gaining block?" + ), + "best_play": choice( + "Which single card from `combat.hand` best advances winning this fight?", + { + "Strike": "Deal 6 damage to one enemy.", + "Bash": "Deal 8 damage and apply 2 Vulnerable to one enemy.", + "Defend": "Gain 5 Block.", + "Cleave": "Deal 8 damage to every enemy.", + }, + ), + "target_priority": choice( + "Which enemy should a single-target attack hit first?", + { + "JAW_WORM_0": "12 HP, attacking for 11 this turn.", + "CULTIST_1": "30 HP, buffing strength this turn.", + }, + ), + }, + ) + + print(response.summary()) + play = response["best_play"] + print(f"[gate] best_play auto-act={'yes' if gate(play) else 'no'}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(_selftest())