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

630 lines
24 KiB
Python
Raw Blame History

#!/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())
Reference in a new issue View git blame Copy permalink