0xrsydn/sts2-bot
1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add TypeSafe System One (Jev) client with call tracing

Browse source 
Batched question API (noul/choice/score), retrying transport, and a
margin-based confidence gate. Every call appends one JSON line of
questions and parsed answers to $JEV_TRACE when set; tracing never
raises, so it cannot break a run.
This commit is contained in:
0xrsydn 2026-09-21 17:09:43 +07:00
commit 1623377769
1 changed files with 462 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

462
jev.py Normal file
View file

@ -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())