1623377769
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.
462 lines
16 KiB
Python
462 lines
16 KiB
Python
#!/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())
|