#!/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. Question shapes follow the documented "advanced: structure" rules: the `instructions` of every primitive, each Choice option description, each Score level description and the Noul `criteria.true` / `criteria.false` entries all accept a string, an object, or an array. Build them with `ask()` and `entry()`: noul(ask("Would `offered.card0` make this deck stronger?", focus="Judge deck fit, not raw power.", inspect="`offered.card0`, `deck`"), true=entry("It adds damage, block or scaling this deck lacks", examples=["A 2-cost attack that scales with Strength"]), false=entry("It is off-plan, redundant, or too slow to matter", not_for="A card that is merely different")) choice(ask("Which card best improves this deck, or is skipping better?"), {f"card{i['index']}": entry(f"{i['name']}: {i['description']}", not_for="...") for i in cards} | {"skip": entry("Take nothing; keep the deck lean")}) A threshold tuned on a Noul is never reused on a Choice: state each gate's floor and margin at the call site (`gate_choice`). 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 math 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 # Choice gates are stated per call site, because a threshold tuned on a Noul # must not be carried over to a Choice: the two answer different questions and # are not on a comparable scale. These are the DEFAULTS for a low-stakes Choice # (a card play). Irreversible decisions (events) pass their own stricter pair. CHOICE_TOP_MIN = 0.45 CHOICE_MARGIN_MIN = 0.20 class JevError(RuntimeError): """Raised for transport, auth, and protocol failures.""" # -------------------------------------------------------------------------- # Question builders # -------------------------------------------------------------------------- def entry(what: Any = None, *, not_for: Any = None, examples: Any = None, signals: Any = None, **more: Any) -> Any: """ One structured description entry, per the documented shapes. `instructions`, Choice option descriptions, Score level descriptions and Noul `criteria.true` / `criteria.false` all accept a string, an object, or an array. Use the object form when a description needs several kinds of guidance: what it covers, what belongs elsewhere, and examples. Returns the bare string when only `what` is given, so a simple description stays simple. The documented field names are used consistently across entries so the model can compare them directly. """ fields = {"what": what, "not_for": not_for, "examples": examples, "signals": signals, **more} present = {k: v for k, v in fields.items() if v is not None} if not present: return None if set(present) == {"what"} and isinstance(present["what"], str): return present["what"] return present def ask(question: Any, *, focus: Any = None, inspect: Any = None, compare: Any = None, **more: Any) -> Any: """ Structured instructions: the question plus named guidance fields. `focus` says what to judge, `inspect` / `compare` point at the exact state paths the answer depends on (backticked dot-and-index paths). Use it when a bare sentence would blur several instructions together; a short, unambiguous question stays a string. """ fields = {"focus": focus, "inspect": inspect, "compare": compare, **more} present = {k: v for k, v in fields.items() if v is not None} if not present: return question return {"question": question, **present} def noul_criteria(true: Any = None, false: Any = None) -> dict | None: """ Contrastive Noul criteria: what a yes means, what a no means. The instruction and the criteria must ask for the same thing — a Noul whose `true` maps to "no" is a documented failure mode. """ criteria = {k: v for k, v in (("true", true), ("false", false)) if v is not None} return criteria or None def noul(instructions: Any, criteria: dict | None = None, *, true: Any = None, false: Any = None) -> dict: """ Yes/no question. Returns only P(yes); there is no confidence field. Pass `true=` / `false=` for the contrastive criteria, or `criteria=` for a prebuilt mapping. """ q: dict[str, Any] = {"type": "noul", "instructions": instructions} built = criteria if criteria is not None else noul_criteria(true, false) if built is not None: q["criteria"] = built return q def choice(instructions: Any, criteria: dict) -> dict: """ Pick one option from a map of option -> description (max 255). Each description is an `entry()`. Give the model the full list rather than a shortlist, and add an explicit "none of the above"-style option whenever the set might not cover every input. """ return {"type": "choice", "instructions": instructions, "criteria": criteria} def score(instructions: Any, criteria: list) -> dict: """ Position on an ordered list of levels, each an `entry()`. Use a Score when the answer is a position on a spectrum. The returned `score` is a fractional position on the level scale, not an index. """ 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" @property def level(self) -> int: """Nearest level index. `score` is a FRACTIONAL position on the scale.""" return int(round(self.score)) def level_entry(self) -> Any: """The legend entry for the nearest level, keyed by index as a string.""" return self.legend.get(str(self.level)) 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, UnicodeDecodeError) as exc: raise JevError(f"non-JSON response: {raw[:200]!r}") from exc response = self._parse(data, elapsed) if set(response.answers) != set(questions): raise JevError("response question IDs do not match the request") for qid, answer in response.answers.items(): question = questions[qid] if answer.kind != question.get("type"): raise JevError(f"answer type does not match question {qid!r}") if isinstance(answer, ChoiceAnswer) and set(answer.probabilities) != set(question["criteria"]): raise JevError(f"answer options do not match question {qid!r}") if isinstance(answer, ScoreAnswer) and len(answer.legend) != len(question["criteria"]): raise JevError(f"answer levels do not match question {qid!r}") _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: def number(value, field: str, *, probability: bool = False) -> float: if isinstance(value, bool) or not isinstance(value, (int, float)): raise JevError(f"missing or non-numeric {field}") if not math.isfinite(value) or (probability and not 0 <= value <= 1): raise JevError(f"invalid {field}: expected a finite {'probability' if probability else 'number'}") return float(value) def probabilities(item: dict) -> dict[str, float]: values = item.get("probabilities") if not isinstance(values, dict) or not values: raise JevError("missing or invalid probability distribution") if not all(isinstance(k, str) for k in values): raise JevError("probability keys must be strings") return {k: number(v, f"probabilities[{k!r}]", probability=True) for k, v in values.items()} if not isinstance(data, dict): raise JevError("expected a response object") raw_answers = data.get("answers") if not isinstance(raw_answers, dict) or not raw_answers: raise JevError("missing or invalid answers object") answers: dict[str, Answer] = {} for qid, item in raw_answers.items(): if not isinstance(qid, str) or not isinstance(item, dict): raise JevError("invalid question ID or answer object") kind = item.get("type") if kind == "noul": answers[qid] = NoulAnswer(number(item.get("noul"), "noul", probability=True)) elif kind == "choice": values = probabilities(item) chosen = item.get("choice") if not isinstance(chosen, str) or chosen not in values: raise JevError(f"invalid choice for question {qid!r}") answers[qid] = ChoiceAnswer( choice=chosen, probabilities=values, confidence=number(item.get("confidence"), "confidence", probability=True), ) elif kind == "score": values = probabilities(item) legend = item.get("legend") if not isinstance(legend, dict) or not 2 <= len(legend) <= 10: raise JevError(f"invalid score legend for question {qid!r}") levels = {str(i) for i in range(len(legend))} if set(legend) != levels or set(values) != levels: raise JevError(f"invalid score levels for question {qid!r}") value = number(item.get("score"), "score") if not 0 <= value <= len(legend) - 1: raise JevError(f"score outside its levels for question {qid!r}") answers[qid] = ScoreAnswer( score=value, legend=legend, probabilities=values, confidence=number(item.get("confidence"), "confidence", probability=True), ) else: raise JevError(f"unknown answer type {kind!r} for question {qid!r}") usage = data.get("usage", {}) if not isinstance(usage, dict): raise JevError("invalid token usage object") counts = [usage.get(key, 0) for key in ("input_tokens", "output_tokens")] if any(isinstance(n, bool) or not isinstance(n, int) or n < 0 for n in counts): raise JevError("invalid token usage counts") return JevResponse( answers=answers, model=data.get("model", "?"), input_tokens=counts[0], output_tokens=counts[1], latency_s=elapsed, ) # -------------------------------------------------------------------------- # Trace log (one JSON line per call, path from $JEV_TRACE) # -------------------------------------------------------------------------- def answer_record(a: Answer) -> dict: """Machine-readable form of one answer, for a decision log. Carries the gate outcome for every primitive, so a gate can be analysed offline from the log alone instead of being recomputed from the value. """ if isinstance(a, NoulAnswer): return {"kind": "noul", "noul": a.noul, "yes": a.yes, "gated": gate(a)} 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_choice(answer: ChoiceAnswer, top_min: float, margin_min: float) -> bool: """ Gate a Choice with thresholds stated for THIS decision. A threshold tuned on a Noul must not be carried over to a Choice: the two answer different questions and are not on a comparable scale. So each call site names the floor and the margin that decision actually needs. The margin is the scale-free signal; `confidence` is peakedness and falls as the option count rises, so it is never the primary gate. """ top = answer.probabilities.get(answer.choice, 0.0) return top >= top_min and margin(answer) >= margin_min 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): return gate_choice(answer, CHOICE_TOP_MIN, CHOICE_MARGIN_MIN) 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())